Spaces:
Sleeping
Sleeping
| """ | |
| Processor/Extension framework | |
| Adapated from https://github.com/gdiepen/python_processor_example | |
| """ | |
| import inspect | |
| import pkgutil | |
| from src.logger import logger | |
| class Processor: | |
| """Base class that each processor must inherit from.""" | |
| def __init__( | |
| self, | |
| options=None, | |
| relative_dir=None, | |
| image_instance_ops=None, | |
| ): | |
| self.options = options | |
| self.relative_dir = relative_dir | |
| self.image_instance_ops = image_instance_ops | |
| self.tuning_config = image_instance_ops.tuning_config | |
| self.description = "UNKNOWN" | |
| class ProcessorManager: | |
| """Upon creation, this class will read the processors package for modules | |
| that contain a class definition that is inheriting from the Processor class | |
| """ | |
| def __init__(self, processors_dir="src.processors"): | |
| """Constructor that initiates the reading of all available processors | |
| when an instance of the ProcessorCollection object is created | |
| """ | |
| self.processors_dir = processors_dir | |
| self.reload_processors() | |
| def get_name_filter(processor_name): | |
| def filter_function(member): | |
| return inspect.isclass(member) and member.__module__ == processor_name | |
| return filter_function | |
| def reload_processors(self): | |
| """Reset the list of all processors and initiate the walk over the main | |
| provided processor package to load all available processors | |
| """ | |
| self.processors = {} | |
| self.seen_paths = [] | |
| logger.info(f'Loading processors from "{self.processors_dir}"...') | |
| self.walk_package(self.processors_dir) | |
| def walk_package(self, package): | |
| """walk the supplied package to retrieve all processors""" | |
| imported_package = __import__(package, fromlist=["blah"]) | |
| loaded_packages = [] | |
| for _, processor_name, ispkg in pkgutil.walk_packages( | |
| imported_package.__path__, imported_package.__name__ + "." | |
| ): | |
| if not ispkg and processor_name != __name__: | |
| processor_module = __import__(processor_name, fromlist=["blah"]) | |
| # https://stackoverflow.com/a/46206754/6242649 | |
| clsmembers = inspect.getmembers( | |
| processor_module, | |
| ProcessorManager.get_name_filter(processor_name), | |
| ) | |
| for _, c in clsmembers: | |
| # Only add classes that are a sub class of Processor, but NOT Processor itself | |
| if issubclass(c, Processor) & (c is not Processor): | |
| self.processors[c.__name__] = c | |
| loaded_packages.append(c.__name__) | |
| logger.info(f"Loaded processors: {loaded_packages}") | |
| # Singleton export | |
| PROCESSOR_MANAGER = ProcessorManager() | |