Implementing a Device Simulator

As mentioned before, FcfSim is built as a framework, therefore there are several ways to implement a Device Simulator. However, using the offered ServerInterface class and some decorators for the class and methods will fulfill most of the needs. Below we will show an implementation using ServerInterface; a more step-by-step way can be understood by code introspection or in the SDK chapter (FCFsim SDK Description).

First thing to know, everything needed to implement a device simulator is located in one api module:

from ifw.fcfsim import api as dvs

Note: here dvs stands for DeVice Simulation; choose any alias you prefer.

Implement Simulator for fcfsimServer

As an example we will make a simulator for a simple fictitious Switch device.

The fcfsimServer command-line tool relies on a configuration file that contains any desired key/value pairs to run the simulation properly. The only requirement for the fcfsimServer app is that the config file contains a “factory” keyword which points to a python function. This factory function will build a ServerInterface instance, e.g.:

factory: fcs1.simulators.switch:build_switch_interface

This will load the module fcs1.simulators.switch and look for build_switch_interface with the signature f(cfg: dict) -> IServerInterface, where the cfg is a dictionary containing the rest of the configuration file.

Example of a factory:

from ifw.fcfsim import api as dvs
import typing

def build_switch_interface(
    cfg_dict: dict[str,typing.Any]
) -> dvs.IServerInterface:
    ...

The rest of the tutorial here explains how to build the ServerInterface.

To make a proper simulator we need two mandatory things:

• populate the OPC-UA server with nodes and RPC methods.

• make the business logic interact with the server.

What fcfsim offers is:

• easy ways to build an OPC-UA namespace profile

• an automatic connection between business logic (attributes and methods) and the OPC-UA server.

Optionally:

• build extra threads to take care of some simulation behavior

• build and run an scxml4py state machine.

Below is an example where the OPC-UA profile is defined within python classes using typing annotations for attributes and decorators for (RPC) methods.

Simple Simulator

from ifw.fcfsim.api import opcua as ua
from typing import Annotated, Any

@ua.node.attrconnect
class SwitchStat:
    """Switch Status class"""
    state: Annotated[int, ua.Ua("nState", ua.Int16)] = 0
    state_text: Annotated[str, ua.Ua("sState")] = "off"
    counter: Annotated[int, ua.Ua("nCounter")] = 0

class SwitchEngine:
     stat: Annotated[SwitchStat, ua.Ua(access='r')]

     def __init__(self):
         self.stat = SwitchStat()

     @ua.uamethod("RPC_TurnOn")
     def turn_on(self) -> int:
         self.stat.state = 1
         self.stat.state_text = "on"
         return 0 # no error

     @ua.uamethod("RPC_TurnOff")
     def turn_off(self) -> int:
         self.stat.state = 0
         self.stat.state_text = "off"
         return 0 # no error

     def load_cfg(self, cfg: dict[str,Any]) -> None:
         """load configuration (e.g. from configng)"""
         if cfg.get("auto_on", False):
             self.stat.state = 1
             self.stat.state_text = "on"

def build_switch_interface(
     cfg_dict: dict[str,Any]
) -> ua.IServerInterface[SwitchEngine]:
     switch = SwitchEngine()
     switch.load_cfg(cfg_dict)
     profile = ua.extract_profile(SwitchEngine)

     return ua.ServerInterface(
        cfg_dict["prefix"], switch, profile
     )

That’s it: this is a simulator with two nodes and two RPC methods.

Some notes about the code:

• @ua.node.attrconnect is a class decorator. It allows executing callback methods when attributes are changed. In our case, the callbacks update the node in the OPC-UA server. These callbacks are automatically installed by the ua.ServerInterface.

• ua.extract_profile is a function that inspects the annotations of the class and decorated methods to extract the OPC-UA namespace profile (definition of nodes and RPCs) as well as the mapping between the “Engine” and OPC-UA nodes and methods. Also available is the ua.cache_profile class decorator, which extracts the profile and stores it in the class. If no profile is given to the ServerInterface, it will look for the class’s cached profile. Example:

@ua.cache_profile
class SwitchEngine:
   ... # see example above

def build_switch_interface(
    cfg_dict: dict[str,Any]
) -> ua.IServerInterface[SwitchEngine]:
    switch = SwitchEngine()
    switch.load_cfg(cfg_dict)

    return ua.ServerInterface(
       cfg_dict["prefix"], switch # profile cached in switch class
    )

Using standard config and interface

The class dvs.server.ServerInterface offers a classmethod from_cfg which builds a ServerInterface from a config dictionary based on the FcfSimStdCfg type defined in the configng schema schema/ifw/fcfsim/simlib/fcfsim.schema.yaml.

Following our example:

from dataclasses import dataclass, field

@dataclass
class SwitchServerInterface(dvs.server.ServerInterface):
    engine: SwitchEngine = field(default_factory=SwitchEngine)

And for the configng side:

!cfg.include schema/ifw/fcfsim/simlib/fcfsim.schema.yaml:

!cfg.typedef FcfSimSwitchCfg(FcfSimStdCfg):
  factory: !cfg.type:string "fcs1.simulators.switch:SwitchServerInterface.from_cfg"
  auto_on: !cfg.type:boolean false

Then the config file instance will look like :

!cfg.include schema/mic/fcs1/simulators/switch.schema.yaml

switch1: !cfg.type:FcfSimSwitchCfg
  device_name:      Switch1
  prefix:     MAIN.Switch1
  namespace:  4
  auto_on: false

These configs attributes above are part of the ServerInterface and understood by the .from_cfg classmethod.

More custom configuration parameters can be added, they will be passed to the load_cfg method of the engine.

Running the simulator

Using the config file :

> fcfsimServer --cfg config/mic/fcs1/switch/switch.cfg.yaml --port 4840

Or it can be included with other devices using the fcfsim manager, for instance:

# content of config/mic/fcs1.cfg.yaml

!cfg.include schema/ifw/fcfsim/mgr/simmgr.schema.yaml:

fcs1: !cfg.type:FcfSimMgrCfg
   devices:
      - name: lamp1
        cfgfile: 'config/ifw/fcfsim/example/lamp.cfg.yaml'
        overrides:
            - {name: prefix, value: MAIN.Lamp001}
      - name: switch1
        cfgfile: config/mic/fcs1/switch/switch.cfg.yaml
        overrides:
            - {name: prefix, value: MAIN.Switch001}

Then

> fcfsimServer --cfg config/mic/fcs1.cfg.yaml --port 4840

Adding a threaded task: a Runner

One may want to add a thread: a method running a task cyclically (as a PLC does) to handle simulated device behavior. This is easy to do, again with the help of a decorator.

Following the example above:

@ua.cache_profile
class SwitchEngine:
     ... # above code here

     @ua.core.task_method
     async def next(self) -> None:
         self.stat.counter += 1

     ...

The task method can have an initialiser called when the server starts. The frequency is set to the server frequency by default, but this can be changed:

@ua.cache_profile
class SwitchEngine:
     ...

     @ua.core.task_method(update_frequency=2.0) # 2.0 Hz
     async def next(self) -> None:
         self.stat.counter += 1

     @next.initialiser
     async def initialise(self) -> None:
         self.stat.counter = 0

     ...

The task methods are understood by the ServerInterface to install runners (threads) accordingly. The threads are started when the server starts, with all thread initialisers executed before any cycling tasks.

Note, the function does not have to be async, however it is good practice to have an async method here, for future-proofing.

For further, more complex behavior, or if the programmer does not want to use a decorator, one can still re-implement the add_to_server method of the Interface class and declare the Runner threads manually.

Add a State Machine

If one has an scxml state machine, this can be used and easily installed in the engine business logic.

The activity, action and listener business parts of the state machine can be simply declared on the engine with method decorators as well.

The following example assumes that the state machine defines a “TurnOn.Execute” action and a “TurningOn” activity.

The additional requirement asked by the standard ServerInterface is that the engine must have a method get_event_queue returning an event queue (dvs.sm.EventQueue) used to share events between the engine business logic and the state machine.

@ua.cache_profile
class SwitchEngine:
    def __init__(self):
        self.event_queue = dvs.sm.EventQueue()
        self.stat = SwitchStat()

    def get_event_queue(self) -> dvs.sm.IEventQueue:
        return self.event_queue

    ...

    @sm.event_listener_method
    def listen_event(self, event: dvs.sm.Event) -> None:
        # do something: log, update device data ...
        ...

    @sm.status_listener_method
    def listen_state(self,
      status: set[dvs.sm.State]
    ) -> None:
       # Listen state change, used to inject state machine new state inside
       # the device data
       ...

    @sm.action_method("TurnOn.Execute")
    def turn_on_action(
       self, handler: dvs.sm.IActionHandler, context: dvs.sm.Context
    ) -> None:
       ...

    @sm.activity_method("TurningOn")
    def turning_on_activity(self,  handler: sm.IActivityHandler) -> None:
       ...

Explanation:

• @sm.event_listener_method decorates a method as an event listener for the state machine. Most likely only one is defined. The method must have one single input argument which is a scxml4py Event.

• @sm.status_listener_method decorates a method as a status listener. When the state machine state changes, a listener can log or update things (e.g. stat.state, stat.substate) inside the engine data. This method receives a set of scxml4py State.

• @sm.action_method decorates the method as an action. The Action id is given to the decorator. The method receives a dvs.sm.IActionHandler (see below) and a scxml4py Context.

• @sm.activity_method decorates the method as an activity. The Activity id is given to the decorator. The method receives a dvs.sm.IActivityHandler.

• get_event_queue returns the event queue used to share events between the engine and the state machine.

The last step is to provide a scxml file containing the state machine inside the config file schema:

!cfg.include schema/ifw/fcfsim/simlib/fcfsim.schema.yaml:

!cfg.typedef FcfSimSwitchCfg(FcfSimStdCfg):
  factory: !cfg.type:string "fcs1.simulators.switch:SwitchServerInterface.from_cfg"
  auto_on: !cfg.type:boolean false
  state_machine_scxml: !cfg.type:string "config/mic/fcs1/simulator/switch/switch.scxml.xml"

Lastly, one could also want to align defaults in configng and the interface class defaults:

from typing import TextIO

@dataclass
class SwitchServerInterface(dvs.server.ServerInterface):
    engine: SwitchEngine = field(default_factory=SwitchEngine)
    state_machine_scxml: str | TextIO = "config/mic/fcs1/simulator/switch/switch.scxml.xml"

This allows running the device simulator properly without the need for a config file (e.g. in unit tests).

Subclassing existing simulator

As an example, let us add a “lrVoltage” to the standard lamp device. As usual there isn’t a single way to do it, here is one using class inheritance:

from dataclasses import dataclass, field
from ifw.fcfsim import api as dvs
from ifw.fcfsim.devices.lamp import sim as lamp


@dvs.opcua.cache_profile
@dvs.node.attrconnect # for the new voltage attribute
class MyLampEngine(lamp.LampSimEngine):
     voltage: Annotated[float, ua.Ua("stat.lrVoltage")] = 0.0

     @dvs.core.task_method
     async def next(self) -> None:
         await super().next() # Lamp has a next() async task method
         self.voltage = self.stat.intensity / 12.34

@dataclass
class MyLampInterface(lamp.LampSimInterface):
    engine: MyLampEngine = field(default_factory = MyLampEngine)

A new MyLamp config file should now include the new factory, e.g. factory: "my_module:MyLampInterface.from_cfg".

Composition

Again we can use an example for the description: let us imagine a special device (the equivalent of a TC3 Function Block) containing a Lamp and a Motor.

What is written below concerns only the composition part of the two devices as an illustration; most likely the special device will have additional business logic.

import functools
from dataclasses import dataclass, field
from ifw.fcfsim import api as dvs
from ifw.fcfsim.api import opcua as ua
from ifw.fcfsim.devices.motor import sim as motor
from ifw.fcfsim.devices.lamp import sim as lamp

@dvs.node.attrconnect
class ComboStat:
    """Lamp and motor combined Status"""
    state: int = 0
    substate: int = 0

@ua.cache_profile
class ComboSimEngine:
    motor: Annotated[motor.MotorSimEngine, ua.Ua("fbMotor")]
    lamp: Annotated[lamp.LampSimEngine, ua.Ua("fbLamp")]
    stat: Annotated[ComboStat, ua.Ua("stat")]

    def __init__(self):
        self.service = dvs.Service() # contains, log, event queue and signals
        self.motor = motor.MotorSimEngine()
        self.lamp =  lamp.LampSimEngine()
        self.stat = ComboStat()


    @dvs.core.task_method
    async def next(self) -> None:
        await self.motor.next()
        await self.lamp.next()
        self.stat.substate = min(self.motor.stat.substate, self.lamp.stat.substate)
        self.stat.state = min(self.motor.stat.state, self.lamp.stat.state)

    @next.initialiser
    async def initialise(self) -> None:
        await self.motor.initialise()
        await self.lamp.initialise()

    @ua.uamethod("RPC_Reset")
    def reset(self) -> Annotated[int,ua.Int16]:
        self.motor.reset()
        self.lamp.reset()
        return 0

    @ua.uamethod("RPC_Init")
    def init(self) -> Annotated[int,ua.Int16]:
        self.motor.init()
        self.lamp.init()
        return 0

    #etc ...

@dataclass
class ComboSimInterface(dvs.server.ServerInterface):
    engine: ComboSimEngine = field(default_factory=ComboSimEngine)

    def add_to_server(self, server: dvs.opcua.IServerService) -> None:
        # super will add nodes and rpcs, including motor's and lamp's because
        # they are part of the cached, annotated, profile
        super().add_to_server(server)

        # We still need to attach motor and lamp in order to have the
        # state machine running, but we will **excludes** 'opcua' and 'engine'
        server.attach_all(
          motor.MotorSimInterface(self.prefix+".fbMotor", self.engine.motor),
          lamp.LampSimInterface(self.prefix+".fbLamp", self.engine.lamp),
          excludes = {"opcua", "engine"}
        )

Notes:

• add_to_server this method populates OPC-UA server nodes and RPCs, and adds other potential runners (business thread, state machine, etc …).

• excludes Here, for the lamp and motor, we exclude the attachment of “opcua” because it is part of the Combo annotated namespace profile already. We also exclude “engine” (the thread inside motor and lamp) because they are executed within the Combo (by the .next() task method). However “sm” (state machine) is not excluded, so the lamp and motor state machine runners will be executed by the server.

Note

Another possible implementation could have been:

@ua.cache_profile
class ComboSimEngine:
    motor: motor.MotorSimEngine
    lamp: lamp.LampSimEngine
    stat: Annotated[ComboStat, ua.Ua("stat")]

... # see above

@dvs.core.task_method
async def next(self) -> None:
      self.stat.substate = min(self.motor.stat.substate, self.lamp.stat.substate)
      self.stat.state = min(self.motor.stat.state, self.lamp.stat.state)

@dataclass
class ComboSimInterface(dvs.server.ServerInterface):
    engine: ComboSimEngine = field(default_factory=ComboSimEngine)

    def add_to_server(self, server: dvs.opcua.IServerService) -> None:
        super().add_to_server(server)
        server.attach_all(
          motor.MotorSimInterface(self.prefix+".fbMotor", self.engine.motor),
          lamp.LampSimInterface(self.prefix+".fbLamp", self.engine.lamp),
        )

The differences are:

• motor and lamp namespace profiles are not added to the ComboEngine (no ua.Ua() annotation)

• the namespace profile is handled in the ServerInterface add_to_server method

• next() tasks of motor and lamp are handled in their own thread.

For another example of composition, one can look at a multi-axis device such as the ADC.