Safecor Package

class safecor.Api(*args, **kwargs)

Bases: object

This class allows a third-party program to send commands or receive notifications without having to use the socket directly.

The API provides a simple set of functions for sending commands and receiving notifications. However, it does not handle the formatting of commands, which is managed by the RequestFactory, ResponseFactory, NotificationFactory or ApiHelper classes.

The API can only be used within a user Domain (DomU).

To use the API, simply instantiate the Api class and open the socket by calling the start() function. Then, the other functions allow sending and receiving messages and notifications.

See also

ApiHelper - The API helper class

Singleton

This class is a singleton because it keeps information about the context. There is only one instance in an application.

Example of usage :

Api().add_ready_callback(self.__on_ready)
Api().start()

Asynchronous

All commands operate asynchronously. The result of executing a command will only be communicated through a notification or a response. Therefore, a callback function must be provided to the API in order to receive responses. The internal topics provided in the Topics use the suffix /request and /response to differenciate the query and the answer. When an answer is expected on a topic, it will always be suffixed with /response.

The callbacks are:

• Api.add_message_callback() - A message has been received

• Api.add_ready_callback() - The API is ready

• Api.add_restart_callback() - The restart has been accepted or refused by the core

• Api.add_shutdown_callback() - The shutdown has been accepted or refused by the core

• Api.add_subscription_callback() - The broker acknowledged a subscription

Subscription

The messages and notifications can be received only when the following conditions are met:

• The topic has been subscribed (see Api.subscribe())

• A message callback function has been provided (see Api.add_message_callback())

How to use the API

The sequence to correctly handle the connection and subscriptions on a broker is the following:

1. start

2. on ready

   • subscribe

3. on subscribed

   • continue the app

Here is an example:

from safecor import MqttFactory, Api

subscriptions = [ "my-topic-1", "my-topic-2" ]

def start():
    Api().start(domain_identifier="my-domain")
    Api().add_ready_callback(on_connected)

def on_connected():
    Api().add_subscription_callback(on_subscribed)
    Api().add_message_callback(self.on_message_received)
    success, mid = Api().subscribe(<topic name>)
    if success:
        subscriptions.append(mid)

def on_subscribed(mid):
    if mid in subscriptions:
        <continue app starting>

if __name__ == "__main__":
    start()

add_message_callback(callback_fn)

Adds a callback for receiving messages

The callback will receive the following arguments:

• topic:str The topic of the message

• payload:dict The payload as a JSON object

Parameters:

callback_fn (Callable) – A function which will be called by the API when a message arrives.

add_ready_callback(callback_fn)

Adds a callback to notify when the API is connected to the MQTT broker and ready.

The callback won’t receive any argument

Parameters:

callback_fn (Callable) – A function which will be called by the API when it is connected to the MQTT broker.

add_restart_callback(callback_fn)

Adds a callback to notify when the system is restarting a component.

Parameters:

callback_fn (Callable) – A function which will be called by the API when the system is restarting a component.

add_shutdown_callback(callback_fn)

Adds a callback to notify when the system has started shutting down.

Parameters:

callback_fn (Callable) – A function which will be called by the API when the system is shutting down.

add_subscription_callback(callback_fn)

Adds a callback to notify when a subscription has been acknowledged by the broker.

The callback will receive the following arguments:

• topic:str The topic subscribed

Parameters:

callback_fn (Callable) – A function which will be called by the API when the subscription has been acknowledged.

archive_extensions_handled()

Returns the list of file extensions handled by the mount function

clear_sys_usb_queues()

Queries sys-usb to clear its queues.

copy_file(source_disk: str, filepath: str, destination_disk: str)

Copies a file from a disk to another.

When the file is created the Topic.NEW_FILE notification is sent.

Parameters:

• source_disk (str) – The source disk.

• filepath (str) – The full path of the file to be copied.

• destination_disk (str) – The destination disk.

create_file(filepath: str, disk: str, contents: bytes, binary=False)

Creates a new file on a disk by providing raw data.

The directories will be created if necessary before creating the file on the disk.

Binary/string data

Both binary and string data can be provided but when providing binary data, they will be encoded in Base64 before being sent to the broker because only text messages can be exchanged in MQTT. If binary data are provided, the argument binary must be set to True.

When the file has been created, the notification Topic.NEW_FILE is sent.

Parameters:

• filepath (str) – The full path of the file.

• disk (str) – The disk on which the file will be created.

• contents (bytes) – The data to write in the file.

• binary (bool, optional) – If True, it means that the data provided are binary.

critical(message: str, module: str = '')

Publish a critical message.

Parameters:

• message (str) – The message to publish

• module (str, optional) – The name of the component involved.

debug(message: str, module: str = '')

Publish a debug message.

Parameters:

• message (str) – The message to publish

• module (str, optional) – The name of the component involved.

delete_file(filepath: str, disk: str)

Removes a file from a disk (including the storage).

When the storage is involved, please use the Constantes.REPOSITORY.

In case of an error, the Topic.ERROR notification is sent.

discover_components() → None

Asks all the components of the system to notify their information and state.

The response is provided on the topic Topic.DISCOVER_COMPONENTS

error(message: str, module: str = '')

Publish an error message.

Parameters:

• message (str) – The message to publish

• module (str, optional) – The name of the component involved.

get_current_language()

Asks for the current system language

get_default_language()

Asks for the default language of the system

get_disks_list()

Asks the system to provide the list of external disks connected.

The response is provided on the topic Topic.LIST_DISKS.

get_file_fingerprint(filepath: str, disk: str)

Compute the fingerprint of a file.

The response is provided on the topic Topic.FILE_FINGERPRINT.

Parameters:

• filepath (str) – The full path of the file.

• disk (str) – The disk on which the file is located.

get_files_list(disk: str, recursive: bool = False, from_dir: str = '')

Asks the system to provide the list of files of an external disk or the repository.

The response is provided on the topic Topics.LIST_FILES

By default the listing is done from the root of the disk. This can be changed by setting the argument from_dir to a specifi directory of the disk.

By default the listing is not recursive, which means that the system will only look in the directory provided and not in the subdirectories if any. This can be changed by setting the argument from_dir.

Performance consideration

If the disk has a large number of files and nestes folders, the recursive listing will take a long time and return a single response with a large JSON string. This should be avoided for performance reasons.

Always prefer a pure asynchronous method of listing all the files and directories by listing the files only not recursively and send new queries for the remaining directories.

Thus, the system will always be very efficient and responsive instead of waiting for a big result to be provided.

Please also consider using a lazy-loading strategy which minimizes the load on the system by avoiding the listing of files which are not necessary and delays the listing when it is necessary.

The response is provided on the topic Topic.LIST_FILES.

Parameters:

• disk (str) – The name of the disk.

• recursive (bool) – Looks for files recursively if True.

• from_dir (str, optional) – The starting directory for the listing.

get_languages_list()

Asks for the languages defined in the product

get_mqtt_client()

Returns the current MQTT client instance.

info(message: str, module: str = '')

Publish a info message.

Parameters:

• message (str) – The message to publish

• module (str, optional) – The name of the component involved.

mount_file(disk: str, filepath: str)

Tries to mount a virtual disk (ISO, VMDK, etc).

If the mount succeeded, a disk state notification will be sent.

Parameters:

• disk (str) – The disk on which the file is located.

• filepath (str) – The path of the file to mount.

notify_disk_added(disk)

Sends a notification to inform the other components an external storage has been added to the system.

This notification is for internal use only

notify_disk_removed(disk)

Sends a notification to inform the other components an external storage has been removed from the system.

This notification is for internal use only

notify_gui_ready() → None

Sends a notification to inform the other components the Graphical User Interface is ready.

This notification is mandatory to make the splash screen disappear when the GUI is ready, otherwise the GUI of the system would remain behind the splash screen

ping(target_domain: str, data: str = '')

Sends a ping request to a specific user Domain

The ping request is peer-to-peer, so the topic format is a little different from other topics. The name of the target is in the topic so it can be routed by the broker.

The request should be constructed with RequestFactory.create_request_ping().

Parameters:

• target_domain (str) – The name of the targetted Domain.

• data (str, optional) – Data to be sent to the target.

publish(topic: str, payload: dict)

Publish a message on the MQTT broker.

The message is formatted in JSON and can provided directly as a dict.

Parameters:

• topic (str) – The topic to publish to.

• payload (dict) – The message formatted in JSON.

publish_components(components: list) → None

Publishes a list of components and their state.

The payload should be created using the fonction ResponseFactory.create_response_component_state().

The format of a component state structure is the following:

{
    "id": "unique identifier of the component,
    "domain_name": "automatically provided by Safecor using the hostname",
    "label": "The name of the component",
    "type": "a key identifying the type or category of the component",
    "state": "The current state of the component. See EtatComposant",
    "version": "The software version of the component",
    "description": "A paragraph describing the component"
}

Parameters:

components (list) – a List of components in the form of dictionaries.

read_file(disk: str, filepath: str)

Copies a file into the repository.

When the file is created the Topic.NEW_FILE notification is sent.

Parameters:

• disk (str) – The source disk.

• filepath (str) – The full path of the file to be copied.

request_energy_state() → None

Asks the system the state of the power supply.

The response is provided on the topic Topic.ENERGY_STATE

request_system_info() → None

Asks the system to provide technical information.

Here is an example of the result provided:

{
    "core": {
        "version": "1.1",
        "debug_on": false
    },
    "system": {
        "os": {
            "name": "Linux",
            "release": "6.12.20-0-lts",
            "version": "#1-Alpine SMP PREEMPT_DYNAMIC 2025-03-24 08:09:11"
        },
        "machine": {
            "arch": "x86_64",
            "processor": "",
            "platform": "Linux-6.12.20-0-lts-x86_64-with",
            "cpu": {
                "count": 12,
                "freq_current": 1689.5970000000004,
                "freq_min": 0.0,
                "freq_max": 0.0,
                "percent": 0.0
            },
            "memory": {
                "total": 405987328,
                "available": 103829504,
                "percent": 74.4,
                "used": 252207104,
                "free": 20271104
            },
            "load": {
                "1": 0.4306640625,
                "5": 0.16650390625,
                "15": 0.0595703125
            }
        },
        "boot_time": 1747205332.0,
        "uuid": "11ec0800-4fb9-11ef-bd38-ad993f2e7700",
        "storage": {
            "total": 12345678,
            "used": 0,
            "free": 12345678,
            "files": 0
        }
    }
}

The response is provided on the topic Topic.SYSTEM_INFO

restart_domain(domain_name: str)

Makes a Domain restart.

The response is sent on the topic Topics.RESTART_DOMAIN with the following payload:

{
    "domain_name": "The Domain name",
    "state": "accepted or refused",
    "reason": "The reason why the restart was refused"
}

Parameters:

domain_name (str) – The name of the Domain to restart.

set_current_language(lang: str)

Sets the current system language

shutdown()

Makes the system shutdown.

When a shutdown is asked by a Domain, typically the GUI, a notification is sent to all the components to inform them of the shutdown so they can get prepared.

The response is sent on the topic Topics.SHUTDOWN with the following payload:

{
    "state": "accepted or refused"
    "reason": "the reason why the shutdown was refused"
}

start(mqtt_client: MqttClient | None, domain_identifier: str = 'undefined', recording: bool = False, logfile: str = '/var/log/safecor/safecor.log')

Starts the API by connecting to the MQTT broker and opening a log file if asked.

When no mqtt_client argument is provided the API tries to create a DomU client with the identifier provided.

Parameters:

• mqtt_client (MqttClient) – The instance of the MqttClient class which handles the connexion to the MQTT broker.

• domain_identifier (str) – The unique identifier (label) for the domain connection.

• recording (bool) – If true, the events will be recorded in a log file.

• logfile (str) – If recording is true, the location of the log file can provided with this parameter.

stop()

Stops the Api by disconnecting from the MQTT broker.

subscribe(topic: str) → tuple[bool, int | None] | None

Subscribes to a topic on the broker.

If the function subscribed to is a Safecor function, the Topics class provides the reference of the topic names.

Parameters:

topic (str) – The topic to subscribe to.

Returns:

A tuple containing the result as bool and the ID of the subscription

Return type:

tuple[bool, int | None]

unmount(disk: str)

Unmounts a disk that has been previously mounted

If the unmount succeeded, a disk state notification will be sent.

Parameters:

disk (str) – The disk name.

warn(message: str, module: str = '')

Publish a warn message.

Parameters:

• message (str) – The message to publish

• module (str, optional) – The name of the component involved.

class safecor.ApiHelper

Bases: object

This class provides static functions to help extracting information from the API.

The functions extract data from the payload contained in the API messaged.

Example:

def on_message_received(self, topic:str, payload:dict):
    if topic == Topics.DISK_STATE:
        disk_name = ApiHelper.get_disk_name(payload)

static get_disk_name(payload: dict) → str

Returns the name of the disk.

Associated topics: Topics.DISK_STATE

Parameters:

payload – The payload received from the broker

static get_disk_state(payload: dict) → DiskState

Returns the state of the disk (= “connected”)

Associated topics: Topics.DISK_STATE

Parameters:

payload – The payload received from the broker

static is_disk_connected(payload: dict) → bool

Returns true if the disk is connected

Associated topics: Topics.DISK_STATE

Parameters:

payload – The payload received from the broker

class safecor.BenchmarkId

Bases: object

FILES = 'files'

INPUTS = 'inputs'

class safecor.ComponentState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ERROR = 'error'

READY = 'ready'

STARTING = 'starting'

UNKNOWN = 'unknown'

class safecor.ComponentsHelper

Bases: object

This class helps extracting information about components

It handles a list of system components that can be updated on demand (update()). The information stored can the be extracted using the other functions.

See also

• Api.discover_components() - Discover components with the API

• ComponentState - Component state

clear()

Resets all components in the list

get_by_id(component_id: str) → dict

Returns a component by its ID

get_components()

Returns all the components stored by this instance

get_ids() → list[str]

Returns a list of all components IDs

get_ids_by_type(component_type: str) → list[str]

Returns all components IDs that match a specific type

For exemple, to query all the core components:

def query(self):
    Api().add_message_callback(self.on_message)
    Api().discover_components()

def on_message(self, topic:str, payload:dict):
    if topic == Topics.DISCOVER_COMPONENTS:
        hlp = ComponentsHelper()
        hlp.update(payload)
        core_components = hlp.get_ids_by_type("core")

get_state(component_id: str) → ComponentState

Returns the state of a component

See also

• ComponentState

get_states() → dict

Returns a dict of components states

Example:

{
    "sys-usb": "ready",
    "sys-gui": "ready"
}

See also

• ComponentState

get_type(component_id: str) → str

Returns the type of a component

The types of components are free for the products based on Safecor, but for the components of the Safecor core the value is core.

update(updates: list) → None

Updates the components list with a new list

See also

• Protocol documentation <https://github.com/TristanIsrael/Safecor/wiki/Protocol> chapter Discover the components of the system

class safecor.Configuration(name: str, identifier: dict, settings: dict)

Bases: object

This class encapsulates the data of a configuration

A configuration is made of: - a name (field name) - an identifier for the hardware (field identifier) - a set of parameters (field settings) - the languages used by the system (field languages) - the default language defined for the system (field default_language)

default_language = 'en'

identifier = {}

languages = []

name = ''

settings = {}

class safecor.ConfigurationHelper

Bases: object

This class handles different sets of parameters defined for hardware configurations.

static apply_configuration(topology: Topology) → Topology

Returns the configuration for the running system.

The configuration returned will be composed of the keys from topology.json: - the global configuration (all keys except “configuration”) - the specific configuration which matches the identifier defined in the configuration section

The global settings are loaded and then overwritten by a specific configuration settings if any is applicable.

Some settings cannot be overwritten: - product name - domains - configurations - vpcu

class safecor.ConnectionType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

@brief Cette énumération permet d’identifier le type de connexion utilisée.

SERIAL_PORT = 'serial_port'

TCP_DEBUG = 'tcp'

UNIX_SOCKET = 'unix_socket'

class safecor.Constants

Bases: object

This class defined constants for the whole system

ARCHIVE_EXTENSIONS_HANDLED = ['.7z', '.a', '.apk', '.ar', '.ear', '.tar', '.tar.bz2', '.tar.gz', '.tar.lz4', '.tar.zst', '.tar.xz', '.bz2', '.cab', '.cpio', '.cpio.gz', '.deb', '.gz', '.iso', '.jar', '.lha', '.lzh', '.lz', '.lz4', '.lzma', '.mtree', '.pkg', '.rar', '.tgz', '.war', '.xar', '.xz', '.zip', '.Z', '.zst']

BENCHMARK_INPUTS_ITERATIONS = 1000

DOM0_REPOSITORY_PATH = '/usr/lib/safecor/storage'

DOMU_INPUT_SOCKET_FILEPATH = '/dev/hvc2'

DOMU_REPOSITORY_PATH = '/mnt/storage'

ENABLE_LOCAL_LOG = True

FINGERPRINT_METHOD = 'md5'

FRAME_SIZE = 1024

GUI_DOMAIN_NAME = 'sys-gui'

IO_BENCHMARK = 'safecor_io_benchmark'

LOCAL_LOG_FILEPATH = '/var/log/safecor/safecor.log'

LOG_STRING_FORMAT_DEBUG = '%(asctime)s %(levelname)-8s %(domaine)-10s [%(entite)-20s] %(message)s'

LOG_STRING_FORMAT_PRODUCTION = '%(asctime)s %(levelname)-8s %(domaine)-10s [%(entite)-20s] %(message)s'

MQTT_LOG_BROKER_SOCKET = '/tmp/mqtt_log.sock'

MQTT_LOG_SOCKET_FILTER = '/var/run/safecor/*-log.sock'

MQTT_MSG_BROKER_SOCKET = '/tmp/mqtt_msg_dom0.sock'

MQTT_MSG_BROKER_SOCKETS = '/tmp/mqtt_msg*.sock'

MQTT_MSG_SOCKET_FILTER = '/var/run/safecor/*-msg.sock'

PID_FILES_PATH = '/tmp'

SAFECOR_DISK_CONTROLLER = 'safecor_disk_controller'

SAFECOR_INPUT_CONTROLLER = 'safecor_input_controller'

SAFECOR_SYSTEM_CONTROLLER = 'safecor_system_controller'

SERIAL_PORT_LOG = '/dev/hvc2'

SERIAL_PORT_MSG = '/dev/hvc1'

STR_REPOSITORY = '__repository__'

SYS_USB_INPUT_SOCKET_FILEPATH = '/var/run/safecor/sys-usb-input.sock'

USB_MOUNT_POINT = '/media/usb'

XEN_SOCKETS_PATH = '/var/run/safecor'

class safecor.Debugging

Bases: object

This class provides facilities for product debugging.

Before using the functions of this class you have to start the API.

See also

• Api - API Documentation

benchmark_messaging(target_domain: str, callback_fn, iterations: int = 100, max_data_size_in_bytes: int = 1024)

Sends ping requests to a specific Domain, verifies the data and measure the performance.

The payload will be randomly generated. The requests are all sent in one time, in a bruteforce style.

The API must be ready.

Parameters:

• target_domain (str) – The name of the Domain which will receive the messages.

• callback_fn (Callable) – The function which will receive the results of the benchmark.

• iterations (int, optional) – The number of iterations for the test. Default is 100.

• max_data_size_in_bytes (int, optional) – The maximum length of data generated.

class safecor.DiskMonitor(folder: str, mqtt_client: MqttClient)

Bases: object

This class monitors a folder in order to detect the addition and removal of files and folders.

It works in polling mode because it needs to be compatible with XEN’s 9pfs protocol which does not handle filesystem notifications.

This class must be instanciated with the mount points directory that must be monitored (for instance /mnt).

Use case: Storage connection

When an external storage is connected, a new mount point is automatically created by the system (see udev or mdev) in the mount points directory. For instance, if the storage is named NO NAME, the mount point will be /mnt/NO NAME.

In such a case, the notification DISK_STATE will be sent.

See also

• Topics

Use case: Storage disconnection

When an external storage is removed, the mount point is removed by the system in the mount points directory (/mnt). In such a case, the notification DISK_STATE will be sent.

device_event(action, device)

start()

stop()

class safecor.DiskState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

CONNECTED = 'connected'

DISCONNECTED = 'disconnected'

MOUNTED = 'mounted'

UNKNOWN = 'unknown'

UNMOUNTED = 'unmounted'

class safecor.Dom0Controller(mqtt_client: MqttClient)

Bases: object

This class handles some of the commands sent by Domains that involve the repository and the system in general (supervision, configuration, etc)

The capabilities of the Dom0 controller are:

• List the files of the repository (Topics.LIST_FILES).

• Delete a file in the repository (Topics.DELETE_FILE).

• Calculate a file footprint (Topics.FILE_FOOTPRINT).

• Shut the system down (Topics.SHUTDOWN).

• Restart a Domain (Topics.RESTART_DOMAIN). ** The Dom0 cannot be restarted **.

• Get the system information (Topics.SYSTEM_INFO).

• Get the energy state (Topics.ENERGY_STATE).

• Discover the components of the system (Topics.DISCOVER_COMPONENTS).

• Ping a Domain (Topics.PING).

• Notify that the GUI is ready (Topics.GUI_READY).

All the function can be called using the API or the protocol.

See also

Api - The API class

__init__(mqtt_client: MqttClient)

Instanciates the Dom0 controller.

Parameters:

mqtt_client (MqttClient) – The instance of the MqttClient class which handles the connexion to the MQTT broker.

start()

Starts the Dom0 controller.

When the Dom0 controlelr starts it automatically subscribes to the following topics:

• Topics.LIST_FILES

• Topics.FILE_FOOTPRINT

• Topics.SHUTDOWN

• Topics.RESTART_DOMAIN

• Topics.GUI_READY

• Topics.SYSTEM_INFO

• Topics.ENERGY_STATE

• Topics.DELETE_FILE

• Topics.DISCOVER_COMPONENTS

• Topics.PING

After the Dom0 controller is started it is able to answer requests on these topics.

class safecor.Domain(domain_name: str, domain_type: DomainType)

Bases: object

This class encapsulates information about a Domain

cpu_affinity = []

domain_type = 2

id = 0

memory = 1000

name = 'NoName'

package = ''

temp_disk_size = 0

vcpu_group = 'group1'

vcpus = 1

class safecor.DomainType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

This enumeration defines a Domain”s type

BUSINESS = 2

CORE = 1

UNKNOWN = 0

class safecor.FileHelper

Bases: object

This class defines helper function for querying the system about disks and files

Most of these functions are used on the sys-usb Domain by SysUsbController or on the Dom0 by Dom0Controller.

static calculate_fingerprint(filepath: str) → str

Calculates the fingerprint of a file.

Parameters:

filepath (str) – The full path of the file of which the fingerprint must be calculated.

Returns:

A sha-512 encoded fingerprint

Return type:

str

See also

hashlib.file_digest()

static copy_file(source_disk: str, filepath: str, destination_disk: str, source_fingerprint: str) → str

Copies a file to another place.

The destination directory must exist.

The fingerprint of both files will compared at the end. The fingerprint is returned if they are identical.

Parameters:

• source_disk (str) – The disk that contains the source file

• filepath (str) – The path of the file on the source disk

• destination_disk (str) – The disk that will contain the copy

• source_fingerprint (str) – The fingerprint of the source file

Returns:

The fingerprint of the file if both fingerprints are equals, else an empty string

Return type:

str

static copy_file_to_repository(source_disk: str, filepath: str, fingerprint: str)

Copies a file in the repository of the system.

Parameters:

• source_disk – The disk that contains the source file

• filepath (str) – The path of the sourc file

• fingerprint (str) – The fingerprint of the source file

static create_file(filepath: str, size_ko: int)

Creates a new file on a disk

If size_ko is different from 0, the file will be filled with random data.

Parameters:

• filepath (str) – The path of the file to create

• size_ko (int) – The initial size of the new file

static get_disks_list() → list

This function returns the list of external disks connected to the system.

The list of disks is obtained from the directories contained in Constants.USB_MOUNT_POINT, by checking whether they are mount points.

This function does not return the list of files mounted.

static get_files_list(disk: str, recursive: bool, from_dir: str = '') → list

This function returns the complete directory tree of a mount point.

The path used is built by concatenating the Constants.USB_MOUNT_POINT constant with the name passed as an argument.

The list can be queried for external storages and files mounted.

For each file or folder, a dictionary is created:

{ “type”: “folder”, “path”: “/”, “name”: “dossier 1” }, { “type”: “file”, “path”: “/dossier 1”, “name”: “fichier 1”, “size”: 333344 }, { “type”: “file”, “path”: “/dossier 1/dossier 2”, “name”: “fichier 4”, “size”: 12 }

static get_folder_contents(path: str, contents_list: list, cutting: int = 0, recursive: bool = False, from_dir: str = '')

Queries the contents of a folder

Parameters:

• path – The path of the folder

• type – str

• contents_list – The list in which the new contents will be appended

• type – list

• cutting – If True the contents entry will be cut starting at this index

• type – bool

• recursive – If True the contents of each folder will by analyzed recursively

• type – bool

• from_dir – Specifies a start directory for the analysis. If not specified the analysis will start at the root of the disk.

• type – str

static get_mount_point(disk: str) → str | None

Returns the real mount point in the system

An external storage is mounted under /media/usb and a file is mounted in /media/loop. this function helps finding the correct mount point for a disk.

Parameters:

disk – The name of the mount point

static is_archive_file(filename: str) → bool

Verifies whether the file is (supposed to be) an archive file

The verification is done with the file extension, not using the MIME type nor controlling the MAGIC od the file.

static remove_file(filepath: str) → bool

Removes a file

Parameters:

filepath (str) – The path of the file to remove

Returns:

True if the file has been removed, else False

Return type:

bool

class safecor.InputType

Bases: object

KEYBOARD = 1

MOUSE = 2

TOUCH = 3

UNKNOWN = 0

class safecor.InputsDaemon(*args, **kwargs)

Bases: object

This class monitors mouse, touch device and keyboard inputs and serialize filtered information through the XenBus.

The communication channel (pvchan) inputs is used between sys-usb and the Dom0.

This class is not intended to be used in a project, it is directly managed by the core.

Mouse

Information on mouse position and buttons are all transmitted each time an event occurs. It means that the information are sent as a data structure containing x and y position and the buttons state.

start(mqtt_client: MqttClient)

Starts the inputs daemon

As soon as it starts, the daemon looks for mouses, touchscreens and keyboards. When it finds a device it starts monitoring it in a thread and serializes all input events on the pv channel inputs.

stop()

Stops the inputs daemon

class safecor.KeymapFR

Bases: object

Cette classe permet de convertir un code clavier en caractère

KEYMAP = {2: '1', 3: '2', 4: '3', 5: '4', 6: '5', 7: '6', 8: '7', 9: '8', 10: '9', 11: '0', 13: '=', 15: '\t', 16: 'A', 17: 'Z', 18: 'E', 19: 'R', 20: 'T', 21: 'Y', 22: 'U', 23: 'I', 24: 'O', 25: 'P', 28: '\r', 30: 'Q', 31: 'S', 32: 'D', 33: 'F', 34: 'G', 35: 'H', 36: 'J', 37: 'K', 38: 'L', 39: ';', 44: 'W', 45: 'X', 46: 'C', 47: 'V', 48: 'B', 49: 'N', 50: 'M', 51: ',', 52: '.', 53: '/', 55: '*', 57: ' ', 74: '-', 78: '+', 83: '.', 96: '\r', 98: '/', 117: '=', 121: ',', 434: '$', 435: '€', 512: '0', 513: '1', 514: '2', 515: '3', 516: '4', 517: '5', 518: '6', 519: '7', 520: '8', 521: '9', 522: '*', 523: '£', 524: 'A', 525: 'B', 526: 'C', 527: 'D', 620: '11', 621: '12'}

static convertit(evcode)

class safecor.LibvirtHelper

Bases: object

static get_cpu_allocation_for_domain(domain_name: str) → list

static get_cpu_count() → int

@brief Returns the number of CPU in the system

In case of error the function returns 0.

static get_domains() → dict

@brief Returns the list of Domains declared on the system and their status

static get_domains_count() → int

@brief Returns the number of domains

static reboot_domain(domain_name: str) → bool

@brief Reboots a domain.

Returns True if the reboot has been acknowledged by the hypervisor

class safecor.Logger(*args, **kwargs)

Bases: object

This class provides mechanisms for logging and recording log files

The logging facility is intended to store messages and write them in a log file if asked.

For sending logging messages the best practice is to use the class Api and the functions Api.info(), Api.debug(), Api.warning(), etc. The embedded core logger will handle all messages for a system.

This class can be used for an additionnal facility for a system based on Safecor.

clear_log()

Clears the current log

critical(description: str, module: str = '')

Sends a critical message

debug(description: str, module: str = '')

Sends a debugging message

domain_name() → str

error(description: str, module: str = '')

Sends an error message

filename() → str

static format_logline(message: str) → str

Formats a message in the log file

The default format is %y-%m-%d %H:%M:%S.%f - Message text

Example:

2025-01-25 13:12:31.123 - The system has started

info(description: str, module: str = '')

Sends an information message

is_recording()

is_setup() → bool

log_level()

static loglevel_from_topic(topic: str) → int

static loglevel_value(level: str) → int

module_name()

static print(message: str) → None

Prints a formatted log in the standard output

See also

• format_logline() - Log line format

reset()

Resets the logging instance

set_log_level(log_level)

setup(module_name: str, mqtt_client: MqttClient, log_level: int = 20, recording: bool = False, filename: str = '/var/log/safecor/safecor.log')

Sets up the logging facility.

Parameters:

• module_name (str) – The module information is free. A module should be considered as a logical part of the product or of a component.

• mqtt_client (str) – The MQTT client that handles the connection to the broker.

• log_level (int) – The minimum logging level that this facility will take into account. When the log level of a message is lower that this, it is ignored.

• recording (bool) – If True, the logging messages received will be recorded in a file.

• filename (str) – The path of the file that will record all filtered messages.

See also

• logging - Standard Python logging facility

warn(description: str, module: str = '')

Sends a warning message

Synonym of warning().

warning(description: str, module: str = '')

Sends a warning message

class safecor.MockSysUsbController(sync_lock: Event)

Bases: object

start(source_disks_list: dict, storage_path: str, destination_disk_path: str)

class safecor.Mouse(move: MouseMove = 0, x: int = 0, y: int = 0, buttons: int = 0, wheel: int = 0)

Bases: object

This class encapsulates information about a mouse or a touch event.

The properties are public in order to make the changes quicker because the events are very numerous and need to be sent very quickly.

button_equals(other, button) → bool

Return True if a button is pressed in this instance and another one.

Parameters:

• other (Mouse) – Another instance of Mouse

• button (MouseButton) – The button to compare

button_pressed(button) → bool

Returns True if the button is pressed.

Parameters:

button (MouseButton) – The button to verify

buttons = 0

buttons_equal(other) → bool

Returns True if the buttons are the same between this instance and another instance.

Parameters:

other (Mouse) – Another instance of Mouse

equals(other) → bool

Returns True if the position and the buttons are the same between this instance and another instance.

Parameters:

other (Mouse) – Another instance of Mouse

static from_data(data: bytes)

Deserializes the data received from a serial bus in this instance.

move = 0

serialize() → bytes

Serializes this instance’s data in order to send it on a serial bus.

wheel = 0

wheel_equals(other) → bool

Returns True if the wheel state is the same between this instance and another one.

Parameters:

other (Mouse) – Another instance of Mouse

wheel_moved() → bool

Returns True if the wheel is moved

x: int = 0

y: int = 0

class safecor.MouseButton

Bases: object

This class codes the buttons of a mouse

LEFT = 1

MIDDLE = 2

RIGHT = 4

UNKNOWN = 0

class safecor.MouseMove

Bases: object

This class codes the mouse positing strategy used by the system

In relative positioning, an negative x value means that the mouse is moving left. A negative y value means that the mouse is moving up.

ABSOLUTE = 1

Used in systems with touch screens or touch pads. The coordinates are absolute to the screen and depends from the resolution.

RELATIVE = 0

Used in systems with regular mouse. The coordinates are relative from the previous value.

class safecor.MouseWheel

Bases: object

This class codes the mouse wheel movement

DOWN = -1

NO_MOVE = 0

UP = 1

class safecor.MqttClient(identifier: str, connection_type: ConnectionType = ConnectionType.UNIX_SOCKET, connection_string: str = '', debugging=False)

Bases: object

@brief Cette classe permet d’échanger des messages MQTT au sein du système.

add_connected_callback(callback)

add_message_callback(callback)

connected = False

connection_string: str = ''

connection_type: ConnectionType = 'unix_socket'

del_message_callback(callback)

identifier: str = 'unknown'

is_starting = False

on_connected: Callable[[], None] | None = None

on_log: Callable[[int, str], None] | None = None

on_message: Callable[[str, dict], None] | None = None

on_subscribed: Callable[[], None] | None = None

publish(topic: str, payload: dict)

reset_message_callbacks()

start()

stop()

subscribe(topic: str) → tuple[MQTTErrorCode, int | None]

unsubscribe(topic: str)

Unsubscribes a client from a specific topic

unsubscribe_all()

Unsubscribes a client from all topics

class safecor.MqttFactory

Bases: object

static create_mqtt_client_dom0(identifier: str) → MqttClient

static create_mqtt_client_domu(identifier: str) → MqttClient

static create_mqtt_network_dev(identifier: str) → MqttClient

class safecor.MqttHelper

Bases: object

This class contains helper functions for the MQTT facility

static check_payload(payload: dict, expected_keys: list)

Verifies that the payload contains some necessary keys

class safecor.NotificationFactory

Bases: object

This class helps creating notifications

All functions of this class are static.

static create_notification_deleted_file(disk: str, filepath: str) → dict

Create a notification for a file deletion

static create_notification_disk_state(disk: str, state: DiskState) → dict

Creates a notification for a disk’s state

static create_notification_energy_state(battery) → dict

Create a notification for the energy state

static create_notification_error(disk: str, filepath: str, error: str) → dict

Create a notification for an error

static create_notification_new_file(disk: str, filepath: str, source_fingerprint: str, dest_fingerprint: str) → dict

Create a notification for a new file

static create_notification_setting_changed(key: str, value) → dict

Create a notification for a setting of which value has changed

class safecor.RequestFactory

Bases: object

The class MessageFactory generates notifications, commands and errors used in the communication between Domains.

The emitter of the command is automatically added during the creation.

Reminder Domains involved in the foundation are: - sys-usb which manages USB devices

Notification are:

• A DomU notifies Dom0 he has started (DomU to Dom0)

• sys-usb notifies Dom0 that a USB storage is connected (DomU to Dom0)

• sys-usb notifies Dom0 that a USB storage is disconnected (DomU to Dom0)

Commands are:

• Logging

  • A Domain adds an entry in the log (DomU to Dom0)

  • Dom0 records the log on a USB storage (DomU to Dom0)

• System

  • A DomU asks for the debug status (DomU to Dom0)

  • A DomU asks for to reboot of the system (DomU to Dom0)

  • A DomU asks for to shut the system down (DomU to Dom0)

  • A DomU asks for the reset of another DomU (DomU to Dom0)

  • A DomU asks for the batery level (DomU to Dom0)

  • A DomU asks for the charging state (DomU to Dom0)

• Files

  • A DomU asks for the list of storages (DomU to Dom0 to sys-usb)

  • A DomU asks for the creation of a secure archive on a USB storage (DomU to Dom0 to sys-usb)

  • A DomU asks for adding a file to a secure archive (DomU to Dom0 to sys-usb)

  • A DomU asks for the copy of a file (DomU to Dom0 to sys-usb)

• Notifications

  • To do

static create_request_copy_file(source_disk: str, filepath: str, destination_disk: str) → dict

Create a request to copy a file from a storage to another, including the repository.

Example :

{ filepath: "External storage:/a path/a filename", disk_destination: "Another storage" }

static create_request_create_file(filepath: str, disk: str, contents: bytes, compressed: bool = False) → dict

Create a request to create a new file on a storage

Parameters:

• filepath (str) – The filepath of file to create

• disk (str) – The storage on which the file should be created

• contents (bytes) – The data to write in the file as a byte array

• compressed (bool) – Indicates whether the data should be compressed - Default is False

static create_request_delete_file(filepath: str, disk: str = '__repository__') → dict

Create a request to delete a file from a storage

Parameters:

• filepath – The filepath of the file to delete

• disk (str) – The storage name - Default value is the local repository

static create_request_files_list(disk: str, recursive: bool = False, from_dir: str = '') → dict

Creates a request to receive the files list of a storage

static create_request_get_file_fingerprint(filepath: str, disk: str) → dict

Create a request to get a file’s fingerprint

static create_request_mount_file(disk: str, filepath: str)

Create a request to mount a file

static create_request_ping(ping_id, source_name, data, datetime)

Create a request to ping a Domain

When a Domain is pinged it should answer automatically with a ping response.

static create_request_read_file(disk: str, filepath: str) → dict

Creates a request to read a file’s content

static create_request_restart_domain(domain_name: str)

Create a request to restart a Domain

static create_request_save_log(disk: str, filename: str)

Create a request to save the log file

static create_request_set_language(lang: str)

Create a request to set the language of the system

static create_request_set_log_level(log_level)

Create a request to set the log level

Log levels are defined in the default logging module.

static create_request_start_benchmark(id_benchmark: str)

Create a request to start the benchmark

static create_request_unmount(disk: str)

Create a request to unmount a disk

class safecor.ResponseFactory

Bases: object

This class helps generating commands for API reponses.

static create_entry_component_state(component_id: str, component_label: str, domain_name: str, state: ComponentState, component_type='core') → dict

static create_reponse_benchmark_fichiers_demarre() → dict

static create_reponse_benchmark_fichiers_erreur(erreur: str) → dict

static create_reponse_benchmark_fichiers_termine(metrics: list) → dict

static create_reponse_benchmark_inputs(duree: int, iterations: int) → dict

static create_response_component_state(components: list[dict]) → dict

static create_response_copy_file(filepath: str, disk: str, success: bool, fingerprint: str) → dict

static create_response_create_file(filepath: str, disk: str, fingerprint: str, success: bool) → dict

static create_response_disks_list(disks_list: list = []) → dict

static create_response_file_fingerprint(filepath: str, disk: str, fingerprint: str) → dict

static create_response_get_setting(key: str, value)

static create_response_language_default(language: str)

static create_response_languages_list(languages: list)

static create_response_list_files(disk: str, files: list = []) → dict

static create_response_ping(ping_id, source_name, data, sent_at)

static create_response_restart_domain(domain_name: str, accepted: bool, reason: str)

static create_response_set_setting(key: str, value)

static create_response_shutdown(accepted: bool)

class safecor.Screen

Bases: object

This class encapsulates information about a screen

height = 0

rotation = 0

width = 0

class safecor.SerialMQTTClient(path: str, baudrate: int, *args, **kwargs)

Bases: Client

@brief Cette classe permet au client MQTT de communiquer sur un port série.

close()

disconnect(reasoncode: ReasonCode | None = None, properties: Properties | None = None)

Disconnect a connected client from the broker.

Parameters:

• reasoncode (ReasonCode) – (MQTT v5.0 only) a ReasonCode instance setting the MQTT v5.0 reasoncode to be sent with the disconnect packet. It is optional, the receiver then assuming that 0 (success) is the value.

• properties (Properties) – (MQTT v5.0 only) a Properties instance setting the MQTT v5.0 properties to be included. Optional - if not set, no properties are sent.

loop_start() → MQTTErrorCode

This is part of the threaded client interface. Call this once to start a new thread to process network traffic. This provides an alternative to repeatedly calling loop() yourself.

Under the hood, this will call loop_forever in a thread, which means that the thread will terminate if you call disconnect()

loop_stop() → MQTTErrorCode

This is part of the threaded client interface. Call this once to stop the network thread previously created with loop_start(). This call will block until the network thread finishes.

This don’t guarantee that publish packet are sent, use wait_for_publish or on_publish to ensure publish are sent.

on_connection_lost: Callable[[], None] | None = None

class safecor.Settings

Bases: object

This class defines the settings keys reference

CURRENT_LANGUAGE = 'current_language'

DEFAULT_LANGUAGE = 'default_language'

class safecor.SingletonMeta

Bases: type

This class is a metaclass for developing Singleton classes.

Use use this way:

class MyClass(metaclass=SingletonMeta):

    # There won't be any __init__() method.

    def function(self):
        pass

# Use the Singleton
MyClass().function()

class safecor.SysLogger(module_name: str)

Bases: object

This class is a logger that sends messages to the system logger facility (syslog)

critical(description: str)

Sends a critical message

debug(description: str)

Sends a debugging message

error(description: str)

Sends an error message

info(description: str)

Sends an information message

warn(description: str)

Sends a warning message

Synonym of warning().

warning(description: str)

Sends a warning message

class safecor.SysUsbController(mqtt_client: MqttClient)

Bases: object

This class handles messages exchanged between sys-usb and the Dom0 or the other Domains

sanitize_filename(s: str) → str

Replaces all non-printable chars of a filename with ‘_’

start()

stop()

class safecor.System(*args, **kwargs)

Bases: object

The System class provides functions for querying or modifying the system’s state.

Some of the functions belong to the main system (Dom0) and some other belong to the virtual machines (DomU)

compute_cpus_for_group(group_name: str, groups: dict) → list[int]

Computes the CPUs (or cores) which will be pinned to the Domains of the group.

The first CPU is assigned to Dom0 and sys-usb Domain.

If there are at least 4 CPUs the second CPU is also assigned to Dom0 and sys-usb.

The other CPUs are assigned to sys-gui and the other groups by trying to avoid overlapping.

This function must be ran in the Dom0.

static compute_vcpus_for_group(group_name: str, groups: dict) → int

Computes the number of vCPUs which will be pinned to each Domain of a group.

The number of vCPUs depends on the value of the parameter vcpu.groups defined in the file topology.json.

This function must be ran in the Dom0.

static cpu_affinity_to_string(cpu_affinity: list) → str

Converts the cpu affinity of the Domain definition to a string

Example: >>>

cpu_affinity_to_string([1,2,3,4]) ‘1-4’

static debug_activated()

Returns whether the debugging has been activated

This function must be ran in the Dom0.

static domain_name()

Returns the Domain’s name

This function must be ran in a DomU.

get_cpu_allocation() → dict

Provides information about CPU allocation for all the Domains

This function must be ran in the Dom0.

static get_framebuffer_dimension(fbdev: int = 0) → tuple[int, int]

Returns the framebuffer dimensions as a tuple (width, height)

By default, the framebuffer fb0 is queried. This can be overriden by setting the parameter fbdev

get_platform_cpu_count() → int

Returns the CPU count of the machine

The CPU count includes all cores.

get_screen_height()

Returns the system main screen’s resolution height

This function must be ran in a DomU.

get_screen_rotation() → int

Returns the Domain’s screen rotation angle

The rotation angle is defined in the topology file in the setting gui.screen.rotation.

This function must be ran in a DomU.

Possible values are: 0, 90, 180, 270.

get_screen_width() → int

Returns the system main screen’s resolution width

This function must be ran in a DomU.

get_setting(key: str)

get_settings() → dict

static get_splash_bgcolor_from_topology(override_topology_file: str = '') → int

Returns the background color for the plash defined in the topology file

This function is necessary for treatments that occur before libvirt and xenstore are loaded. The function get_topology_struct() is one of those.

static get_system_information() → dict

Returns a JSON struct containing the information on the system.

This function must be ran in the Dom0.

A typical struct is:

{
"core": {
    "version": "1.1",
    "debug_on": false
},
"system": {
    "os": {
        "name": "Linux",
        "release": "6.12.20-0-lts",
        "version": "#1-Alpine SMP PREEMPT_DYNAMIC 2025-03-24 08:09:11"
    },
    "machine": {
        "arch": "x86_64",
        "processor": "",
        "platform": "Linux-6.12.20-0-lts-x86_64-with",
        "cpu": {
            "count": 12,
            "freq_current": 1689.5960000000002,
            "freq_min": 0.0,
            "freq_max": 0.0,
            "percent": 0.0
        },
        "memory": {
            "total": 405987328,
            "available": 96657408,
            "percent": 76.2,
            "used": 256733184,
            "free": 12472320
        },
        "load": {
            "1": 0.5244140625,
            "5": 0.21875,
            "15": 0.08154296875
        }
    },
    "boot_time": 1748036696.0,
    "uuid": "11ec0800-4fb9-11ef-bd38-ad993f2e7700"
    "storage": {
        "total": 12345678,
        "used": 0,
        "free": 12345678,
        "files_count": 0
    }
}

}

get_system_uuid()

Returns the system’s UUID

The UUID is queried from the Linux kernel’s /sys/class/dmi/id/product_uuid

static get_topology(override_topology_file: str = '') → Topology

Returns a Topology object initialized with the contents of the topology file

This function must be ran in the Dom0.

static get_topology_data(override_topology_file: str = '') → dict

Interprets the topology file data as a JSON object

This function must be ran in the Dom0.

static get_topology_struct(override_topology_file: str = '') → dict

Returns the topology of the current system

The topology structure is different from the configuration file topology.json because the file will evolve from its original format and it must be dissociated from the internal structure to avoid future compatibility problems.

The topology is defined in the file topology.json. This function returns a data structure representing the topology as a dict. Instead of returning the JSON data as the function read_topology_file() does, it returns a structure representing the objects:

{
    "domains": [
        "my-domain": {
            "vcpu_group": "group1",
            "memory": 4000,
            "vcpus": 2,
            "cpus": "3-4",
            "package": "",
            "temp_disk_size": 4096
        }
    ],
    "system": {
        "use_usb": 1,
        "use_gui": 1,
        "screen_rotation": 0,
        "gui_app_package": "",
        "gui_memory": 1000,
    },
    "product": {
        "name": "Safecor"
    },
    "configurations": [
        {
            "name": "",
            ...
        }
    ]
}

All the keys are guaranteed to exist with a default value if necessary.

monitor_file(path: str, filename: str, fn_callback)

Starts the monitoring of a single file or a whole directory.

When a file is created or removed, the callback function is called with the following parameters: - path:str - The directory path - filename:str - The file name - exists:bool - True if the file has been created, else False

This function is uninterruptible.

Parameters:

• path (str) – The directory to monitor

• filename (str) – A file name to monitor, set to None if the whole directory should be monitored

• fn_callback – The callback function

parse_kernel_command_line_settings()

Returns the settings defined on the command line

All the keys are turned to lowercase.

static parse_range(value: str) → tuple[int, ...]

Converts an int range represented as a string into a tuple of all values between min and max

Examples

>>> parse_range("1-2")
    (1,2)
>>> parse_range("3-8")
    (3,4,5,6,7,8)

The left value must be lower that the right value.

static read_topology_file(override_topology_file: str = '') → str

Reads the topology file describing the product

This function must be ran in the Dom0.

static reset_topology()

Resets the current topology

set_setting(key: str, value)

class safecor.TaskRunner(max_tasks: int = 0)

Bases: object

run_task(task, args)

start()

stop()

class safecor.Topics

Bases: object

This class defines MQTT topics for the API

BENCHMARK = 'system/misc/benchmark'

COPY_FILE = 'system/disks/copy_file'

CREATE_FILE = 'system/disks/create_file'

CURRENT_LANGUAGE = 'system/settings/languages/current'

DEBUG = 'system/events/debug'

DEBUGGING = 'system/debugging'

DEFAULT_LANGUAGE = 'system/settings/languages/default'

DELETED_FILE = 'system/disks/deleted_file'

DELETE_FILE = 'system/disks/delete_file'

DISCOVER = 'system/discover'

DISCOVER_COMPONENTS = 'system/discover/components'

DISKS = 'system/disks'

DISK_STATE = 'system/disks/state'

ENERGY = 'system/energy'

ENERGY_STATE = 'system/energy/state'

ERROR = 'system/events/error'

EVENTS = 'system/events'

FILE_FINGERPRINT = 'system/disks/file_fingerprint'

GUI_READY = 'system/workflow/gui_ready'

INFO = 'system/events/info'

KEEPALIVE = 'system/misc/ping'

LANGUAGES = 'system/settings/languages'

LIST_DISKS = 'system/disks/list_disks'

LIST_FILES = 'system/disks/list_files'

LIST_LANGUAGES = 'system/settings/languages/list'

MISC = 'system/misc'

MOUNT_FILE = 'system/disks/mount_file'

NEW_FILE = 'system/disks/new_file'

PING = 'system/debugging/ping'

READ_FILE = 'system/disks/read_file'

RESTART_DOMAIN = 'system/workflow/restart_domain'

SAVE_LOG = 'system/events/save_log'

SETTINGS = 'system/settings'

SETTING_CHANGED = 'system/settings/changed'

SET_LANGUAGE = 'system/settings/languages/set'

SET_LOG_LEVEL = 'system/events/set_loglevel'

SHUTDOWN = 'system/workflow/shutdown'

SYSTEM = 'system'

SYSTEM_INFO = 'system/info'

SYS_USB = 'system/sys-usb'

SYS_USB_CLEAR_QUEUES = 'system/sys-usb/clear-queues'

UNMOUNT = 'system/disks/unmount'

WARNING = 'system/events/warning'

WORKFLOW = 'system/workflow'

class safecor.Topology(other=None)

Bases: object

This class encapsulates and handles information about the topology of a product

The topology is usually defined in the file /etc/safecor/topology.json

The topology object corresponds to the configuration that should be applied to the device.

__init__(other=None)

Create a new Topology from scratch or as a deep copy if other is not None

add_color(color_name: str, color_value_as_hex: str)

Adds a named color to the list

add_domain(domain: Domain)

Adds a Domain’s definition to the list

business_domains() → list[Domain]

Returns only the business domains

color_as_hex(color_name: str) → str

Returns a named color as an hex value

color_as_rgba(color_name: str) → tuple[int, int, int, int]

Returns an RGBA named color as a tuple value

colors() → dict

Returns the RGBA colors list as a list of 8 bit tuples (r, g, b, a)

See also

colors_as_hex()

colors_as_hex() → dict

Returns the colors list as hexadecimal values (#rrggbbaa)

See also

colors()

configurations = []

default_language = 'en'

domain(domain_name: str) → Domain | None

Returns the Domain object with the specified name

domain_names() → list

Returns the list of Domain names

domains

alias of List[Domain]

gui: Gui

initialized() → bool

Returns whether the object is initialized

is_pci_bus_blacklisted(bus_bdf: str) → False

Returns true if the bus is in the blacklist

languages = []

pci: Pci

product_name = 'No Name'

screen: Screen

set_initialized(initialized: bool)

Sets the objet initialized

use_gui = False

use_usb = False

uuid = ''