tip/RefactoringRope/RefactoringClient.py

# -*- coding: utf-8 -*-

# Copyright (c) 2017 - 2025 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the refactoring client interface to rope.
"""

import contextlib
import os
import sys
import traceback

modulePath = sys.argv[-1]  # it is always the last parameter
sys.path.append(modulePath)

try:
    import rope.base.libutils
    import rope.base.project
    import rope.contrib.findit
    import rope.refactor.change_signature
    import rope.refactor.encapsulate_field
    import rope.refactor.extract
    import rope.refactor.inline
    import rope.refactor.introduce_factory
    import rope.refactor.introduce_parameter
    import rope.refactor.localtofield
    import rope.refactor.method_object
    import rope.refactor.move
    import rope.refactor.rename
    import rope.refactor.restructure
    import rope.refactor.topackage
    import rope.refactor.usefunction

    from rope.refactor.importutils import ImportOrganizer
except ImportError:
    sys.exit(42)

from ProgressHandle import ProgressHandle

from eric7.EricNetwork.EricJsonClient import EricJsonClient


class RefactoringClient(EricJsonClient):
    """
    Class implementing the refactoring client interface to rope.
    """

    def __init__(self, host, port, projectPath):
        """
        Constructor

        @param host ip address the background service is listening
        @type str
        @param port port of the background service
        @type int
        @param projectPath path to the project
        @type str
        """
        from FileSystemCommands import RefactoringClientFileSystemCommands  # noqa: I102

        super().__init__(host, port)

        self.__methodMapping = {
            "AbortAction": self.__abortAction,
            "CloseProject": self.__closeProject,
            "Validate": self.__validate,
            "QueryReferences": self.__queryReferences,
            "QueryDefinition": self.__queryDefinition,
            "QueryImplementations": self.__queryImplementations,
            "GetConfig": self.__getConfig,
            "ConfigChanged": self.__configChanged,
            "PerformSoa": self.__performSOA,
            "ReportChanged": self.__reportChanged,
            "History": self.__processHistory,
            "PreviewChanges": self.__previewChanges,
            "ApplyChanges": self.__applyChanges,
            "ClearChanges": self.__clearChanges,
            "CalculateRenameChanges": self.__calculateRenameChanges,
            "CalculateChangeOccurrencesChanges": (
                self.__calculateChangeOccurrencesChanges
            ),
            "CalculateExtractChanges": self.__calculateExtractChanges,
            "RequestInlineType": self.__requestInlineType,
            "CalculateInlineChanges": self.__calculateInlineChanges,
            "RequestMoveType": self.__requestMoveType,
            "CalculateMoveChanges": self.__calculateMoveChanges,
            "RequestUseFunction": self.__requestUseFunction,
            "CalculateUseFunctionChanges": self.__calculateUseFunctionChanges,
            "CalculateIntroduceFactoryChanges": self.__calculateIntroduceFactoryChanges,
            "CalculateIntroduceParameterChanges": (
                self.__calculateIntroduceParameterChanges
            ),
            "CalculateImportsChanges": self.__calculateImportsChanges,
            "CalculateRestructureChanges": self.__calculateRestructureChanges,
            "RequestSignature": self.__requestSignature,
            "CalculateSignatureChanges": self.__calculateSignatureChanges,
            "CalculateInlineArgumentDefaultChanges": (
                self.__calculateInlineArgumentDefaultChanges
            ),
            "CalculateModuleToPackageChanges": self.__calculateModuleToPackageChanges,
            "RequestFieldName": self.__requestFieldName,
            "CalculateEncapsulateFieldChanges": self.__calculateEncapsulateFieldChanges,
            "CalculateLocalToAttributeChanges": self.__calculateLocalToAttributeChanges,
            "CalculateMethodObjectChanges": self.__calculateMethodObjectChanges,
        }

        self.__fsCommands = RefactoringClientFileSystemCommands(self)

        self.__projectpath = projectPath
        self.__project = rope.base.project.Project(
            self.__projectpath, fscommands=self.__fsCommands
        )

        self.__progressHandle = None

        self.__changes = {}
        # dict storing the retrieved changes for various refactorings

    def handleCall(self, method, params):
        """
        Public method to handle a method call from the server.

        @param method requested method name
        @type str
        @param params dictionary with method specific parameters
        @type dict
        """
        self.__methodMapping[method](params)

    def __handleRopeError(self, err):
        """
        Private method to process a rope error.

        @param err rope exception object
        @type Exception
        @return dictionary containing the error information
        @rtype dict
        """
        tb = traceback.format_tb(sys.exc_info()[2])
        ropeError = str(type(err)).split()[-1]
        ropeError = ropeError[1:-2].split(".")[-1]
        errorDict = {
            "Error": ropeError,
            "ErrorString": str(err),
            "Traceback": tb,
        }
        if ropeError == "ModuleSyntaxError":
            errorDict["ErrorFile"] = err.filename
            errorDict["ErrorLine"] = err.lineno

        return errorDict

    def __abortAction(self, params):  # noqa: U100
        """
        Private method to abort the current action.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        if self.__progressHandle is not None and not self.__progressHandle.is_stopped():
            self.__progressHandle.stop()

    def __validate(self, params):  # noqa: U100
        """
        Private slot to validate the project.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        self.__project.validate(self.__project.root)

    def __closeProject(self, params):  # noqa: U100
        """
        Private slot to validate the project.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        self.__project.close()

    def __getConfig(self, params):  # noqa: U100
        """
        Private method to send some configuration data to the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        result = {
            "RopeFolderName": self.__project.ropefolder.real_path,
            "RopeHelpFile": os.path.join(
                os.path.dirname(__file__), "Documentation", "rope", "overview.rst"
            ),
            "RopeInfo": rope.INFO,
            "RopeVersion": rope.VERSION,
            "RopeCopyright": rope.COPYRIGHT,
            "PythonVersion": "Python{0}".format(sys.version_info[0]),
        }
        try:
            # rope version < 1.2.0
            result["DefaultConfig"] = self.__project._default_config()
        except AttributeError:
            # rope version >= 1.2.0
            # read our own copy of default_config derived from the default
            # settings in rope.base.prefs.Prefs
            with open(
                os.path.join(os.path.dirname(__file__), "default_config.py"),
                "r",
                encoding="utf-8",
            ) as f:
                result["DefaultConfig"] = f.read()

        self.sendJson("Config", result)

    def __configChanged(self, params):  # noqa: U100
        """
        Private method to handle a change of the configuration file.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        self.__project.close()
        self.__project = rope.base.project.Project(
            self.__projectpath, fscommands=self.__fsCommands
        )

    def __queryReferences(self, params):
        """
        Private method to handle the Find References action.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        occurrences = []

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            occurrences = rope.contrib.findit.find_occurrences(
                self.__project,
                resource,
                offset,
                unsure=True,
                in_hierarchy=True,
                task_handle=self.__progressHandle,
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        result = {
            "Title": title,
            "EntriesCount": len(occurrences),
            "Entries": [
                [occurrence.resource.real_path, occurrence.lineno, occurrence.unsure]
                for occurrence in occurrences
            ],
        }
        result.update(errorDict)

        self.sendJson("QueryReferencesResult", result)

    def __queryDefinition(self, params):
        """
        Private method to handle the Find Definition action.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        source = params["Source"]

        errorDict = {}
        location = None

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            location = rope.contrib.findit.find_definition(
                self.__project, source, offset, resource
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result = {
            "Title": title,
        }
        if location is not None:
            result["Location"] = [location.resource.real_path, location.lineno]
        result.update(errorDict)

        self.sendJson("QueryDefinitionResult", result)

    def __queryImplementations(self, params):
        """
        Private method to handle the Find Implementations action.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        occurrences = []

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            occurrences = rope.contrib.findit.find_implementations(
                self.__project, resource, offset, task_handle=self.__progressHandle
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        result = {
            "Title": title,
            "EntriesCount": len(occurrences),
            "Entries": [
                [occurrence.resource.real_path, occurrence.lineno, occurrence.unsure]
                for occurrence in occurrences
            ],
        }
        result.update(errorDict)

        self.sendJson("QueryImplementationsResult", result)

    def __performSOA(self, params):
        """
        Private method to perform SOA on all modules.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        title = params["Title"]

        errorDict = {}

        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            rope.base.libutils.analyze_modules(
                self.__project, task_handle=self.__progressHandle
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        result = {
            "Title": title,
        }
        result.update(errorDict)

        self.sendJson("SoaFinished", result)

    def __reportChanged(self, params):
        """
        Private method to register some changed sources.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        filename = params["FileName"]
        oldSource = params["OldSource"]

        with contextlib.suppress(Exception):
            rope.base.libutils.report_change(self.__project, filename, oldSource)

    def __processHistory(self, params):
        """
        Private method to process the various history related requests.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        subcommand = params["Subcommand"]
        if subcommand == "Get":
            changes = {}
            if params["Filename"]:
                # file history
                resource = rope.base.libutils.path_to_resource(
                    self.__project, params["Filename"]
                )
                undoList = []
                for change in reversed(self.__project.history.undo_list):
                    if resource in change.get_changed_resources():
                        undoList.append(change)
                redoList = []
                for change in self.__project.history.redo_list:
                    if resource in change.get_changed_resources():
                        redoList.append(change)
            else:
                # project history
                undoList = list(reversed(self.__project.history.undo_list))
                redoList = self.__project.history.redo_list

            result = {
                "Subcommand": "Histories",
                "Undo": [],
            }
            for change in undoList:
                changes[id(change)] = change
                result["Undo"].append([str(change), id(change)])
            result["Redo"] = []
            for change in redoList:
                changes[id(change)] = change
                result["Redo"].append([str(change), id(change)])

            self.__changes["History"] = changes

            self.sendJson("HistoryResult", result)

        elif subcommand == "GetChange":
            result = {
                "Subcommand": "ChangeDescription",
                "Description": self.__changes["History"][
                    params["Id"]
                ].get_description(),
            }

            self.sendJson("HistoryResult", result)

        elif subcommand in ["Undo", "Redo"]:
            change = self.__changes["History"][params["Id"]]
            self.__progressHandle = ProgressHandle(self, change.description, False)
            if subcommand == "Undo":
                self.__project.history.undo(change, task_handle=self.__progressHandle)
            else:
                self.__project.history.redo(change, task_handle=self.__progressHandle)
            self.__progressHandle.reset()
            self.__progressHandle = None

            result = {
                "Subcommand": subcommand,
                "ChangedFiles": [
                    res.real_path for res in change.get_changed_resources()
                ],
            }

            self.sendJson("HistoryResult", result)

        elif subcommand == "Clear":
            self.__project.history.clear()

        elif subcommand == "ClearChanges":
            with contextlib.suppress(KeyError):
                del self.__changes["History"]

    def __clearChanges(self, params):
        """
        Private method to clear the changes cache of a given change group.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        with contextlib.suppress(KeyError):
            del self.__changes[params["ChangeGroup"]]

    def __applyChanges(self, params):
        """
        Private method to apply the changes of a given change group.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        errorDict = {}

        self.__progressHandle = ProgressHandle(self, params["Title"], False)
        try:
            changes = self.__changes[params["ChangeGroup"]]
            if changes is not None:
                self.__project.do(changes, self.__progressHandle)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        result = {
            "Subcommand": "ChangesApplied",
            "ChangeGroup": params["ChangeGroup"],
            "Title": params["Title"],
            "ChangedFiles": [res.real_path for res in changes.get_changed_resources()],
        }
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __previewChanges(self, params):
        """
        Private method to determine the changes data for a preview.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        try:
            changes = self.__changes[params["ChangeGroup"]]
            description = changes.description
        except KeyError:
            changes = None
            description = ""

        changesData = []
        if changes is not None:
            for change in changes.changes:
                changeTitle = str(change)
                try:
                    changeText = change.get_description()
                except AttributeError:
                    changeText = None
                changesData.append([changeTitle, changeText])

        result = {
            "Subcommand": "PreviewChanges",
            "ChangeGroup": params["ChangeGroup"],
            "Description": description,
            "Changes": changesData,
        }

        self.sendJson("Changes", result)

    def __calculateRenameChanges(self, params):
        """
        Private method to calculate the rename changes based on the parameters
        sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        isLocal = params["LocalRename"]
        newName = params["NewName"]
        renameHierarchy = params["RenameHierarchy"]
        renameInStrings = params["RenameInStrings"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)

        resources = [resource] if isLocal else None

        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            renamer = rope.refactor.rename.Rename(self.__project, resource, offset)
            changes = renamer.get_changes(
                newName,
                resources=resources,
                in_hierarchy=renameHierarchy,
                unsure=lambda o: self.__confirmUnsure(o, changeGroup),
                docs=renameInStrings,
                task_handle=self.__progressHandle,
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __confirmUnsure(self, occurrence, changeGroup):
        """
        Private method to confirm unsure occurrences.

        @param occurrence reference to the occurrence object
        @type rope.refactor.occurrences.Occurrence
        @param changeGroup name of the change group
        @type str
        @return flag indicating an occurrence
        @rtype bool
        """
        filename = occurrence.resource.real_path
        start, end = occurrence.get_primary_range()

        self.sendJson(
            "Changes",
            {
                "Subcommand": "ConfirmUnsure",
                "ChangeGroup": changeGroup,
                "FileName": filename,
                "StartOffset": start,
                "EndOffset": end,
            },
        )

        answer = self.poll(waitMethod="ConfirmUnsure")

        return answer["Answer"]

    def __calculateChangeOccurrencesChanges(self, params):
        """
        Private method to calculate the 'Change Occurrences' changes based on
        the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        newName = params["NewName"]
        onlyCalls = params["OnlyCalls"]
        reads = params["Reads"]
        writes = params["Writes"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            renamer = rope.refactor.rename.ChangeOccurrences(
                self.__project, resource, offset
            )
            changes = renamer.get_changes(
                newName, only_calls=onlyCalls, reads=reads, writes=writes
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateExtractChanges(self, params):
        """
        Private method to calculate the 'Extract' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        @exception Exception raised to indicate an invalid extraction type
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        startOffset = params["StartOffset"]
        endOffset = params["EndOffset"]
        kind = params["Kind"]
        newName = params["NewName"]
        similar = params["Similar"]
        global_ = params["Global"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            if kind not in ("variable", "method"):
                raise Exception("Invalid extraction kind <{0}>.".format(kind))
            elif kind == "variable":
                extractor = rope.refactor.extract.ExtractVariable(
                    self.__project, resource, startOffset, endOffset
                )
            elif kind == "method":
                extractor = rope.refactor.extract.ExtractMethod(
                    self.__project, resource, startOffset, endOffset
                )
            changes = extractor.get_changes(newName, similar=similar, global_=global_)
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __requestInlineType(self, params):
        """
        Private method to determine the 'Inline' changes type based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        result = {
            "Subcommand": "InlineType",
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            inliner = rope.refactor.inline.create_inline(
                self.__project, resource, offset
            )
            result.update(
                {
                    "Name": inliner.name,
                    "Kind": inliner.get_kind(),
                }
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateInlineChanges(self, params):
        """
        Private method to calculate the 'Inline' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        kind = params["Kind"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            inliner = rope.refactor.inline.create_inline(
                self.__project, resource, offset
            )
            opts = (
                {
                    "in_hierarchy": params["Hierarchy"],
                }
                if kind == "parameter"
                else {
                    "remove": params["Remove"],
                    "only_current": params["OnlyCurrent"],
                }
            )
            changes = inliner.get_changes(task_handle=self.__progressHandle, **opts)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __requestMoveType(self, params):
        """
        Private method to determine the 'Move Method' changes type based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        result = {
            "Subcommand": "MoveType",
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            mover = rope.refactor.move.create_move(self.__project, resource, offset)
            if isinstance(mover, rope.refactor.move.MoveGlobal):
                result.update(
                    {
                        "Kind": "move_global_method",
                        "Method": "",
                    }
                )
            else:
                result.update(
                    {
                        "Kind": "move_method",
                        "Method": mover.get_method_name(),
                    }
                )
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateMoveChanges(self, params):
        """
        Private method to calculate the 'Move ...' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        kind = params["Kind"]
        newName = params["NewName"]
        attribute = params["Attribute"]
        destination = params["DestinationModule"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            mover = rope.refactor.move.create_move(self.__project, resource, offset)
            if kind == "move_method":
                changes = mover.get_changes(
                    attribute, newName, task_handle=self.__progressHandle
                )
            else:
                if kind == "move_global_method":
                    dest = self.__project.get_pycore().find_module(
                        os.path.splitext(destination)[0]
                    )
                else:
                    # move_module
                    if destination.endswith(os.sep):
                        destination = destination[:-1]
                    dest = self.__project.get_pycore().find_module(destination)
                changes = mover.get_changes(dest, task_handle=self.__progressHandle)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __requestUseFunction(self, params):
        """
        Private method to determine the 'Use Function' function name based on
        the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        result = {
            "Subcommand": "UseFunctionName",
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            user = rope.refactor.usefunction.UseFunction(
                self.__project, resource, offset
            )
            result["FunctionName"] = user.get_function_name()
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateUseFunctionChanges(self, params):
        """
        Private method to calculate the 'Use Function' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            user = rope.refactor.usefunction.UseFunction(
                self.__project, resource, offset
            )
            changes = user.get_changes(task_handle=self.__progressHandle)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateIntroduceFactoryChanges(self, params):
        """
        Private method to calculate the 'Introduce Factory' changes based on
        the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        factoryName = params["Name"]
        globalFactory = params["GlobalFactory"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            introducer = rope.refactor.introduce_factory.IntroduceFactoryRefactoring(
                self.__project, resource, offset
            )
            changes = introducer.get_changes(
                factoryName,
                global_factory=globalFactory,
                task_handle=self.__progressHandle,
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateIntroduceParameterChanges(self, params):
        """
        Private method to calculate the 'Introduce Parameter' changes based on
        the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        parameterName = params["Name"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            introducer = rope.refactor.introduce_parameter.IntroduceParameter(
                self.__project, resource, offset
            )
            changes = introducer.get_changes(parameterName)
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateImportsChanges(self, params):
        """
        Private method to calculate the 'Import' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        methodName = params["MethodName"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        method = {
            "organize_imports": ImportOrganizer.organize_imports,
            "expand_star_imports": ImportOrganizer.expand_star_imports,
            "relatives_to_absolutes": ImportOrganizer.relatives_to_absolutes,
            "froms_to_imports": ImportOrganizer.froms_to_imports,
            "handle_long_imports": ImportOrganizer.handle_long_imports,
        }[methodName]
        importOrganizer = ImportOrganizer(self.__project)
        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            changes = method(importOrganizer, resource, offset=offset)
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        # send the change description first
        description = changes.description if changes else ""
        self.sendJson(
            "Changes",
            {
                "ChangeGroup": changeGroup,
                "Title": title,
                "Subcommand": "ChangeDescription",
                "Description": description,
            },
        )

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateRestructureChanges(self, params):
        """
        Private method to calculate the 'Restructure' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        pattern = params["Pattern"]
        goal = params["Goal"]
        args = params["Args"]
        imports = params["Imports"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        self.__project.validate(self.__project.root)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            restructuring = rope.refactor.restructure.Restructure(
                self.__project, pattern, goal, args=args, imports=imports
            )
            changes = restructuring.get_changes(task_handle=self.__progressHandle)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __requestSignature(self, params):
        """
        Private method to calculate the 'Signature' based on the parameters
        sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        result = {
            "Subcommand": "Signature",
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            changer = rope.refactor.change_signature.ChangeSignature(
                self.__project, resource, offset
            )
            result["DefinitionInfo"] = changer.get_args()
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateSignatureChanges(self, params):
        """
        Private method to calculate the 'Signature' changes based on the
        parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        removals = params["Removals"]
        additions = params["Additions"]
        newOrdering = params["Ordering"]
        autodef = params["AutoDef"]
        doHierarchy = params["Hierarchy"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        changers = []

        # removals
        for index in removals:
            remover = rope.refactor.change_signature.ArgumentRemover(index)
            changers.append(remover)
        # additions
        for index, name, default, value in additions:
            adder = rope.refactor.change_signature.ArgumentAdder(
                index, name, default, value
            )
            changers.append(adder)
        # new ordering
        changers.append(
            rope.refactor.change_signature.ArgumentReorderer(
                newOrdering, autodef=autodef
            )
        )

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            changer = rope.refactor.change_signature.ChangeSignature(
                self.__project, resource, offset
            )
            changes = changer.get_changes(
                changers, in_hierarchy=doHierarchy, task_handle=self.__progressHandle
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateInlineArgumentDefaultChanges(self, params):
        """
        Private method to calculate the 'Inline Argument Default' changes
        based on the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        argumentIndex = params["Index"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            changer = rope.refactor.change_signature.ChangeSignature(
                self.__project, resource, offset
            )
            inliner = rope.refactor.change_signature.ArgumentDefaultInliner(
                argumentIndex
            )
            changes = changer.get_changes([inliner], task_handle=self.__progressHandle)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateModuleToPackageChanges(self, params):
        """
        Private method to calculate the 'Module to Package' changes
        based on the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            changes = rope.refactor.topackage.ModuleToPackage(
                self.__project, resource
            ).get_changes()
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        # send the change description first
        description = changes.description if changes else ""
        self.sendJson(
            "Changes",
            {
                "ChangeGroup": changeGroup,
                "Title": title,
                "Subcommand": "ChangeDescription",
                "Description": description,
            },
        )

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __requestFieldName(self, params):
        """
        Private method to calculate the 'Field Name' based on the parameters
        sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        result = {
            "Subcommand": "FieldName",
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            encapsulateField = rope.refactor.encapsulate_field.EncapsulateField(
                self.__project, resource, offset
            )
            result["Name"] = encapsulateField.get_field_name()
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateEncapsulateFieldChanges(self, params):
        """
        Private method to calculate the 'Encapsulate Field' changes based on
        the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        getter = params["Getter"]
        setter = params["Setter"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        self.__progressHandle = ProgressHandle(self, title, True)
        try:
            encapsulateField = rope.refactor.encapsulate_field.EncapsulateField(
                self.__project, resource, offset
            )
            changes = encapsulateField.get_changes(
                getter=getter, setter=setter, task_handle=self.__progressHandle
            )
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        self.__progressHandle.reset()
        self.__progressHandle = None

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateLocalToAttributeChanges(self, params):
        """
        Private method to calculate the 'Local Variabe to Attribute' changes
        based on the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            changes = rope.refactor.localtofield.LocalToField(
                self.__project, resource, offset
            ).get_changes()
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        # send the change description first
        description = changes.description if changes else ""
        self.sendJson(
            "Changes",
            {
                "ChangeGroup": changeGroup,
                "Title": title,
                "Subcommand": "ChangeDescription",
                "Description": description,
            },
        )

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)

    def __calculateMethodObjectChanges(self, params):
        """
        Private method to calculate the 'Method to Method Object' changes
        based on the parameters sent by the server.

        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        changeGroup = params["ChangeGroup"]
        title = params["Title"]
        filename = params["FileName"]
        offset = params["Offset"]
        name = params["Name"]

        errorDict = {}
        changes = None
        result = {
            "ChangeGroup": changeGroup,
            "Title": title,
        }

        resource = rope.base.libutils.path_to_resource(self.__project, filename)
        try:
            converter = rope.refactor.method_object.MethodObject(
                self.__project, resource, offset
            )
            changes = converter.get_changes(name)
        except Exception as err:
            errorDict = self.__handleRopeError(err)

        self.__changes[changeGroup] = changes

        result["Subcommand"] = "ChangesCalculated"
        result.update(errorDict)

        self.sendJson("Changes", result)


if __name__ == "__main__":
    if len(sys.argv) != 5:
        print("Host, port, project path and module path parameters are missing. Abort.")
        sys.exit(1)

    host, port, projectPath = sys.argv[1:-1]

    # Create a Qt6 application object in order to allow the processing of
    # modules containing Qt stuff.
    try:
        from PyQt6.QtCore import QCoreApplication
    except (ImportError, RuntimeError):
        QCoreApplication = None
    if QCoreApplication is not None:
        app = QCoreApplication([])

    client = RefactoringClient(host, int(port), projectPath)
    # Start the main loop
    client.run()

    sys.exit(0)

#
# eflag: noqa = M801