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, a device type, 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, renaming, and deleting directories. Each of these properties returns one: files for a dataset’s files, configurations for a device’s configurations, configurations for a device type’s configurations, and assets for the org’s assets.

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, device type, 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("rover-01")
>>> directory = device.configurations.create_directory("calib")
>>> print(directory.relative_path)
calib

>>> directory = device.configurations.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 can be created in a device’s or a device type’s configurations or in the org’s assets (device.configurations, device_type.configurations, org.assets), not in a dataset’s files. A link lets one file appear in many devices’ configurations without being copied.

A link 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 point it at the new version, 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 association is a dataset, or 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

>>> import pathlib
>>> from roboto.domain import devices, orgs
>>> runbook = orgs.Org.from_id("og_abc123").assets.get_file_by_path("runbooks/field_service.pdf")
>>> device = devices.Device.from_id("rover-01")
>>> link = device.configurations.create_link(runbook, "runbooks/field_service.pdf")
>>> link.download(pathlib.Path("/tmp/field_service.pdf"))
delete_directories(directory_paths)#

Delete directories from the association’s files, with every file and directory beneath them.

The directories stop appearing in listings when this returns. The files and directories beneath them are removed in the background, so they can still appear for a short while. A path that names nothing is ignored; a path that names a file deletes that file. A link beneath a deleted directory is deleted; the file it points at is kept unless it, too, is beneath a deleted directory. Links elsewhere that point at deleted files stop working. On a device’s configurations, only the device’s own directories are deleted, never its device type’s.

Parameters:

directory_paths (list[str]) – Paths of the directories to delete, relative to the root of the association’s files.

Raises:
Return type:

None

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("rover-01")
>>> device.configurations.delete_directories(["calib/old", "scratch"])
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. A matching link is deleted; the file it points at is not.

On a device’s configurations, only the device’s own files are considered. The configurations of its device type are never deleted here; delete those through device_type.configurations.

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.assets.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])
download_files(out_path, include_patterns=None, exclude_patterns=None, print_progress=True, *, include_inherited=False)#

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. A link is saved at its own path with the contents of the file version it points at.

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.

  • include_inherited (bool) – When these are a device’s configurations, also download those of its device type, selected as list_files() selects them. Refused for any other association.

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("rover-01")
>>> downloaded = device.configurations.download_files(
...     pathlib.Path("/tmp/rover-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.

On a device’s configurations, only the device’s own files are looked up. A file the device inherits from its device type is not found here; look it up through device_type.configurations instead.

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("rover-01")
>>> file = device.configurations.get_file_by_path("manifest.json")
>>> print(file.file_id)
fl_xyz789

>>> old_file = device.configurations.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("rover-01")
>>> for directory in device.configurations.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, *, include_inherited=False)#

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. For a device, a device type or an org, links are listed as well as files; is_link tells them apart.

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.

  • include_inherited (bool) – When these are a device’s configurations, also list those of its device type. Refused for any other association. Where the device and its type both have a file at the same path, only the device’s is listed. file.association.is_device_type tells which files came from the type. Only list_files and download_files() include the type’s configurations: get_file_by_path(), rename_file(), delete_files() and delete_directories() act on the device’s own configurations only, and delete() on a file from the type deletes it from the type, for every device of that type.

Yields:

File instances that match the specified patterns.

Raises:
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("rover-01")
>>> for file in device.configurations.list_files():
...     print(file.relative_path)
manifest.json
calib/front_cam.yaml

>>> for file in device.configurations.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("rover-01")
>>> device.configurations.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 keeps its ID and versions; 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("rover-01")
>>> record = device.configurations.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("rover-01")
>>> device.configurations.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("rover-01")
>>> device.configurations.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("rover-01")
>>> file_ids = device.configurations.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