Output 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
# Internal imports.
import config
from modules.base.outputs.base import AbstractOutputModule, models
from models.validations import OneOf
class OutputModule(AbstractOutputModule):
"""
An output 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."""
can_be_buffer: bool = False
"""If True, the child has to implement 'store_buffer_data' and 'get_buffer_data'."""
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.OutputModule):
"""
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)
@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']
}
def _run(self, data: models.Data):
"""
Method for executing the module.
:param data: The data object.
"""
try:
self.logger.info(self._dyn(self.configuration.level))
# TODO: You custom logic here...
except Exception as e:
self.logger.error("Something went wrong while trying to process data: {0}"
.format(str(e)), exc_info=config.EXC_INFO)
self._buffer(data=data, invalid=False)
Previous
Buffering
Buffering
Next
Introduction
Introduction