# filesystem Python API: `luxonis_ml.utils.filesystem` ## Classes ### FSType Filesystem backend type. #### Attributes ##### FSSPEC fsspec-compatible filesystem. ##### MLFLOW MLflow artifact storage. ### LuxonisFileSystem 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. The backend is chosen by the protocol of the URL: ```pycon file:// s3:/// gcs:/// mlflow://// ``` A path with no protocol is treated as a local one. For MLflow the `` is optional and defaults to the run's artifact root, so `mlflow:///` is itself a valid destination. A bare `mlflow://` binds the filesystem to whichever run is currently active instead, and requires `allow_active_mlflow_run=True`. MLflow artifacts cannot be removed, so the `delete_*` methods raise `NotImplementedError` for them. 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** > ```pycon >>> @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 ``` Uploading to the artifact store of a specific MLflow run, which returns the URL of the new artifact. ```pycon fs = LuxonisFileSystem( "mlflow://12/abc123/checkpoints", tracking_uri="http://localhost:5000", ) fs.put_file("model.ckpt", "best.ckpt") # -> "mlflow://12/abc123/checkpoints/best.ckpt" ``` #### Methods ##### init ```python def __init__(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, tracking_uri: str | None = None): ``` Initialize the `LuxonisFileSystem`. Parameters * `path` (`str`): Input path consisting of a protocol and path, or only a path for local files. MLflow paths are of the form `mlflow:////`, where the artifact path is optional, or a bare `mlflow://` to use the active run. * `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 an `fsspec` `filecache` directory, used only by the `s3://` and `gcs://` backends. No cache is used if not set, and it is ignored, with a warning, for `file://` and `mlflow://`. * `put_file_plugin` (`str | None`): Name of a registered function in [PUT_FILE_REGISTRY](https://docs.luxonis.com/software-v3/ai-inference/model-source/training/luxonis-ml/luxonis-ml-api-reference/utils/filesystem.md) to use instead of [LuxonisFileSystem.put_file](https://docs.luxonis.com/software-v3/ai-inference/model-source/training/luxonis-ml/luxonis-ml-api-reference/utils/filesystem.md). The registered function must conform to the [PutFile](https://docs.luxonis.com/software-v3/ai-inference/model-source/training/luxonis-ml/luxonis-ml-api-reference/utils/filesystem.md) protocol. * `tracking_uri` (`str | None`): MLflow tracking URI to use. If omitted, `MLFLOW_TRACKING_URI` from the environment is used. Raises * `ValueError`: 1. If the protocol is not supported. 2. 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. ##### delete_dir ```python def delete_dir(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 MLflow, whose artifacts cannot be deleted. ##### delete_file ```python def delete_file(remote_path: PosixPathType): ``` Delete a single file from remote storage. Parameters * `remote_path` (`PosixPathType`): Relative path to the remote file. Raises * `NotImplementedError`: If using MLflow, whose artifacts cannot be deleted. ##### delete_files ```python def delete_files(remote_paths: list[PosixPathType]): ``` Delete multiple files from remote storage. Parameters * `remote_paths` (`list[PosixPathType]`): Relative paths to remote files. Raises * `NotImplementedError`: If using MLflow, whose artifacts cannot be deleted. ##### download ```python def download(url: str, dest: PathType | None, tracking_uri: str | None = 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. * `tracking_uri` (`str | None`): MLflow tracking URI to use for `mlflow://` URLs. Returns * `Path`: Path to the downloaded file or directory. When the destination already exists, the data is nested inside it and the nested path is returned. ##### exists ```python def exists(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. Raises * `MlflowException`: If the lookup itself fails, e.g. on a connection, authentication or server error. Only a definitive "this does not exist" answer from MLflow is reported as `False`. ##### get_dir ```python def get_dir(remote_paths: PosixPathType | Iterable[PosixPathType], local_dir: PathType, mlflow_instance: ModuleType | None = None) -> Path: ``` Copy many files from remote storage to local storage. > **Note** > An MLflow download is staged next to the destination, so a large artifact needs free space on the destination's own filesystem rather than in `$TMPDIR`. 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`): MLflow instance to use when downloading from an active run. Returns * `Path`: Path to the downloaded directory. Raises * `ValueError`: If using MLflow and the run cannot be resolved. ##### get_file ```python def get_file(remote_path: PosixPathType, local_path: PathType, mlflow_instance: ModuleType | None = None) -> Path: ``` Copy a single file from remote storage. > **Note** > An MLflow download is staged next to the destination, so a large artifact needs free space on the destination's own filesystem rather than in `$TMPDIR`. Parameters * `remote_path` (`PosixPathType`): Relative path to the remote file. * `local_path` (`PathType`): Path to the local file. * `mlflow_instance` (`ModuleType | None`): MLflow instance to use when downloading from an active run. Returns * `Path`: Path to the downloaded file. Raises * `ValueError`: If using MLflow and either no relative artifact path is specified or the run cannot be resolved. ##### get_file_uuid ```python def get_file_uuid(path: PathType, local: bool = False) -> str: ``` Read a file and returns the (unique) UUID generated from file bytes. > **Note** > The file is hashed in chunks, so memory use does not grow with its size. MLflow is the exception: its artifacts are downloaded through `read_to_byte_buffer`, which holds the whole file in memory before hashing begins. 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 * `ValueError`: If using MLflow and either no relative artifact path is specified or the run cannot be resolved. ##### get_file_uuids ```python def get_file_uuids(paths: Iterable[PathType], local: bool = False) -> dict[str, str]: ``` Compute the UUIDs for all files stored in the filesystem. > **Note** > The files are hashed concurrently, so the memory bound of `get_file_uuid` applies once per thread in the pool. 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. ##### get_protocol ```python 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. ##### init_fsspec_filesystem ```python 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. ##### is_directory ```python def is_directory(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. Raises * `FileNotFoundError`: If the path cannot be inspected on an `fsspec` filesystem. * `MlflowException`: If the MLflow run cannot be inspected, e.g. because it does not exist. ##### put_bytes ```python def put_bytes(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`): MLflow instance to use when uploading to an active run. Raises * `ValueError`: If using MLflow and either no relative artifact path is specified or the run cannot be resolved. ##### put_dir ```python def put_dir(local_paths: PathType | Iterable[PathType], remote_dir: PosixPathType, uuid_dict: dict[str, str] | None = None, mlflow_instance: ModuleType | None = None, copy_contents: bool = False) -> dict[str, str] | None: ``` Upload files to remote storage. > **Note** > A directory is uploaded to MLflow in one call and is never staged. Passing individual files uploads them one by one through `put_file` instead, so with `uuid_dict` set the staging described there applies to each of them. 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 * `ValueError`: 1. * `If`local_paths`is a single path but not a`: directory. * `If uploading to an active MLflow run without an`: MLflow instance. ##### put_file ```python def put_file(local_path: PathType, remote_path: PosixPathType, mlflow_instance: ModuleType | None = None) -> str: ``` Copy a single file to remote storage. > **Note** > MLflow stores a file under its local base name, so uploading under a different one is staged through a temporary copy. It is placed next to the source, meaning a large artifact needs free space on the source's own filesystem rather than in `$TMPDIR`. If that directory is not writable the staging falls back to `$TMPDIR`, which on most Linux distributions is a RAM-backed `tmpfs`, so a large artifact may then fail with `No space left on device`. 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. ##### read_text ```python def read_text(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 * `ValueError`: If using MLflow and either no relative artifact path is specified or the run cannot be resolved. ##### read_to_byte_buffer ```python def read_to_byte_buffer(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`: 1. If using MLflow but no relative artifact path is specified. 2. If using MLflow but the run cannot be resolved. ##### split_full_path ```python 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. ##### upload ```python def upload(local_path: PathType, url: str, tracking_uri: str | None = None): ``` Upload file or directory to remote storage. Useful for uploading a single local object without having to create a [LuxonisFileSystem](https://docs.luxonis.com/software-v3/ai-inference/model-source/training/luxonis-ml/luxonis-ml-api-reference/utils/filesystem.md) instance. Parameters * `local_path` (`PathType`): Path to the local file or directory. * `url` (`str`): URL to the remote file or directory. * `tracking_uri` (`str | None`): MLflow tracking URI to use for `mlflow://` URLs. ##### walk_dir ```python def walk_dir(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]` Yields * Relative paths to the files in the remote directory. #### Attributes ##### artifact_path MLflow artifact path. Returns * The artifact path relative to the run's artifact root, or `None` if the filesystem is not an MLflow filesystem or no artifact path was given. ##### experiment_id MLflow experiment ID. Returns * The experiment ID, or `None` if the filesystem is not an MLflow filesystem or the experiment is not known. ##### full_path Full remote path. Returns * Full remote path prefixed with the protocol. ##### is_fsspec Check whether the filesystem uses fsspec. Returns * `True` if the filesystem uses fsspec. ##### is_mlflow Check whether the filesystem is an MLflow filesystem. Returns * `True` if the filesystem is an MLflow filesystem, ##### protocol Returns the protocol of the filesystem. @type: str ##### run_id MLflow run ID. Returns * The run ID, or `None` if the filesystem is not an MLflow filesystem or the run is not known. ##### tracking_uri MLflow tracking URI. Returns * The tracking URI, or `None` if the filesystem is not an MLflow filesystem. ### PutFile Protocol for the `put_file` plugins. ## Attributes ### PUT_FILE_REGISTRY