diff --git a/IoTuring/ClassManager/ClassManager.py b/IoTuring/ClassManager/ClassManager.py index ea30c2763..a522fc0de 100644 --- a/IoTuring/ClassManager/ClassManager.py +++ b/IoTuring/ClassManager/ClassManager.py @@ -1,56 +1,107 @@ from __future__ import annotations -import os -from pathlib import Path -from os import path import importlib.util import importlib.machinery import sys import inspect +from pathlib import Path + +from IoTuring.ClassManager.consts import * from IoTuring.Logger.LogObject import LogObject +from IoTuring.MyApp.App import App + -# from IoTuring.ClassManager import consts +class ClassManager(LogObject): + """Base class for ClassManagers + + This class is used to find and load classes without importing them + The important this is that the class is inside a folder that exactly the same name of the Class and of the file (obviously not talking about extensions) + """ -# This is a parent class + def __init__(self, class_key:str) -> None: -# Implement subclasses in this way: + if class_key not in CLASS_PATH: + raise Exception(f"Invalid class key {class_key}") + else: + self.classesRelativePath = CLASS_PATH[class_key] -# def __init__(self): -# ClassManager.__init__(self) -# self.baseClass = Entity : Select the class to find -# self.GetModulesFilename(consts.ENTITIES_PATH) : Select path where it should look for classes and add all classes to found list + # Store loaded classes here: + self.loadedClasses = [] -# This class is used to find and load classes without importing them -# The important this is that the class is inside a folder that exactly the same name of the Class and of the file (obviously not talking about extensions) + # Collect paths + self.moduleFilePaths = self.GetModuleFilePaths() + def GetModuleFilePaths(self) -> list[Path]: + """Get the paths of of python files of this class + + Raises: + Exception: If path not defined or exists + FileNotFoundError: No module in the dir + + Returns: + list[Path]: List of paths of python files + """ + + if not self.classesRelativePath: + raise Exception("Path to deployments not defined") + + # Get the absolute path of the dir of files: + classesRootPath = App.getRootPath().joinpath(self.classesRelativePath) + + if not classesRootPath.exists: + raise Exception(f"Path does not exist: {classesRootPath}") + + self.Log(self.LOG_DEVELOPMENT, + f'Looking for python files in "{classesRootPath}"...') + + python_files = classesRootPath.rglob("*.py") + + # Check if a py files are in a folder with the same name !!! (same without extension) + filepaths = [f for f in python_files if f.stem == f.parent.stem] + + if not filepaths: + raise FileNotFoundError( + f"No module files found in {classesRootPath}") + + self.Log(self.LOG_DEVELOPMENT, + f"Found {str(len(filepaths))} modules files") + + return filepaths + + def GetClassFromName(self, wantedName: str) -> type | None: + """Get the class of given name, and load it + + Args: + wantedName (str): The name to look for + + Returns: + type | None: The class if found, None if not found + """ + + # Check from already loaded classes: + module_class = next( + (m for m in self.loadedClasses if m.__name__ == wantedName), None) + + if module_class: + return module_class + + modulePath = next( + (m for m in self.moduleFilePaths if m.stem == wantedName), None) + + if modulePath: + + loadedModule = self.LoadModule(modulePath) + loadedClass = self.GetClassFromModule(loadedModule) + self.loadedClasses.append(loadedClass) + return loadedClass -class ClassManager(LogObject): - def __init__(self): - self.modulesFilename = [] - module_path = sys.modules[self.__class__.__module__].__file__ - if not module_path: - raise Exception("Error getting path: " + str(module_path)) else: - self.mainPath = path.dirname(path.abspath(module_path)) - # THIS MUST BE IMPLEMENTED IN SUBCLASSES, IS THE CLASS I WANT TO SEARCH !!!! - self.baseClass = None - - def GetClassFromName(self, wantedName) -> type | None: - # From name, load the correct module and extract the class - for module in self.modulesFilename: # Search the module file - moduleName = self.ModuleNameFromPath(module) - # Check if the module name matches the given name - if wantedName == moduleName: - # Load the module - loadedModule = self.LoadModule(module) - # Now get the class - return self.GetClassFromModule(loadedModule) - return None - - def LoadModule(self, path): # Get module and load it from the path + return None + + def LoadModule(self, module_path: Path): # Get module and load it from the path try: loader = importlib.machinery.SourceFileLoader( - self.ModuleNameFromPath(path), path) + module_path.stem, str(module_path)) spec = importlib.util.spec_from_loader(loader.name, loader) if not spec: @@ -58,50 +109,25 @@ def LoadModule(self, path): # Get module and load it from the path module = importlib.util.module_from_spec(spec) loader.exec_module(module) - moduleName = os.path.split(path)[1][:-3] - sys.modules[moduleName] = module + sys.modules[module_path.stem] = module return module except Exception as e: - self.Log(self.LOG_ERROR, "Error while loading module " + - path + ": " + str(e)) + self.Log(self.LOG_ERROR, + f"Error while loading module {module_path.stem}: {str(e)}") # From the module passed, I search for a Class that has className=moduleName def GetClassFromModule(self, module): for name, obj in inspect.getmembers(module): if inspect.isclass(obj): - if(name == module.__name__): + if (name == module.__name__): return obj raise Exception(f"No class found: {module.__name__}") - # List files in the _path directory and get only files in subfolders - def GetModulesFilename(self, _path): - classesRootPath = path.join(self.mainPath, _path) - if os.path.exists(classesRootPath): - self.Log(self.LOG_DEVELOPMENT, - "Looking for python files in \"" + _path + os.sep + "\"...") - result = list(Path(classesRootPath).rglob("*.py")) - entities = [] - for file in result: - filename = str(file) - # TO check if a py files is in a folder !!!! with the same name !!! (same without extension) - pathList = filename.split(os.sep) - if len(pathList) >= 2: - if pathList[len(pathList)-1][:-3] == pathList[len(pathList)-2]: - entities.append(filename) - - self.modulesFilename = self.modulesFilename + entities - self.Log(self.LOG_DEVELOPMENT, "Found " + - str(len(entities)) + " modules files") - - def ModuleNameFromPath(self, path): - classname = os.path.split(path) - return classname[1][:-3] - - def ListAvailableClassesNames(self) -> list: - res = [] - for py in self.modulesFilename: - res.append(path.basename(py).split(".py")[0]) - return res - def ListAvailableClasses(self) -> list: - return [self.GetClassFromName(n) for n in self.ListAvailableClassesNames()] + """Get all classes of this ClassManager + + Returns: + list: The list of classes + """ + + return [self.GetClassFromName(f.stem) for f in self.moduleFilePaths] diff --git a/IoTuring/ClassManager/EntityClassManager.py b/IoTuring/ClassManager/EntityClassManager.py deleted file mode 100644 index 5f6dee5e6..000000000 --- a/IoTuring/ClassManager/EntityClassManager.py +++ /dev/null @@ -1,12 +0,0 @@ -from IoTuring.ClassManager.ClassManager import ClassManager -from IoTuring.ClassManager import consts -from IoTuring.Entity.Entity import Entity - - -# Class to load Entities from the Entitties dir and get them from name -class EntityClassManager(ClassManager): - def __init__(self): - ClassManager.__init__(self) - self.baseClass = Entity - self.GetModulesFilename(consts.ENTITIES_PATH) - # self.GetModulesFilename(consts.CUSTOM_ENTITIES_PATH) # TODO Decide if I'll use customs diff --git a/IoTuring/ClassManager/WarehouseClassManager.py b/IoTuring/ClassManager/WarehouseClassManager.py deleted file mode 100644 index 0bedb2ae6..000000000 --- a/IoTuring/ClassManager/WarehouseClassManager.py +++ /dev/null @@ -1,11 +0,0 @@ -from IoTuring.ClassManager.ClassManager import ClassManager -from IoTuring.ClassManager import consts -from IoTuring.Warehouse.Warehouse import Warehouse - - -# Class to load Entities from the Entitties dir and get them from name -class WarehouseClassManager(ClassManager): - def __init__(self): - ClassManager.__init__(self) - self.baseClass = Warehouse - self.GetModulesFilename(consts.WAREHOUSES_PATH) diff --git a/IoTuring/ClassManager/consts.py b/IoTuring/ClassManager/consts.py index 5a424e024..9e073220c 100644 --- a/IoTuring/ClassManager/consts.py +++ b/IoTuring/ClassManager/consts.py @@ -1,2 +1,7 @@ -ENTITIES_PATH = "../Entity/Deployments/" -WAREHOUSES_PATH = "../Warehouse/Deployments/" +KEY_ENTITY = "entity" +KEY_WAREHOUSE = "warehouse" + +CLASS_PATH = { + KEY_ENTITY: "Entity/Deployments", + KEY_WAREHOUSE: "Warehouse/Deployments" +} diff --git a/IoTuring/Configurator/Configurator.py b/IoTuring/Configurator/Configurator.py index 136f1aa83..06954df18 100644 --- a/IoTuring/Configurator/Configurator.py +++ b/IoTuring/Configurator/Configurator.py @@ -6,8 +6,7 @@ from IoTuring.Logger.LogObject import LogObject from IoTuring.Exceptions.Exceptions import UserCancelledException -from IoTuring.ClassManager.EntityClassManager import EntityClassManager -from IoTuring.ClassManager.WarehouseClassManager import WarehouseClassManager +from IoTuring.ClassManager.ClassManager import ClassManager, KEY_ENTITY, KEY_WAREHOUSE from IoTuring.Configurator import ConfiguratorIO from IoTuring.Configurator import messages @@ -111,7 +110,7 @@ def Menu(self) -> None: def ManageEntities(self) -> None: """ UI for Entities settings """ - ecm = EntityClassManager() + ecm = ClassManager(KEY_ENTITY) manageEntitiesChoices = [] @@ -146,21 +145,19 @@ def ManageEntities(self) -> None: def ManageWarehouses(self) -> None: """ UI for Warehouses settings """ - wcm = WarehouseClassManager() + wcm = ClassManager(KEY_WAREHOUSE) manageWhChoices = [] - availableWarehouses = wcm.ListAvailableClassesNames() - for whName in availableWarehouses: - short_wh_name = whName.replace("Warehouse", "") + availableWarehouses = wcm.ListAvailableClasses() + for whClass in availableWarehouses: - enabled_sign = " " - if self.IsWarehouseActive(short_wh_name): - enabled_sign = "X" + enabled_sign = "X" \ + if self.IsWarehouseActive(whClass.NAME) else " " manageWhChoices.append( - {"name": f"[{enabled_sign}] - {short_wh_name}", - "value": short_wh_name}) + {"name": f"[{enabled_sign}] - {whClass.NAME}", + "value": whClass}) choice = self.DisplayMenu( choices=manageWhChoices, @@ -170,7 +167,7 @@ def ManageWarehouses(self) -> None: if choice == CHOICE_GO_BACK: self.Menu() else: - self.ManageSingleWarehouse(choice, wcm) + self.ManageSingleWarehouse(choice) def DisplayHelp(self) -> None: self.DisplayMessage(messages.HELP_MESSAGE) @@ -195,10 +192,10 @@ def WriteConfigurations(self) -> None: """ Save to configurations file """ self.configuratorIO.writeConfigurations(self.config) - def ManageSingleWarehouse(self, warehouseName, wcm: WarehouseClassManager): + def ManageSingleWarehouse(self, whClass): """UI for single Warehouse settings""" - if self.IsWarehouseActive(warehouseName): + if self.IsWarehouseActive(whClass.NAME): manageWhChoices = [ {"name": "Edit the warehouse settings", "value": "Edit"}, {"name": "Remove the warehouse", "value": "Remove"} @@ -209,24 +206,24 @@ def ManageSingleWarehouse(self, warehouseName, wcm: WarehouseClassManager): choice = self.DisplayMenu( choices=manageWhChoices, - message=f"Manage warehouse {warehouseName}" + message=f"Manage warehouse {whClass.NAME}" ) if choice == CHOICE_GO_BACK: self.ManageWarehouses() elif choice == "Edit": - self.EditActiveWarehouse(warehouseName, wcm) + self.EditActiveWarehouse(whClass.NAME) elif choice == "Add": - self.AddActiveWarehouse(warehouseName, wcm) + self.AddActiveWarehouse(whClass) elif choice == "Remove": confirm = inquirer.confirm(message="Are you sure?").execute() if confirm: - self.RemoveActiveWarehouse(warehouseName) + self.RemoveActiveWarehouse(whClass.NAME) else: self.ManageWarehouses() - def ManageSingleEntity(self, entityConfig, ecm: EntityClassManager): + def ManageSingleEntity(self, entityConfig, ecm: ClassManager): """ UI to manage an active warehouse (edit config/remove) """ manageEntityChoices = [ @@ -251,7 +248,7 @@ def ManageSingleEntity(self, entityConfig, ecm: EntityClassManager): else: self.ManageEntities() - def SelectNewEntity(self, ecm: EntityClassManager): + def SelectNewEntity(self, ecm: ClassManager): """ UI to add a new Entity """ # entity classnames without unsupported entities: @@ -286,7 +283,7 @@ def SelectNewEntity(self, ecm: EntityClassManager): else: self.AddActiveEntity(choice, ecm) - def ShowUnsupportedEntities(self, ecm: EntityClassManager): + def ShowUnsupportedEntities(self, ecm: ClassManager): """ UI to show unsupported entities """ # entity classnames without unsupported entities: @@ -306,7 +303,7 @@ def ShowUnsupportedEntities(self, ecm: EntityClassManager): self.ManageEntities() - def AddActiveEntity(self, entityName, ecm: EntityClassManager): + def AddActiveEntity(self, entityName, ecm: ClassManager): """ From entity name, get its class and retrieve the configuration preset, then add to configuration dict """ entityClass = ecm.GetClassFromName(entityName) try: @@ -368,10 +365,9 @@ def IsWarehouseActive(self, warehouseName) -> bool: return True return False - def AddActiveWarehouse(self, warehouseName, wcm: WarehouseClassManager) -> None: + def AddActiveWarehouse(self, whClass) -> None: """ Add warehouse to the preferences using a menu with the warehouse preset if available """ - whClass = wcm.GetClassFromName(warehouseName + "Warehouse") try: preset = whClass.ConfigurationPreset() # type: ignore @@ -385,7 +381,7 @@ def AddActiveWarehouse(self, warehouseName, wcm: WarehouseClassManager) -> None: "No configuration needed for this Warehouse :)") # Save added settings - self.WarehouseMenuPresetToConfiguration(warehouseName, preset) + self.WarehouseMenuPresetToConfiguration(whClass.NAME, preset) except UserCancelledException: self.DisplayMessage("Configuration cancelled", force_clear=True) @@ -395,7 +391,7 @@ def AddActiveWarehouse(self, warehouseName, wcm: WarehouseClassManager) -> None: self.ManageWarehouses() - def EditActiveWarehouse(self, warehouseName, wcm: WarehouseClassManager) -> None: + def EditActiveWarehouse(self, warehouseName) -> None: """ UI for single Warehouse settings edit """ self.DisplayMessage( "You can't do that at the moment, change the configuration file manually. Sorry for the inconvenience") @@ -406,7 +402,7 @@ def EditActiveWarehouse(self, warehouseName, wcm: WarehouseClassManager) -> None # WarehouseMenuPresetToConfiguration appends a warehosue to the conf so here I should remove it to read it later # TO implement only when I know how to add removable value while editing configurations - def EditActiveEntity(self, entityConfig, ecm: WarehouseClassManager) -> None: + def EditActiveEntity(self, entityConfig, ecm: ClassManager) -> None: """ UI for single Entity settings edit """ self.DisplayMessage( "You can't do that at the moment, change the configuration file manually. Sorry for the inconvenience") diff --git a/IoTuring/Configurator/ConfiguratorLoader.py b/IoTuring/Configurator/ConfiguratorLoader.py index 0c02f7ec6..152642eab 100644 --- a/IoTuring/Configurator/ConfiguratorLoader.py +++ b/IoTuring/Configurator/ConfiguratorLoader.py @@ -4,8 +4,7 @@ from IoTuring.Entity.Entity import Entity from IoTuring.Logger.LogObject import LogObject from IoTuring.Configurator.Configurator import KEY_ENTITY_TYPE, Configurator, KEY_ACTIVE_ENTITIES, KEY_ACTIVE_WAREHOUSES, KEY_WAREHOUSE_TYPE -from IoTuring.ClassManager.WarehouseClassManager import WarehouseClassManager -from IoTuring.ClassManager.EntityClassManager import EntityClassManager +from IoTuring.ClassManager.ClassManager import ClassManager, KEY_ENTITY, KEY_WAREHOUSE from IoTuring.Warehouse.Warehouse import Warehouse @@ -18,7 +17,7 @@ def __init__(self, configurator: Configurator) -> None: # Return list of instances initialized using their configurations def LoadWarehouses(self) -> list[Warehouse]: warehouses = [] - wcm = WarehouseClassManager() + wcm = ClassManager(KEY_WAREHOUSE) if not KEY_ACTIVE_WAREHOUSES in self.configurations: self.Log( self.LOG_ERROR, "You have to enable at least one warehouse: configure it using -c argument") @@ -41,7 +40,7 @@ def LoadWarehouses(self) -> list[Warehouse]: # Return list of entities initialized def LoadEntities(self) -> list[Entity]: entities = [] - ecm = EntityClassManager() + ecm = ClassManager(KEY_ENTITY) if not KEY_ACTIVE_ENTITIES in self.configurations: self.Log( self.LOG_ERROR, "You have to enable at least one entity: configure it using -c argument") diff --git a/IoTuring/MyApp/App.py b/IoTuring/MyApp/App.py index 0467394ce..b4f2f52f2 100644 --- a/IoTuring/MyApp/App.py +++ b/IoTuring/MyApp/App.py @@ -1,4 +1,5 @@ from importlib.metadata import metadata +from pathlib import Path class App(): METADATA = metadata('IoTuring') @@ -40,6 +41,15 @@ def getUrlHomepage() -> str: def getUrlReleases() -> str: return App.URL_RELEASES + @staticmethod + def getRootPath() -> Path: + """Get the project root path + + Returns: + Path: The path to th project root as a pathlib.Path + """ + return Path(__file__).parents[1] + def __str__(self) -> str: msg = "" msg += "Name: " + App.getName() + "\n" diff --git a/tests/ClassManager/test_ClassManager.py b/tests/ClassManager/test_ClassManager.py new file mode 100644 index 000000000..bdd2d342d --- /dev/null +++ b/tests/ClassManager/test_ClassManager.py @@ -0,0 +1,13 @@ +from IoTuring.ClassManager.ClassManager import ClassManager, KEY_ENTITY, KEY_WAREHOUSE + + +class TestClassManager: + def testClassCount(self): + for class_key in [KEY_ENTITY, KEY_WAREHOUSE]: + + cm = ClassManager(class_key) + assert bool(cm.loadedClasses) == False + + class_num = len(cm.ListAvailableClasses()) + assert class_num == len(cm.GetModuleFilePaths()) + assert class_num == len(cm.loadedClasses)