radiens_drive_catalog.drive

drive.py

Google Drive API interactions: recursive scanning and file download.

xdat recording file naming convention (3 files per base_name): {base_name}_data.xdat {base_name}.xdat.json {base_name}_timestamp.xdat

Classes

DriveItem

Bases: TypedDict

A single item returned by the Drive API files.list endpoint.

DriveItemEntry

Bases: TypedDict

One non-recording Drive item stored in the catalog.

Represents a single file or folder found on Drive that is not part of an xdat recording — for example a logs/ directory, a config file, or a writeup. No structural or semantic classification is applied; the consumer decides what these items mean.

Attributes:

Name	Type	Description
`name`	`str`	The file or folder name as it appears on Drive (e.g. `"logs"`).
`is_folder`	`bool`	`True` if this item is a Drive folder.
`drive_path`	`str`	Slash-joined path from the root folder to the parent folder of this item. Together with `name`, this uniquely identifies an item.
`drive_id`	`str`	Google Drive ID of this file or folder.
`mime_type`	`str`	MIME type as reported by the Drive API.
`local_path`	`str \| None`	Absolute local path to the downloaded file or folder, or `None` if not yet downloaded.
`upload_time`	`str \| None`	ISO 8601 `createdTime` from the Drive API, or `None`.

RecordingEntry

Bases: TypedDict

One xdat recording dataset stored in the catalog.

Represents a single neural recording. The three xdat files (_data.xdat, .xdat.json, _timestamp.xdat) share a common base_name stem and are treated as a single unit.

Attributes:

Name	Type	Description
`base_name`	`str`	Shared filename stem across all three xdat files. Note: `base_name` is not globally unique; the pair `(drive_path, base_name)` uniquely identifies a recording.
`drive_path`	`str`	Slash-joined path from the root folder to the folder containing the recording files.
`drive_file_ids`	`dict[str, str]`	Maps file type labels (`"data"`, `"meta"`, `"timestamp"`) to their Google Drive file IDs.
`local_path`	`str \| None`	Absolute path to the local directory where the recording has been downloaded, or `None` if not yet downloaded.
`upload_time`	`str \| None`	ISO 8601 `createdTime` from the Drive API, or `None`.

Functions

download_item

download_item(service, item, local_data_dir, quiet=False)

Download a Drive item (file or folder) mirroring the Drive path locally.

Items are stored under {local_data_dir}/{drive_path}/{name}, keeping them in the same local directory tree as recordings.

For folder items the entire subtree is downloaded recursively.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`item`	`DriveItemEntry`	The :class:`DriveItemEntry` to download.	required
`local_data_dir`	`str`	Root local data directory.	required
`quiet`	`bool`	If `True`, suppress tqdm progress bars.	`False`

Returns:

Type	Description
`str`	The absolute path to the downloaded file or folder as a string.

download_recording

download_recording(
    service, recording, local_data_dir, quiet=False
)

Download the xdat files for a recording into a path-mirrored local directory.

Files are stored under {local_data_dir}/{drive_path}/, mirroring the Drive folder hierarchy.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`recording`	`RecordingEntry`	The :class:`RecordingEntry` whose files should be downloaded.	required
`local_data_dir`	`str`	Root local data directory.	required
`quiet`	`bool`	If `True`, suppress tqdm progress bars.	`False`

Returns:

Type	Description
`str`	The absolute path to the directory where the files were written.

scan_drive

scan_drive(service, root_folder_id, *, quiet=False)

Recursively scan Drive and return xdat recordings and all other items.

Traverses the full subtree rooted at root_folder_id.

Recording collection: xdat files are recognized at any depth by their suffix and merged by (drive_path, base_name) into :class:RecordingEntry records. The three component files are treated as a single unit.

Item collection: Every non-xdat file and every folder (at any depth) is cataloged as a :class:DriveItemEntry. No depth-based filtering or structural classification is applied — the consumer decides what items mean.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`root_folder_id`	`str`	Google Drive folder ID of the root folder to scan.	required

Returns:

Type	Description
`list[RecordingEntry]`	A tuple `(recordings, items)` where `recordings` is a list of
`list[DriveItemEntry]`	class:`RecordingEntry` dicts and `items` is a list of
`tuple[list[RecordingEntry], list[DriveItemEntry]]`	class:`DriveItemEntry` dicts. All entries have `local_path=None`.

scan_drive_flat

scan_drive_flat(service, root_folder_id, *, quiet=False)

Scan Drive without traversing from a root folder.

Flat-lists every file visible to the account, reconstructs folder paths via the parents field, and applies the same classification rules as :func:scan_drive: xdat files become :class:RecordingEntry records; everything else becomes a :class:DriveItemEntry.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`root_folder_id`	`str`	The logical root folder ID — used as the stop sentinel when resolving paths.	required

Returns:

Type	Description
`tuple[list[RecordingEntry], list[DriveItemEntry]]`	Same `(recordings, items)` tuple as :func:`scan_drive`.

xdat_recording_exists_locally

xdat_recording_exists_locally(recording, local_dir)

Return True if all xdat files for a recording exist under local_dir.

Parameters:

Name	Type	Description	Default
`recording`	`RecordingEntry`	The catalog entry to check.	required
`local_dir`	`str \| Path`	Directory expected to contain the three xdat files.	required

Returns:

Type	Description
`bool`	`True` if every file in `XDAT_SUFFIXES` exists at
`bool`	`local_dir/{base_name}{suffix}`; `False` if any are missing.

Raises:

Type	Description
`ValueError`	If `drive_file_ids` contains an unrecognized file type.