vyra_base.helper package

Submodules

vyra_base.helper.env_handler module

class vyra_base.helper.env_handler.EnvHandler

Bases: object

Class to handle the environment variables for the project.

__init__()

Initialize the EnvHandler class.

env: dict = {}

async load_env(env_path)

Load the environment variables from a file.

Parameters:

env_path (Path) – Path to the environment file directory.

vyra_base.helper.env_handler.get_env_required(name)

Return type:

str

Parameters:

name (str)

vyra_base.helper.error_handler module

class vyra_base.helper.error_handler.ErrorTraceback

Bases: object

Class to check traceback errors and feedback them to the logger.

static check_error_exist(error_details=[], log_print=False)

Check if an error occurred and log it.

Parameters:

• error_details (list) – List to append error details to.

• log_print (bool)

Returns:

True if an error exists, otherwise False.

Return type:

bool

static w_check_error_exist(func)

Decorator to check if an error occurred after function execution.

Parameters:

func (callable) – The function to decorate.

Returns:

The wrapped function.

Return type:

callable

vyra_base.helper.file_lock module

async vyra_base.helper.file_lock.get_lock_for_file(file_path)

Get (or create) a lock for the specified file.

Parameters:

file_path (Union[Path, AsyncPath]) – The path to the file.

Returns:

An asyncio.Lock object for the file.

Return type:

Lock

async vyra_base.helper.file_lock.release_lock_for_file(file_path)

Remove the lock from the dictionary if no tasks are waiting for it.

Parameters:

file_path (Union[Path, AsyncPath]) – The path to the file.

vyra_base.helper.file_lock.get_lock_for_file_sync(file_path)

Get (or create) a threading lock for the specified file (synchronous).

Parameters:

file_path (Union[Path, str]) – The path to the file.

Returns:

A threading.Lock object for the file.

Return type:

allocate_lock

vyra_base.helper.file_lock.release_lock_for_file_sync(file_path)

Remove the threading lock from the dictionary if not locked (synchronous).

Parameters:

file_path (Union[Path, str]) – The path to the file.

vyra_base.helper.file_reader module

class vyra_base.helper.file_reader.FileReader

Bases: object

File reader.

Reads file content from locally stored files into the module.

async classmethod open_json_file(config_file, config_default=PosixPath('.'))

Reads a JSON file.

To ensure cross platform compatibility, JSON files are opened with utf8-sig encoding.

Parameters:

• config_file (Path) – JSON formatted file.

• config_default (Path) – JSON file as a default configuration for the module.

Returns:

Parsed JSON content.

Return type:

Any

Raises:

• IOError – If an unexpected IO error occurs.

• UnicodeDecodeError – If a decoding error occurs.

• json.decoder.JSONDecodeError – If a JSON decoding error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

async classmethod open_markdown_file(config_file, config_default='')

Reads a markdown file.

Parameters:

• config_file (Path) – Markdown formatted file.

• config_default (str) – Markdown file as a default configuration for the module.

Returns:

File content.

Return type:

Any

Raises:

• IOError – If an unexpected IO error occurs.

• UnicodeDecodeError – If a decoding error occurs.

• json.decoder.JSONDecodeError – If a JSON decoding error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

async classmethod open_env_file(env_path)

Reads an environment (.env) file.

Parameters:

env_path (Path) – Path to the directory containing the .env file.

Returns:

Dictionary of environment variables.

Return type:

dict

Raises:

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

async classmethod open_toml_file(config_file)

Reads a TOML file.

Parameters:

config_file (Path) – TOML formatted file.

Returns:

Parsed TOML content as a dictionary.

Return type:

Any

Raises:

• ImportError – If tomli is not installed for Python < 3.11.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

async classmethod open_ini_file(config_file)

Reads an INI file.

Parameters:

config_file (Path) – INI formatted file.

Returns:

Parsed INI content as a dictionary.

Return type:

Any

Raises:

• ImportError – If configparser is not available.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

async classmethod open_yaml_file(config_file)

Reads a YAML file.

Parameters:

config_file (Path) – YAML formatted file.

Returns:

Parsed YAML content as a dictionary.

Return type:

Any

Raises:

• ImportError – If PyYAML is not installed.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_json_file_sync(config_file, config_default=PosixPath('.'))

Reads a JSON file (synchronous).

Parameters:

• config_file (Path) – JSON formatted file.

• config_default (Path) – JSON file as a default configuration for the module.

Returns:

Parsed JSON content.

Return type:

Any

Raises:

• IOError – If an unexpected IO error occurs.

• UnicodeDecodeError – If a decoding error occurs.

• json.decoder.JSONDecodeError – If a JSON decoding error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_markdown_file_sync(config_file, config_default='')

Reads a markdown file (synchronous).

Parameters:

• config_file (Path) – Markdown formatted file.

• config_default (str) – Markdown file as a default configuration for the module.

Returns:

File content.

Return type:

Any

Raises:

• IOError – If an unexpected IO error occurs.

• UnicodeDecodeError – If a decoding error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_env_file_sync(env_path)

Reads an environment (.env) file (synchronous).

Parameters:

env_path (Path) – Path to the directory containing the .env file.

Returns:

Dictionary of environment variables.

Return type:

Optional[dict]

Raises:

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_toml_file_sync(config_file)

Reads a TOML file (synchronous).

Parameters:

config_file (Path) – TOML formatted file.

Returns:

Parsed TOML content as a dictionary.

Return type:

Optional[Any]

Raises:

• ImportError – If tomli is not installed for Python < 3.11.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_ini_file_sync(config_file)

Reads an INI file (synchronous).

Parameters:

config_file (Path) – INI formatted file.

Returns:

Parsed INI content as a dictionary.

Return type:

Optional[Any]

Raises:

• ImportError – If configparser is not available.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

classmethod open_yaml_file_sync(config_file)

Reads a YAML file (synchronous).

Parameters:

config_file (Path) – YAML formatted file.

Returns:

Parsed YAML content as a dictionary.

Return type:

Optional[Any]

Raises:

• ImportError – If PyYAML is not installed.

• IOError – If an unexpected IO error occurs.

• TypeError – If an unexpected type error occurs.

• FileNotFoundError – If the file is not found.

vyra_base.helper.file_writer module

class vyra_base.helper.file_writer.FileWriter

Bases: object

File writer.

Writes Python objects to a file with both async and sync methods.

async classmethod write_json_file(file, file_content)

Writes a JSON file from a dictionary (asynchronous).

Parameters:

• file (Path) – The path to the file to write.

• file_content (dict) – The dictionary to serialize as JSON.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

async classmethod write_binary_file(file, content)

Writes binary content to a file (asynchronous).

Parameters:

• file (Path) – The path to the file to write.

• content (bytes) – The binary content to write.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

async classmethod write_yaml_file(file, file_content)

Writes a YAML file from a dictionary (asynchronous).

Parameters:

• file (Path) – The path to the file to write.

• file_content (dict) – The dictionary to serialize as YAML.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

classmethod write_json_file_sync(file, file_content)

Writes a JSON file from a dictionary (synchronous).

Parameters:

• file (Path) – The path to the file to write.

• file_content (dict) – The dictionary to serialize as JSON.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

classmethod write_binary_file_sync(file, content)

Writes binary content to a file (synchronous).

Parameters:

• file (Path) – The path to the file to write.

• content (bytes) – The binary content to write.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

classmethod write_yaml_file_sync(file, file_content)

Writes a YAML file from a dictionary (synchronous).

Parameters:

• file (Path) – The path to the file to write.

• file_content (dict) – The dictionary to serialize as YAML.

Returns:

True if writing finished successfully, else False.

Return type:

Optional[bool]

vyra_base.helper.log_filter module

class vyra_base.helper.log_filter.LogFilter(param='')

Bases: Filter

Custom logging filter for filtering log records based on parameters.

Extends Python’s logging.Filter to provide VYRA-specific log filtering capabilities.

Variables:

param – Filter parameter string.

Parameters:

param (str)

param: str

__init__(param='')

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Parameters:

param (str)

Return type:

None

Module contents

Helper Utilities for VYRA Base

Collection of utility functions and helpers for various tasks.

vyra_base.helper.update_ament_prefix_path(workspace_install_dir)

Add workspace install directory to AMENT_PREFIX_PATH.

This makes ROS2 packages in the workspace discoverable to rosidl_runtime_py utilities like get_message(), get_service().

Parameters:

workspace_install_dir (str | Path) – Path to workspace install directory (e.g., “/workspace/install”)

Return type:

bool

Returns:

True if path was added, False if already present

Examples

>>> update_ament_prefix_path("/workspace/install")
True

vyra_base.helper.update_python_path(package_site_packages_dir, prepend=True)

Add package site-packages directory to sys.path for Python imports.

This enables importing generated Python modules (*_pb2.py, ROS2 .py files) from newly built packages.

Parameters:

• package_site_packages_dir (str | Path) – Path to site-packages directory (e.g., “/workspace/install/v2_modulemanager_interfaces/lib/python3.12/site-packages”)

• prepend (bool) – If True, add to front of sys.path (default), else append

Return type:

bool

Returns:

True if path was added, False if already present

Examples

>>> update_python_path("/workspace/install/mypackage/lib/python3.12/site-packages")
True

vyra_base.helper.detect_python_version()

Detect current Python version (e.g., “3.12”).

Return type:

str

Returns:

Python version string in format “X.Y”

vyra_base.helper.find_package_site_packages(workspace_install_dir, package_name, python_version=None)

Find site-packages directory for a package in workspace.

Searches for typical ROS2 package structure: <workspace>/install/<package>/lib/python<version>/site-packages

Parameters:

• workspace_install_dir (str | Path) – Workspace install directory

• package_name (str) – Name of the package

• python_version (Optional[str]) – Python version (e.g., “3.12”), auto-detected if None

Return type:

Optional[Path]

Returns:

Path to site-packages if found, None otherwise

Examples

>>> find_package_site_packages("/workspace/install", "v2_modulemanager_interfaces")
Path('/workspace/install/v2_modulemanager_interfaces/lib/python3.12/site-packages')

vyra_base.helper.ensure_interface_package_discoverable(workspace_install_dir, package_name, python_version=None)

Make an interface package discoverable to running Python process.

Updates both AMENT_PREFIX_PATH (for ROS2 introspection) and sys.path (for Python imports of generated modules).

Parameters:

• workspace_install_dir (str | Path) – Workspace install directory

• package_name (str) – Name of the interface package

• python_version (Optional[str]) – Python version (auto-detected if None)

Return type:

bool

Returns:

True if package is now discoverable, False on failure

Examples

>>> ensure_interface_package_discoverable(
...     "/workspace/install",
...     "v2_modulemanager_interfaces"
... )
True

vyra_base.helper.ensure_workspace_discoverable(workspace_install_dir, python_version=None)

Make all packages in a workspace install directory discoverable.

Scans workspace install directory and updates paths for all found packages.

Parameters:

• workspace_install_dir (str | Path) – Workspace install directory

• python_version (Optional[str]) – Python version (auto-detected if None)

Return type:

int

Returns:

Number of packages made discoverable

Examples

>>> ensure_workspace_discoverable("/workspace/install")
5  # Made 5 packages discoverable

vyra_base.helper.get_current_paths_info()

Get current AMENT_PREFIX_PATH and sys.path for debugging.

Return type:

dict[str, str]

Returns:

Dict with ‘ament_prefix_path’ and ‘sys_path’ entries

class vyra_base.helper.VyraLoggingConfig

Bases: object

Centralized logging configuration for vyra_base library.

This class manages the logging configuration for the entire vyra_base library, providing consistent logging across all modules and components.

classmethod initialize(log_level=None, log_directory=None, log_config=None, enable_console=True, enable_file=True, enable_colors=True, enable_json=False, rate_limit=None)

Initialize the logging configuration.

Parameters:

• log_level (Optional[str]) – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Can also be set via VYRA_LOG_LEVEL environment variable.

• log_directory (Optional[Path]) – Directory where log files will be stored.

• log_config (Optional[Dict[str, Any]]) – Optional custom logging configuration dict.

• enable_console (bool) – Whether to enable console logging.

• enable_file (bool) – Whether to enable file logging.

• enable_colors (bool) – Whether to use colored output (console only).

• enable_json (bool) – Whether to use JSON formatting (file only).

• rate_limit (Optional[float]) – Minimum seconds between duplicate messages (None = disabled).

Return type:

None

classmethod get_logger(name)

Get a logger instance.

Ensures logging is initialized before returning the logger.

Parameters:

name (str) – Logger name (typically __name__ of the module).

Return type:

Logger

Returns:

Logger instance.

classmethod set_level(level, logger_name=None)

Dynamically change logging level.

Parameters:

• level (str) – New logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL).

• logger_name (Optional[str]) – Specific logger to update, or None for root vyra_base logger.

Return type:

None

classmethod add_handler_to_logger(logger_name, handler)

Add a custom handler to a specific logger.

This is useful for adding additional logging outputs (e.g., to external systems).

Parameters:

• logger_name (str) – Name of the logger to add handler to.

• handler (Handler) – Handler instance to add.

Return type:

None

classmethod reset()

Reset the logging configuration.

This is mainly useful for testing.

Return type:

None

vyra_base.helper.get_logger(name)

Convenience function to get a logger.

Parameters:

name (str) – Logger name (typically __name__).

Return type:

Logger

Returns:

Logger instance.

Example

>>> logger = get_logger(__name__)
>>> logger.info("Application started")

vyra_base.helper.deep_merge(d1, d2)

Recursively merges two dictionaries, combining nested dictionaries. If a key exists in both dictionaries and both values are dictionaries, they are merged recursively. Otherwise, the value from the second dictionary overwrites the value from the first dictionary. :type d1: dict :param d1: The first dictionary. :type d2: dict :param d2: The second dictionary.

Return type:

dict

Returns:

A new dictionary that is the result of merging d1 and d2.

Parameters:

• d1 (dict)

• d2 (dict)

vyra_base.helper.fuzzy_match(word, possibilities, n=3, cutoff=0.6)

Find the closest matches to word from a list of possibilities.

Uses difflib.get_close_matches() under the hood, providing a convenient wrapper with sensible defaults for industrial automation diagnostics (e.g. suggesting valid interface names when a lookup fails).

Parameters:

• word (str) – The string to match against the possibilities.

• possibilities (Sequence[str]) – The pool of candidate strings to search in.

• n (int) – Maximum number of close matches to return. Defaults to 3.

• cutoff (float) – Minimum similarity score in the range [0, 1]. Strings with a ratio below this threshold are excluded. Defaults to 0.6.

Returns:

A list of the best matches (may be empty if nothing is close enough), sorted by decreasing similarity.

Return type:

list[str]

Usage example:

from vyra_base.helper.func import fuzzy_match

candidates = ["state_feeder", "news_feeder", "error_feeder"]
suggestions = fuzzy_match("stae_feeder", candidates)
# → ["state_feeder"]