Wed, 21 Sep 2022 15:30:34 +0200
Reformatted source code with 'Black'.
# -*- coding: utf-8 -*- # Copyright (c) 2017 - 2022 Detlev Offenbach <detlev@die-offenbachs.de> # """ Module implementing the refactoring client interface to rope. """ import sys import os import contextlib modulePath = sys.argv[-1] # it is always the last parameter sys.path.append(modulePath) try: import rope.base.project import rope.base.libutils 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 """ 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, } from FileSystemCommands import RefactoringClientFileSystemCommands 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 """ import traceback 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): """ 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): """ 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): """ 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): """ 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): """ 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 = [] import rope.contrib.findit 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 import rope.contrib.findit 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 = [] import rope.contrib.findit 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, } import rope.refactor.rename 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, } import rope.refactor.rename 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, } import rope.refactor.extract 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, } import rope.refactor.inline 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, } import rope.refactor.extract 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, } import rope.refactor.move 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, } import rope.refactor.move 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, } import rope.refactor.usefunction 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, } import rope.refactor.usefunction 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, } import rope.refactor.introduce_factory 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, } import rope.refactor.introduce_parameter 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, } from rope.refactor.importutils import ImportOrganizer 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, } import rope.refactor.restructure 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, } import rope.refactor.change_signature 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 = [] import rope.refactor.change_signature # 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, } import rope.refactor.topackage 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, } import rope.refactor.encapsulate_field 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, } import rope.refactor.encapsulate_field 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, } import rope.refactor.localtofield 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, } import rope.refactor.method_object 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