# 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. 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 ``` #### 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): ``` 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](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. 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 a protocol that is not yet supported. ##### 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 a protocol that is not yet supported. ##### 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 a protocol that is not yet supported. ##### download ```python 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. ##### 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. ##### 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. 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. ##### 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. 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. ##### 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. 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. ##### 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. 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. ##### 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`): Currently unused. MLflow uploads from bytes are not implemented. Raises * `NotImplementedError`: If using a protocol that is not yet supported. ##### 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. 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. ##### 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. 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 * `NotImplementedError`: If using a protocol that is not yet supported. ##### 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 while there is already an active run. 2. If using MLflow but no relative artifact path is specified. 3. If using MLflow but no [run_id](https://docs.luxonis.com/software-v3/ai-inference/model-source/training/luxonis-ml/luxonis-ml-api-reference/utils/filesystem.md) is specified. ##### 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): ``` 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. ##### 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. Raises * `NotImplementedError`: If walking an MLflow artifact directory, which is not supported. #### Attributes ##### 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. ##### 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, ##### path The path component of the input URL, with the protocol stripped. ##### protocol Returns the protocol of the filesystem. @type: str ##### run_id MLflow run ID when using MLflow filesystem. ##### tracking_uri MLflow tracking URI when using MLflow filesystem. ##### url The original input URL. ### PutFile Protocol for the `put_file` plugins. ## Attributes ### PUT_FILE_REGISTRY