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][source]¶
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)[source]¶
Define the arguments for the argparse parser.
- Parameters:
parser – The argparse parser.
- inet_nm.check.check_filter_args(parser: ArgumentParser)[source]¶
- 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][source]¶
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)[source]¶
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][source]¶
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][source]¶
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][source]¶
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_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][source]¶
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], List[NmNode]][source]¶
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, the third list is the missing nodes, the forth is the total node count.
- inet_nm.check.get_nodes_with_state(nodes: List[NmNode], connected=True) List[NmNode][source]¶
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.cli_check module¶
inet_nm.cli_commission module¶
inet_nm.cli_decommission module¶
inet_nm.cli_exec module¶
inet_nm.cli_export module¶
inet_nm.cli_fake_usb module¶
inet_nm.cli_free module¶
inet_nm.cli_inventory module¶
inet_nm.cli_set_location module¶
inet_nm.cli_show_location module¶
inet_nm.cli_tmux module¶
inet_nm.cli_tty_from_uid module¶
inet_nm.cli_update_cache module¶
inet_nm.cli_update_from_os module¶
inet_nm.commissioner module¶
This module commissions USB boards.
- inet_nm.commissioner.add_node_to_nodes(nodes: List[NmNode], node: NmNode) List[NmNode][source]¶
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.is_uninitialized_sn(node: NmNode, sns: List = None)[source]¶
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.mock_device()[source]¶
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][source]¶
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[source]¶
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.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.BoardInfoCommitHash(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass for handling the board info configuration hash.
This is just the git commit used for the board info generation.
- class inet_nm.config.BoardInfoConfig(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass 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
- class inet_nm.config.EnvConfig(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass 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[source]¶
Load the env configuration.
- Returns:
The loaded env data.
- save(data: EnvConfigFormat)[source]¶
Save the env configuration.
- Parameters:
data – The env data to save.
- class inet_nm.config.LocationCache(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass for handling the location cache.
- Parameters:
config_dir – Directory for the configuration files.
- file_path¶
Path to the env configuration file.
- Type:
Path
- class inet_nm.config.LocationConfig(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass for handling the location mapping.
- Parameters:
config_dir – Directory for the configuration files.
- file_path¶
Path to the env configuration file.
- Type:
Path
- class inet_nm.config.NodesConfig(config_dir: Path | str)[source]¶
Bases:
_ConfigFileClass 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
- inet_nm.config.check_commit_hash(config_dir: Path | str) bool[source]¶
Check if the commit hash is the same as the current commit hash.
- Parameters:
config_dir – Directory for the configuration files.
- Returns:
True if the commit hash is the same as the current commit hash, False otherwise.
- inet_nm.config.config_arg(parser: ArgumentParser)[source]¶
Add a configuration argument to the provided parser.
- Parameters:
parser (argparse.ArgumentParser) – ArgumentParser object.
inet_nm.data_types module¶
This module contains the data types used by the inet-nm module.
- class inet_nm.data_types.DictSerializable[source]¶
Bases:
objectMixin class to provide to_dict and from_dict methods.
- class inet_nm.data_types.EnvConfigFormat(shared: Dict[str, str], nodes: Dict[str, Dict[str, str]], patterns: List[Dict])[source]¶
Bases:
DictSerializableA representation of the environment configuration file.
- patterns¶
List of patterns to match against the environment variables.
- Type:
List[Dict]
- 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)[source]¶
Bases:
DictSerializableA representation of an embedded dev board connected via USB.
inet_nm.fake_usb module¶
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)[source]¶
Bases:
objectProvides a context manager-based file lock mechanism.
- file_name¶
The name of the file to be used as the lock.
- acquire(timeout: int = None, poll_interval: float = 0.05) None[source]¶
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.
inet_nm.graph module¶
Graph module for the inet_nm package.
This module provides a simple implementation of a graph data structure.
- inet_nm.graph.filter_valid_locations(locations)[source]¶
Filter a list of locations into valid and invalid locations.
- Parameters:
locations – List of location strings.
- Returns:
A tuple of valid and invalid locations.
Examples
locations: [‘1.1.1’, ‘3.1.2’, ‘1.3.4’, ‘garbage’] valid_locs: [(1, 1, 1), (3, 1, 2), (1, 3, 4)] invalid_locs: [‘garbage’]
- inet_nm.graph.parse_locations(locations)[source]¶
Parse a list of locations into a grid string.
- Parameters:
locations – List of locations to parse.
- Returns:
A grid string representing the locations.
Examples
locations: [‘1.1.1’, ‘3.1.2’, ‘1.3.4’] ┌─┬─┬─┐ │ │ │ │ │ │ │ │ │x│ │ │ │ │ │ │ ┼─┼─┼─┼ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ┼─┼─┼─┼ │ │ │x│ │ │ │ │ │ │ │ │ │x│ │ │ └─┴─┴─┘
inet_nm.location module¶
- inet_nm.location.get_location_cache(nodes: List[NmNode], id_paths: Dict)[source]¶
Get the location cache for a list of NmNode objects.
- Parameters:
nodes – List of NmNode objects.
id_paths – List of id_paths to check.
- Returns:
The location cache.
- inet_nm.location.merge_location_cache_chunks(caches: List[List[Dict]])[source]¶
Merge location cache chunks into a single cache.
Due to only being able to power on a chunk at a time we need to sort through each of the location caches and look through all id_paths that have a missing state and see if they are available in another chunk. If they are then probably they were just powered off.
- Parameters:
caches – List of location cache chunks.
- Returns:
The merged location cache.
inet_nm.locking module¶
This module contains functions for locking nodes.
- inet_nm.locking.get_lock_path(node: NmNode) Path[source]¶
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][source]¶
Get the list of UIDs of currently locked nodes.
- Returns:
A sorted list of UIDs of locked nodes.
inet_nm.power_control module¶
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)[source]¶
Bases:
NmNodesRunnerRuns 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])[source]¶
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()[source]¶
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)[source]¶
Bases:
NmNodesRunnerBase class for tmux runners.
This class inherits from NmNodesRunner and sets up a tmux session.
- SETUP_WAIT = 0.2¶
- class inet_nm.runner_apps.NmTmuxPanedRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)[source]¶
Bases:
NmTmuxBaseRunnerRun tmux with paned windows.
This class inherits from NmTmuxBaseRunner and sets up a tmux session with panes.
- class inet_nm.runner_apps.NmTmuxWindowedRunner(nodes: List[NmNode], default_timeout: int = None, seq=False, force=False, extra_env: EnvConfigFormat = None)[source]¶
Bases:
NmTmuxBaseRunnerRuns tmux with individual windows for each node.
This class inherits from NmTmuxBaseRunner and sets up a tmux session with individual windows for each node.
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)[source]¶
Bases:
objectClass 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)[source]¶
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.
inet_nm.runner_helper module¶
inet_nm.usb_ctrl module¶
- exception inet_nm.usb_ctrl.TtyNotPresent[source]¶
Bases:
ExceptionException to be raised when a TTY device is not found for a given NmNode.
- inet_nm.usb_ctrl.get_connected_id_paths() Set[str][source]¶
Get the ID_PATHs of all connected USB devices.
- Returns:
A list of ID_PATHs of all connected USB devices.
- inet_nm.usb_ctrl.get_connected_uids() List[str][source]¶
Get the UIDs of all connected USB devices.
- Returns:
A list of UIDs of all connected USB devices.
- inet_nm.usb_ctrl.get_devices_from_tty(saved_nodes: List[NmNode] | None = None) List[NmNode][source]¶
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.usb_ctrl.get_id_path_from_node(node: NmNode) str[source]¶
Get the ID_PATH of a connected USB device.
- Parameters:
node – The node to get the ID_PATH for.
- Returns:
The ID_PATH of the connected USB device.
- inet_nm.usb_ctrl.get_tty_from_nm_node(nm_node: NmNode) str[source]¶
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.usb_ctrl.get_ttys_from_nm_node(nm_node: NmNode) List[str][source]¶
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.usb_ctrl.get_uid_from_id_path(id_path: str) str[source]¶
Get the UID of a connected USB device.
- Parameters:
id_path – The ID_PATH of the connected USB device.
- Returns:
The UID of the connected USB device.
- Raises:
Exception – If the node is not found, maybe not connected.
- inet_nm.usb_ctrl.get_usb_info_from_node(node: NmNode)[source]¶
Read the USB information from a node.
- Parameters:
node – The node to read the USB information from.
- Returns:
A tuple containing the ID_PATH, hub, and port of the connected USB device.
- Raises:
Exception – If the node is not found, maybe not connected.