Input 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 time
from threading import Thread
import datetime

# Internal imports.
import config
from models.validations import Range, OneOf
from modules.base.inputs.base import AbstractInputModule, AbstractVariableModule, AbstractTagModule, models

__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."""


class InputModule(AbstractInputModule):
    """
    An input module.

    :param configuration: The configuration object of the module.
    """
    version: int = __version__
    """The version of the module."""
    public: bool = __public__
    """Is this module public?"""
    description: str = __description__
    """A short description."""
    author: str = __author__
    """The author name."""
    email: str = __email__
    """The email address of the author."""
    deprecated: bool = __deprecated__
    """Is this module deprecated."""
    third_party_requirements: list[str] = __third_party_requirements__
    """Define your requirements here."""

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

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

    @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))))

    @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 {
            "key": ['debug', 'info', 'error', 'warning', 'critical']
        }

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

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


class VariableModule(AbstractVariableModule):
    """
    A variable module.

    :param configuration: The configuration object of the module.
    :param input_module_instance: The instance of the parent input module if it exists.
    """
    version: int = __version__
    """The version of the module."""
    public: bool = __public__
    """Is this module public?"""
    description: str = __description__
    """A short description."""
    author: str = __author__
    """The author name."""
    email: str = __email__
    """The email address of the author."""
    deprecated: bool = __deprecated__
    """Is this module deprecated."""
    third_party_requirements: list[str] = __third_party_requirements__
    """Define your requirements here."""

    @dataclass
    class Configuration(models.VariableModule):
        """
        The configuration model of the module.
        """
        input_module: str = field(
            metadata=dict(description="The id of the input module. The required input module has to be "
                                      "module_name: inputs.mqtt.client_1",
                          required=True),
            default=None)
        interval: float = field(
            metadata=dict(description="Interval in seconds in which the module provides data objects. "
                                      "If interval is 0, the module is executed once.",
                          required=False,
                          validate=Range(min=0, max=1000, exclusive=False)),
            default=1)
        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, input_module_instance=None):
        super().__init__(configuration=configuration,
                         input_module_instance=input_module_instance)

    @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))))

    @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']
        }

    def start(self):
        """
        Start the module. Raise an exception if something went wrong.
        """
        # The start time of the execution. Needed for the try to implement a non-shifting execution routine.
        start_time = datetime.datetime.now()
        while self.active:
            try:
                value = self._dyn(self.configuration.level)
                data = models.Data(fields={self.input_module_instance.configuration.key: value},
                                   measurement=self.configuration.measurement)
                self._call_links(data)
            except Exception as e:
                self.logger.error("Could not generate data: {0}".format(str(e)), exc_info=config.EXC_INFO)
            finally:
                # This should prevent shifting a little.
                required_time = datetime.datetime.now() - start_time
                if self.configuration.interval == 0:
                    break
                if self.configuration.interval > float(required_time.seconds):
                    time.sleep(self.configuration.interval - float(required_time.seconds))
                start_time = datetime.datetime.now()

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


class TagModule(AbstractTagModule):
    """
    A tag module.

    :param configuration: The configuration object of the module.
    :param input_module_instance: The instance of the parent input module if it exists.
    """
    version: int = __version__
    """The version of the module."""
    public: bool = __public__
    """Is this module public?"""
    description: str = __description__
    """A short description."""
    author: str = __author__
    """The author name."""
    email: str = __email__
    """The email address of the author."""
    deprecated: bool = __deprecated__
    """Is this module deprecated."""
    third_party_requirements: list[str] = __third_party_requirements__
    """Define your requirements here."""

    @dataclass
    class Configuration(models.TagModule):
        """
        The configuration model of the module.
        """
        input_module: str = field(
            metadata=dict(description="The id of the input module. The required input module has to be "
                                      "module_name: inputs.mqtt.client_1",
                          required=True),
            default=None)
        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, input_module_instance=None):
        super().__init__(configuration=configuration,
                         input_module_instance=input_module_instance)

    @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))))

    @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']
        }

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

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

    def _run(self) -> dict[str, Any]:
        """
        Method generates/requests the data for this module and returns it.

        :returns: A dict containing the generated key-value pairs.
        """
        return {self.input_module_instance.configuration.key: self._dyn(self.configuration.level)}