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:
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:
# $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:
"""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:
"""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:
"""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:
Reacting to events and property changes¶
As in C++, you can attach a callback to react to events and/or property changes. For example:
"""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:
"""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:
# $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:
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.
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:
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:
./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.