Skip to content
Roboto
Esc
↑↓navigate↵open⌘Jpreview
On this page

roboto.domain.files.file_system

Module Contents

FileSystem

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

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

roboto_client Optional[roboto.http.RobotoClient]
file_service Optional[roboto.storage.FileService]
org_id Optional[str]

Properties

FileSystem.association

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

FileSystem.create_directory()

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

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

If the directory already exists and error_if_exists is True.

If the caller lacks permission to create the directory.

If the directory name is invalid or the parent path does not exist (when create_intermediate_dirs is False).

Returns

DirectoryRecord of the created directory.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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
create_link(target, relative_path)#View Source

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

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

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.

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

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

Usage

from roboto.domain import devices, orgs
urdf = orgs.Org.from_id("og_abc123").files.get_file_by_path("urdf/rover/rover.urdf")
device = devices.Device.from_id("rover-01")
link = device.files.create_link(urdf, "urdf/rover.urdf")
link.download(pathlib.Path("/tmp/rover.urdf"))

FileSystem.delete_files()

delete_files(include_patterns=None, exclude_patterns=None)#View Source

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

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.

Usage

from roboto.domain import orgs
org = orgs.Org.from_id("og_abc123")
org.files.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])

FileSystem.download_files()

download_files(out_path, include_patterns=None, exclude_patterns=None, print_progress=True)#View Source

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[tuple[roboto.domain.files.record.FileRecord, pathlib.Path]]

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

Raises

A selected file’s path resolves outside out_path; nothing is downloaded.

Caller lacks permission to download files.

Notes

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

Usage

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

FileSystem.get_file_by_path()

get_file_by_path(relative_path, version_id=None)#View Source

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

The association has no file at the given path.

Caller lacks permission to access the file.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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

FileSystem.list_directories()

list_directories()#View Source

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

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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]

FileSystem.list_files()

list_files(include_patterns=None, exclude_patterns=None)#View Source

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

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.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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

FileSystem.rename_directory()

rename_directory(old_path, new_path)#View Source

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

Raises

No directory exists at old_path.

new_path conflicts with an existing node or contains a cycle.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-01")
device.files.rename_directory("calib/front", "front_calib")

FileSystem.rename_file()

rename_file(file_id, new_path)#View Source

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

No file with file_id exists.

new_path conflicts with an existing file, the parent directory does not exist, or the move would create a cycle.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-01")
record = device.files.rename_file("fl_xyz789", "manifest.json")
record.relative_path
# 'manifest.json'

FileSystem.upload_directory()

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)#View Source

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.

Usage

import pathlib
from roboto.domain import devices
device = devices.Device.from_id("rover-01")
device.files.upload_directory(
    pathlib.Path("/path/to/calibration"),
    exclude_patterns=["**/*.log"],
)

FileSystem.upload_file()

upload_file(file_path, file_destination_path=None, print_progress=True, device_id=None)#View Source

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

The upload reported success without reporting a file ID.

Usage

import pathlib
from roboto.domain import devices
device = devices.Device.from_id("rover-01")
device.files.upload_file(pathlib.Path("/path/to/manifest.json"))

FileSystem.upload_files()

upload_files(files, file_destination_paths={}, max_batch_size=MAX_FILES_PER_MANIFEST, print_progress=True, device_id=None)#View Source

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

dict[pathlib.Path, str]

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

Usage

import pathlib
from roboto.domain import devices
device = devices.Device.from_id("rover-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'

MAX_FILES_PER_MANIFEST

roboto.domain.files.file_system.MAX_FILES_PER_MANIFEST = 500#View Source

Was this page helpful?