radiens_drive_catalog.drive

drive.py

Google Drive API interactions: recursive scanning and file download.

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

Classes

AssetEntry

Bases: TypedDict

One non-xdat asset record stored in the catalog.

Represents a single file or folder found on Drive that is not an xdat dataset — for example a logs/ directory, a PowerPoint, or a writeup.

Attributes:

Name	Type	Description
`asset_name`	`str`	The file or folder name as it appears on Drive (e.g. `"logs"`, `"notes.pptx"`).
`asset_type`	`Literal['folder', 'file']`	`"folder"` or `"file"`.
`drive_path`	`str`	Slash-joined path from the root folder to the parent folder of this asset (e.g. `"2026-02-15_batch/reaching"`). Together with `asset_name`, this uniquely identifies an asset.
`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` if not available.

DatasetEntry

Bases: TypedDict

One xdat dataset record stored in the catalog.

Represents a single neural recording dataset. The three xdat files (_data.xdat, .xdat.json, _timestamp.xdat) share a common base_name stem.

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 dataset files (e.g. `"2026-02-15_batch/reaching/probe1"`).
`drive_file_ids`	`dict[str, str]`	Maps file type labels (`"data"`, `"meta"`, `"timestamp"`) to their Google Drive file IDs. May contain fewer than three keys if only some files were found during scanning.
`local_path`	`str \| None`	Absolute path to the local directory where the dataset has been downloaded, or `None` if not yet downloaded.
`upload_time`	`str \| None`	ISO 8601 `createdTime` from the Drive API for one of the dataset files, or `None` if not available.

DriveItem

Bases: TypedDict

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

id, name, and mimeType are always present in well-formed responses. parents is only populated when explicitly requested in the fields parameter (used by the flat-scan fallback). createdTime is populated when requested; absent on older API responses.

Functions

download_asset

download_asset(service, asset, local_data_dir, quiet=False)

Download an asset (file or folder) into the assets subdirectory.

Assets are stored under {local_data_dir}/assets/{drive_path}/{asset_name}, keeping them separate from the xdat dataset files.

For folder assets the entire subtree is downloaded recursively, mirroring the Drive folder structure under the asset's local directory.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`asset`	`AssetEntry`	The :class:`AssetEntry` to download.	required
`local_data_dir`	`str`	Root local data directory (same as used for datasets).	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_dataset

download_dataset(
    service, dataset, local_data_dir, quiet=False
)

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

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

{local_data_dir}/{drive_path}/{base_name}_data.xdat
{local_data_dir}/{drive_path}/{base_name}.xdat.json
{local_data_dir}/{drive_path}/{base_name}_timestamp.xdat

The directory is created if it does not already exist.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`dataset`	`DatasetEntry`	The `DatasetEntry` 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 datasets and non-xdat assets.

Traverses the full subtree rooted at root_folder_id.

Dataset collection: xdat files are recognized at any depth by their suffix and merged by (drive_path, base_name) into :class:DatasetEntry records.

Asset collection: Non-xdat content is cataloged according to depth:

Folders encountered inside an experiment folder (depth ≥ 3 from root) are cataloged as "folder" assets and also recursed into so that any xdat files they contain are still found.
Non-xdat files encountered inside a date folder or experiment folder (depths 2-3 from root) are cataloged as "file" assets.
Content deeper than the experiment level is not separately cataloged (it is part of a parent folder asset and downloaded with it).

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[DatasetEntry]`	A tuple `(datasets, assets)` where `datasets` is a list of
`list[AssetEntry]`	class:`DatasetEntry` dicts and `assets` is a list of
`tuple[list[DatasetEntry], list[AssetEntry]]`	class:`AssetEntry` 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.

This is the fallback used when the service account cannot access root_folder_id (404 / 403). It flat-lists every file visible to the account, reconstructs folder paths via the parents field, and applies the same depth-based classification rules as :func:scan_drive.

Parameters:

Name	Type	Description	Default
`service`	`DriveService`	Authenticated Drive API service.	required
`root_folder_id`	`str`	The logical root folder ID. Not used for API calls — only as a stop sentinel when resolving paths so that depth assignments match the recursive scan.	required

Returns:

Type	Description
`tuple[list[DatasetEntry], list[AssetEntry]]`	Same `(datasets, assets)` tuple as :func:`scan_drive`.

xdat_dataset_exists_locally

xdat_dataset_exists_locally(dataset, local_dir)

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

Checks for each of the three canonical xdat suffixes (_data.xdat, .xdat.json, _timestamp.xdat). If drive_file_ids contains a file-type label that is not one of the known types, raises rather than silently returning False.

Parameters:

Name	Type	Description	Default
`dataset`	`DatasetEntry`	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 a file-type label that is not one of the known xdat types.