Skip to content

Screenshot

The Python component

Sen allows you to use Python by embedding an interpreter and giving you native access to all objects, sessions and buses. In particular:

  • Sen objects will be represented as regular Python objects. You can use properties, react to events and call methods.
  • You can create queries to look for objects.
  • You can import your own Sen C++ packages and directly create, manipulate and publish your own objects.
  • You can fetch information of the execution context (application name, current time, etc.)
  • You can install callbacks for your Python code to be called on certain events or property changes.

One of the key reasons for having Python is to leverage on all its existing libraries and scripting power to let you implement custom, complex, or highly customizable logic with ease. Examples of this might be tests ( at unit, integration, behavioral or sub-system level), orchestration logic, visualization, analysis, ML, etc.

Sen will try to find the following functions in your Python module:

Name Description
run() Called once at the start of the component execution.
update() Called on every iteration. The execution rate is defined as freqHz in the config file.
stop() Called once at the end of the execution.

The sen.api attribute is a Python object that allows you to interact with Sen. It defines the following members:

Name Type Description
time timedelta The current time.
defaultTimeout timedelta Default timeout used in waitUntil(), if none given.
appName string The name of the application where the script is being executed.
config dictionary Configuration passed to the component to parametrize your module.
getBus(name) sen.Bus Returns an object representing a bus. Use the "session.bus" format.
open(query) sen.ObjectList Returns a list containing the objects matching the query.
make(type, name) sen.Object Creates a new object. Constructor args go last and are keyed.
waitUntil(cond, time) Boolean Holds the execution until the condition is met (timeout is opt.).
requestKernelStop() None Asks the kernel to shut down.

The sen.Bus class has the following members:

Name Description
add(object) Registers an object. Does nothing if already registered.
remove(object) Unregisters an object. Does nothing if not present.

The sen.ObjectList is a regular Python sequence with the following extra members:

Name Description
onAdded(callback) Registers a function to be called when a new object gets added.
onRemoved(callback) Registers a function to be called when an object gets removed.
waitUntilEmpty(timeout) Holds the execution until the list is empty.
waitUntilNotEmpty(timeout) Holds the execution until the list is not empty.
waitUntilSizeIs(count, timeout) Holds the execution until the list size is equal to count.

The three waitUntil members take the same optional timeout as sen.api.waitUntil(), and return False when it elapses.

The sen.Object class contains all the properties and methods of the corresponding class. In addition, you can register callbacks to react to events and property changes:

object.on<EventName>(callback)
object.on<PropertyName>Changed(callback)

Independently of their class, all sen.Object instances have the following members:

Name Type Description
name string The name of this instance, as provided by its owner.
localName string The name of this instance, as seen by this component.
id integral Unique identifier.
lastCommitTime timedelta Time point of the last commit made on this object.
className string The name of the class.

We will now go over a set of simple examples to illustrate the API.

Examples

You need to define your Python module, which consists of a main .py file. Then you load the py component as follows:

1_python_hello.yaml
# $schema: ../base/schema.json
include:
  - ../base/term.yaml

load:
  - name: py
    group: 3
    freqHz: 1
    module: hello_python

This will instantiate the interpreter in a component that will run your module. module names the Python module, so it carries no .py extension. The include pulls in the term, which is how the shipped examples avoid repeating it in every configuration.

Example module:

hello_python.py
"""Example module that demonstrates how to use the python component."""

import sen


# this is executed only once (at the start of the component execution)
def run():
    """Sen run: to setup the initial component state."""
    print("Python: run")
    print(f"Python: the config is: {sen.api.config}")
    print(f"Python: the app name is: {sen.api.appName}")


# this is executed every cycle
def update():
    """Sen update: triggers test execution."""
    print(f"Python: update (current time: {sen.api.time})")


# this is executed only once (at the end of the component execution)
def stop():
    """Sen stop: trigger that the execution stops."""
    print("Python: stop called")

NOTE: Remember to add the folder where your scripts are located to the PYTHONPATH environment variable.

This would print:

Python: run
Python: the config is: ...
Python: the app name is: ...
Python: update (current time: ...)
Python: update (current time: ...)
...
Python: stop called

Inspecting objects

Use the sen.api.open(query) to get access to objects. This returns a list that is automatically updated. You can also define callbacks to react on objects being added or removed.

The list can also be iterated as a normal Python sequence.

For example:

inspecting_objects.py
"""Example module that demonstrates how to inspect objects."""

import sen

# to store the objects that we want to inspect
list = None


def run():
    """Sen run: to setup the initial component state."""
    global list  # refer to the global variable defined above  # noqa: PLW0603

    list = sen.api.open("SELECT * FROM local.kernel")  # open it

    # register some callbacks to show changes in the list
    list.onAdded(lambda obj: print(f"Python: object added {obj}"))
    list.onRemoved(lambda obj: print(f"Python: object removed {obj}"))


def update():
    """Sen update: triggers test execution."""
    print(f"Python: printing the list at: {sen.api.time})")
    print(list)

    print("Python: iterating over the objects in the list:")
    for obj in list:
        print(f"Python: * object {obj.name}")
        print(f"Python:   - class: {obj.className}")
        print(f"Python:   - id:    {obj.id}")
        print(f"Python:   - time:  {obj.lastCommitTime}")

All objects define the name, className, id and lastCommitTime as read-only properties. The next example shows a more complete representation of accessing object members.

Interacting with objects

You can call C++ methods on objects from Python. For example:

interacting_with_objects.py
"""Example module that demonstrates how to interact with objects."""

import sen

# to store the object
obj = None


def run():
    """Sen run: to setup the initial component state."""
    global obj  # refer to the global variable defined above  # noqa: PLW0603
    obj = sen.api.open('SELECT * FROM local.shell WHERE name = "shell_impl"')


def update():
    """Sen update: triggers test execution."""
    print("Python: update")

    # if the object is present, do something with it
    if len(obj) != 0:
        print("Python: interacting with the object")
        obj[0].info("i16")  # print the info of the i16 type
        obj[0].ls()  # list the current objects in the terminal
        obj[0].history()  # show the current history

        print("Python: asking the process to shut down")
        obj[0].shutdown()  # trigger the process shutdown

If you need to process the return value of a method call (which is asynchronous), you can pass a callback. For example:

obj.addNumbers(2, 2, lambda result: print(f"the result is {result}"))

Reacting to events and property changes

As in C++, you can attach a callback to react to events and/or property changes. For example:

reacting_to_events_and_changes.py
"""Example module that demonstrates how to react on events and changes (e.g., object added)."""

import sen

# to store the objects
teachers = None


def teacherDetected(teacher) -> None:
    """Prints information about the detected teacher."""
    teacher.onStressLevelPeaked(lambda args: print(f"Python: {teacher.name} stress level peaking to {args}"))
    teacher.onStatusChanged(lambda: print(f"Python: {teacher.name} status changed to {teacher.status}"))


def run() -> None:
    """Sen run: to setup the initial component state."""
    global teachers  # refer to the global variable defined above  # noqa: PLW0603

    print("Python: run")

    # select the object and install some callbacks
    teachers = sen.api.open("SELECT school.Teacher FROM school.primary")
    teachers.onAdded(teacherDetected)

Creating and publishing objects

You can import your Sen packages and instantiate objects from Python. For example:

creating_objects.py
"""Example module that demonstrates how to create objects."""

import sen

myObject = testBus = None


def run():
    """Sen run: to setup the initial component state."""
    global myObject, testBus  # refer to the globals defined above  # noqa: PLW0603

    type = {
        "entityKind": 1,
        "domain": 2,
        "countryCode": 198,
        "category": 1,
        "subcategory": 3,
        "specific": 0,
        "extra": 0,
    }

    id = {"entityNumber": 1, "federateIdentifier": {"siteID": 1, "applicationID": 1}}

    print("Python: creating and publishing the object")
    myObject = sen.api.make(
        "aircrafts.DummyAircraft", "myAircraft", entityType=type, alternateEntityType=type, entityIdentifier=id
    )
    testBus = sen.api.getBus("my.tutorial")
    testBus.add(myObject)

    # setting the speed property to 150
    myObject.speed = {"north": 150, "east": 0, "down": 0}


def update():
    """Sen update: triggers test execution."""
    print(myObject)


def stop():
    """Sen stop: trigger that the execution stops."""
    global testBus, myObject  # refer to the globals defined above  # noqa: PLW0603

    print("Python: deleting the object")
    testBus.remove(myObject)
    myObject = None
    testBus = None

In this case we had to import the my_package Sen package using your configuration file. It looked like this:

3_python_create_objects.yaml
# $schema: ../base/schema.json
include:
  - ../base/term.yaml

load:
  - name: py
    group: 3
    freqHz: 1
    module: creating_objects
    imports: [aircrafts]

Using enumerations

You can use native enumerations in Python, as Sen automagically defines them for the Sen packages that you import. To do so, you need to use the import sen.<package> directive in Python. For example, say that we have a package called my_package that defines an enumeration type called MyEnum with the corresponding value:

import sen.my_package


def run():
    print(f"Python: enum 'sen.my_package.MyEnum.first' has the value '{sen.my_package.MyEnum.first}'")

This would print:

Python: enum 'sen.my_package.MyEnum.first' has the value 'first'

The value is the enumerator's name, not its ordinal. Sen converts that name back when the value crosses into a Sen object, so passing an integer in its place fails rather than being interpreted.

Making your script fully sequential

During testing, you might be in the need of performing fully sequential calls.

If you need to ensure that the method call is completed. You can set the syncCalls flag to True and treat it as a regular function call.

sen.api.syncCalls = True

result = obj.addNumbers(2, 2)
print(f"the result is {result}")

If you need to wait until a certain condition is achieved, use the sen.api.waitUntil(condition, timeout = 0) function. This function will hold the script execution until the condition you specify (as a function) is fulfilled.

The timeout is a duration that, when elapsed, will fail the waitUntil call, making it return False.

Testing with python

You can use python for testing your sen objects in a flexible manner. Using pytest framework, you can manipulate sen objects in order to verify complex behaviors, making this a powerful tool for automated integration testing.

The use of pytest-bdd allows the users to automate project requirements testing in a flexible way using the Gherkin language.

Using Python from your package or component

The Python component will create an "Interpreter" object and publish it in a given bus if you define the "bus" configuration option.

It provides the following interface:

python_interpreter.stl
package sen.components.py;

import "configuration.stl"

// represents the embedded Python interpreter
class PythonInterpreter
{
  // evaluates a Python expression
  fn eval(expr: string) -> string;

  // executes a Python code block
  fn exec(code: string);

  // executes a Python file
  fn execFile(file: string);
}

For example, you can now call Python from the term:

test.yaml
load:
  - name: term
    open: [local.py]
  - name: py
    group: 3
    freqHz: 30
    bus: local.py
./sen run test.yaml

  Sen v0.0.1  GNU-11.4.0 [debug]
  ▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬▬
  Chance favours the prepared mind  - Louis Pasteur
  Type 'help' for a list of commands.

✓ local.py.interpreter.exec "c = 2"
✓ local.py.interpreter.eval "c"
  "2"
✓ local.py.interpreter.exec "c = c + 2"
✓ local.py.interpreter.eval "c"
  "4"
✓ local.py.interpreter.exec "import my_module"
✓ local.py.interpreter.eval "my_module.add(2,2)"
  "4"

With this approach, you are also enabled to call the interpreter if you want to delegate some functionality into Python. Just find the interpreter in the selected bus, compose your code, evaluate your expressions and interpret the result.