luxonis_ml.utils.LuxonisFileSystem

class documentation

class LuxonisFileSystem:

Constructor: LuxonisFileSystem(path, allow_active_mlflow_run, allow_local, cache_storage, put_file_plugin)

An abstraction over remote and local sources.

This class provides a unified interface for file operations across different storage backends, including local filesystems, S3, GCS, and MLflow artifact storage.

For more flexibility, users can register custom implementations of the put_file method in PUT_FILE_REGISTRY. Its name can then be passed as the put_file_plugin argument when initializing LuxonisFileSystem. This allows for custom upload logic, such as additional processing before upload or integration with other services.

Example

>>> @PUT_FILE_REGISTRY.register()
... def put_file_plugin(*args, **kwargs) -> str:
...     print("Custom put_file called!")
...     return "remote_path"
...
>>> fs = LuxonisFileSystem(
...     "file:///tmp/luxonis-ml",
...     put_file_plugin="put_file_plugin",
... )
>>> remote_path = fs.put_file("local/file.txt", "remote/file.txt")
Custom put_file called!
>>> print(remote_path)
remote_path

Static Method	`download`	Download file or directory from remote storage.
Static Method	`get_protocol`	Extract the detected protocol from a path.
Static Method	`split_full_path`	Split the full path into protocol and absolute path.
Static Method	`upload`	Upload file or directory to remote storage.
Method	`__init__`	Initialize the `LuxonisFileSystem`.
Method	`delete_dir`	Delete a directory and all its contents from remote storage.
Method	`delete_file`	Delete a single file from remote storage.
Method	`delete_files`	Delete multiple files from remote storage.
Method	`exists`	Check whether the given remote path exists.
Method	`get_dir`	Copy many files from remote storage to local storage.
Method	`get_file`	Copy a single file from remote storage.
Method	`get_file_uuid`	Read a file and returns the (unique) UUID generated from file bytes.
Method	`get_file_uuids`	Compute the UUIDs for all files stored in the filesystem.
Method	`init_fsspec_filesystem`	Initialize an `fsspec` filesystem for the configured protocol.
Method	`is_directory`	Check whether the given remote path is a directory.
Method	`put_bytes`	Upload a file to remote storage directly from file bytes.
Method	`put_dir`	Upload files to remote storage.
Method	`put_file`	Copy a single file to remote storage.
Method	`read_text`	Read a file into a string.
Method	`read_to_byte_buffer`	Read a file into a byte buffer.
Method	`walk_dir`	Walk through the individual files in a remote directory.
Instance Variable	`allow_active_mlflow_run`	Whether operations are allowed on the active MLflow run.
Instance Variable	`allow_local`	Whether operations are allowed on the local file system.
Instance Variable	`artifact_path`	MLflow artifact path when using MLflow filesystem.
Instance Variable	`cache_storage`	Path to cache storage, or `None` if no cache is used.
Instance Variable	`experiment_id`	MLflow experiment ID when using MLflow filesystem.
Instance Variable	`fs`	Initialized fsspec filesystem when using FSSPEC.
Instance Variable	`fs_type`	Type of the filesystem, either MLFLOW or FSSPEC.
Instance Variable	`path`	The path component of the input URL, with the protocol stripped.
Instance Variable	`run_id`	MLflow run ID when using MLflow filesystem.
Instance Variable	`tracking_uri`	MLflow tracking URI when using MLflow filesystem.
Instance Variable	`url`	The original input URL.
Property	`full_path`	Full remote path.
Property	`is_fsspec`	Check whether the filesystem uses fsspec.
Property	`is_mlflow`	Check whether the filesystem is an MLflow filesystem.
Property	`protocol`	Returns the protocol of the filesystem.
Static Method	`_split_mlflow_path`	Split an MLflow path into experiment, run, and artifact parts.
Instance Variable	`_allow_active_mlflow_run`	Undocumented
Instance Variable	`_allow_local`	Undocumented
Instance Variable	`_cache_storage`	Undocumented
Instance Variable	`_fs`	Undocumented
Instance Variable	`_fs_type`	Undocumented
Instance Variable	`_is_mlflow_active_run`	Undocumented
Instance Variable	`_path`	Undocumented
Instance Variable	`_protocol`	Undocumented
Instance Variable	`_url`	Undocumented

@staticmethod

def download(url: str, dest: PathType | None) -> Path: ¶

Download file or directory from remote storage.

Intended for downloading a single remote object without needing to create a LuxonisFileSystem instance.

Parameters
url:`str`	URL to the file or directory.
dest:`PathType \| None`	Destination directory. If `None`, the current working directory is used.

Returns
`Path`	Path to the downloaded file or directory.

@staticmethod

def get_protocol(path: str) -> str: ¶

Extract the detected protocol from a path.

Parameters
path:`str`	Path optionally containing the protocol.

Returns
`str`	Detected protocol. Defaults to `"file"` if no protocol is specified.

@staticmethod

def split_full_path(path: PathType) -> tuple[str, str]: ¶

Split the full path into protocol and absolute path.

Parameters
path:`PathType`	Full path optionally containing the protocol.

Returns
`tuple[str, str]`	The used protocol and absolute path.

@staticmethod

def upload(local_path: PathType, url: str): ¶

Upload file or directory to remote storage.

Useful for uploading a single local object without having to create a LuxonisFileSystem instance.

Parameters
local_path:`PathType`	Path to the local file or directory.
url:`str`	URL to the remote file or directory.

@typechecked

def __init__(self, path: str, allow_active_mlflow_run: bool | None = False, allow_local: bool | None = True, cache_storage: str | None = None, put_file_plugin: str | None = None): ¶

Initialize the LuxonisFileSystem.

Parameters
path:`str`	Input path consisting of a protocol and path, or only a path for local files.
allow_active_mlflow_run:`bool \| None`	Whether operations are allowed on the active MLflow run.
allow_local:`bool \| None`	Whether operations are allowed on the local file system.
cache_storage:`str \| None`	Path to cache storage. No cache is used if not set.
put_file_plugin:`str \| None`	Name of a registered function in `PUT_FILE_REGISTRY` to use instead of `LuxonisFileSystem.put_file`. The registered function must conform to the `PutFile` protocol.

Raises
`ValueError`	If the protocol is not supported. If the protocol is `"file"` but local files are not allowed. If the protocol is `"mlflow"` but no MLflow path is specified and using the active MLflow run is not allowed. If the protocol is `"mlflow"` but there is no `"MLFLOW_TRACKING_URI"` in environment variables.

def delete_dir(self, remote_dir: PosixPathType = '', allow_delete_parent: bool = False): ¶

Delete a directory and all its contents from remote storage.

Parameters
remote_dir:`PosixPathType`	Relative path to the remote directory.
allow_delete_parent:`bool`	Whether to allow deleting the parent directory.

Raises
`ValueError`	If no directory is specified and deleting the parent directory is not allowed.
`NotImplementedError`	If using a protocol that is not yet supported.

def delete_file(self, remote_path: PosixPathType): ¶

Delete a single file from remote storage.

Parameters
remote_path:`PosixPathType`	Relative path to the remote file.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def delete_files(self, remote_paths: list[PosixPathType]): ¶

Delete multiple files from remote storage.

Parameters
remote_paths:`list[PosixPathType]`	Relative paths to remote files.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def exists(self, remote_path: PosixPathType = '') -> bool: ¶

Check whether the given remote path exists.

Parameters
remote_path:`PosixPathType`	Relative path to the remote file. Defaults to an empty string, which represents the root path of the filesystem.

Returns
`bool`	`True` if the path exists, `False` otherwise.

def get_dir(self, remote_paths: PosixPathType | Iterable[PosixPathType], local_dir: PathType, mlflow_instance: ModuleType | None = None) -> Path: ¶

Copy many files from remote storage to local storage.

Parameters
remote_paths:`PosixPathType \| Iterable[PosixPathType]`	Either a path specifying a directory to walk, or an iterable of files that may be in different directories.
local_dir:`PathType`	Path to the local directory.
mlflow_instance:`ModuleType \| None`	Currently unused. MLflow downloads through this method are not implemented.

Returns
`Path`	Path to the downloaded directory.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def get_file(self, remote_path: PosixPathType, local_path: PathType, mlflow_instance: ModuleType | None = None) -> Path: ¶

Copy a single file from remote storage.

Parameters
remote_path:`PosixPathType`	Relative path to the remote file.
local_path:`PathType`	Path to the local file.
mlflow_instance:`ModuleType \| None`	Currently unused. MLflow downloads through this method are not implemented.

Returns
`Path`	Path to the downloaded file.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def get_file_uuid(self, path: PathType, local: bool = False) -> str: ¶

Read a file and returns the (unique) UUID generated from file bytes.

Parameters
path:`PathType`	Relative path to the remote file, or a local path when `local` is `True`.
local:`bool`	Specifies a local path as opposed to a remote path.

Returns
`str`	UUID generated from the file bytes.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def get_file_uuids(self, paths: Iterable[PathType], local: bool = False) -> dict[str, str]: ¶

Compute the UUIDs for all files stored in the filesystem.

Parameters
paths:`Iterable[PathType]`	Relative remote paths, or local paths when `local` is `True`.
local:`bool`	Specifies local paths as opposed to remote paths.

Returns
`dict[str, str]`	Dictionary mapping paths to their UUIDs.

def init_fsspec_filesystem(self) -> fsspec.AbstractFileSystem: ¶

Initialize an fsspec filesystem for the configured protocol.

Returns
`fsspec.AbstractFileSystem`	Initialized `fsspec` filesystem.

Raises
`NotImplementedError`	If the protocol is not supported by `fsspec`.
`RuntimeError`	If the credentials for the protocol are not properly set in environment variables.

def is_directory(self, remote_path: PosixPathType) -> bool: ¶

Check whether the given remote path is a directory.

Parameters
remote_path:`PosixPathType`	Relative path to the remote path.

Returns
`bool`	`True` if the path is a directory, `False` otherwise.

def put_bytes(self, file_bytes: bytes, remote_path: PosixPathType, mlflow_instance: ModuleType | None = None): ¶

Upload a file to remote storage directly from file bytes.

Parameters
file_bytes:`bytes`	File contents to upload.
remote_path:`PosixPathType`	Relative path to the remote file.
mlflow_instance:`ModuleType \| None`	Currently unused. MLflow uploads from bytes are not implemented.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

@overload

def put_dir(self, local_paths: Iterable[PathType], remote_dir: PosixPathType, uuid_dict: dict[str, str] | None = None, mlflow_instance: ModuleType | None = None, copy_contents: bool = False) -> dict[str, str]:

@overload

def put_dir(self, local_paths: PathType, remote_dir: PosixPathType, uuid_dict: dict[str, str] | None = None, mlflow_instance: ModuleType | None = None, copy_contents: bool = False):

Upload files to remote storage.

Parameters
local_paths:`PathType \| Iterable[PathType]`	Either a path specifying a directory to walk, or an iterable of files that may be in different directories.
remote_dir:`PosixPathType`	Relative path to the remote directory.
uuid_dict:`dict[str, str] \| None`	Stores paths as keys and corresponding UUIDs as values to replace the file basename.
mlflow_instance:`ModuleType \| None`	MLflow instance to use when uploading to an active run.
copy_contents:`bool`	If `True`, only copy the contents of the folder specified in `local_paths`.

Returns
`dict[str, str] \| None`	Mapping of local paths to remote paths if `local_paths` is an iterable of files, otherwise `None`.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.
`ValueError`	If `local_paths` is not a directory.

def put_file(self, local_path: PathType, remote_path: PosixPathType, mlflow_instance: ModuleType | None = None) -> str: ¶

Copy a single file to remote storage.

Parameters
local_path:`PathType`	Path to the local file.
remote_path:`PosixPathType`	Relative path to the remote file.
mlflow_instance:`ModuleType \| None`	MLflow instance to use when uploading to an active run.

Returns
`str`	Full remote path of the uploaded file.

Raises
`ValueError`	If using MLflow and there is no active run or no MLflow instance provided.

def read_text(self, remote_path: PosixPathType) -> str | bytes: ¶

Read a file into a string.

Parameters
remote_path:`PosixPathType`	Relative path to the remote file.

Returns
`str \| bytes`	The contents of the file.

Raises
`NotImplementedError`	If using a protocol that is not yet supported.

def read_to_byte_buffer(self, remote_path: PosixPathType | None = None) -> BytesIO: ¶

Read a file into a byte buffer.

Parameters
remote_path:`PosixPathType \| None`	Relative path to the remote file. If omitted, reads from `self.path`.

Returns
`BytesIO`	Byte buffer containing the file contents.

Raises
`ValueError`	If using MLflow while there is already an active run. If using MLflow but no relative artifact path is specified. If using MLflow but no `run_id` is specified.

def walk_dir(self, remote_dir: PosixPathType, recursive: bool = True, typ: Literal['file', 'directory', 'all'] = 'file') -> Iterator[str]: ¶

Walk through the individual files in a remote directory.

Parameters
remote_dir:`PosixPathType`	Relative path to the remote directory.
recursive:`bool`	If True, walks through the directory recursively. Defaults to True.
typ:`Literal['file', 'directory', 'all']`	Type of entries to yield. Corresponds to the `"type"` field in fsspec's ls output. Defaults to `"file"`.

Returns
`Iterator[str]`	Undocumented

Yields
Relative paths to the files in the remote directory.

Raises
`NotImplementedError`	If walking an MLflow artifact directory, which is not supported.

allow_active_mlflow_run = ¶

Whether operations are allowed on the active MLflow run.

allow_local = ¶

Whether operations are allowed on the local file system.

artifact_path = ¶

MLflow artifact path when using MLflow filesystem.

cache_storage = ¶

Path to cache storage, or None if no cache is used.

experiment_id = ¶

MLflow experiment ID when using MLflow filesystem.

fs = ¶

Initialized fsspec filesystem when using FSSPEC.

fs_type = ¶

Type of the filesystem, either MLFLOW or FSSPEC.

path = ¶

The path component of the input URL, with the protocol stripped.

run_id = ¶

MLflow run ID when using MLflow filesystem.

tracking_uri = ¶

MLflow tracking URI when using MLflow filesystem.

url = ¶

The original input URL.

@property

full_path: str = ¶

Full remote path.

Returns
Full remote path prefixed with the protocol.

@property

is_fsspec: bool = ¶

Check whether the filesystem uses fsspec.

Returns
`True` if the filesystem uses fsspec.

@property

is_mlflow: bool = ¶

Check whether the filesystem is an MLflow filesystem.

Returns
`True` if the filesystem is an MLflow filesystem,

@property

protocol: str = ¶

Returns the protocol of the filesystem.

@type: str

@staticmethod

def _split_mlflow_path(path: PathType) -> list[str | None]: ¶

Split an MLflow path into experiment, run, and artifact parts.

Parameters
path:`PathType`	MLflow path to split.

Returns
`list[str \| None]`	Three path components. Missing components are returned as `None`.

_allow_active_mlflow_run = ¶

Undocumented

_allow_local = ¶

Undocumented

_cache_storage = ¶

Undocumented

_fs = ¶

Undocumented

_fs_type = ¶

Undocumented

_is_mlflow_active_run: bool = ¶

Undocumented

_path = ¶

Undocumented

_protocol = ¶

Undocumented

_url = ¶

Undocumented