Skip to main content

io.dandi

logger_config​

logger​

REMOTE_PREFIXES​

is_remote_url​

def is_remote_url(path) -> bool

True if path is a remote URL string that should be streamed.

open_remote_nwbfile​

def open_remote_nwbfile(url: str) -> Tuple[pynwb.NWBFile, pynwb.NWBHDF5IO]

Open a remote NWB file for lazy streaming.

Returns (nwbfile, io). Keep io open until the data you need has been read into memory, then close it.

_short_err​

def _short_err(exc: BaseException, limit: int = 400) -> str

Concise one-line error string. pynwb/hdmf errors (e.g. ConstructError) embed the whole file's builder tree, so we keep the type plus a bounded message (head + tail, since the real cause is usually at the end).

_get_client​

def _get_client(dandi_instance: str = "dandi")

Client for a DANDI instance (e.g. "dandi" or "dandi-sandbox").

It authenticates when the instance's API key environment variable is set (DANDI_API_KEY for "dandi", DANDI_SANDBOX_API_KEY for "dandi-sandbox"), which is required to read embargoed dandisets.

_resolve_assets​

def _resolve_assets(client,
                    dandiset: str,
                    version: str,
                    files: Optional[List[str]] = None,
                    limit: Optional[int] = None) -> Dict

Map asset path -> asset for the .nwb assets of a dandiset.

If files is given, only those asset paths are resolved. If limit is given, stop after collecting that many .nwb assets.

_stream_url​

def _stream_url(client, asset) -> str

S3 URL to stream an asset from.

_resolve_asset_urls​

def _resolve_asset_urls(dandiset: str,
                        version: str,
                        files: Optional[List[str]] = None,
                        limit: Optional[int] = None,
                        dandi_instance: str = "dandi") -> Dict[str, str]

Map asset path -> stream URL for the .nwb assets of a dandiset.

If files is given, only those asset paths are resolved. If limit is given, stop after collecting that many .nwb assets.

_open_asset​

def _open_asset(
    client,
    asset,
    download_dir: Optional[Path] = None
) -> Tuple[pynwb.NWBFile, pynwb.NWBHDF5IO]

Open an NWB asset by streaming it, or by downloading it into download_dir when given.

Returns (nwbfile, io); close io when done.

_find_pose_estimations​

def _find_pose_estimations(
        nwbfile: pynwb.NWBFile) -> List[Tuple[str, str, List[str]]]

Locate ndx-pose PoseEstimation containers in the file.

Scans all processing modules and returns one entry per ndx_pose.PoseEstimation as (processing_module_key, pose_estimation_key, [series_names]).

_load_movement_ds​

def _load_movement_ds(nwbfile: pynwb.NWBFile,
                      requested_series: List[str],
                      progress_prefix: str = "") -> xr.Dataset

Build a movement dataset from the file, keeping only the requested series.

Dataset construction is delegated to movement's from_nwb_file (one call per PoseEstimation container, merged); series selection is applied afterward as an xarray subset. from_nwb_file reads all keypoints, so selection here is a post-filter, not a fetch-filter. The NWB file must still be open.

dandiset_parse​

def dandiset_parse(dandiset: str,
                   version: str,
                   n_items: Optional[int] = None,
                   dandi_instance: str = "dandi") -> dict

Scan a dandiset for NWB files containing ndx-pose PoseEstimation data.

Each .nwb asset is streamed open (metadata only) and inspected for ndx_pose.PoseEstimation containers across all processing modules.

Parameters

  • dandiset (str): Dandiset identifier, e.g. "000689".
  • version (str): Published version (e.g. "0.240530.1923") or "draft".
  • n_items (int, optional): If given, scan only the first n_items NWB assets (handy for quick checks/tests). Default None scans the entire dandiset.
  • dandi_instance (str, optional): DANDI instance hosting the dandiset, e.g. "dandi" (default) or "dandi-sandbox". Embargoed dandisets need the instance's API key in the environment (DANDI_API_KEY for "dandi", DANDI_SANDBOX_API_KEY for "dandi-sandbox").

Returns

  • dict: {"valid_files": [...], "pose_estimation_series": [{"name", "count"}, ...]} where valid_files are asset paths that contain PoseEstimation data, and each entry of pose_estimation_series is a series (keypoint) name with the number of valid files it appears in, ordered by count desc.

dandiset_load​

def dandiset_load(dandiset: str,
                  version: str,
                  files: List[str],
                  pose_estimation_series: List[str],
                  out_dir: Path | str = ".",
                  dandi_instance: str = "dandi",
                  download: bool = False) -> List[str]

Stream selected files/series from a dandiset into local movement .nc files.

Only the requested files and the requested pose series (keypoints) are kept. The resulting .nc files can be passed straight to vame.init_new_project(..., source_software="movement").

Parameters

  • dandiset (str): Dandiset identifier, e.g. "000689".
  • version (str): Published version or "draft".
  • files (List[str]): Asset paths to load (e.g. the valid_files from :func:dandiset_parse).
  • pose_estimation_series (List[str]): Series (keypoint) names to include; others are not fetched.
  • out_dir (Path or str, optional): Directory to write the .nc files into (created if missing). Default ".".
  • dandi_instance (str, optional): DANDI instance hosting the dandiset, e.g. "dandi" (default) or "dandi-sandbox". Embargoed dandisets need the instance's API key in the environment (DANDI_API_KEY for "dandi", DANDI_SANDBOX_API_KEY for "dandi-sandbox").
  • download (bool, optional): If True, download each NWB file to a temporary folder and read it locally, deleting it afterwards. Faster than streaming for files that hold little besides pose data. Default False streams the files.

Returns

  • List[str]: Paths to the written .nc files, in the order of files.