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"`.
`date_folder`	`str \| None`	Name of the top-level date-prefixed folder (depth 1), or `None` if the asset is at the Drive root.
`experiment`	`str \| None`	Name of the depth-2 subfolder, or `None` if the asset lives directly inside the date folder.
`drive_path`	`str`	Slash-joined path from the root folder to the parent folder of this asset (e.g. `"2026-02-15_batch/reaching"`).
`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.

DatasetEntry

Bases: TypedDict

One xdat dataset record stored in the catalog.

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

Attributes:

Name	Type	Description
`base_name`	`str`	Shared filename stem across all three xdat files. This is the canonical identifier used throughout the package.
`date_folder`	`str \| None`	Name of the top-level date-prefixed folder (depth 1), or `None` if the dataset is at the Drive root.
`experiment`	`str \| None`	Name of the depth-2 subfolder (e.g. `"reaching"`), or `None` if the dataset lives directly inside the date folder.
`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.

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).

Functions

download_asset

download_asset(service, asset, local_data_dir)

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 flat 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

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)

Download the xdat files for a dataset into local_data_dir.

Files are written flat (not in a per-dataset subdirectory):

{local_data_dir}/{base_name}_data.xdat
{local_data_dir}/{base_name}.xdat.json
{local_data_dir}/{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`	Path to the local directory where files will be written.	required

Returns:

Type	Description
`str`	The `local_data_dir` path as a string.

scan_drive

scan_drive(service, root_folder_id)

Recursively scan Drive and return xdat datasets and non-xdat assets.

Traverses the full subtree rooted at root_folder_id.

Dataset collection (unchanged): xdat files are recognized at any depth by their suffix and merged by 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).

The depth thresholds correspond to len(path_components) inside _recurse: 0 = root, 1 = inside date folder, 2 = inside experiment.

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 (one per unique `base_name`) and
`tuple[list[DatasetEntry], list[AssetEntry]]`	`assets` is a list of :class:`AssetEntry` dicts. All entries have
`tuple[list[DatasetEntry], list[AssetEntry]]`	`local_path=None`; that field is set later by :class:`Catalog`.

scan_drive_flat

scan_drive_flat(service, root_folder_id)

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`.