rsio.utils package

Submodules

rsio.utils.auxiliary module

class rsio.utils.auxiliary.Component(name: str, path: Path, formalism: Formalism)

Bases: object

class rsio.utils.auxiliary.Launch(nodes: list[Component])

Bases: object

class rsio.utils.auxiliary.MqttConfiguration(host: str = 'localhost', port: int = 1883, keepalive: int = 5)

Bases: object

host: str = 'localhost'

keepalive: int = 5

port: int = 1883

class rsio.utils.auxiliary.RedisConfiguration(host: str = 'localhost', port: int = 6379, db: int = 0, socket_timeout: int = 5)

Bases: object

db: int = 0

host: str = 'localhost'

port: int = 6379

socket_timeout: int = 5

rsio.utils.auxiliary.activate_virtual_environment(venv_name='venv')

Activates the virtual environment. :param venv_name: The name of the virtual environment directory. Defaults to “venv”.

rsio.utils.auxiliary.build_docker_image(docker_file_path: Path, image_name: str) → bool

Build a Docker image for a Python module at a given path.

Parameters:

• docker_file_path – Path Dockerfile

• image_name – Name of the Docker image to be created

rsio.utils.auxiliary.compress_folder(folder_path)

rsio.utils.auxiliary.create_virtual_environment(venv_name='venv')

Creates a Python virtual environment.

Parameters:

venv_name – The name of the virtual environment directory. Defaults to “venv”.

rsio.utils.auxiliary.deactivate_virtual_environment()

Deactivates the virtual environment.

rsio.utils.auxiliary.decompress_folder(data, output_path)

rsio.utils.auxiliary.execute_commands(commands: list[str])

Execute multiple command line functions concurrently.

Parameters: commands (list of str): List of command line commands to execute.

rsio.utils.auxiliary.extract_commands(launch_description)

rsio.utils.auxiliary.get_custom_code(text: str, tag: str) → list | None

rsio.utils.auxiliary.get_docker_version() → str | bool

Check if Docker is installed on the system :return: String with the docker version if Docker is installed, False otherwise

rsio.utils.auxiliary.get_pip_path(venv_name: str) → Path

Returns the path to the pip executable based on the operating system.

rsio.utils.auxiliary.get_python_version() → str

Check and return the current Python version installed on the system. :return: string representing the Python version. :rtype: str or None

rsio.utils.auxiliary.install_python_packages_from_requirements_file(requirements_file: Path, venv_name: str | None = 'venv') → bool

Installs packages listed in a requirements.txt file into the virtual or native environment.

Parameters:

• venv_name – The name of the virtual environment directory. Defaults to “venv”.

• requirements_file – The path to the requirements.txt file. Defaults to “requirements.txt”.

rsio.utils.auxiliary.is_mqtt_reachable(mqtt_configuration: MqttConfiguration = MqttConfiguration(host='localhost', port=1883, keepalive=5)) → bool

Check if MQTT is running and reachable :return: True if MQTT broker is running and reachable, False otherwise

rsio.utils.auxiliary.is_python_package_installed(package: str) → bool

rsio.utils.auxiliary.is_redis_reachable(redis_configuration: RedisConfiguration = RedisConfiguration(host='localhost', port=6379, db=0, socket_timeout=5)) → bool

Check if Redis is running and reachable :return: True if Redis is running and reachable, False otherwise

rsio.utils.auxiliary.parse_launch_xml(file_path: Path, formalism: Formalism = Formalism.PYTHON)

rsio.utils.auxiliary.replace_custom_code(text: str, tag: str, replacement: str) → str

rsio.utils.auxiliary.run_command(command: str)

rsio.utils.auxiliary.run_docker_container(image_name: str, container_name: str | None = None, ports: dict[int, int] | None = None) → bool

Run a Docker container from an existing image.

Parameters:

• image_name – Name of the Docker image to run

• container_name – Optional name for the container

• ports – Optional dictionary mapping container ports to host ports (e.g., {“8080”: “8080”})

Returns:

None

rsio.utils.constants module

class rsio.utils.constants.Deployment(*values)

Bases: Enum

CONTAINER = 'containerized'

NATIVE = 'native'

VIRTUAL = 'virtualenv'

class rsio.utils.constants.ExecutionPatterns

Bases: object

IOP = 'IOP'

IPO = 'IPO'

TRIGGERED = 'TRIGGERED'

class rsio.utils.constants.Formalism(*values)

Bases: Enum

CPP = 'C++'

CSHARP = 'C#'

GO = 'Go'

JAVA = 'Java'

JAVASCRIPT = 'JavaScript'

KOTLIN = 'Kotlin'

PYTHON = 'Python'

RUST = 'Rust'

class rsio.utils.constants.MonitorType

Bases: object

POSTPROCESSING = 'postprocessing'

RUNTIME = 'runtime'

class rsio.utils.constants.OperatingSystem(*values)

Bases: Enum

UNIX = 'unix'

WINDOWS = 'windows'

class rsio.utils.constants.OrchestrationType

Bases: object

DISTRIBUTED = 'DISTRIBUTED'

DISTRIBUTED_FIXED_IO = 'DISTRIBUTED_FIXED_IO'

FMI_local = 'FMI_LOCAL'

class rsio.utils.constants.ProjectDirectories(name: str, deploy_file_path: pathlib._local.Path = PosixPath('Realization/ManagingSystem/Actions/deploy.py'), build_file_path: pathlib._local.Path = PosixPath('Realization/ManagingSystem/Actions/build.py'), run_file_path: pathlib._local.Path = PosixPath('Realization/ManagingSystem/Actions/run.py'), platform_config_template: str = 'Realization/ManagingSystem/Platform/{name}/config.yaml', launch_xml_file_path_template: str = 'Realization/ManagingSystem/Platform/{name}/launch.xml')

Bases: object

build_file_path: Path = PosixPath('Realization/ManagingSystem/Actions/build.py')

deploy_file_path: Path = PosixPath('Realization/ManagingSystem/Actions/deploy.py')

launch_xml_file_path_template: str = 'Realization/ManagingSystem/Platform/{name}/launch.xml'

name: str

property platform_config: Path

platform_config_template: str = 'Realization/ManagingSystem/Platform/{name}/config.yaml'

run_file_path: Path = PosixPath('Realization/ManagingSystem/Actions/run.py')

class rsio.utils.constants.StepStatus

Bases: object

FAILED = 'Failed'

PASSED = 'Passed'

PENDING = 'Pending'

RUNNING = 'Running'

class rsio.utils.constants.VerificationMethods

Bases: object

STATICBOUND = 'StaticBound'

STATICLOWERBOUND = 'StaticLowerBound'

STATICUPPERBOUND = 'StaticUpperBound'

rsio.utils.constants.deployment_from_name(name: str) → Deployment

Convert a case-insensitive name to a Deployment.

rsio.utils.constants.detect_operating_system() → OperatingSystem

rsio.utils.constants.enum_from_name(name: str, enum: Type[EnumType]) → EnumType

Convert a case-insensitive name to an EnumType.

rsio.utils.constants.formalism_from_name(name: str) → Formalism

Convert a case-insensitive name to a Formalism.

Module contents