File size: 2,834 Bytes
b400ace | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """
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()
@staticmethod
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()
|