roboto.domain.files.file_system#

Module Contents#

class roboto.domain.files.file_system.FileSystem(association, roboto_client=None, file_service=None, org_id=None)#

The files and directories under one association: a dataset, a device, or the org itself.

A FileSystem holds that association’s directory tree and the operations on it: listing, uploading, downloading, renaming, and deleting files, and creating and renaming directories. Datasets, devices, and orgs each return one as files, files, and files.

Every relative path a method takes or returns is relative to the root of the association’s tree, which files of other associations do not share.

Parameters:
property association: roboto.association.Association#

The dataset, device, or org whose files this object works on.

Return type:

roboto.association.Association

create_directory(name, error_if_exists=False, create_intermediate_dirs=False, parent_path=None, origination=None)#

Create a directory among the association’s files.

Parameters:

  • name (str) – Name of the directory to create.

  • error_if_exists (bool) – If True, raises an exception if the directory already exists.

  • parent_path (Optional[pathlib.Path]) – Path of the parent directory. If None, creates the directory at the root of the association’s files.

  • origination (Optional[str]) – Optional string describing the source or context of the directory creation.

  • create_intermediate_dirs (bool) – If True, creates intermediate directories in the path if they don’t exist. If False, requires all parent directories to already exist.

Raises:
Returns:

DirectoryRecord of the created directory.

Return type:

roboto.domain.files.record.DirectoryRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> directory = device.files.create_directory("calib")
>>> print(directory.relative_path)
calib

>>> directory = device.files.create_directory(
...     name="final",
...     parent_path=pathlib.Path("path/to/deep"),
...     create_intermediate_dirs=True,
... )
>>> print(directory.relative_path)
path/to/deep/final

Put a link to another file at relative_path among the association’s files.

A link lets one file, such as a URDF in the org’s own files, appear in many devices’ files without being copied. It pins one version of its target: passing a File pins that file’s version, and passing a file ID pins the target’s current version. Later versions of the target do not move the link; create the link again at the same path to re-point it, which adds a version to the link unless it already pins that target version. Missing parent directories are created. Downloading the link, or asking it for a signed URL, fetches the pinned version of the target.

Parameters:

  • target (Union[roboto.domain.files.file.File, str]) – The file to link to, or its ID. It must be a file in the same org, not a link or a directory.

  • relative_path (str) – Where the link sits, relative to the root of the association’s files.

Returns:

The link, whose is_link is True.

Raises:

  • RobotoConflictException – A file or a directory already occupies relative_path. The reverse is refused too: uploading a file to a link’s path is a conflict until the link is deleted.

  • RobotoInvalidRequestException – The target is a link or a directory, is in another org, or does not exist at the version to pin.

  • RobotoUnauthorizedException – The caller cannot edit the association’s files or view the target.

Return type:

roboto.domain.files.file.File

Examples

>>> from roboto.domain import devices, orgs
>>> urdf = orgs.Org.from_id("og_abc123").files.get_file_by_path("urdf/lemi/lemi.urdf")
>>> device = devices.Device.from_id("lemi-01")
>>> link = device.files.create_link(urdf, "urdf/lemi.urdf")
>>> link.download(pathlib.Path("/tmp/lemi.urdf"))
delete_files(include_patterns=None, exclude_patterns=None)#

Delete the association’s files that match the given patterns.

Deletes files that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.

Parameters:

  • include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None or empty, all files are considered for deletion. An empty list is treated as no filter (all files), not as “include nothing”.

  • exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from deletion. Takes precedence over include patterns. If None or empty, no files are excluded.

Raises:

RobotoUnauthorizedException – Caller lacks permission to delete files.

Return type:

None

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Examples

>>> from roboto.domain import orgs
>>> org = orgs.Org.from_id("og_abc123")
>>> org.files.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])
download_files(out_path, include_patterns=None, exclude_patterns=None, print_progress=True)#

Download the association’s files to a local directory.

Downloads files that match the specified patterns to the given local directory. The files’ directory structure is preserved in the download location. If the output directory doesn’t exist, it will be created. Files are found with list_files(), which does not return links yet, so no link is downloaded; download one with download().

Parameters:

  • out_path (pathlib.Path) – Local directory path where files should be downloaded.

  • include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None or empty, all files are downloaded. An empty list is treated as no filter (all files), not as “include nothing”.

  • exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from download. Takes precedence over include patterns. If None or empty, no files are excluded.

  • print_progress (bool) – Whether to show a progress bar during download.

Returns:

List of tuples containing (FileRecord, local_path) for each downloaded file.

Raises:
Return type:

list[tuple[roboto.domain.files.record.FileRecord, pathlib.Path]]

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> downloaded = device.files.download_files(pathlib.Path("/tmp/lemi-01"), include_patterns=["calib/**"])
>>> print(f"Downloaded {len(downloaded)} files")
Downloaded 2 files
get_file_by_path(relative_path, version_id=None)#

Get a File instance for the association’s file at the specified path.

Parameters:

  • relative_path (Union[str, pathlib.Path]) – Path of the file relative to the root of the association’s files.

  • version_id (Optional[int]) – Specific version of the file to retrieve. If None, gets the latest version.

Returns:

File instance representing the file at the specified path.

Raises:
Return type:

roboto.domain.files.file.File

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> file = device.files.get_file_by_path("manifest.json")
>>> print(file.file_id)
fl_xyz789

>>> old_file = device.files.get_file_by_path("manifest.json", version_id=1)
>>> print(old_file.version)
1
list_directories()#

Yield every directory among the association’s files, at any depth.

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> for directory in device.files.list_directories():
...     print(directory.relative_path)
calib
urdf
Return type:

collections.abc.Generator[roboto.domain.files.record.DirectoryRecord, None, None]

list_files(include_patterns=None, exclude_patterns=None)#

List the association’s files with optional pattern-based filtering.

Returns all of the association’s files that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.

Parameters:

  • include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None or empty, all files are considered. An empty list is treated as no filter (all files), not as “include nothing”.

  • exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude. Takes precedence over include patterns. If None or empty, no files are excluded.

Yields:

File instances that match the specified patterns.

Raises:

RobotoUnauthorizedException – Caller lacks permission to list files.

Return type:

collections.abc.Generator[roboto.domain.files.file.File, None, None]

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Files appear in this list shortly after their upload completes, not instantly.

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> for file in device.files.list_files():
...     print(file.relative_path)
manifest.json
calib/front_cam.yaml

>>> for file in device.files.list_files(include_patterns=["calib/**"], exclude_patterns=["**/*.bak"]):
...     print(file.relative_path)
calib/front_cam.yaml
rename_directory(old_path, new_path)#

Rename or move a directory among the association’s files.

Both old_path and new_path are relative to the root of the association’s files. Pass a new_path with fewer path components to move the directory up the tree, or a different leaf name at the same depth to rename in place.

Parameters:

  • old_path (str) – Current relative path of the directory (e.g. "logs/session1").

  • new_path (str) – Target relative path of the directory (e.g. "session1" to move up one level).

Returns:

Updated DirectoryRecord reflecting the new path.

Raises:
Return type:

roboto.domain.files.record.DirectoryRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.rename_directory("calib/front", "front_calib")
rename_file(file_id, new_path)#

Rename or move a file among the association’s files.

new_path is relative to the root of the association’s files. Pass a path with fewer components to move the file up the tree, a different name at the same depth to rename in place, or a path under a different directory to move sideways.

The file’s storage URI is unchanged; only its relative path changes.

Parameters:

  • file_id (str) – ID of the file to rename or move.

  • new_path (str) – Target relative path for the file (e.g. "file.bag" to move to the root, or "other_dir/file.bag" to move into an existing directory).

Returns:

Updated FileRecord reflecting the new path.

Raises:
Return type:

roboto.domain.files.record.FileRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> record = device.files.rename_file("fl_xyz789", "manifest.json")
>>> record.relative_path
'manifest.json'
upload_directory(directory_path, include_patterns=None, exclude_patterns=None, delete_after_upload=False, max_batch_size=MAX_FILES_PER_MANIFEST, print_progress=True, device_id=None)#

Upload all files and directories recursively from the specified directory path.

Use include_patterns and exclude_patterns to control what files and directories are uploaded, and delete_after_upload to clean up your local filesystem after the uploads succeed.

Parameters:

  • directory_path (pathlib.Path) – Local directory whose contents are uploaded, keeping its layout.

  • include_patterns (Optional[list[str]]) – gitignore-style patterns for files to include. If None, every file is included.

  • exclude_patterns (Optional[list[str]]) – gitignore-style patterns for files to exclude. Takes precedence over include_patterns.

  • delete_after_upload (bool) – If True, each uploaded local file is deleted once the uploads succeed.

  • max_batch_size (int) – Maximum number of files per upload transaction.

  • print_progress (bool) – Whether to display an upload progress bar.

  • device_id (Optional[str]) – Optional identifier of the device that generated this data.

Return type:

None

Notes

Both pattern lists follow the gitignore pattern format described in https://git-scm.com/docs/gitignore#_pattern_format.

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.upload_directory(
...     pathlib.Path("/path/to/calibration"),
...     exclude_patterns=["**/*.log"],
... )
upload_file(file_path, file_destination_path=None, print_progress=True, device_id=None)#

Upload a single file associated with association.

Parameters:

  • file_path (pathlib.Path) – Local file to upload.

  • file_destination_path (Optional[str]) – Destination path among the association’s files. Defaults to the file’s own name at the root.

  • print_progress (bool) – Whether to display an upload progress bar.

  • device_id (Optional[str]) – Optional identifier of the device that generated this data.

Returns:

The uploaded file. Its record is fetched from the platform the first time it is read.

Raises:

RobotoInternalException – The upload reported success without reporting a file ID.

Return type:

roboto.domain.files.file.File

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.upload_file(pathlib.Path("/path/to/manifest.json"))
upload_files(files, file_destination_paths={}, max_batch_size=MAX_FILES_PER_MANIFEST, print_progress=True, device_id=None)#

Upload multiple files associated with association.

Parameters:

  • files (collections.abc.Iterable[pathlib.Path]) – Local files to upload.

  • file_destination_paths (collections.abc.Mapping[pathlib.Path, str]) – Mapping from local path to destination path among the association’s files. Files not in the mapping upload to the root under their own name.

  • max_batch_size (int) – Maximum number of files per upload transaction.

  • print_progress (bool) – Whether to display an upload progress bar.

  • device_id (Optional[str]) – Optional identifier of the device that generated this data.

Returns:

Mapping from each uploaded local path to the ID of the file record it created.

Return type:

dict[pathlib.Path, str]

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> file_ids = device.files.upload_files(
...     [pathlib.Path("/path/to/front_cam.yaml")],
...     file_destination_paths={pathlib.Path("/path/to/front_cam.yaml"): "calib/front_cam.yaml"},
... )
>>> file_ids[pathlib.Path("/path/to/front_cam.yaml")]
'fl_0123456789abcdef'
roboto.domain.files.file_system.MAX_FILES_PER_MANIFEST = 500#
On this page