luxonis_ml.utils

package documentation

Shared utility entry point for Luxonis ML internals and integrations.

The luxonis_ml.utils package gathers cross-cutting helpers used throughout the data, tracker, telemetry, and archive modules. It includes configuration loading, environment-backed settings, a local/remote filesystem abstraction, graph traversal helpers, logging/deprecation utilities, and registry primitives for plugin-style extension points.

Example

Access the shared filesystem abstraction from the package import.

from luxonis_ml.utils import LuxonisFileSystem

fs = LuxonisFileSystem("file:///tmp/dataset")
files = fs.walk_dir(".")

Note

Some utilities rely on optional dependencies. Install luxonis-ml[utils] when using filesystem, config, and logging helpers outside environments that already provide those packages.

See Also

luxonis_ml.utils.config, luxonis_ml.utils.environ, luxonis_ml.utils.filesystem, and luxonis_ml.utils.registry for the implementation-level contracts.

Module	`config`	Undocumented
Module	`environ`	No module docstring; 0/2 variable, 1/1 function, 1/2 class documented
Module	`filesystem`	No module docstring; 1/3 function, 2/2 classes documented
Module	`graph`	Undocumented
Module	`logging`	Undocumented
Module	`path`	No module docstring; 3/3 functions documented
Module	`registry`	Undocumented
Module	`__main__`	No module docstring; 0/1 variable, 4/4 functions documented

From __init__.py:

Class	`AutoRegisterMeta`	Metaclass for automatically registering modules.
Class	`BaseModelExtraForbid`	Base model with extra fields forbidden.
Class	`LuxonisConfig`	Class for storing configuration.
Class	`LuxonisFileSystem`	An abstraction over remote and local sources.
Class	`Registry`	A registry to store and retrieve modules.
Function	`deprecated`	Mark a function or its parameters as deprecated.
Function	`is_acyclic`	Test if graph is acyclic.
Function	`log_once`	Log a message only once.
Function	`setup_logging`	Set up global logging using loguru and rich.
Function	`traverse_graph`	Traverse the graph in topological order, starting from the nodes with no predecessors.
Constant	`PUT_FILE_REGISTRY`	Undocumented

def deprecated(*args: str, suggest: dict[str, str] | None = None, additional_message: str | None = None, altogether: bool = False) -> Callable[[Callable], Callable]: ¶

Mark a function or its parameters as deprecated.

Example

decorator = deprecated(
    "old_arg",
    "another_old_arg",
    suggest={"old_arg": "new_arg"},
    additional_message="Usage of 'old_arg' is discouraged.",
)

def my_func(old_arg, another_old_arg, new_arg=None):
    pass

my_func = decorator(my_func)
my_func("foo")

# DeprecationWarning: Argument 'old_arg'
# in function `my_func` is deprecated and
# will be removed in future versions.
# Use 'new_arg' instead.
# Usage of 'old_arg' is discouraged.

Parameters
*args:`str`	Names of the deprecated parameters.
suggest:`dict[str, str] \| None`	Suggested replacement parameters.
additional_message:`str \| None`	Additional message to append to the warning message.
altogether:`bool`	If True, marks the whole function as deprecated.

Returns
`Callable[[Callable], Callable]`	Decorator that emits deprecation warnings.

def is_acyclic(graph: dict[str, list[str]]) -> bool: ¶

Test if graph is acyclic.

Examples

>>> is_acyclic({"a": ["b"], "b": []})
True
>>> is_acyclic({"a": ["b"], "b": ["a"]})
False
>>> is_acyclic({})
True

Parameters
graph:`dict[str, list[str]]`	Graph represented as a dictionary of predecessors.

Returns
`bool`	`True` if graph is acyclic, `False` otherwise.

def log_once(logger: Callable[[str], None], message: str): ¶

Log a message only once.

Parameters
logger:`Callable[[str], None]`	The logger to use.
message:`str`	The message to log.

def setup_logging(*, level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = None, file: PathType | None = None, use_rich: bool = True, **kwargs): ¶

Set up global logging using loguru and rich.

Parameters
level:`Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] \| None`	Logging level. If not set, reads from the environment variable `LOG_LEVEL`.
file:`PathType \| None`	Path to the log file. If provided, logs will be saved to this file.
use_rich:`bool`	If True, uses rich for logging.
**kwargs	Additional keyword arguments to pass to `RichHandler`.

Raises
`ValueError`	If the logging level is invalid.

def traverse_graph(graph: dict[str, list[str]], nodes: Mapping[str, T]) -> Iterator[tuple[str, T, list[str], list[str]]]: ¶

Traverse the graph in topological order, starting from the nodes with no predecessors.

Example

>>> graph = {"a": ["b"], "b": []}
>>> nodes = {"a": 1, "b": 2}
>>> for name, value, preds, rem in traverse_graph(graph, nodes):
...     print(name, value, preds, rem)
b 2 [] ['a']
a 1 ['b'] []

Parameters
graph:`dict[str, list[str]]`	Graph represented as a dictionary of predecessors. Keys are node names, values are node predecessors.
nodes:`Mapping[str, T]`	Dictionary mapping node names to values.

Returns
`Iterator[tuple[str, T, list[str], list[str]]]`	Undocumented

Yields
Tuples containing node name, node value, node predecessors, and remaining unprocessed nodes.

Raises
`RuntimeError`	If the graph is malformed.

PUT_FILE_REGISTRY: Registry[PutFile] = ¶

Undocumented

Value

Registry(name='put_file')