FCFsim SDK Description

Philosophy

Here is a description of the FcfSim architecture and components. This is not full documentation of the modules, classes and functions.

FcfSim is fully typed; all variables, attributes and arguments are typed, with a few exceptions of ‘Any’ type annotations.
It uses typing.Protocol to describe python object interfaces when this is suitable. This allows the user to alter FcfSim components as long as they respect their protocol, following the so-called Duck Typing philosophy.
It is made of many files, but the packages and sub-packages are assembled inside api modules, offering all public classes and functions.
Example :
from ifw.fcfsim.opcua import api as ua
from ifw.fcfsim import api as dvs

Note, a dedicated api module file is used instead of putting the api into the __init__ file. This is for unit test and integration purposes.

ifw.fcfsim.core.node

from ifw.fcfsim.core.node import api as node

The node sub-package offers classes and methods in order to access and connect data (stored somewhere in python object) with a clear interface (e.g. IDataNode).

This is useful for FcfSim to reproduce TwinCAT OPC-UA handling, where variables are automatically exposed and linked to the OPC-UA server.

However, as it can be used for other purposes, the node module is generic and has nothing to do with OPC-UA.

The main components are:

attrconnect() is a class decorator. It will modify the __setattr__ method of the class in order to be able to attach callbacks to object attribute changes.
Array This is a list-like object but with a fixed size, and item callback connection capabilities.
Map This is a dictionary with item callback connection capabilities.
data_node() Builds a IDataNode from an object and a string path. This object can be used as getter, setter and connector, with a total abstraction of where the data is.

In a nutshell :

from ifw.fcfsim.core.node import api as node
from typing import Any

@node.attrconnect
class StatData:
    x: float = 0.0
    y: float = 0.0

    def __init__(self):
       self.switches = node.Array(3, bool)

class Data:
   def __init__(self):
       self.stat = StatData()


some_database = {}

def set_db(key: str, value: Any) -> None:
   some_database[key] = value

data = Data()
node.data_node(data, "stat.x").connect(lambda value:set_db("X", value), group="db")
node.data_node(data, "stat.switches[1]").connect(lambda value:set_db("S_1", value), group="db")

data.stat.x = 10
data.stat.switches[1] = 100
assert some_database['X'] == 10
assert some_database['S_1'] == 100

# detach all subscription labelled "db"
node.subscription_group("db").disconnect()

ifw.fcfsim.core

The core api contains useful classes and functions which can be imported and used outside the scope of fcfsim.

from ifw.fcfsim.core import api as core

Exceptions

RpcError: Subclassed from RuntimeError. This is dedicated to Remote Procedure Calls; it holds an integer error code. It is intended to be caught by a server in order to reply correctly to a client with an error code.
StopActivity, ActivityError: Subclassed from RuntimeError. To be raised when a threaded activity is done (Stop) or failed (Error), e.g. a motor movement. SafeActivity is a context manager used to catch any activity stop or error.
StopTask: Subclassed from RuntimeError. To be used to stop a cycling task from within the task executor function.

The following are error providers; they are exception factories built from an error code. The built Exception will contain a proper error message.

RpcErrorProvider
ActivityErrorProvider

Example

from ifw.fcfsim.core import api as core
class MotorRpcError(core.CodeEnum):
    OK = 0, "OK"
    NOT_OP = -1, "Cannot control motor. Not in OP state."
    NOT_NOTOP_READY = -2, "Call failed. Not in NOTOP_READY."
    NOT_NOTOP_NOTREADY = -3, "Call failed. Not in NOTOP_NOTREADY/ERROR."
    # etc ...
motor_errors = core.RpcErrorProvider(MotorRpcError)

Is equivalent to:

from ifw.fcfsim.core import api as core
import enum
class MotorRpcError(enum.IntEnum):
    OK = 0
    NOT_OP = -1
    NOT_NOTOP_READY = -2
    NOT_NOTOP_NOTREADY = -3
    # etc ...
motor_errors = core.RpcErrorProvider(
    texts = {
      MotorRpcError.OK: "OK",
      MotorRpcError.NOT_OP: "Cannot control motor. Not in OP state.",
      MotorRpcError.NOT_NOTOP_READY: "Call failed. Not in NOTOP_READY.",
      MotorRpcError.NOT_NOTOP_NOTREADY: "Call failed. Not in NOTOP_NOTREADY/ERROR."
    }
)

Usage:

raise motor_errors.error(MotorRpcError.NOT_OP)

log

A set of log interfaces. fcfsim is using the python logging system, not the CII’s yet.

get_logger equivalent of python logging.getLogger
init_loggers() configure given logger(s)
print_existing_loggers()
LogParser A class to provide standard log option for command line arguments and initialise logs accordingly.

Runner & Task

A set of tools to run something continuously.

Please keep in mind that the examples below are illustrations taken outside of their context, so they may seem overkill or pointless.

Signals

Signals offers a simple interface to listen to incoming OS signals (SIGTERM, SIGINT) and propagate values to whatever subscribed to it. This is used mostly to shut down running threads properly.

A Signal object has the interface ISignals.

A AppSignals is by default handling SIGTERM and SIGINT (e.g. ctrl-c) signals to propagate an “exit” to subscribers, with the proper code.

import time
from ifw.fcfsim.core import api as core
signals = core.AppSignals()
alive = True

def receive_exit(code: int) -> None:
  global alive
  alive = False
  print("Receive Exit code", code)
signals.subscribe("exit", receive_exit)

counter = 0
while alive:
  print("hello", counter)
  if counter>=20:
    signals.trigger("exit", 0)
  counter += 1
  time.sleep(1.0)

The above program can be stopped with ctrl-c before an “exit” signal is triggered after 20 iterations.

ifw.fcfsim.core.api.ExitListener is a small helper to listen for the exit signal.

import time
from ifw.fcfsim.core import api as core
signals = core.AppSignals()
listener = core.ExitListener(signals)
counter = 0
while listener.is_alive():
  print("hello", counter)
  if counter>=20:
    signals.trigger("exit", 0)
  counter += 1
  time.sleep(1.0)
print("App terminated with exit code", listener.get_exit_code())

A global instance of an AppSignals is offered and available from the get_global_signals()

Also available is the SlaveSignals which will subscribe to a master Signals in order to propagate signals to its subscribers.

import time
from ifw.fcfsim.core import api as core

master = core.get_global_signals()
slave = core.SlaveSignals(master)

def receiver(code: int) -> None:
  print("Receive", code)

slave.subscribe("exit", receiver)
master.trigger("exit", 0)
# print Receive 0

Looper

A looper is defined by an ILooper. Its role is to run an async method cyclically. It is intended to be used within a Runner (see below). A generic looper is provided; it executes a given async method with a given period and will listen to the “exit” of a given Signals (global signals by default). Looper

import time
import asyncio
from ifw.fcfsim.core import api as core

signals = core.AppSignals()
counter = 0

async def task() -> None:
  global counter
  print("hello", counter)
  counter += 1
  if counter>20:
    signals.trigger("exit", 0)

asyncio.run( core.Looper(0.5, signals).run(task) )

ctrl-c can terminate the above program.

Runner

A runner is defined by IRunner, it just aims to run something (an async method) in two steps: initialise, then run. A generic runner using one (or two) function(s) as input is provided Runner

To group several runners into one, use RunnerGroup.

Typically, FcfSim uses runners to execute background tasks on top of the OPC-UA server, like the state machine execution or other user periodic tasks.

from ifw.fcfsim.core import api as core
import time
import asyncio

signals = core.AppSignals()

async def periodic_func():
  print(time.time())

runner = core.Runner(periodic_func, looper=core.Looper(0.1, signals=signals))

core.run_in_thread(runner)
time.sleep(2.0)
signals.exit(0)

Any class can contain some methods flagged as “task method” which can be collected into runners:

from ifw.fcfsim.core import api as core
import time
import asyncio

class Device:
  start_time = 0.0

  @core.task_method
  async def next(self):
    t = time.time() - self.start_time
    print(t)
    if t > 2.0:
      raise core.StopTask()

  @next.initialiser
  async def init(self):
    self.start_time = time.time()


device = Device()
runner = core.RunnerGroup(*core.collect_runners(device, core.Looper(0.1)))
core.run_in_thread(runner)

Input / output

io module

The io module offers simplified tools and an interface to load configng documents into dictionaries. A loader is defined as ifw.fcfsim.core.api.ILoader.

ifw.fcfsim.core.api.ConfigLoader is a loader to handle configng documents.

import io
from ifw.fcfsim.core import api as core

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

loader = core.io.ConfigLoader(check_document=True)
data = loader.load(config_file)

assert data["name"] == "Example"

By default the first element of the config file is loaded but this can be changed:

import io
from ifw.fcfsim.core import api as core

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

loader = core.io.ConfigLoader(prefix="example", check_document=True)
data = loader.load(config_file)

Prefix is also hierarchical, so it can be used to load a specific value, for instance:

import io
from ifw.fcfsim.core import api as core

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

timeout_loader = core.io.ConfigLoader(prefix="example.timeout")
assert timeout_loader.load(config_file) == 3000

A “wrapper” can be used to parse the raw data into something else. Note that the clipper below also needs import typing:

import io
from ifw.fcfsim.core import api as core

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

def clipper(maxi: int) -> typing.Callable[[int], int]:
  def clip(value):
    return min(value, maxi)
  return clip

timeout_loader = core.io.ConfigLoader(
    wrapper=clipper(1000), prefix="example.timeout"
)
assert timeout_loader.load(config_file) == 1000

import io
from ifw.fcfsim.core import api as core
from dataclasses import dataclass

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

@dataclass
class Data:
  name: str
  timeout: int = 1000

loader = core.io.ConfigLoader(wrapper=lambda cfg: Data(**cfg))
data = loader.load(config_file)
assert data.timeout == 3000

Config

Core provide a base ifw.fcfsim.core.api.Config object with protocol ifw.fcfsim.core.api.IConfig.

ifw.fcfsim.core.api.Config has factory classmethods in order to load the content of a user configuration (file or dictionary) into a clean configuration data class.

This is useful to handle a typed, static object as configuration instead of an untyped configng document instance or a python dictionary.

import io
from ifw.fcfsim.core import api as core
from dataclasses import dataclass

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")

@dataclass
class MyConfig(core.Config):
  name: str
  timeout: int = 1000

config = MyConfig.from_cfg_file(config_file)
assert config.timeout == 3000
assert config.name  == "Example"

Also things that cannot be checked in configng schema can be deserialized using class method.

import io
import urllib
import typing

from ifw.fcfsim.core import api as core
from dataclasses import dataclass

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    endpoint: "opc.tcp://localhost:4840"
""")

@dataclass
class MyConfig(core.Config):
  name: str
  url: urllib.parse.ParseResult

  @classmethod
  def deserialise_entry(
    cls, name: str, value: typing.Any
  ) -> typing.Iterable[tuple[str, typing.Any]]:
    match name:
      case "endpoint":
        yield "url", urllib.parse.urlparse(value)
      case _:
        # note super method is just:  `yield name, value` in core.Config
        yield from super().deserialise_entry(name, value)

config = MyConfig.from_cfg_file(config_file)
assert config.url.scheme == "opc.tcp"

In the above example we intercept the endpoint, parse it and change the attribute to “url”. yield is used instead of return in order to be able to set several attributes from one single key/value input.

By default the fields of the config object are all annotated class values. This can be changed with the get_class_fields classmethod, which should return a set of strings.

One can also load a config on an already instantiated config object.

import io
from ifw.fcfsim.core import api as core
from dataclasses import dataclass

# mimic a configng file
config_file = io.StringIO("""
  example:
    name: "Example"
    timeout: 3000
""")
loader = core.io.ConfigLoader(check_document=True)

class MyConfig(core.Config):
  name: str = "Undefined"
  timeout: int = 1000

config = MyConfig()
assert config.name == "Undefined"
config.load_cfg( loader.load(config_file) )
assert config.name  == "Example"

wait

This is a simple tool to wait for the result of one or several methods to be true within a given timeout.

from ifw.fcfsim.core import api as core
import time
import threading

class MyDevice:
  state: int = 0
  def __init__(self, init_time: float = 1.0):
    self.init_time = init_time

  def is_initialised(self) -> bool:
    return bool(self.state)

  def init(self) -> None:
    self.state = 0
    time.sleep(self.init_time)
    self.state = 1

device1 = MyDevice(1.0)
device2 = MyDevice(3.0)

threading.Thread(target=device1.init).start()
threading.Thread(target=device2.init).start()

t = time.time()
core.wait(device1.is_initialised, device2.is_initialised, timeout=60)
print( "Waited", time.time()-t, "seconds")

Above should wait approx 3 seconds.

The default operand is “all”: all shall be true to return. This can be changed

threading.Thread(target=device1.init).start()
threading.Thread(target=device2.init).start()

t = time.time()
core.wait(device1.is_initialised, device2.is_initialised, operand="any")
print( "Waited", time.time()-t, "seconds")

Above should return after approx one second.