inet_nm package

Submodules

inet_nm.check module

This module contains functions for checking the state of the boards.

This is meant for evaluating inventory.

inet_nm.check.all_features_from_nodes(nodes: List[NmNode]) → List[str]

Get a list of all features provided by the nodes.

Parameters:

nodes – A list of nodes.

Returns:

A list of all features.

inet_nm.check.check_args(parser: ArgumentParser)

Define the arguments for the argparse parser.

Parameters:

parser – The argparse parser.

inet_nm.check.check_filter_args(parser: ArgumentParser)

inet_nm.check.check_nodes(nodes: List[NmNode], all_nodes: bool = False, missing: bool = False, feat_filter: List[str] = None, feat_eval: str = None, used: bool = False, skip_dups: bool = False, only_used: bool = False, boards: List[str] = None, uids: List[str] = None, locked_nodes: List[str] = None) → List[NmNode]

Get a filtered list of nodes based on the provided parameters.

Parameters:

• nodes – A list of nodes to check.

• all_nodes – If True, all nodes will be returned.

• missing – If True, only the nodes that are missing will be returned.

• feat_filter – A list of features. Only the nodes that contain all of these features will be returned.

• feat_eval – The function to use to evaluate the features.

• used – If True, used nodes will also be returned.

• skip_dups – If True, duplicate nodes will be removed.

• only_used – If True, only the used nodes will be returned.

• boards – A list of boards to use.

• uids – A list of UIDs of nodes to use.

• locked_nodes – A list of UIDs of nodes that are locked.

Returns:

A list of filtered nodes.

inet_nm.check.eval_features(nodes: List[NmNode], features: set, eval_func)

Evaluate features of nodes using a provided function.

Parameters:

• nodes – A list of nodes to evaluate.

• features – A set of features to evaluate.

• eval_func – The function to use to evaluate the features.

Returns:

A list of nodes that meet the criteria of the evaluation function.

Raises:

ValueError – If the evaluation function cannot be evaluated.

inet_nm.check.filter_nodes(nodes: List[NmNode], must_contain: List[str]) → List[NmNode]

Filter the nodes based on the provided features.

Parameters:

• nodes – A list of nodes to filter.

• must_contain – A list of features. Only the nodes that contain all of these features will be returned.

Returns:

A list of filtered nodes.

inet_nm.check.filter_used_nodes(nodes: List[NmNode], used_uids: List[str], remove=True) → List[NmNode]

Filter the nodes based on the provided used UIDs.

Parameters:

• nodes – A list of nodes to filter.

• used_uids – A list of UIDs of nodes that are already used.

• remove – If True, remove the used nodes. If False, remove the unused nodes.

Returns:

A list of nodes that are not already used.

Return type:

List[NmNode]

inet_nm.check.get_all_features(nodes: List[NmNode]) → List[str]

Get a list of all features provided by the nodes.

Parameters:

nodes – A list of nodes.

Returns:

A list of all features.

inet_nm.check.get_connected_uids() → List[str]

Get the UIDs of all connected USB devices.

Returns:

A list of UIDs of all connected USB devices.

inet_nm.check.get_filtered_nodes(config: str, all_nodes: bool = False, missing: bool = False, feat_filter: List[str] = None, feat_eval: str = None, used: bool = False, skip_dups: bool = False, only_used: bool = False, boards: List[str] = None, uids: List[str] = None) → List[NmNode]

Get a list of nodes based on the provided parameters.

Parameters:

• config – The configuration path for the nodes.

• all_nodes – If True, all nodes will be returned.

• missing – If True, only the nodes that are missing will be returned.

• feat_filter – A list of features. Only the nodes that contain all of these features will be returned.

• feat_eval – The function to use to evaluate the features.

• used – If True, used nodes will also be returned.

• skip_dups – If True, duplicate nodes will be removed.

• only_used – If True, only the used nodes will be returned.

• boards – A list of boards to use.

• uids – A list of UIDs of nodes to use.

Returns:

A list of filtered nodes.

inet_nm.check.get_inventory_nodes(config: str, feat_filter: List[str] = None, feat_eval: str = None, boards: List[str] = None, uids: List[str] = None) → Tuple[List[NmNode], List[NmNode], List[NmNode]]

Get a list of nodes based on the provided parameters.

Parameters:

• config – The configuration path for the nodes.

• feat_filter – A list of features. Only the nodes that contain all of these features will be returned.

• feat_eval – The function to use to evaluate the features.

• boards – A list of boards to use.

• uids – A list of UIDs of nodes to use.

Returns:

A tuple of lists of nodes. The first list is the available nodes, the

second list is the used nodes, and the third list is the missing nodes.

inet_nm.check.get_nodes_with_state(nodes: List[NmNode], connected=True) → List[NmNode]

Get a list of nodes that are connected or not connected.

Parameters:

• nodes – A list of nodes to filter.

• connected – If True, return the nodes that are connected. If False, return the nodes that are not connected.

Returns:

A list of filtered nodes.

inet_nm.check.nodes_to_boards(nodes: List[NmNode]) → Dict[str, int]

Convert a list of nodes to a dictionary of board counts.

Parameters:

nodes – A list of nodes.

Returns:

A dictionary where the keys are the names of the boards and the values

are the counts of those boards.

inet_nm.check.skip_duplicate_boards(nodes: List[NmNode]) → List[NmNode]

Remove duplicate boards from a list of nodes.

Parameters:

nodes – A list of nodes.

Returns:

A list of nodes without duplicates.

inet_nm.cli_check module

inet_nm.cli_check.main()

CLI entrypoint for counting the inventory of boards.

inet_nm.cli_commission module

inet_nm.cli_commission.main()

CLI to commission a USB board.

inet_nm.cli_decommission module

inet_nm.cli_decommission.main()

CLI to commission a USB board.

inet_nm.cli_exec module

inet_nm.cli_exec.main()

CLI entrypoint for executing a command on each node.

inet_nm.cli_export module

inet_nm.cli_export.main()

CLI entrypoint for entering env vars.

inet_nm.cli_free module

inet_nm.cli_free.main()

Release all locks by deleting all lock files and print the process.

inet_nm.cli_inventory module

inet_nm.cli_inventory.main()

CLI entrypoint for counting the inventory of boards.

inet_nm.cli_tmux module

inet_nm.cli_tmux.main()

CLI entrypoint for starting tmux sessions.

inet_nm.cli_tty_from_uid module

inet_nm.cli_tty_from_uid.main()

CLI to get the TTY device node string for a given NmNode UID.

inet_nm.cli_update_commissioned module

inet_nm.cli_update_commissioned.main()

Go through the list of commissioned nodes and update features.

inet_nm.cli_update_from_os module

inet_nm.cli_update_from_os.main()

CLI entrypoint for updating the board info from the OS.

inet_nm.commissioner module

This module commissions USB boards.

exception inet_nm.commissioner.TtyNotPresent

Bases: Exception

Exception to be raised when a TTY device is not found for a given NmNode.

inet_nm.commissioner.add_node_to_nodes(nodes: List[NmNode], node: NmNode) → List[NmNode]

Add a new NmNode to a list of existing NmNode objects.

Replaces any existing NmNode with the same UID.

Parameters:

• nodes – List of existing NmNode objects.

• node – New NmNode to be added.

Returns:

Updated list of NmNode objects.

inet_nm.commissioner.check_and_set_uninitialized_sn(node: NmNode, sns: List = None)

Check if a given NmNode has an uninitialized serial number and prompt the user to set it.

Parameters:

• node – An NmNode object.

• sns – List of serial numbers to check against.

inet_nm.commissioner.get_devices_from_tty(saved_nodes: List[NmNode] | None = None) → List[NmNode]

Retrieve connected TTY devices as a list of NmNode objects.

Parameters:

saved_nodes – List of previously saved NmNode objects.

Returns:

List of NmNode objects representing connected TTY devices.

Raises:

TtyNotPresent – If a TTY device could not be found for a given NmNode.

inet_nm.commissioner.get_tty_from_nm_node(nm_node: NmNode) → str

Retrieve the TTY device node string for a given NmNode.

Parameters:

nm_node – An NmNode object.

Returns:

The TTY device node string.

Raises:

TtyNotPresent – If a TTY device could not be found for the given NmNode.

inet_nm.commissioner.get_ttys_from_nm_node(nm_node: NmNode) → List[str]

Retrieve the TTY device node string for a given NmNode.

Parameters:

nm_node – An NmNode object.

Returns:

The TTY device node string.

Raises:

TtyNotPresent – If a TTY device could not be found for the given NmNode.

inet_nm.commissioner.mock_device()

Mock a device, useful for boards that do not have USB interfaces

inet_nm.commissioner.remove_node_to_nodes(nodes: List[NmNode], node: NmNode) → List[NmNode]

Remove an NmNode from a list of existing NmNode objects.

Parameters:

• nodes – List of existing NmNode objects.

• node – NmNode to be removed.

Returns:

Updated list of NmNode objects.

inet_nm.commissioner.select_available_node(nodes: List[NmNode]) → NmNode

Prompt the user to select an available node from a list of NmNode objects.

Parameters:

nodes – List of available NmNode objects.

Returns:

The selected NmNode.

inet_nm.commissioner.select_board(board_names: List[str], node: NmNode) → str

Prompts the user to select a board for a given NmNode.

Parameters:

• board_names – List of available board names.

• node – NmNode for which to select a board.

Returns:

The selected board name.

inet_nm.config module

Provide utilities for interacting with the inet_nm configuration.

It includes functionality for saving and loading information about the boards and nodes, as well as for handling the path configuration.

class inet_nm.config.BoardInfoConfig(config_dir: Path | str)

Bases: _ConfigFile

Class for handling the board info configuration.

The board info configuration is a JSON file containing a dictionary with the board name as key and a list of features provided by the board as value.

The board info configuration file is located in the config directory and is named board_info.json.

Parameters:

config_dir – Directory for the configuration files.

file_path

Path to the board info configuration file.

Type:

Path

load() → Dict[str, str | int]

Load the board info configuration.

Returns:

The loaded board info data.

save(data: Dict[str, str | int])

Save the board info configuration.

Parameters:

data – The board info data to save.

class inet_nm.config.EnvConfig(config_dir: Path | str)

Bases: _ConfigFile

Class for handling the environment configuration.

The env configuration is a JSON file containing a env vars.

The env configuration file is located in the config directory and is named env.json.

Parameters:

config_dir – Directory for the configuration files.

file_path

Path to the env configuration file.

Type:

Path

load() → EnvConfigFormat

Load the env configuration.

Returns:

The loaded env data.

save(data: EnvConfigFormat)

Save the env configuration.

Parameters:

data – The env data to save.

class inet_nm.config.NodesConfig(config_dir: Path | str)

Bases: _ConfigFile

Class for handling the nodes configuration.

The nodes configuration is a JSON file containing a list of NmNode objects.

The nodes configuration file is located in the config directory and is named nodes.json.

Parameters:

config_dir – Directory for the configuration files.

file_path

Path to the nodes configuration file.

Type:

Path

load() → List[NmNode]

Load the nodes configuration.

Returns:

The loaded nodes data.

save(data: List[NmNode])

Save the nodes configuration.

Parameters:

data – The nodes data to save.

inet_nm.config.config_arg(parser: ArgumentParser)

Add a configuration argument to the provided parser.

Parameters:

parser (argparse.ArgumentParser) – ArgumentParser object.

inet_nm.config.get_default_path() → Path

Return the default path for the configuration files.

Returns:

The default path for the configuration files.

Return type:

Path

inet_nm.data_types module

This module contains the data types used by the inet-nm module.

class inet_nm.data_types.DictSerializable

Bases: object

Mixin class to provide to_dict and from_dict methods.

classmethod from_dict(data: Dict[str, Any]) → DictSerializable

Create an instance from a dictionary.

Parameters:

data – Dictionary with data to populate the instance.

Returns:

An instance of the class that inherits this mixin.

to_dict() → Dict[str, Any]

Convert the instance to a dictionary.

Returns:

Dictionary representation of the instance.

class inet_nm.data_types.EnvConfigFormat(shared: Dict[str, str], nodes: Dict[str, Dict[str, str]], patterns: List[Dict])

Bases: DictSerializable

A representation of the environment configuration file.

shared

Shared environment variables.

Type:

Dict[str, str]

nodes

Environment variables for each node.

Type:

Dict[str, Dict[str, str]]

patterns

List of patterns to match against the environment variables.

Type:

List[Dict]

nodes: Dict[str, Dict[str, str]]

patterns: List[Dict]

shared: Dict[str, str]

class inet_nm.data_types.NmNode(serial: str, vendor_id: str, product_id: str, vendor: str, driver: str, model: str | None = None, uid: str | None = None, board: str | None = None, features_provided: List[str] | None = None, mock: bool = False, ignore: bool = False)

Bases: DictSerializable

A representation of an embedded dev board connected via USB.

serial

Serial number of the device.

Type:

str

vendor_id

Vendor ID of the device.

Type:

str

product_id

Product ID of the device.

Type:

str

vendor

Name of the vendor.

Type:

str

driver

Name of the driver.

Type:

str

model

Model name of the device.

Type:

str | None

uid

Unique ID of the device

Type:

str | None

board

Name of the board.

Type:

str | None

features_provided

List of features provided by the device.

Type:

List[str] | None

mock

Whether the node is a mock node.

Type:

bool

ignore

Whether the node should be ignored.

Type:

bool

board: str | None = None

static calculate_uid(product_id: str, vendor_id: str, serial: str) → str

Calculate a unique ID for a device.

Parameters:

• product_id – Product ID of the device.

• vendor_id – Vendor ID of the device.

• serial – Serial number of the device.

Returns:

Calculated unique ID.

driver: str

features_provided: List[str] | None = None

ignore: bool = False

mock: bool = False

model: str | None = None

product_id: str

serial: str

uid: str | None = None

vendor: str

vendor_id: str

class inet_nm.data_types.NodeEnv(NM_IDX: int, NM_UID: str, NM_SERIAL: str, NM_BOARD: str, NM_PORT: str)

Bases: DictSerializable

Environment variables for a node.

NM_IDX

Index of the node.

Type:

int

NM_UID

Unique ID of the node.

Type:

str

NM_SERIAL

Serial number of the node.

Type:

str

NM_BOARD

Board of the node.

Type:

str

NM_PORT

Port of the node.

Type:

str

NM_BOARD: str

NM_IDX: int

NM_PORT: str

NM_SERIAL: str

NM_UID: str

to_dict() → Dict[str, str]

Convert the instance to a dictionary.

Returns:

Dictionary representation of the instance.

Return type:

Dict

inet_nm.filelock module

This module provides a simple implementation of a file-based locking mechanism.

The main class FileLock uses OS-level file system operations for locking and unlocking. This kind of lock can be used to prevent the simultaneous execution of a piece of code by different processes.

class inet_nm.filelock.FileLock(file_name: str, timeout: int = 10)

Bases: object

Provides a context manager-based file lock mechanism.

file_name

The name of the file to be used as the lock.

timeout

The maximum time to wait for the lock to be released.

Type:

int

acquire(timeout: int = None, poll_interval: float = 0.05) → None

Acquire the file lock.

If the lock is currently in use, wait until it is released, or until the specified timeout has passed.

Parameters:

• timeout – The maximum time to wait for the lock to be released. Default is None, which means use self.timeout.

• poll_interval – Time interval to check the lock file’s existence.

Raises:

FileLockTimeout – If the timeout has passed and the lock has not been released.

property is_locked: bool

Check if the file lock is locked.

Returns:

True if the lock file exists, False otherwise.

Return type:

bool

release(force: bool = False) → None

Release the file lock.

Parameters:

force – If True, force release the lock even if the lock is held by others.

exception inet_nm.filelock.FileLockTimeout(message='Timeout occurred.')

Bases: Exception

Exception raised when a file lock operation fails.

message -- explanation of the error

inet_nm.locking module

This module contains functions for locking nodes.

inet_nm.locking.get_lock_path(node: NmNode) → Path

Get the path to the lock file for a given node.

Parameters:

node – The node for which to get the lock file path.

Returns:

The path to the lock file for the node.

inet_nm.locking.get_locked_uids() → List[str]

Get the list of UIDs of currently locked nodes.

Returns:

A sorted list of UIDs of locked nodes.

inet_nm.locking.locks_dir() → Path

Get the directory for lock files.

If the directory does not exist, create it.

Returns:

The path to the lock files directory.

inet_nm.locking.release_all_locks()

Release all locks by deleting all lock files.

inet_nm.runner_apps module

This module includes classes that are used to run applications on nodes.

Classes:

NmShellRunner: Runs shell commands on nodes. NmTmuxPanedRunner: Runs tmux with paned windows. NmTmuxWindowedRunner: Runs tmux with individual windows for each node.

class inet_nm.runner_apps.NmShellRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)

Bases: NmNodesRunner

Runs shell commands on nodes.

This class inherits from NmNodesRunner and overrides the func method to execute shell commands on nodes.

cmd

Command to execute on nodes.

SETUP_WAIT = 0.1

cmd = 'echo $NM_IDX'

func(node: NmNode, idx: int, env: Dict[str, str])

Execute shell commands on nodes.

Parameters:

• node – Node to run the command on.

• idx – Index of the node.

• env – Environment variables for the command.

json_filter = False

output_filter = None

post()

Run after the operations on nodes have completed.

It prints the results of the commands executed on nodes.

results = []

class inet_nm.runner_apps.NmTmuxBaseRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)

Bases: NmNodesRunner

Base class for tmux runners.

This class inherits from NmNodesRunner and sets up a tmux session.

SETUP_WAIT = 0.2

cmd: str = None

post()

Run after the operations on nodes have completed.

It attaches to the tmux session and keeps checking if the session is still active.

session_name: str = 'default'

class inet_nm.runner_apps.NmTmuxPanedRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)

Bases: NmTmuxBaseRunner

Run tmux with paned windows.

This class inherits from NmTmuxBaseRunner and sets up a tmux session with panes.

func(node: NmNode, idx: int, env: Dict[str, str])

Set up a tmux session with paned windows.

Parameters:

• node – Node to run the command on.

• idx – Index of the node.

• env – Environment variables for the command.

class inet_nm.runner_apps.NmTmuxWindowedRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)

Bases: NmTmuxBaseRunner

Runs tmux with individual windows for each node.

This class inherits from NmTmuxBaseRunner and sets up a tmux session with individual windows for each node.

func(node: NmNode, idx: int, env: Dict[str, str])

Sets up a tmux session with individual windows for each node.

Parameters:

• node – Node to run the command on.

• idx – Index of the node.

• env – Environment variables for the command.

inet_nm.runner_base module

Runs applications on nodes.

This module provides the NmNodesRunner class which handles running operations on multiple nodes. An operation is defined by a method func which is to be implemented in the subclass.

class inet_nm.runner_base.NmNodesRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)

Bases: object

Class to handle running operations on nodes.

The class manages locking/unlocking nodes, running operations on each node concurrently in separate threads, and handles cleanup after operations have completed.

The operations to be run are defined by a method func which is to be implemented in the subclass.

acquire(timeout: float = None)

Acquire file locks for all nodes.

This method must be called before running operations on nodes.

Parameters:

timeout (float) – Timeout value for file lock acquisition. If None, default_timeout is used.

func(node: NmNode, idx: int, env: Dict[str, str])

Function to execute.

It must be implemented in the subclass.

Parameters:

• node – The node to run the function on.

• idx – The index of the node in the list of nodes.

• env – A dictionary of environment variables.

post()

Override in the subclass if post-operation steps are needed.

pre()

Override in the subclass if pre-operation steps are needed.

release()

Release all acquired file locks.

run()

Run operations on all nodes.

This method spawns a new thread for each node and runs the operation concurrently on all nodes. The operation to run is defined by the func method.

inet_nm.runner_helper module

inet_nm.runner_helper.do_nothing(signum, frame)

Does nothing.

inet_nm.runner_helper.node_env_vars(config: str)

inet_nm.runner_helper.sanity_check(cmd, **kwargs)

Checks if the command is available and if there are any nodes available.

Parameters:

• cmd (str) – Command to check.

• **kwargs – Keyword arguments to pass to check.get_filtered_nodes.

Returns:

List of NmNode objects.

Module contents