From 4c4932bac7ceaeb78ebfa73afc45e5ff4b771705 Mon Sep 17 00:00:00 2001 From: Massimo Capodiferro <77293250+maxcapodi78@users.noreply.github.com> Date: Fri, 24 May 2024 16:09:55 +0200 Subject: [PATCH] REFACTOR: Refactor grpc plugin (#4719) Co-authored-by: maxcapodi78 Co-authored-by: Kathy Pippert <84872299+PipKat@users.noreply.github.com> --- _unittest/test_01_Design.py | 1 + pyaedt/application/Design.py | 1 - pyaedt/desktop.py | 105 +++--- pyaedt/generic/grpc_plugin.py | 261 --------------- pyaedt/generic/grpc_plugin_dll.py | 104 ------ pyaedt/generic/grpc_plugin_dll_class.py | 403 ++++++++++++++++++++++++ 6 files changed, 449 insertions(+), 426 deletions(-) delete mode 100644 pyaedt/generic/grpc_plugin.py delete mode 100644 pyaedt/generic/grpc_plugin_dll.py create mode 100644 pyaedt/generic/grpc_plugin_dll_class.py diff --git a/_unittest/test_01_Design.py b/_unittest/test_01_Design.py index eddd0ace585..bcf6d53cf0f 100644 --- a/_unittest/test_01_Design.py +++ b/_unittest/test_01_Design.py @@ -305,6 +305,7 @@ def test_27_odesktop(self): "", "", "", + "", ] def test_28_get_pyaedt_app(self): diff --git a/pyaedt/application/Design.py b/pyaedt/application/Design.py index 6da363380b7..34c3b50f2a8 100644 --- a/pyaedt/application/Design.py +++ b/pyaedt/application/Design.py @@ -3340,7 +3340,6 @@ def _insert_design(self, design_type, design_name=None): if not is_windows and settings.aedt_version and self.design_type == "Circuit Design": time.sleep(1) self.odesktop.CloseAllWindows() - if new_design is None: # pragma: no cover new_design = self.desktop_class.active_design(self.oproject, unique_design_name, self.design_type) if new_design is None: diff --git a/pyaedt/desktop.py b/pyaedt/desktop.py index 84d941dda43..553984209da 100644 --- a/pyaedt/desktop.py +++ b/pyaedt/desktop.py @@ -261,32 +261,12 @@ def _close_aedt_application(desktop_class, close_desktop, pid, is_grpc_api): return True except Exception: # pragma: no cover warnings.warn("Something went wrong closing AEDT. Exception in `_main.oDesktop.QuitApplication()`.") - elif _desktop_sessions and len(_desktop_sessions) > 1 and not desktop_class.parent_desktop_id: - pyaedt_logger.error("Release is not allowed when multiple desktop sessions are available.") - pyaedt_logger.error("Closing Desktop session.") - try: - os.kill(pid, 9) - if _desktop_sessions: - for v in _desktop_sessions.values(): - if pid in v.parent_desktop_id: - del v.parent_desktop_id[v.parent_desktop_id.index(pid)] - return True - except Exception: # pragma: no cover - warnings.warn("Something went wrong closing AEDT. Exception in `_main.oDesktop.QuitApplication()`.") - elif _desktop_sessions and len(_desktop_sessions) > 1: - pyaedt_logger.error("A child desktop session is linked to this session.") - pyaedt_logger.error("Multiple desktop sessions must be released in reverse order.") - return False else: - try: - import pyaedt.generic.grpc_plugin as python_grpc_wrapper - - python_grpc_wrapper.AedtAPI.ReleaseAll() - return True - except Exception: # pragma: no cover - warnings.warn( - "Something went wrong releasing AEDT. Exception in `StandalonePyScriptWrapper.Release()`." - ) + for k, d in _desktop_sessions.items(): + if k == pid: + d.grpc_plugin.recreate_application(True) + d.grpc_plugin.Release() + return True elif not inside_desktop: if close_desktop: try: @@ -455,16 +435,17 @@ def __new__(cls, *args, **kwargs): # machine = kwargs.get("machine") or "" if (not args or len(args)<6) else args[5] port = kwargs.get("port") or 0 if (not args or len(args) < 7) else args[6] aedt_process_id = kwargs.get("aedt_process_id") or None if (not args or len(args) < 8) else args[7] - if settings.use_multi_desktop and is_windows and not inside_desktop and new_desktop_session: + if settings.use_multi_desktop and not inside_desktop and new_desktop_session: pyaedt_logger.info("Initializing new Desktop session.") return object.__new__(cls) elif len(_desktop_sessions.keys()) > 0: - if settings.use_multi_desktop and is_windows and (port or aedt_process_id): + if settings.use_multi_desktop and (port or aedt_process_id): for el in list(_desktop_sessions.values()): if (el.port != 0 and el.port == port) or ( el.aedt_process_id and el.aedt_process_id == aedt_process_id ): return el + return object.__new__(cls) sessions = list(_desktop_sessions.keys()) try: process_id = _desktop_sessions[sessions[0]].odesktop.GetProcessID() @@ -491,18 +472,22 @@ def __init__( port=0, aedt_process_id=None, ): - if _desktop_sessions and (specified_version is None or not settings.use_grpc_api): + if _desktop_sessions and specified_version is None: specified_version = list(_desktop_sessions.values())[-1].aedt_version_id if aedt_process_id: # pragma no cover aedt_process_id = int(aedt_process_id) if getattr(self, "_initialized", None) is not None and self._initialized: + try: + self.grpc_plugin.recreate_application(True) + except Exception: + pass return else: self._initialized = True self._initialized_from_design = True if Desktop._invoked_from_design else False Desktop._invoked_from_design = False self.parent_desktop_id = [] - + self._odesktop = None self._connected_app_instances = 0 """Initialize desktop.""" @@ -998,41 +983,42 @@ def _initialize( os.environ["DesktopPluginPyAEDT"] = os.path.join(settings.aedt_install_dir, "PythonFiles", "DesktopPlugin") launch_msg = "AEDT installation Path {}".format(base_path) self.logger.info(launch_msg) - import pyaedt.generic.grpc_plugin as python_grpc_wrapper + from pyaedt.generic.grpc_plugin_dll_class import AEDT - if _desktop_sessions: - last_session = list(_desktop_sessions.values())[-1] - all_desktop = [i for i in last_session.odesktop.GetRunningInstancesMgr().GetAllRunningInstances()] - for desktop in all_desktop: - try: - if port and desktop.GetGrpcServerPort() == port: - self.isoutsideDesktop = True - self.odesktop = desktop - self.aedt_process_id = self.odesktop.GetProcessID() - self.is_grpc_api = True - last_session.parent_desktop_id.append(self.aedt_process_id) - return True - except Exception: - messages = desktop.GetMessages("", "", 0) - for message in messages: - if "gRPC server running on port: " in message and str(port) in message: - self.isoutsideDesktop = True - self.odesktop = desktop - self.aedt_process_id = self.odesktop.GetProcessID() - self.is_grpc_api = True - last_session.parent_desktop_id.append(self.aedt_process_id) - return True - if new_session: - self.launched_by_pyaedt = new_session - oapp = python_grpc_wrapper.CreateAedtApplication(machine, port, non_graphical, new_session) + if settings.use_multi_desktop: + os.environ["DesktopPluginPyAEDT"] = os.path.join( + list(installed_versions().values())[0], "PythonFiles", "DesktopPlugin" + ) + self.grpc_plugin = AEDT(os.environ["DesktopPluginPyAEDT"]) + oapp = self.grpc_plugin.CreateAedtApplication(machine, port, non_graphical, new_session) if oapp: self.isoutsideDesktop = True - self.odesktop = oapp.GetAppDesktop() self.aedt_process_id = self.odesktop.GetProcessID() self.is_grpc_api = True return True + @property + def odesktop(self): + """AEDT instance containing all projects and designs. + + Examples + -------- + Get the COM object representing the desktop. + + >>> from pyaedt import Desktop + >>> d = Desktop() + >>> d.odesktop + """ + try: + return self.grpc_plugin.odesktop + except Exception: + return self._odesktop + + @odesktop.setter + def odesktop(self, val): + self._odesktop = val + def _init_grpc(self, non_graphical, new_aedt_session, version, student_version, version_key): if settings.remote_rpc_session: # pragma: no cover settings.remote_api = True @@ -1052,12 +1038,12 @@ def _init_grpc(self, non_graphical, new_aedt_session, version, student_version, socket.getfqdn(), socket.getfqdn().split(".")[0], ]: - self.machine = "" + self.machine = "127.0.0.1" else: settings.remote_api = True if not self.port: - if self.machine: - self.logger.error("New Session of AEDT cannot be started on remote machine from Desktop Class.") + if self.machine and self.machine != "127.0.0.1": + self.logger.error("New session of AEDT cannot be started on remote machine from Desktop Class.") self.logger.error("Either use port argument or start an rpc session to start AEDT on remote machine.") self.logger.error("Use client = pyaedt.common_rpc.client(machinename) to start a remote session.") self.logger.error("Use client.aedt(port) to start aedt on remote machine before connecting.") @@ -1571,7 +1557,6 @@ def release_desktop(self, close_projects=True, close_on_exit=True): for a in props: self.__dict__.pop(a, None) - self.odesktop = None gc.collect() return result diff --git a/pyaedt/generic/grpc_plugin.py b/pyaedt/generic/grpc_plugin.py deleted file mode 100644 index 025e259651f..00000000000 --- a/pyaedt/generic/grpc_plugin.py +++ /dev/null @@ -1,261 +0,0 @@ -import re -import types - -from pyaedt.generic.general_methods import GrpcApiError -from pyaedt.generic.general_methods import _retry_ntimes -from pyaedt.generic.general_methods import settings -import pyaedt.generic.grpc_plugin_dll as AedtAPI - -logger = settings.logger -__all__ = ["CreateAedtApplication", "Release"] - - -def CreateAedtObj(objectID, bIsPropSvr, listFuncs): - # print("Create " + str(objectID)) - if bIsPropSvr: - return AedtPropServer(objectID, listFuncs) - return AedtObjWrapper(objectID, listFuncs) - - -def CreateAedtBlockObj(list): - count = len(list) - if count > 1: - if isinstance(list[0], str): - toks = list[0].split(":") - if len(toks) == 2: - start = -1 - if count % 2 == 0: - if toks[1] == "=": - start = 2 - elif count > 2: - if toks[0] == "NAME": - start = 1 - if start > 0: - isBlock = True - for i in range(start, count - 1, 2): - if isinstance(list[i], str): - toks = list[i].split(":") - if len(toks) != 2 or toks[1] != "=": - isBlock = False - break - else: - isBlock = False - break - if isBlock: - return AedtBlockObj(list) - return list - - -def GetAedtObjId(obj): - if isinstance(obj, AedtObjWrapper): - return obj.objectID - return None - - -class AedtBlockObj(list): - def GetName(self): - if len(self) > 0: - f = self[0] - if isinstance(f, str): - d = f.split(":") - if len(d) == 2: - if d[0] == "NAME": - return d[1] - - def __GetValueIdxByKey__(self, keyName): - for i in range(0, len(self) - 1): - if isinstance(self[i], str): - toks = self[i].split(":") - if len(toks) == 2: - if toks[1] == "=" and toks[0] == keyName: - return i + 1 - raise GrpcApiError(keyName + " is not a key!") - - def __getitem__(self, idxOrKey): - if isinstance(idxOrKey, str): - idx = self.__GetValueIdxByKey__(idxOrKey) - if idx != None: - return super().__getitem__(idx) - return super().__getitem__(idxOrKey) - - def __setitem__(self, idxOrKey, newVal): - if isinstance(idxOrKey, int): - if idxOrKey >= 0 or idxOrKey < len(self): - oldItem = self.__getitem__(idxOrKey) - if isinstance(oldItem, str): - toks = oldItem.split(":") - if len(toks) == 2 and (toks[1] == "=" or toks[0] == "NAME"): - raise GrpcApiError("The element is a key should not be overwritten!") - return super().__setitem__(idxOrKey, newVal) - if isinstance(idxOrKey, str): - idx = self.__GetValueIdxByKey__(idxOrKey) - if idx != None: - return super().__setitem__(idx, newVal) - raise GrpcApiError("Key not found") - raise GrpcApiError("Must be key name or index") - - def keys(self): - arr = [] - for i in range(0, len(self) - 1): - if isinstance(self[i], str): - toks = self[i].split(":") - if len(toks) == 2 and toks[1] == "=": - arr.append(toks[0]) - return arr - - -class AedtObjWrapper: - def __init__(self, objID, listFuncs): - self.__dict__["objectID"] = objID # avoid derive class overwrite __setattr__ - self.__dict__["__methodNames__"] = listFuncs - # print(self.objectID) - - def __str__(self): - return "Instance of an Aedt object:" + str(self.objectID) - - def __Invoke__(self, funcName, argv): - if settings.enable_debug_grpc_api_logger: - settings.logger.debug("{} {}".format(funcName, argv)) - try: - return _retry_ntimes( - settings.number_of_grpc_api_retries, AedtAPI.InvokeAedtObjMethod, self.objectID, funcName, argv - ) # Call C function - except Exception: # pragma: no cover - raise GrpcApiError("Failed to execute grpc AEDT command: {}".format(funcName)) - - def __dir__(self): - return self.__methodNames__ - - def __GetObjMethod__(self, funcName): - try: - - def DynamicFunc(self, *args): - return self.__Invoke__(funcName, args) - - return types.MethodType(DynamicFunc, self) - except (AttributeError, GrpcApiError): - raise GrpcApiError("This Aedt object has no attribute '" + funcName + "'") - - def __getattr__(self, funcName): - try: - if funcName == "ScopeID": # backward compatible for IronPython wrapper. - return self.objectID - return self.__GetObjMethod__(funcName) - except Exception: - raise GrpcApiError("Failed to get grpc API AEDT attribute {}".format(funcName)) - - def __setattr__(self, attrName, val): - if attrName == "objectID" or attrName == "__methodNames__": - raise GrpcApiError("Modify this attribute is not allowed") - elif attrName in self.__methodNames__: - raise GrpcApiError(attrName + " is a function name") - else: - super().__setattr__(attrName, val) - - def __del__(self): - AedtAPI.ReleaseAedtObject(self.objectID) - - def match(self, patternStr): # IronPython wrapper implemented this function return IEnumerable. - class IEnumerable(list): - def __getattr__(self, key): - if key == "Count": - return len(self) - - pattern = re.compile(patternStr) - found = IEnumerable() - allMethods = self.__methodNames__ - for method in allMethods: - if pattern.match(method): - found.append(method) - return found - - def GetHashCode(self): # IronPython build in function - return self.__hash__() - - -class AedtPropServer(AedtObjWrapper): - def __init__(self, objID, listFuncs): - AedtObjWrapper.__init__(self, objID, listFuncs) - self.__dict__["__propMap__"] = None - self.__dict__["__propNames__"] = None - - def __GetPropAttributes(self): - if self.__propMap__ == None: - propMap = {} - propNames = self.GetPropNames() - for prop in propNames: - attrName = "" - if prop[0].isdigit(): - attrName += "_" - for c in prop: - if c.isalnum() == True: - attrName += c - else: - attrName += "_" - propMap[attrName] = prop - self.__propMap__ = propMap - return self.__propMap__ - - def __dir__(self): - ret = super().__dir__().copy() - for attrName in self.__GetPropAttributes().keys(): - ret.append(attrName) - return ret - - def __getattr__(self, attrName): - try: - return super().__getattr__(attrName) - except AttributeError: - # if AedtAPI.IsAedtObjPropName(self.objectID, attrName, False): - # return self.GetPropValue(attrName) - propMap = self.__GetPropAttributes() - if attrName in propMap: - return self.GetPropValue(propMap[attrName]) - raise GrpcApiError("Failed to retrieve attribute {} from GRPC API".format(attrName)) - - def __setattr__(self, attrName, val): - if attrName in self.__dict__: - self.__dict__[attrName] = val - return - - # if AedtAPI.IsAedtObjPropName(self.objectID, attrName, True): - # self.SetPropValue(attrName, val) - # return - propMap = self.__GetPropAttributes() - if attrName in propMap: - self.SetPropValue(propMap[attrName], val) - return - super().__setattr__(attrName, val) - - def GetName(self): - return self.__Invoke__("GetName", ()) - - def GetObjPath(self): - return self.__Invoke__("GetObjPath", ()) - - def GetChildNames(self, childType=""): - return self.__Invoke__("GetChildNames", (childType)) - - def GetPropNames(self, includeReadOnly=True): - if includeReadOnly: - if self.__propNames__ == None: - self.__propNames__ = self.__Invoke__("GetPropNames", (includeReadOnly,)) - return self.__propNames__ - return self.__Invoke__("GetPropNames", (includeReadOnly,)) - - def GetPropValue(self, propName=""): - return self.__Invoke__("GetPropValue", (propName,)) - - def SetPropValue(self, propName, val): - return self.__Invoke__("SetPropValue", (propName, val)) - - -def CreateAedtApplication(machine="", port=0, NGmode=False, alwaysNew=True): - return AedtAPI.CreateAedtApplication(machine, port, NGmode, alwaysNew) - - -def Release(): - AedtAPI.ReleaseAll() - - -AedtAPI.SetPyObjCalbacks(CreateAedtObj, CreateAedtBlockObj, GetAedtObjId) diff --git a/pyaedt/generic/grpc_plugin_dll.py b/pyaedt/generic/grpc_plugin_dll.py deleted file mode 100644 index 0c2a4f19821..00000000000 --- a/pyaedt/generic/grpc_plugin_dll.py +++ /dev/null @@ -1,104 +0,0 @@ -from ctypes import CFUNCTYPE -from ctypes import PyDLL -from ctypes import c_bool -from ctypes import c_int -from ctypes import c_wchar_p -from ctypes import py_object -import os - -is_linux = os.name == "posix" -is_windows = not is_linux - -pathDir = os.environ["DesktopPluginPyAEDT"] # DesktopPlugin -pathDir = os.path.dirname(pathDir) # PythonFiles -pathDir = os.path.dirname(pathDir) # DesktopPlugin or Win64 -# dirName = os.path.basename(pathDir) - - -# Plugin filename depends on OS -if is_linux: - pluginFileName = r"libPyDesktopPlugin.so" -else: - pluginFileName = r"PyDesktopPlugin.dll" - -AedtAPIDll_file = os.path.join(pathDir, pluginFileName) # install dir - -if not os.path.isfile(AedtAPIDll_file): - pathDir = os.path.dirname(pathDir) # lib - pathDir = os.path.dirname(pathDir) # core - pathDir = os.path.dirname(pathDir) # view - AedtAPIDll_file = os.path.join(pathDir, r"build_output\64Release\PyDesktopPlugin.dll") # develop dir - # AedtAPIDll_file = os.path.join(pathDir, r"PyAedtStub/x64/Debug/PyAedtStub.dll") #develop dir - -# load dll -if is_windows: - # on windows, modify path - aedtDir = os.path.dirname(AedtAPIDll_file) - originalPath = os.environ["PATH"] - os.environ["PATH"] = originalPath + os.pathsep + aedtDir - AedtAPI = PyDLL(AedtAPIDll_file) - os.environ["PATH"] = originalPath -else: - AedtAPI = PyDLL(AedtAPIDll_file) - -# AedtAPI.SetPyObjCalbacks.argtypes = py_object, py_object, py_object -AedtAPI.SetPyObjCalbacks.restype = None - -# Must use global variable to hold those functions reference -callbackToCreateObj = None -callbackCreateBlock = None -callbackGetObjID = None - - -def SetPyObjCalbacks(CreateAedtObj, CreateAedtBlockObj, GetAedtObjId): - callback_type = CFUNCTYPE(py_object, c_int, c_bool, py_object) - global callbackToCreateObj - global callbackCreateBlock - global callbackGetObjID - callbackToCreateObj = callback_type(CreateAedtObj) # must use global variable to hold this function reference - RetObj_InObj_Func_type = CFUNCTYPE(py_object, py_object) - callbackCreateBlock = RetObj_InObj_Func_type(CreateAedtBlockObj) - callbackGetObjID = RetObj_InObj_Func_type(GetAedtObjId) - AedtAPI.SetPyObjCalbacks(callbackToCreateObj, callbackCreateBlock, callbackGetObjID) - - -# Find the version of AEDT from product info file -version = None -with open(os.path.join(pathDir, "product.info"), "r") as f: - for line in f: - if "AnsProductVersion" in line: - version = line.split("=")[1].strip('\n"') - break - -if version >= "24.1": - AedtAPI.CreateAedtApplication.argtypes = c_wchar_p, py_object, c_bool, c_bool -else: - AedtAPI.CreateAedtApplication.argtypes = c_wchar_p, c_int, c_bool, c_bool - -AedtAPI.CreateAedtApplication.restype = py_object - -AedtAPI.InvokeAedtObjMethod.argtypes = c_int, c_wchar_p, py_object -AedtAPI.InvokeAedtObjMethod.restype = py_object - -AedtAPI.ReleaseAedtObject.argtypes = (c_int,) -AedtAPI.ReleaseAedtObject.restype = None - - -def CreateAedtApplication(machine="", port=0, NGmode=False, alwaysNew=True): - return AedtAPI.CreateAedtApplication(machine, port, NGmode, alwaysNew) - - -def InvokeAedtObjMethod(objectID, funcName, argv): - return AedtAPI.InvokeAedtObjMethod(objectID, funcName, argv) - - -def ReleaseAedtObject(objectID): - AedtAPI.ReleaseAedtObject(objectID) - - -def ReleaseAll(): - AedtAPI.ReleaseAll() - - -def IsEmbedded(): - return False diff --git a/pyaedt/generic/grpc_plugin_dll_class.py b/pyaedt/generic/grpc_plugin_dll_class.py new file mode 100644 index 00000000000..20d8b3c8cc8 --- /dev/null +++ b/pyaedt/generic/grpc_plugin_dll_class.py @@ -0,0 +1,403 @@ +from ctypes import CFUNCTYPE +from ctypes import PyDLL +from ctypes import c_bool +from ctypes import c_int +from ctypes import c_wchar_p +from ctypes import py_object +import os +import re +import types + +from pyaedt.generic.general_methods import GrpcApiError +from pyaedt.generic.general_methods import _retry_ntimes +from pyaedt.generic.general_methods import settings + +logger = settings.logger + + +class AedtBlockObj(list): + def GetName(self): + if len(self) > 0: + f = self[0] + if isinstance(f, str): + d = f.split(":") + if len(d) == 2: + if d[0] == "NAME": + return d[1] + + def __GetValueIdxByKey__(self, keyName): + for i in range(0, len(self) - 1): + if isinstance(self[i], str): + toks = self[i].split(":") + if len(toks) == 2: + if toks[1] == "=" and toks[0] == keyName: + return i + 1 + raise GrpcApiError(keyName + " is not a key!") + + def __getitem__(self, idxOrKey): + if isinstance(idxOrKey, str): + idx = self.__GetValueIdxByKey__(idxOrKey) + if idx != None: + return super().__getitem__(idx) + return super().__getitem__(idxOrKey) + + def __setitem__(self, idxOrKey, newVal): + if isinstance(idxOrKey, int): + if idxOrKey >= 0 or idxOrKey < len(self): + oldItem = self.__getitem__(idxOrKey) + if isinstance(oldItem, str): + toks = oldItem.split(":") + if len(toks) == 2 and (toks[1] == "=" or toks[0] == "NAME"): + raise GrpcApiError("The element is a key. It should not be overwritten.") + return super().__setitem__(idxOrKey, newVal) + if isinstance(idxOrKey, str): + idx = self.__GetValueIdxByKey__(idxOrKey) + if idx != None: + return super().__setitem__(idx, newVal) + raise GrpcApiError("Key is not found.") + raise GrpcApiError("Must be a key name or index.") + + def keys(self): + arr = [] + for i in range(0, len(self) - 1): + if isinstance(self[i], str): + toks = self[i].split(":") + if len(toks) == 2 and toks[1] == "=": + arr.append(toks[0]) + return arr + + +exclude_list = ["GetAppDesktop", "GetProcessID", "GetGrpcServerPort"] + + +class AedtObjWrapper: + def __init__(self, objID, listFuncs, AedtAPI=None): + self.__dict__["objectID"] = objID # avoid derive class overwrite __setattr__ + self.__dict__["__methodNames__"] = listFuncs + self.dllapi = AedtAPI + self.is_linux = True if os.name == "posix" else False + + # print(self.objectID) + + def __str__(self): + return "Instance of an Aedt object:" + str(self.objectID) + + def __Invoke__(self, funcName, argv): + if settings.enable_debug_grpc_api_logger: + settings.logger.debug("{} {}".format(funcName, argv)) + try: + if settings.use_multi_desktop and funcName not in exclude_list: + self.dllapi.recreate_application(True if self.is_linux else False) + ret = _retry_ntimes( + settings.number_of_grpc_api_retries, + self.dllapi.AedtAPI.InvokeAedtObjMethod, + self.objectID, + funcName, + argv, + ) # Call C function + if ret and isinstance(ret, (AedtObjWrapper, AedtPropServer)): + ret.AedtAPI = self.AedtAPI + return ret + except Exception: # pragma: no cover + raise GrpcApiError("Failed to execute gRPC AEDT command: {}".format(funcName)) + + def __dir__(self): + return self.__methodNames__ + + def __GetObjMethod__(self, funcName): + try: + + def DynamicFunc(self, *args): + return self.__Invoke__(funcName, args) + + return types.MethodType(DynamicFunc, self) + except (AttributeError, GrpcApiError): + raise GrpcApiError("This AEDT object has no attribute '" + funcName + "'") + + def __getattr__(self, funcName): + try: + if funcName == "ScopeID": # backward compatible for IronPython wrapper. + return self.objectID + return self.__GetObjMethod__(funcName) + except Exception: + raise GrpcApiError("Failed to get gRPC API AEDT attribute {}".format(funcName)) + + def __setattr__(self, attrName, val): + if attrName == "objectID" or attrName == "__methodNames__": + raise GrpcApiError("This attribute cannot be modified.") + elif attrName in self.__methodNames__: + raise GrpcApiError(attrName + " is a function name.") + else: + super().__setattr__(attrName, val) + + def __del__(self): + if "ReleaseAedtObject" in dir(self.dllapi): + self.dllapi.ReleaseAedtObject(self.objectID) + + def match(self, patternStr): # IronPython wrapper implemented this function return IEnumerable. + class IEnumerable(list): + def __getattr__(self, key): + if key == "Count": + return len(self) + + pattern = re.compile(patternStr) + found = IEnumerable() + allMethods = self.__methodNames__ + for method in allMethods: + if pattern.match(method): + found.append(method) + return found + + def GetHashCode(self): # IronPython build in function + return self.__hash__() + + +class AedtPropServer(AedtObjWrapper): + def __init__(self, objID, listFuncs, aedtapi): + AedtObjWrapper.__init__(self, objID, listFuncs, aedtapi) + self.__dict__["__propMap__"] = None + self.__dict__["__propNames__"] = None + + def __GetPropAttributes(self): + if self.__propMap__ == None: + propMap = {} + propNames = self.GetPropNames() + for prop in propNames: + attrName = "" + if prop[0].isdigit(): + attrName += "_" + for c in prop: + if c.isalnum() == True: + attrName += c + else: + attrName += "_" + propMap[attrName] = prop + self.__propMap__ = propMap + return self.__propMap__ + + def __dir__(self): + ret = super().__dir__().copy() + for attrName, _ in self.__GetPropAttributes(): + ret.append(attrName) + return ret + + def __getattr__(self, attrName): + try: + return super().__getattr__(attrName) + except AttributeError: + # if AedtAPI.IsAedtObjPropName(self.objectID, attrName, False): + # return self.GetPropValue(attrName) + propMap = self.__GetPropAttributes() + if attrName in propMap: + return self.GetPropValue(propMap[attrName]) + raise GrpcApiError("Failed to retrieve attribute {} from gRPC API".format(attrName)) + + def __setattr__(self, attrName, val): + if attrName in self.__dict__: + self.__dict__[attrName] = val + return + propMap = self.__GetPropAttributes() + try: + if attrName in propMap: + self.SetPropValue(propMap[attrName], val) + return + except Exception: + pass + super().__setattr__(attrName, val) + + def GetName(self): + return self.__Invoke__("GetName", ()) + + def GetObjPath(self): + return self.__Invoke__("GetObjPath", ()) + + def GetChildNames(self, childType=""): + return self.__Invoke__("GetChildNames", (childType)) + + def GetPropNames(self, includeReadOnly=True): + if includeReadOnly: + if self.__propNames__ == None: + self.__propNames__ = self.__Invoke__("GetPropNames", (includeReadOnly,)) + return self.__propNames__ + return self.__Invoke__("GetPropNames", (includeReadOnly,)) + + def GetPropValue(self, propName=""): + return self.__Invoke__("GetPropValue", (propName,)) + + def SetPropValue(self, propName, val): + return self.__Invoke__("SetPropValue", (propName, val)) + + +class AEDT: + def __init__(self, pathDir): + is_linux = os.name == "posix" + is_windows = not is_linux + self.original_path = pathDir + self.pathDir = pathDir + self.pathDir = os.path.dirname(self.pathDir) # PythonFiles + self.pathDir = os.path.dirname(self.pathDir) # DesktopPlugin or Win64 + # dirName = os.path.basename(pathDir) + + # Plugin filename depends on OS + if is_linux: + pluginFileName = r"libPyDesktopPlugin.so" + else: + pluginFileName = r"PyDesktopPlugin.dll" + + AedtAPIDll_file = os.path.join(self.pathDir, pluginFileName) # install dir + + if not os.path.isfile(AedtAPIDll_file): + self.pathDir = os.path.dirname(self.pathDir) # lib + self.pathDir = os.path.dirname(self.pathDir) # core + self.pathDir = os.path.dirname(self.pathDir) # view + AedtAPIDll_file = os.path.join(self.pathDir, r"build_output\64Release\PyDesktopPlugin.dll") # develop dir + # AedtAPIDll_file = os.path.join(pathDir, r"PyAedtStub/x64/Debug/PyAedtStub.dll") #develop dir + + # load dll + if is_windows: + # on windows, modify path + aedtDir = os.path.dirname(AedtAPIDll_file) + originalPath = os.environ["PATH"] + os.environ["PATH"] = originalPath + os.pathsep + aedtDir + AedtAPI = PyDLL(AedtAPIDll_file) + os.environ["PATH"] = originalPath + else: + AedtAPI = PyDLL(AedtAPIDll_file) + # AedtAPI.SetPyObjCalbacks.argtypes = py_object, py_object, py_object + AedtAPI.SetPyObjCalbacks.restype = None + + # Must use global variable to hold those functions reference + self.callbackToCreateObj = None + self.callbackCreateBlock = None + self.callbackGetObjID = None + version = None + with open(os.path.join(self.pathDir, "product.info"), "r") as f: + for line in f: + if "AnsProductVersion" in line: + version = line.split("=")[1].strip('\n"') + break + + if version >= "24.1": + AedtAPI.CreateAedtApplication.argtypes = c_wchar_p, py_object, c_bool, c_bool + else: + AedtAPI.CreateAedtApplication.argtypes = c_wchar_p, c_int, c_bool, c_bool + + AedtAPI.CreateAedtApplication.restype = py_object + + AedtAPI.InvokeAedtObjMethod.argtypes = c_int, c_wchar_p, py_object + AedtAPI.InvokeAedtObjMethod.restype = py_object + + AedtAPI.ReleaseAedtObject.argtypes = (c_int,) + AedtAPI.ReleaseAedtObject.restype = None + self.AedtAPI = AedtAPI + self.SetPyObjCalbacks() + self.aedt = None + + def SetPyObjCalbacks(self): + self.callback_type = CFUNCTYPE(py_object, c_int, c_bool, py_object) + self.callbackToCreateObj = self.callback_type( + self.CreateAedtObj + ) # must use global variable to hold this function reference + RetObj_InObj_Func_type = CFUNCTYPE(py_object, py_object) + self.callbackCreateBlock = RetObj_InObj_Func_type(self.CreateAedtBlockObj) + self.callbackGetObjID = RetObj_InObj_Func_type(self.GetAedtObjId) + self.AedtAPI.SetPyObjCalbacks(self.callbackToCreateObj, self.callbackCreateBlock, self.callbackGetObjID) + + def CreateAedtApplication(self, machine="", port=0, NGmode=False, alwaysNew=True): + self.aedt = self.AedtAPI.CreateAedtApplication(machine, port, NGmode, alwaysNew) + self.machine = machine + if port == 0: + self.port = self.aedt.GetAppDesktop().GetGrpcServerPort() + else: + self.port = port + + return self.aedt + + @property + def odesktop(self): + if settings.use_multi_desktop: + self.aedt = self.recreate_application() + return self.aedt.GetAppDesktop() + + def recreate_application(self, force=False): + def run(): + self.ReleaseAedtObject(self.aedt.objectID) + port = self.port + machine = self.machine + self.__init__(self.original_path) + self.port = port + self.machine = machine + self.aedt = self.AedtAPI.CreateAedtApplication(self.machine, self.port, False, False) + return self.aedt + + if force: + return run() + else: + try: + self.aedt.GetAppDesktop() + return self.aedt + except Exception: + return run() + + def InvokeAedtObjMethod(self, objectID, funcName, argv): + return self.AedtAPI.InvokeAedtObjMethod(objectID, funcName, argv) + + def ReleaseAedtObject(self, objectID): + self.AedtAPI.ReleaseAedtObject(objectID) + + def ReleaseAll(self): + self.AedtAPI.ReleaseAll() + + def IsEmbedded(self): + return False + + def CreateAedtObj(self, objectID, bIsPropSvr, listFuncs): + # print("Create " + str(objectID)) + if bIsPropSvr: + return AedtPropServer( + objectID, + listFuncs, + self, + ) + + return AedtObjWrapper( + objectID, + listFuncs, + self, + ) + + def CreateAedtBlockObj(self, list_in): + count = len(list_in) + if count > 1: + if isinstance(list_in[0], str): + toks = list_in[0].split(":") + if len(toks) == 2: + start = -1 + if count % 2 == 0: + if toks[1] == "=": + start = 2 + elif count > 2: + if toks[0] == "NAME": + start = 1 + if start > 0: + isBlock = True + for i in range(start, count - 1, 2): + if isinstance(list_in[i], str): + toks = list_in[i].split(":") + if len(toks) != 2 or toks[1] != "=": + isBlock = False + break + else: + isBlock = False + break + if isBlock: + return AedtBlockObj(list_in) + return list_in + + def GetAedtObjId(self, obj): + if isinstance(obj, AedtObjWrapper): + return obj.objectID + return None + + def Release(self): + self.AedtAPI.ReleaseAll()