Create a custom controller

Copy the included template to create a custom controller.

netdef/Controllers/NewControllerTemplate.py:

import datetime
import logging

from netdef.Controllers import BaseController, Controllers
from netdef.Sources.BaseSource import StatusCode

# import my supported sources
from netdef.Sources.NewSourceTemplate import NewSourceTemplate


@Controllers.register("NewControllerTemplate")
class NewControllerTemplate(BaseController.BaseController):
    def __init__(self, name, shared):
        super().__init__(name, shared)
        self.logger = logging.getLogger(self.name)
        self.logger.info("init")
        self.one_config_entry = self.shared.config.config(
            self.name, "one_config_entry", "default_value"
        )

    def run(self):
        "Main loop. Will exit when receiving interrupt signal"
        self.logger.info("Running")
        while not self.has_interrupt():
            self.loop_incoming()  # dispatch handle_* functions
            self.loop_outgoing()  # dispatch poll_* functions
        self.logger.info("Stopped")

    def handle_readall(self, incoming):
        raise NotImplementedError

    def handle_add_source(self, incoming):
        self.logger.debug("'Add source' event for %s", incoming.key)
        self.add_source(incoming.key, incoming)

    def handle_read_source(self, incoming):
        raise NotImplementedError

    def handle_write_source(self, incoming, value, source_time):
        self.logger.debug(
            "'Write source' event to %s. value: %s at: %s",
            incoming.key,
            value,
            source_time,
        )

    def poll_outgoing_item(self, item):
        if isinstance(item, NewSourceTemplate):  # My
            # TODO: get new value somehow
            address = item.unpack_address()
            new_val = get_the_new_value_somehow(address)
            stime = datetime.datetime.utcnow()
            status_ok = True  # Why not
            cmp_oldew = False  # compare old and new value?

            if self.update_source_instance_value(
                item, new_val, stime, status_ok, cmp_oldew
            ):
                self.send_outgoing(item)

Paste it into your application with a new name:

first_app/Controllers/CmdController.py:

import logging
import datetime
from netdef.Controllers import BaseController, Controllers
from netdef.Sources.BaseSource import StatusCode

# import my supported sources
from netdef.Sources.NewSourceTemplate import NewSourceTemplate

@Controllers.register("CmdController")
class CmdController(BaseController.BaseController):
    def __init__(self, name, shared):
        super().__init__(name, shared)

...

Line 9 and 10 is changed to the same name as the file. Line 7 have to be replaced at a later time to a custom or built-in source

To activate the controller we have to merge following config to default.conf:

[controllers]
CmdController = 1

[CmdController]

Result after merge:

[controllers]
CrontabController = 1
OPCUAServerController = 1
CmdController = 1

[CrontabController]

[OPCUAServerController]

[CmdController]

Create a custom source

Copy the included template to create a custom source for your controller.

netdef/Sources/NewSourceTemplate.py:

from netdef.Interfaces.DefaultInterface import DefaultInterface
from netdef.Sources import BaseSource, Sources


@Sources.register("NewSourceTemplate")
class NewSourceTemplate(BaseSource.BaseSource):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.interface = DefaultInterface

    # TODO: add a address for your new controller
    def unpack_address(self):
        return self.key

Paste it into your application with a new name:

first_app/Sources/CmdSource.py:

from netdef.Sources import BaseSource, Sources
from netdef.Interfaces.DefaultInterface import DefaultInterface

@Sources.register("CmdSource")
class CmdSource(BaseSource.BaseSource):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.interface = DefaultInterface

    # TODO: add a address for your new controller
    def unpack_address(self):
        return self.key

Line 4 and 5 is changed to the same name as the file.

Change line 7 in your custom controller:

first_app/Controllers/CmdController.py:

import logging
import datetime
from netdef.Controllers import BaseController, Controllers
from netdef.Sources.BaseSource import StatusCode

# import my new source
from ..Sources.CmdSource import CmdSource
...

To activate the source we have to merge following config to default.conf:

[sources]
CmdSource = 1

[CmdSource]
controller = CmdController

Result:

[controllers]
CrontabController = 1
OPCUAServerController = 1
CmdController = 1

[sources]
CrontabSource = 1
VariantSource = 1
CmdSource = 1

[CrontabSource]
controller = CrontabController

[VariantSource]
controller = OPCUAServerController

[CmdSource]
controller = CmdController

[CrontabController]

[OPCUAServerController]

[CmdController]

Create a custom rule

Copy the included template to create a custom rule.

netdef/Rules/NewRuleTemplate.py:

import logging
import pathlib

from netdef.Rules import BaseRule, Rules
from netdef.Rules.utils import import_file

SourceInfo = BaseRule.SourceInfo
ExpressionInfo = BaseRule.ExpressionInfo


@Rules.register("NewTemplateRule")
class NewTemplateRule(BaseRule.BaseRule):
    def __init__(self, name, shared):
        super().__init__(name, shared)
        self.logger = logging.getLogger(name)
        self.logger.info("init")

        config = self.shared.config.config
        self.proj_path = pathlib.Path(config("proj", "path", ".")).absolute()

    def setup(self):
        self.logger.info("Running setup")

        # example:
        self.setup_example()

        # sub rule example:
        for name, active in self.shared.config.get_dict(self.name).items():
            if int(active):
                self.setup_sub_rule(name)
        self.logger.info("Done parsing")

    def setup_sub_rule(self, name):
        raise NotImplementedError

    def setup_example(self):
        # example_expression_module = self.import_py_file("config/example_expression.py")

        # config/example_expresion.py:
        # def expression(internal):
        #     if internal.new or internal.update:
        #         print(internal)

        self.add_new_parser("InternalSource")

        source_count = self.add_new_expression(
            ExpressionInfo(
                example_expression_module,
                [SourceInfo("InternalSource", "intern_test_1")],
            )
        )
        self.update_statistics(self.name + ".example", 0, 1, source_count)

    def import_py_file(self, rel_file):
        full_file = pathlib.Path(self.proj_path).joinpath(rel_file)
        nice_name = full_file.name
        return import_file(str(full_file), self.name, nice_name)

    def run(self):
        self.logger.info("Running")
        while not self.has_interrupt():
            self.loop_incoming()  #  dispatch handle_* functions
        self.logger.info("Stopped")

    def handle_run_expression(self, incoming, value, source_time, status_code):
        expressions = self.get_expressions(incoming)
        self.logger.debug(
            "Received %s. Found expressions %s", incoming.key, len(expressions)
        )
        if expressions:
            self.send_expressions_to_engine(
                incoming, expressions, value, source_time, status_code
            )

Paste it into your application with a new name:

first_app/Rules/FirstAppRule.py:

import logging
import pathlib
from .utils import import_file
from . import BaseRule, Rules

SourceInfo = BaseRule.SourceInfo
ExpressionInfo = BaseRule.ExpressionInfo

@Rules.register("FirstAppRule")
class FirstAppRule(BaseRule.BaseRule):
    def __init__(self, name, shared):
        super().__init__(name, shared)
        self.logger = logging.getLogger(name)
        self.logger.info("init")

Line 9 and 10 is changed to the same name as the file.

To activate the rule we have to merge following config to default.conf:

[rules]
FirstAppRule = 1

[FirstAppRule]

Result:

[rules]
FirstAppRule = 1

[FirstAppRule]

[controllers]
CrontabController = 1
OPCUAServerController = 1
CmdController = 1

[sources]
CrontabSource = 1
VariantSource = 1
CmdSource = 1

[CrontabSource]
controller = CrontabController

[VariantSource]
controller = OPCUAServerController

[CmdSource]
controller = CmdController

[CrontabController]

[OPCUAServerController]

[CmdController]