Processor Module Template

"""
Your markdown documentation here...
"""
__version__: int = 1
"""The auto-generated version of the module."""
from dataclasses import dataclass, field
from typing import Any
import os

# Internal imports.
from modules.base.processors.base import AbstractProcessorModule, models
from models.validations import OneOf


class ProcessorModule(AbstractProcessorModule):
    """
    A processor module.

    :param configuration: The configuration object of the module.
    """
    version: int = __version__
    """The version of the module."""
    public: bool = True
    """Is this module public?"""
    description: str = "This module does..."
    """A short description."""
    author: str = "Your Name"
    """The author name."""
    email: str = "your.mail@address.de"
    """The email address of the author."""
    deprecated: bool = False
    """Is this module deprecated."""
    third_party_requirements: list[str] = []
    """Define your requirements here."""
    field_requirements: list[str] = []
    """Data validation requirements for the fields dict."""
    tag_requirements: list[str] = []
    """Data validation requirements for the tags dict."""

    @dataclass
    class Configuration(models.ProcessorModule):
        """
        The configuration of the module.
        """
        level: str = field(
            metadata=dict(
                description="This is an exemplary field configuration parameter.",
                required=False,
                dynamic=True,
                validate=OneOf(['debug', 'info', 'error', 'warning', 'critical'])),
            default="info")

    def __init__(self, configuration: Configuration):
        super().__init__(configuration=configuration, thread_safe=False)

    @classmethod
    def import_third_party_requirements(cls) -> bool:
        """
        Check if all third party requirements are successfully imported.
        Raises an ImportError if the import was not successful.

        :returns: True if the import was successful.
        """
        try:
            # Import third party packages here and mark them as global.
            # global my_module
            # import my_module
            return True
        except Exception:
            raise ImportError("Could not import required packages. Please install '{0}'."
                              .format(' '.join(map(str, cls.third_party_requirements))))

    def start(self):
        """
        Start the module. Raise an exception if something went wrong.
        """
        pass

    def stop(self):
        """
        Stop the module.
        """
        pass

    @staticmethod
    def get_config_data(input_module_instance=None) -> dict[str, Any]:
        """
        Retrieve options for selected configuration parameters of this module.

        :param input_module_instance: If it is a variable or tag module, the input_module_instance.
        :returns: A dictionary containing the parameter as key and a list of options as value.
        """
        return {
            "level": ['debug', 'info', 'error', 'warning', 'critical']
        }

    @classmethod
    def get_test_data(cls) -> list[dict]:
        """
        Provides the test data.

        :returns: A list containing test data with the keys:
                  module_config: The configuration of the module as yaml string.
                  input_data: The input data object with measurement, fields and tags.
                  output_data: The expected output data object with measurement, fields and tags.
        """
        test_1 = {"module_config": cls.Configuration(module_name=os.path.splitext(os.path.basename(__file__))[0],
                                                     level="info"),
                  "input_data": models.Data(measurement="test",
                                            fields={"test3": 1},
                                            tags={"test1": 1, "test2": 2}),
                  "output_data": models.Data(measurement="test",
                                             fields={"test3": 1},
                                             tags={"test1": 1, "test2": 2})}
        return [test_1]

    def _run(self, data: models.Data) -> models.Data:
        """
        Method for executing the module.

        :param data: The data object.
        :returns: The data object after processing.
        """
        self.logger.info(self._dyn(self.configuration.level))
        # TODO: You custom logic here...
        return data