`canvod.store` API reference¶

Versioned data storage using Icechunk.

Package¶

canvod-store: Icechunk storage for GNSS VOD data.

This package provides versioned storage for GNSS VOD data using Icechunk. Manages both RINEX observation storage and VOD analysis results, and stores hemisphere grids and the analysis results computed on them.

`MyIcechunkStore` ¶

Core Icechunk store manager for GNSS data.

This class encapsulates all operations on a single Icechunk repository, providing a clean interface for GNSS data storage and retrieval with integrated logging and proper resource management.

Features: - Automatic repository creation/connection - Group management with validation - Session management with context managers - Integrated logging with file contexts - Configurable compression and chunking

Note on "metadata"

This class manages two distinct things both historically called "metadata":

File registry ({group}/metadata/table): per-file ingest ledger tracking hashes, temporal ranges, filenames, and paths. Managed by append_metadata(), load_metadata(), backup_metadata_table(), etc.
Store metadata (canvod.store_metadata package): store-level provenance (identity, creator, environment, compliance). Written to Zarr root attrs by the orchestrator. See canvod-store-metadata.

Parameters:

Name	Type	Description	Default
`store_path`	`StoreLocation \| Path \| str`	The store: a local directory or `s3://bucket/prefix`.	required
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").	`"gnss_store"`
`compression_level`	`int \| None`	Override default compression level.	`None`
`compression_algorithm`	`str \| None`	Override default compression algorithm.	`None`

Attributes:

Name	Type	Description
`location`	`StoreLocation`	Where the store lives.
`store_path`	`Path \| str`	The store's local directory, or its `s3://` URL.
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").
`compression_level`	`int`	Compression level (1-9).
`compression_algorithm`	`CompressionAlgorithm`	Compression algorithm enum.
`repo`	`Repository`	The Icechunk repository instance.

`settings` `property` ¶

The settings the store writes with.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings. Reading needs none; every write does.

`repo` `property` ¶

Get the repository instance.

`site_name` `property` ¶

The site this store was written for, from its repository metadata.

Never the name of the directory the store sits in, so moving or renaming the directory doesn't change it.

Raises:

Type	Description
`ValueError`	If the store has no repository metadata.

`source_format` `property` ¶

Return the source_format root attribute, or None.

`tree` `property` ¶

Print hierarchical tree of all branches and groups. See print_tree.

`init(store_path, store_type='gnss_store', compression_level=None, compression_algorithm=None, *, config=None)` ¶

Initialize the Icechunk store manager.

Parameters:

Name	Type	Description	Default
`store_path`	`StoreLocation \| Path \| str`	The store: a local directory, `s3://bucket/prefix` or a :class:`~canvod.store_metadata.location.StoreLocation`.	required
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").	`"gnss_store"`
`compression_level`	`int \| None`	Override default compression level.	`None`
`compression_algorithm`	`str \| None`	Override default compression algorithm.	`None`
`config`	`CanvodConfig \| None`	The settings the store writes with; every commit records them. `None` loads the settings file if there is one. Without one the store can be read but not written.	`None`

`chunk_encoding_for(ds)` ¶

Build {var: {"chunks": (...)}} matching self.chunk_strategy.

Only meaningful at group creation -- Zarr fixes an array's chunk shape the moment it's first written, so this must be passed as encoding= on the mode="w" call, never on an append (which must match whatever shape creation already fixed).

Without this, chunk_strategies config only ever affected read-side chunking (the chunks= hint passed to xr.open_zarr elsewhere in this class) -- nothing applied it at write time, so the physical on-disk chunk shape was left to Zarr's own default, silently disconnected from config. Confirmed by grepping this file and processor.py for any encoding["chunks"] assignment: there wasn't one anywhere (2026-07-18).

A dim with a configured chunk size (e.g. epoch: 17280) gets exactly that size regardless of how much data is in this particular write -- Zarr supports writing less than one full chunk in the first call and filling the rest on later appends, which is the whole point of fixing chunk shape once at creation. A dim configured -1 (or unconfigured) gets one chunk spanning whatever this write's current size is (e.g. sid: -1 -> the full SID universe already present at creation).

`readonly_session(branch='main')` ¶

Thin wrapper over repo.readonly_session() for this CM.

`writable_session(branch='main')` ¶

Thin wrapper over repo.writable_session() for this CM.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings.

`set_root_attrs(attrs, branch='main')` ¶

Set root-level Zarr attributes on the store.

Parameters:

Name	Type	Description	Default
`attrs`	`dict[str, Any]`	Key-value pairs to merge into root attrs.	required
`branch`	`str`	Branch to write to.	`"main"`

Returns:

Type	Description
`str`	Snapshot ID from the commit.

`get_root_attrs(branch='main')` ¶

Read root-level Zarr attributes from the store.

Returns:

Type	Description
`dict[str, Any]`	Root attributes (empty dict if none set).

`get_branch_names()` ¶

List all branches in the store.

Returns:

Type	Description
`list[str]`	List of branch names.

`get_group_names(branch=None)` ¶

List all groups in the store.

Parameters:

Name	Type	Description	Default
`branch`	`str \| None`	Repository branch to examine. Defaults to listing groups from all branches.	`None`

Returns:

Type	Description
`dict[str, list[str]]`	Dictionary mapping branch names to lists of group names.

`list_groups(branch='main')` ¶

List all groups in a branch.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name (default: "main")	`'main'`

Returns:

Type	Description
`list[str]`	List of group names in the branch

`list_data_groups(branch='main')` ¶

List the groups that hold data, at any depth.

In a GNSS store these are the receiver groups, as :meth:list_groups gives them; in a VOD store each analysis sits below its calculator (tau_omega/canopy_01_vs_reference_01).

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name (default: "main")	`'main'`

Returns:

Type	Description
`list[str]`	The paths of the groups, sorted.

`print_tree(max_depth=None)` ¶

Display hierarchical tree of all branches, groups, and subgroups.

Parameters:

Name	Type	Description	Default
`max_depth`	`int \| None`	Maximum depth to display. None for unlimited depth. - 0: Only show branches - 1: Show branches and top-level groups - 2: Show branches, groups, and first level of subgroups/arrays - etc.	`None`

`group_exists(group_name, branch='main')` ¶

Check if a group exists.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to check. May be a nested path (e.g. VOD's `"{calculator}/{analysis_name}"`), not just a top-level name.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Returns:

Type	Description
`bool`	True if the group exists, False otherwise.

Notes

Checks membership directly against the opened root group rather than via :meth:get_group_names/:meth:list_groups -- those only enumerate top-level Zarr group keys (root.group_keys()), so for a nested group_name like a VOD analysis group, membership in that flat list is always False even when the group exists. Confirmed 2026-07-21: this previously caused every VOD write after a group's first to take the mode="w" (full overwrite) branch in :meth:write_or_append_vod_group instead of appending, silently destroying all previously-written VOD data on each subsequent write. Zarr's Group.__contains__ supports slash-separated nested paths directly, so this works for both flat GNSS receiver names (no behavior change there) and nested VOD group names (the actual fix).

`preprocessing_records(group_name, start=None, end=None, branch='main')` ¶

Preprocessing records in group_name's log book, without repeats.

In ingest order; rows written before the record existed give "". Each file row's record includes the steps of the group-wide rows written after it (:meth:add_group_variables): it describes the data as they are now. With start and end, only rows whose data overlap that range, without rows of files skipped as already ingested (as :meth:source_file_hashes). A group without a log book gives none.

`check_preprocessing_matches(group_name, dataset, branch='main')` ¶

Refuse data preprocessed differently from the data in group_name.

Compares the operations and settings of the record dataset carries (:data:~canvod.config.models.PREPROCESSING_ATTR; none counts as no preprocessing) with every record in the group's log book. The results of the operations (e.g. the input sampling) may differ. In a VOD store, the records are those of the GNSS data behind each result (vod_preprocessing_record).

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group about to receive `dataset`.	required
`dataset`	`Dataset`	Data to write.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Raises:

Type	Description
`PreprocessingMismatchError`	If the group holds data preprocessed differently.

`check_settings_match(group_name, settings, source='this run has', branch='main')` ¶

Refuse preprocessing settings unlike those of the data in group_name.

Lets a run compare its configured preprocessing (see canvod.ops.planned_settings) with a group before it processes any data; :meth:check_preprocessing_matches applies the same comparison to a dataset at write time. A group that does not exist yet accepts any settings.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group the data will be written to.	required
`settings`	`list[dict]`	Operations and settings, as `canvod.config.models.preprocessing_settings` gives them.	required
`source`	`str`	Names the new data in the error message.	`"this run has"`
`branch`	`str`	Repository branch to examine.	`"main"`

Raises:

Type	Description
`PreprocessingMismatchError`	If the group holds data preprocessed differently.

`add_group_variables(group_name, variables, step, software, branch='main')` ¶

Add variables computed from the data already in a group.

For example the grid cells of every observation (canvod.ops.assign_grid_cells_in_store). Only the new variables are written; the group's other variables, its attributes, its log book and its other subgroups stay as they are. The write is staged on a temporary branch and moved to branch with :meth:promote_branch.

The log book receives one group-wide row (:data:GROUP_WIDE_ACTION) whose preprocessing record holds step. From then on, the preprocessing of the data already in the group counts as their own plus step (:meth:preprocessing_records), so a later write is accepted only if it was preprocessed with step as well, and thus carries the new variables too.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group, e.g. `"canopy_01"` or a VOD analysis group.	required
`variables`	`Dataset`	The new variables, with the group's dimensions and coordinate values as :meth:`read_group` reads them (all of its epochs and signals, in the same order). In a group of day blocks they are written into each day's block, and the padding stays empty.	required
`step`	`dict`	The operation that computed them, as a preprocessing step `{"op", "settings", "result"}` (`canvod.ops.OpResult.step`).	required
`software`	`dict[str, str]`	Version of each package whose code computed them.	required
`branch`	`str`	Branch to change.	`"main"`

Returns:

Type	Description
`str`	The new tip of `branch`.

Raises:

Type	Description
`ValueError`	If the group does not exist, already has one of the variables, or `variables` do not match its dimensions and coordinates; or if the temporary branch already exists.
`BranchMovedError`	If `branch` moved meanwhile; the work stays on the temporary branch.

`read_group(group_name, branch='main', time_slice=None, date=None, chunks=None, drop_empty=False)` ¶

Read data from a group, its epochs in time order.

A group made of day blocks (ADR 0016) has a slot for every epoch of a day at the day's cadence; slots without data hold missing values (NaN, or the _FillValue of integer variables) at their grid times.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to read.	required
`branch`	`str`	Repository branch.	`"main"`
`time_slice`	`slice \| None`	Optional label-based time slice for filtering (passed to `ds.sel`).	`None`
`date`	`str \| None`	YYYYDOY string (e.g. `"2025001"`) selecting a single calendar day, from 00:00 up to, not including, 00:00 of the next day. Converted to a `time_slice`; mutually exclusive with `time_slice`.	`None`
`chunks`	`dict[str, Any] \| None`	Dask chunks for reading. If None, the chunks stored in the group.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read, after `time_slice` (a group written by appending has none; see :func:`~canvod.store.day_blocks.drop_empty_slots`).	`False`

Returns:

Type	Description
`Dataset`	Dataset from the group.

`write_dataset(dataset, group_name, session, mode='a', chunks=None)` ¶

Write a dataset to Icechunk with proper chunking.

Parameters:

Name	Type	Description	Default
`dataset`	`Dataset`	Dataset to write	required
`group_name`	`str`	Group path in store	required
`session`	`Any`	Active writable session or store handle.	required
`mode`	`str`	Write mode: 'w' (overwrite) or 'a' (append)	`'a'`
`chunks`	`dict[str, int] \| None`	Chunking spec. If None, uses store's chunk_strategy. Example: {'epoch': 17280, 'sid': -1}	`None`

`grow_day_blocks(days, *, message, branch='main')` ¶

Give days a block in existing day-block groups, in one commit.

A run calls this once per batch, before reading files, for the days whose Phase 1 succeeded (ADR 0016). A day that has a block keeps it, so a block grown for a day that failed later is reused on the next run. When another commit changed the branch meanwhile and the two conflict, the growth starts again on the new tip: it rereads the epoch axis, reuses the blocks added meanwhile and grows the rest.

Parameters:

Name	Type	Description	Default
`days`	`Mapping[str, Mapping[datetime64, float]]`	For each group, the cadence in seconds of each day to grow.	required
`message`	`str`	Commit message.	required
`branch`	`str`	Branch to commit to.	`"main"`

Returns:

Type	Description
`str or None`	Snapshot ID of the growth commit; `None` if every day already had a block, in which case nothing is committed.

Raises:

Type	Description
`LayoutError`	If a group does not exist (its first write creates it) or is not made of day blocks.
`CadenceError`	If a cadence does not fit the chunks, or a day has a block at a coarser cadence.
`RuntimeError`	If the growth still conflicts after `GROWTH_ATTEMPTS` attempts, with :func:`~canvod.store_metadata.io.wait_after_conflict` between them.

`backup_metadata_table(group_name, session)` ¶

Backup the metadata table to a Polars DataFrame.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name.	required
`session`	`Any`	Active session for reading.	required

Returns:

Type	Description
`DataFrame \| None`	DataFrame with metadata rows, or None if missing.

Raises:

Type	Description
`RuntimeError`	If the table exists but cannot be read.

`restore_metadata_table(group_name, df, session)` ¶

Restore the metadata table from a Polars DataFrame.

This recreates the full Zarr structure for the metadata table.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name.	required
`df`	`DataFrame`	Metadata table to restore.	required
`session`	`Any`	Active session for writing.	required

Returns:

Type	Description
`None`

`rewrite_group(session, group_name, dataset, drop_ranges, metadata_parts=None)` ¶

Rewrite group_name with dataset and keep its metadata/.

A mode="w" write deletes the group's subgroups. The log book comes back unchanged; every other metadata dataset (e.g. sbf_obs) comes back from the session's base snapshot without its epochs inside drop_ranges (the replaced data), merged in epoch order with the datasets in metadata_parts (name -> per-file datasets). A metadata dataset left without epochs is not written back. Writes into session without committing.

Raises:

Type	Description
`LayoutError`	If the group is made of day blocks: its files are written into their day's block instead, and a whole-group write would drop its layout and block index.

`get_group_info(group_name, branch='main')` ¶

Get metadata about a group.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Returns:

Type	Description
`dict[str, Any]`	Group metadata.

Raises:

Type	Description
`ValueError`	If the group does not exist.

`metadata_dataset_exists(group_name, name, branch='main')` ¶

Return True if a metadata dataset name exists for group_name.

`write_metadata_dataset(meta_ds, group_name, name, branch='main')` ¶

Write a pre-concatenated metadata dataset to {group_name}/metadata/{name}.

Always writes with mode="w" (full overwrite for the day).

Parameters:

Name	Type	Description	Default
`meta_ds`	`Dataset`	Pre-concatenated `(epoch, sid)` metadata dataset.	required
`group_name`	`str`	Target group (receiver name).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch to write to.	`"main"`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`write_metadata_parts(parts, group_name, name, session)` ¶

Add per-file metadata datasets to {group_name}/metadata/{name}.

Writes into session without committing, so the caller commits the metadata together with the observations it belongs to. Creates the dataset if it does not exist yet, otherwise appends along epoch.

Parameters:

Name	Type	Description	Default
`parts`	`list[Dataset]`	Per-file metadata datasets with an `epoch` dim, one per file whose observations are written in the same session.	required
`group_name`	`str`	Target group (the observations' store group).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`session`	`Session or ForkSession`	Open writable session of the observations' write.	required

Returns:

Type	Description
`int`	Number of epochs written.

`append_metadata_datasets(parts, group_name, name, branch='main')` ¶

Append metadata datasets in their own commit: no in-memory concat.

Creates {group_name}/metadata/{name} if it does not exist yet, otherwise appends along epoch. Prefer :meth:write_metadata_parts inside the session that writes the observations, so both land in one commit.

Parameters:

Name	Type	Description	Default
`parts`	`list[Dataset]`	Individual per-file metadata datasets with an `epoch` dim.	required
`group_name`	`str`	Target group (receiver name).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch to write to.	`"main"`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`read_metadata_dataset(group_name, name, branch='main', chunks=None)` ¶

Read a metadata dataset name for group_name.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group (receiver) name.	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch.	`"main"`
`chunks`	`dict \| None`	Dask chunks for reading. If None, the chunks stored in the dataset.	`None`

Returns:

Type	Description
`Dataset`	Lazy `(epoch, sid)` metadata dataset.

`get_metadata_dataset_info(group_name, name, branch='main')` ¶

Get info about metadata dataset name for group_name.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group (receiver) name.	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch.	`"main"`

Returns:

Type	Description
`dict[str, Any]`	Info dict with the same structure as `get_group_info()`.

Raises:

Type	Description
`ValueError`	If the metadata dataset does not exist.

`sbf_metadata_exists(group_name, branch='main')` ¶

Return True if an SBF metadata dataset exists for group_name.

`write_sbf_metadata(meta_ds, group_name, branch='main')` ¶

Write SBF metadata dataset.

`read_sbf_metadata(group_name, branch='main', chunks=None)` ¶

Read SBF metadata dataset.

`get_sbf_metadata_info(group_name, branch='main')` ¶

Get SBF metadata info.

`rel_path_for_commit(file_path)` ¶

Generate relative path for commit messages.

Parameters:

Name	Type	Description	Default
`file_path`	`Path`	Full file path.	required

Returns:

Type	Description
`str`	Relative path string with log_path_depth parts.

`get_store_stats()` ¶

Get statistics about the store.

Returns:

Type	Description
`dict[str, Any]`	Store statistics.

`append_metadata(group_name, rinex_hash, start, end, snapshot_id, action, commit_msg, dataset_attrs, branch='main', canonical_name=None, physical_path=None)` ¶

Standalone metadata-row append: commits on its own.

For ingest paths (write_initial_group, append_to_group) use _append_metadata_row directly inside the data-write session to avoid an extra commit per file.

`append_metadata_bulk(group_name, rows, session=None)` ¶

Append multiple metadata rows in one commit.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name (e.g. "canopy", "reference")	required
`rows`	`list[dict[str, Any]]`	List of metadata records matching the schema used in append_metadata().	required
`session`	`Session or ForkSession`	If provided, rows are written into this session (caller commits later). If None, this method opens its own writable session and commits once.	`None`

`should_skip_vod_write(group_name, source_file_hashes, time_start, time_end, branch='main')` ¶

Check whether a VOD write should be skipped as a duplicate.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	`{model}/{analysis_name}` group path.	required
`source_file_hashes`	`dict[str, str]`	`{receiver_name: "File Hash"}` for every receiver the computation read from.	required
`time_start`	`datetime64`	Epoch range of the incoming VOD dataset.	required
`time_end`	`datetime64`	Epoch range of the incoming VOD dataset.	required
`branch`	`str`		`"main"`

Returns:

Type	Description
`tuple[bool, str]`	`(True, "hash_match")`, `(True, "temporal_overlap")`, or `(False, "")`.

`commit_vod_days(session, forks, days, branch='main')` ¶

Merge the forks of a unit of VOD days and commit them once (ADR 0021).

forks come from session, which had no changes of its own when they were made; they hold the days the worker pool wrote (:mod:canvodpy.orchestrator.vod_phase). Under the store's commit lock: merge them, rebase onto commits made since the session was opened, write the days not in a fork (VodDayWrite.in_fork), add a log-book row per day and commit.

Returns:

Type	Description
`str or None`	Snapshot ID of the commit; `None` without days.

Raises:

Type	Description
`PreprocessingMismatchError`	If a group holds results of data preprocessed differently.
`ValueError`	If a group holds results of another calculator, or other calculator parameters.
`LayoutError`	If a day's result covers more than its one day.
`(RebaseFailedError, ConflictError)`	If another writer changed the same chunks first; nothing is committed.

`write_or_append_vod_groups_batch(items, branch='main', dedup=True)` ¶

Write/append multiple VOD analysis groups in one fork/merge batch.

The same cross-group parallelization RinexDataProcessor uses for GNSS receiver groups (_write_receiver_batch_forked), applied to VOD analysis groups: any brand-new analysis group is created directly on its own clean session that commits alone (pre-pass) -- verified empirically (see _write_receiver_batch_forked's docstring) that Session.merge() silently drops a fork's data if the base session it merges into had its own pending uncommitted writes before .fork() was called, so new-group creation and forked appends can never safely share one session. Only groups that already exist go through the fork/merge phase; a brand-new group's pre-pass write is its entire contribution to this batch (a VOD write is exactly one dataset per group per call, unlike GNSS's many-files-per-group-per- day case, so there is never leftover data for a newly-created group to append afterward).

Fail-fast: if any group's forked write raises, no merge or commit happens for the fork/merge phase -- a partial merge would silently drop a group's data from an apparently-successful commit. A pre-pass commit that already landed before a later fork failure is safe to leave in place: the next retry's dedup check sees that group's data as already-committed and won't re-attempt it.

`load_metadata(store, group_name)` ¶

Load metadata directly from Zarr into a Polars DataFrame.

Parameters:

Name	Type	Description	Default
`store`	`Any`	Zarr store or session store handle.	required
`group_name`	`str`	Group name.	required

Returns:

Type	Description
`DataFrame`	Metadata table.

`read_metadata_table(session, group_name)` ¶

Read the metadata table from a session.

Parameters:

Name	Type	Description	Default
`session`	`Any`	Active session for reading.	required
`group_name`	`str`	Group name.	required

Returns:

Type	Description
`DataFrame`	Metadata table.

`metadata_row_exists(group_name, rinex_hash, start, end, branch='main')` ¶

Check whether a file already exists or temporally overlaps existing data.

Performs two checks in order:

Hash match — if rinex_hash already appears in the metadata table, the file was previously ingested (exact duplicate).
Temporal overlap — if the incoming [start, end] interval overlaps any existing metadata interval, the file covers a time range that is already (partially) present in the store. This catches cases like a daily concatenation file coexisting with the sub-daily files it was built from.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`rinex_hash`	`str`	Hash of the current GNSS dataset.	required
`start`	`datetime64`	Start epoch of the incoming file.	required
`end`	`datetime64`	End epoch of the incoming file.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`

Returns:

Type	Description
`tuple[bool, DataFrame]`	`(True, overlapping_rows)` when the file should be skipped, `(False, empty_df)` when it is safe to ingest.

`should_skip_file(group_name, file_hash, time_start, time_end, branch='main')` ¶

Check whether a file should be skipped before processing and writing.

Thin public wrapper around :meth:metadata_row_exists that returns a simple (skip, reason) pair instead of a DataFrame, suitable for use in Airflow task functions as an early-exit optimization.

This method is an early-exit optimization: it avoids opening a write session for files that are already present. Pass dedup=True to :meth:write_or_append_group to make the store the authoritative final gate (runs the same check again before writing).

Checks performed (in order):

Hash match — file was previously ingested (exact duplicate).
Temporal overlap — incoming time range overlaps existing epochs.

Layer 3 (intra-batch overlap) is not applicable here because task functions process files one at a time.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`file_hash`	`str or None`	Hash from `dataset.attrs["File Hash"]`. When `None` the check is skipped and `(False, "")` is returned.	required
`time_start`	`datetime64`	First epoch of the incoming dataset.	required
`time_end`	`datetime64`	Last epoch of the incoming dataset.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`

Returns:

Type	Description
`tuple[bool, str]`	`(True, "hash_match")`: exact duplicate, skip. `(True, "temporal_overlap")`: overlapping time range, skip. `(False, "")`: safe to ingest.

`dir_entry_counts()` ¶

Cheap on-disk entry counts for manifests/snapshots/transactions.

A directional proxy for Icechunk-internal growth (manifest count, snapshot count) without walking the full store. Deliberately excludes chunks/ (can reach millions of entries -- not cheap) and only applies to local-filesystem-backed stores; returns {} for anything else (e.g. S3/GCS) since os.scandir doesn't apply.

`metadata_row_count(group_name, branch='main')` ¶

Row count of {group_name}/metadata/table, or 0 if absent.

Reads only the index array's shape (no full table load) -- cheap enough to log alongside every dedup check, so growth in batch_check/check_temporal_overlaps duration can be directly correlated against store size instead of assumed.

`source_file_hashes(group_name, start, end, branch='main')` ¶

Hashes of the ingested files whose data overlaps [start, end].

Read from group_name's log book, in ingest order and without duplicates. Rows recording a file that was skipped as already ingested (exists is "True" and nothing was written) are ignored, so each file counts once however often it was re-run.

Used as the provenance of a dataset read back from the store for that range, e.g. the daily inputs of a VOD computation.

`load_metadata_for_dedup(group_name, branch='main')` ¶

{group_name}'s metadata table at the tip of branch.

Read whole the first time, afterwards only the rows committed since (:class:~canvod.store_metadata.log_book.LogBooks): a run asks for it every day, and a whole read grows with the store (ADR 0021).

batch_check_existing/check_temporal_overlaps each used to call load_metadata() independently -- two full-table reads per batch, growing with store size (confirmed against a real backfill: batch_check duration tracked metadata_rows almost exactly, ~5x over 41,597 rows). Callers needing both checks should load once here and pass the result to both via their metadata_df parameter. Returns None (not an empty DataFrame) when the group/metadata doesn't exist yet, so callers can distinguish "nothing to check against" from "failed to load" without re-attempting the load.

`batch_check_existing(group_name, file_hashes, metadata_df=None)` ¶

Check which file hashes already exist in metadata.

Parameters:

Name	Type	Description	Default
`metadata_df`	`DataFrame`	Pre-loaded metadata table (see `load_metadata_for_dedup`) to skip this call's own full-table read. Loads internally (original behavior) when not provided.	`None`

`check_temporal_overlaps(group_name, file_intervals, branch='main', metadata_df=None)` ¶

Check which files temporally overlap existing metadata intervals.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`file_intervals`	`list[tuple[str, datetime64, datetime64]]`	List of `(rinex_hash, start, end)` tuples for incoming files.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`
`metadata_df`	`DataFrame`	Pre-loaded metadata table (see `load_metadata_for_dedup`) to skip this call's own full-table read. Loads internally (original behavior) when not provided.	`None`

Returns:

Type	Description
`set[str]`	Hashes of files whose `[start, end]` overlaps any existing metadata interval. Files whose hash already exists in the store are NOT included (use `batch_check_existing` for those).

`expire_old_snapshots(days=None, branch='main', delete_expired_branches=False, delete_expired_tags=False)` ¶

Expire snapshots older than the given retention period.

Metadata-only "soft delete": rewrites ancestry so expired snapshots become unreachable from any surviving branch/tag. No objects (manifests, chunks, snapshot files) are deleted at this stage — call garbage_collect() separately to reclaim storage. Keeping this expire-only (previously it also ran GC internally) is what lets a caller run GC exactly once via maintenance() instead of twice per call.

Never invoke this automatically inside the pipeline's write path: it's an administrative operation, not safe to run casually alongside active writes. Use a generous days margin (weeks to months, not hours) so it can never touch anything from an in-flight write session.

Parameters:

Name	Type	Description	Default
`days`	`int \| None`	Number of days to retain snapshots. Defaults to 90 (weeks-to- months cadence).	`None`
`branch`	`str`	Branch to apply expiration on.	`"main"`
`delete_expired_branches`	`bool`	Whether to delete branches pointing to expired snapshots. Matches Icechunk's own safe default: flip only deliberately.	`False`
`delete_expired_tags`	`bool`	Whether to delete tags pointing to expired snapshots. Do not set True if using keeper tags (`create_keeper_tag`) — it would delete them along with any other expired-tip tag.	`False`

Returns:

Type	Description
`set[str]`	Expired snapshot IDs.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings.

`garbage_collect(days=90, *, dry_run=False)` ¶

Physically delete objects no longer reachable from any branch/tag.

Only removes what a prior expire_old_snapshots() (or manual ref deletion) already made unreachable — this does not itself expire anything. Always run with dry_run=True first on a store that hasn't been GC'd before, especially one with a large accumulated backlog: there's no documented cost model for GC time/memory at scale, so a first-ever run's duration is unknown until measured.

Not safe to run casually alongside active writes — the cutoff must predate the start of any concurrently-running write session, and only one expire/GC operation should run at a time. Icechunk does have self-protection against races with new commits arriving mid-run (it restarts, bounded retries, rather than deleting something a concurrent writer needs), but that's not a substitute for a generous cutoff margin.

Parameters:

Name	Type	Description	Default
`days`	`int`	Objects older than this (by the same write-time semantics as `expire_old_snapshots`) are eligible for deletion.	`90`
`dry_run`	`bool`	Report what would be deleted without deleting anything.	`False`

Returns:

Type	Description
`dict`	Summary of (would-be-)deleted objects.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings, also for a dry run, so maintenance behaves one way.

`get_history(branch='main', limit=None)` ¶

Return commit ancestry (history) for a branch.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name.	`"main"`
`limit`	`int \| None`	Maximum number of commits to return.	`None`

Returns:

Type	Description
`list[dict]`	Commit info dictionaries (id, message, written_at, parent_ids).

`print_history(branch='main', limit=100)` ¶

Pretty-print the ancestry for quick inspection.

`get_ops_log(limit=None)` ¶

Return the repository operations log (newest first).

The ops log is an immutable audit trail of every structural mutation: commits, branch and tag operations, GC runs, config changes, etc. Unlike :meth:get_history, this is repo-wide (not per-branch) and covers all event types, not only commits.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	Maximum number of entries to return. `None` returns all.	`None`

Returns:

Type	Description
`list[dict]`	Each entry contains `kind` (str), `updated_at` (datetime), and `backup_path` (str).

`print_ops_log(limit=50)` ¶

Pretty-print the repository operations log for quick inspection.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	Maximum number of entries to display (default: 50).	`50`

`maintenance_due(expire_interval_days, gc_delay_days, manifest_count_threshold=None)` ¶

Determine whether scheduled expiration/GC/manifest compaction are due.

Used by canvodpy store maintain-due (a cron-safe, unattended entry point -- see processing.storage.maintenance config). The repo's own ops log (:meth:get_ops_log) is the source of truth for "when did maintenance last actually run": it is written atomically by Icechunk itself as part of the same operation, so it cannot drift the way a separate marker file could (deleted, stale-copied across hosts on a shared mount, or never written if a prior run crashed after the real operation completed but before a marker write). Confirmed empirically (2026-07-20, throwaway store): a real expire_old_snapshots() call always writes an ExpirationRan entry (even when zero snapshots were actually expired -- root/tip snapshots are protected); garbage_collect(dry_run=True) does not write a GCRan entry, only a real (non-dry-run) GC does -- so a dry run correctly never counts as "GC happened" here.

GC is only ever considered due after a qualifying expiration, and never again for the same expiration -- physical deletion should lag the (reversible-in-effect) soft-delete by gc_delay_days, per the skill doc's "expire every 1-2 months, GC every 15-30 days" guidance.

Parameters:

Name	Type	Description	Default
`expire_interval_days`	`int`	Re-run expiration if this many days have passed since the most recent `ExpirationRan` entry (or if there is none yet).	required
`gc_delay_days`	`int`	Run GC this many days after the most recent expiration, once per expiration (skipped if GC already ran since that expiration).	required
`manifest_count_threshold`	`int \| None`	Trigger `rewrite_manifests()` (manifest compaction) when :meth:`dir_entry_counts`'s `manifests` count reaches this. Unlike expiration/GC, `rewrite_manifests()` writes no distinguishing ops-log entry -- it logs as an ordinary `NewCommit`, indistinguishable from a data-ingest commit -- so due-ness is measured directly from current fragmentation rather than time-since-last-ops-log-entry. `None` (default) always reports `manifests_due=False` (caller opted out via `manifests_enabled=False`). Returns `False` (not an error) if the store isn't local-filesystem-backed (:meth:`dir_entry_counts` returns `{}` for S3/GCS).	`None`

Returns:

Type	Description
`dict[str, Any]`	`expire_due` (bool), `gc_due` (bool), `manifests_due` (bool), `last_expire` (datetime \| None), `last_gc` (datetime \| None), `manifest_count` (int \| None).

`compact_manifests(branch='main')` ¶

Consolidate all of a branch's fragmented manifests in one commit.

Thin wrapper over repo.rewrite_manifests(): canvodpy's write pattern (one to_icechunk() call per file within a batch, e.g. 96 fifteen-minute files/receiver-day) creates one new manifest per flush, not per commit -- a single day's ingest for one receiver can create ~96 manifests. This rewrites them back to the currently-configured splitting boundaries.

Uses commit_method="new_commit" deliberately, not "amend": amend replaces the immediately-preceding commit's own message/ metadata rather than merging with it, and _append_to_icechunk already attaches structured provenance (receiver, date, file hashes) to every ingest commit -- amending could silently clobber that. An explicit, separate, auditable commit matches the precedent expire/GC already set. Like every commit, it records software, environment and settings as commit metadata (:func:canvod.store_metadata.rewrite_manifests).

Like expire_old_snapshots()/garbage_collect(), this is an administrative operation -- never call it while a pipeline run is active against this branch (ordinary commit conflict-retry semantics apply, but a still-draining worker pool committing per-file would just make every retry lose, burning cycles for nothing).

Does not shrink on-disk manifest count by itself -- confirmed empirically (2026-07-20, throwaway store, 300 commits): compaction adds a new, consolidated manifest set but leaves the old fragmented ones on disk (399 -> 401 manifests immediately after compacting); only a subsequent expire_old_snapshots() + garbage_collect() actually reclaims them (399 -> 2 in the same test). This mirrors GC's own relationship to expiration -- don't expect dir_entry_counts()['manifests'] to drop right after calling this; it drops after the next expire/GC cycle instead.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch to compact.	`"main"`

Returns:

Type	Description
`str`	Snapshot ID of the compaction commit.

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`str()` ¶

Return a human-readable summary.

Returns:

Type	Description
`str`	Summary string.

`rechunk_group(group_name, chunks, source_branch='main', temp_branch=None, promote_to_main=True, delete_temp_branch=True)` ¶

Rechunk a group with optimal chunk sizes.

Zarr arrays have a fixed chunk shape once created — writing with mode="r+" onto an existing array cannot change chunk boundaries, it can only update values within the existing grid (verified 2026-07-08: a previous version of this method used mode="r+", which silently failed to change on-disk chunking whenever the request didn't align with the current chunks). Actually changing chunk size requires a destructive mode="w" rewrite of the group, which also wipes the group's nested subgroups (metadata/ table, metadata/sbf_obs) — so this method copies that subtree back afterward via a raw recursive Zarr copy (see _copy_zarr_subtree), all on a temporary branch, before main is ever touched. The copy handles canvod-store's two-level-deep metadata/{table,sbf_obs} nesting.

Both the data read and the subtree copy are pinned to the exact snapshot captured at the start of this call. If source_branch moves before the promotion, the promotion is refused with BranchMovedError (the new commits would be dropped) and the rechunked data stays on temp_branch.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to rechunk	required
`chunks`	`dict[str, int]`	Chunking specification, e.g. {'epoch': 180, 'sid': -1}. A group stored in day blocks needs an epoch chunk that divides every day into whole chunks.	required
`source_branch`	`str`	Branch to read original data from (default: "main")	`'main'`
`temp_branch`	`str \| None`	Temporary branch name for rechunked data. If None, uses "{group_name}_rechunked".	`None`
`promote_to_main`	`bool`	If True, move `source_branch` to the rechunked snapshot after writing (see `promote_branch`).	`True`
`delete_temp_branch`	`bool`	If True, delete temporary branch after promotion (only if promote_to_main=True).	`True`

Returns:

Type	Description
`str`	Snapshot ID of the rechunked data

Raises:

Type	Description
`ValueError`	If `temp_branch` already exists, or a day-block group gets no positive epoch chunk.
`CadenceError`	If the epoch chunks would split a day of a day-block group.
`BranchMovedError`	If `source_branch` moved during the rechunk.

`promote_branch(temp_branch, target_branch, expected_tip)` ¶

Move target_branch to the tip of temp_branch.

Only if target_branch still points at expected_tip, the snapshot temp_branch was created from. Otherwise other commits landed on target_branch meanwhile and moving it would drop them.

Parameters:

Name	Type	Description	Default
`temp_branch`	`str`	Branch holding the staged work.	required
`target_branch`	`str`	Branch to move (usually `"main"`).	required
`expected_tip`	`str`	Snapshot ID `target_branch` pointed at when `temp_branch` was created.	required

Returns:

Type	Description
`str`	The new tip of `target_branch`.

Raises:

Type	Description
`BranchMovedError`	If `target_branch` moved. `temp_branch` is left in place.

`create_branch(branch_name, snapshot_id)` ¶

Create a branch pointing at a specific snapshot.

Accepts the full snapshot ID or a prefix of it, such as the 8 characters plot_commit_graph() shows (case-insensitive). The snapshot can be on any branch or tag.

Parameters:

Name	Type	Description	Default
`branch_name`	`str`	Name of the new branch (e.g. `"explore/my-analysis"`).	required
`snapshot_id`	`str`	Full snapshot ID or a prefix of it.	required

Raises:

Type	Description
`ValueError`	If the ID is empty, no snapshot on a branch or tag has this ID or prefix, or the prefix fits several.

`create_release_tag(tag_name, snapshot_id=None)` ¶

Create an immutable tag for an important version.

Parameters:

Name	Type	Description	Default
`tag_name`	`str`	Name for the tag (e.g., "v2024_complete", "before_reprocess")	required
`snapshot_id`	`str \| None`	Snapshot to tag. If None, uses current tip of main branch.	`None`

`list_tags()` ¶

List all tags in the repository.

`delete_tag(tag_name)` ¶

Delete a tag (use with caution - tags are meant to be permanent).

`create_keeper_tag(receiver, yyyydoy, snapshot_id)` ¶

Tag a per-(receiver, date) commit as a permanent retention keeper.

Part of the tag-based retention scheme: tagging keep/{receiver}/{yyyydoy} after each commit means a later expire_old_snapshots(delete_expired_tags=False) run can safely prune everything else while these specific snapshots stay time-travelable forever. Additive and inert on its own — creating a tag doesn't expire or delete anything.

Never raises: a tag name collision (already exists, or was previously deleted — Icechunk tag names are never reusable) is logged and skipped rather than failing the pipeline write that called this.

Parameters:

Name	Type	Description	Default
`receiver`	`str`	Receiver/group name (e.g. `"canopy_01"`).	required
`yyyydoy`	`str`	Date key (e.g. `"2025001"`).	required
`snapshot_id`	`str`	Snapshot to tag — typically the just-committed snapshot.	required

Returns:

Type	Description
`bool`	True if the tag was created, False if skipped.

`plot_commit_graph(branch=None, *, plain=False)` ¶

Visualize commit history as a graph.

Delegates to icechunk's native repo.ancestry_graph(). Renders as an SVG diagram in Jupyter/marimo notebooks, or as colored text in a terminal (print(store.plot_commit_graph())).

Parameters:

Name	Type	Description	Default
`branch`	`str`	Restrict the graph to this branch's linear history. When omitted, all branches are shown as a tree.	`None`
`plain`	`bool`	Render without ANSI colors or SVG fill colors (useful for CI logs or piping to a file).	`False`

Returns:

Type	Description
`AncestryGraph`	Displayable object with `_repr_svg_()` for notebooks.

`cleanup_stale_branches(keep_patterns=None)` ¶

Delete stale temporary branches (e.g., from failed rechunking).

Parameters:

Name	Type	Description	Default
`keep_patterns`	`list[str] \| None`	Patterns to preserve. Default: ["main", "dev"]	`None`

Returns:

Type	Description
`list[str]`	Names of deleted branches

`delete_branch(branch_name)` ¶

Delete a branch.

`get_snapshot_info(snapshot_id)` ¶

Get detailed information about a specific snapshot.

Parameters:

Name	Type	Description	Default
`snapshot_id`	`str`	Snapshot ID to inspect	required

Returns:

Type	Description
`dict`	Snapshot metadata and statistics

`compare_snapshots(snapshot_id_1, snapshot_id_2)` ¶

Compare two snapshots to see what changed.

Parameters:

Name	Type	Description	Default
`snapshot_id_1`	`str`	First snapshot (older)	required
`snapshot_id_2`	`str`	Second snapshot (newer)	required

Returns:

Type	Description
`dict`	Comparison results showing added/removed/modified groups

`maintenance(expire_days=90, cleanup_branches=True, run_gc=True, *, dry_run_gc=False, delete_expired_branches=False, delete_expired_tags=False)` ¶

Run full maintenance on the store: expire, then (optionally) GC.

Must be invoked manually or from a separately scheduled maintenance job: never from the pipeline's own run path, and never on an automatic schedule until garbage_collect(dry_run=True) has been checked against the real store and, if using keeper tags, create_tag() cost at scale has been confirmed. Not safe during active writes — see expire_old_snapshots/ garbage_collect docstrings for the concurrency caveats.

Parameters:

Name	Type	Description	Default
`expire_days`	`int`	Days of snapshot history to keep. Default 90 (weeks-to-months cadence, not the previous 7-day default — 7 days is far too aggressive for a store under sustained daily writes).	`90`
`cleanup_branches`	`bool`	Remove stale temporary branches.	`True`
`run_gc`	`bool`	Run garbage collection after expiration.	`True`
`dry_run_gc`	`bool`	Pass through to `garbage_collect` — report without deleting.	`False`
`delete_expired_branches`	`bool`	Pass through to `expire_old_snapshots`.	`False`
`delete_expired_tags`	`bool`	Pass through to `expire_old_snapshots`. Do not set True if using keeper tags — see `create_keeper_tag`.	`False`

Returns:

Type	Description
`dict`	Summary of maintenance actions.

`GnssResearchSite` ¶

High-level manager for a GNSS research site with dual Icechunk stores.

This class coordinates between RINEX data storage (Level 1) and VOD analysis storage (Level 2), providing a unified interface for site-wide operations.

Architecture: - GNSS Store: Raw/standardized observations per receiver - VOD Store: Analysis products comparing receiver pairs

Features: - Automatic store initialization from config - Receiver management and validation - Analysis workflow coordination - Unified logging and error handling

Parameters:

Name	Type	Description	Default
`site_name`	`str`	Name of the research site (must exist in config).	required

Raises:

Type	Description
`KeyError`	If `site_name` is not found in the RESEARCH_SITES config.

`gnss_store` `cached` `property` ¶

The site's GNSS store, opened on first use.

Raises:

Type	Description
`FileNotFoundError`	If the store doesn't exist yet; a run creates it.

`vod_store` `cached` `property` ¶

The site's VOD store, opened on first use.

Raises:

Type	Description
`FileNotFoundError`	If the store doesn't exist yet; a run creates it.

`site_config` `property` ¶

Get the site configuration as a dictionary.

`receivers` `property` ¶

Get all configured receivers for this site.

`active_receivers` `property` ¶

Get only active receivers for this site.

`vod_analyses` `property` ¶

Get all configured VOD analyses for this site.

Returns auto-derived analyses from scs_from if no explicit vod_analyses are configured.

`active_vod_analyses` `property` ¶

Get only active VOD analyses for this site.

`init(site_name)` ¶

Initialize the site manager.

Parameters:

Name	Type	Description	Default
`site_name`	`str`	Name of the research site.	required

`add_receiver_metadata(dataset, group)` ¶

dataset with the receiver's metadata setting in its attributes.

Attributes the dataset already has are kept; a key of the setting that clashes with one is left out, with a warning. dataset itself is not changed: receivers can share the datasets of one file.

Parameters:

Name	Type	Description	Default
`dataset`	`Dataset`	Data of the receiver.	required
`group`	`str`	The GNSS store group the data go to: a canopy receiver's name, or a reference receiver's group per canopy (`reference_01_canopy_01`), whose setting is `reference_01`'s.	required

`get_receiver_metadata(receiver_name)` ¶

Get freeform metadata for a receiver.

Returns the metadata dict of the receiver in the sites: section of canvod-settings.yaml, or None if not configured.

Parameters:

Name	Type	Description	Default
`receiver_name`	`str`	Name of the receiver.	required

Returns:

Type	Description
`dict or None`	Receiver metadata dict, or None.

`from_gnss_store_path(gnss_store_path)` `classmethod` ¶

Create a GnssResearchSite instance from a GNSS store path.

Parameters:

Name	Type	Description	Default
`gnss_store_path`	`Path`	Path to the RINEX Icechunk store.	required

Returns:

Type	Description
`GnssResearchSite`	Initialized research site manager.

Raises:

Type	Description
`ValueError`	If no matching site is found for the given path.

`get_reference_canopy_pairs()` ¶

Expand scs_from into (reference_name, canopy_name) pairs.

Returns:

Type	Description
`list[tuple[str, str]]`	List of (reference_name, canopy_name) pairs.

`get_auto_vod_analyses()` ¶

Derive VOD analysis pairs from scs_from configuration.

Creates one VOD pair per (canopy, reference_for_that_canopy) combination.

Returns:

Type	Description
`dict[str, VodAnalysisConfig]`	Auto-derived VOD analyses keyed by analysis name.

`validate_site_config()` ¶

Validate that the site configuration is consistent.

Returns:

Type	Description
`bool`	True if configuration is valid.

Raises:

Type	Description
`ValueError`	If configuration is invalid.

`get_receiver_groups()` ¶

Get list of receiver groups that exist in the GNSS store.

Returns:

Type	Description
`list[str]`	Existing receiver group names.

`get_vod_analysis_groups()` ¶

Get list of VOD result groups in the VOD store.

The VOD store nests results as {model}/{analysis_name} (one model per registered VODFactory calculator, disambiguating results from different calculators over the same analysis pair). Top-level list_groups() only returns model names, so this walks one level deeper.

Returns:

Type	Description
`list[str]`	`"{model}/{analysis_name}"` strings for every existing result.

`ensure_store_metadata(store_type, names, source_format)` ¶

Make a store of the site ready for data: the step every run takes first.

Gives a new store its repository metadata from the settings and refuses groups the metadata doesn't list (ADR 0014).

Parameters:

Name	Type	Description	Default
`store_type`	`str`	`"gnss_store"` or `"vod_store"`.	required
`names`	`Sequence[str]`	The groups about to be written: receivers and their pair groups, or VOD analyses.	required
`source_format`	`str \| None`	The reader format (GNSS store) or VOD calculator (VOD store) a new store records; `None` if unknown, which only a store with metadata accepts.	required

Returns:

Type	Description
`dict`	The store's repository metadata.

Raises:

Type	Description
`ValueError`	If the store is new and `source_format` is `None`, the store predates repository metadata (names the migrate command), or a group is not listed in it (names the amend command).

`source_file_hashes_for(group_name, ds)` ¶

Comma-joined hashes of the GNSS files behind ds.

ds is data read from GNSS store group group_name. The hashes come from that group's log book, for every ingested file overlapping ds's epoch range (see MyIcechunkStore.source_file_hashes), so a daily dataset built from many files lists all of them. Falls back to ds.attrs["File Hash"] when the log book has no matching rows.

`preprocessing_records_for(group_name, ds)` ¶

Preprocessing records of the GNSS data behind ds.

ds is data read from GNSS store group group_name; the records come from that group's log book, for the files overlapping ds's epoch range (as :meth:source_file_hashes_for).

`read_receiver_data(receiver_name, time_range=None, drop_empty=True)` ¶

Read data from a specific receiver, its epochs in time order.

Parameters:

Name	Type	Description	Default
`receiver_name`	`str`	Name of the receiver.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the data.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read: the receiver's epochs with data. `False` keeps every slot of a day-block group (see :meth:`MyIcechunkStore.read_group`).	`True`

Returns:

Type	Description
`Dataset`	Dataset containing receiver observations.

Raises:

Type	Description
`ValueError`	If the receiver group does not exist.

`read_vod_analysis(analysis_name, calculator_name, time_range=None, drop_empty=True)` ¶

Read VOD analysis results for one calculator model.

VOD written as day blocks (ADR 0016) has a slot for every epoch of a day at the cadence of its GNSS data; slots without a VOD value hold NaN at their grid times.

Parameters:

Name	Type	Description	Default
`analysis_name`	`str`	Name of the analysis.	required
`calculator_name`	`str`	Registered `VODFactory` name the results were computed with.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the results.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read, after `time_range`: the epochs with VOD. `False` keeps every slot of a day-block group (see :meth:`~canvod.store.store.MyIcechunkStore.read_group`).	`True`

Returns:

Type	Description
`Dataset`	Dataset containing VOD analysis results.

Raises:

Type	Description
`ValueError`	If the analysis group does not exist.

`prepare_vod_input_data(analysis_name, time_range=None)` ¶

Prepare aligned input data for VOD analysis.

Reads data from both receivers specified in the analysis configuration and returns them aligned for VOD processing.

Parameters:

Name	Type	Description	Default
`analysis_name`	`str`	Name of the VOD analysis configuration.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the data.	`None`

Returns:

Type	Description
`tuple of (xr.Dataset, xr.Dataset)`	Tuple of (canopy_dataset, reference_dataset).

Raises:

Type	Description
`ValueError`	If the analysis is not configured or data is missing.

`get_site_summary()` ¶

Get a comprehensive summary of the research site.

Returns:

Type	Description
`dict`	Dictionary with site statistics, data availability, and store paths.

`is_day_complete(yyyydoy, receiver_types=None, completeness_threshold=0.95)` ¶

Check if a day has complete data coverage for all receiver types.

Parameters:

Name	Type	Description	Default
`yyyydoy`	`str`	Date in YYYYDOY format (e.g., "2024256")	required
`receiver_types`	`List[str]`	Receiver types to check. Defaults to ['canopy', 'reference']	`None`
`completeness_threshold`	`float`	Fraction of expected epochs that must exist (default 0.95 = 95%) Allows for small gaps due to receiver issues	`0.95`

Returns:

Type	Description
`bool`	True if all receiver types have complete data for this day

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`str()` ¶

Return a human-readable summary.

Returns:

Type	Description
`str`	Summary string.

`AnalysisStorage` ¶

Manage persistent storage of analysis results for dataset+grid pairs.

Storage layout inside the Icechunk repository::

metadata/{dataset_name}/{grid_name}/
├── weights/              # (ncells,)
│   ├── observation_count
│   ├── solid_angle
│   └── combined
├── filter_masks/         # (epoch, sid)
│   ├── mask_iqr
│   └── mask_zscore
├── spatial_masks/        # (ncells,)
│   ├── mask_north
│   └── mask_high_elevation
└── statistics/           # (ncells,)
    ├── obs_count
    ├── mean_vod
    └── std_vod

Parameters:

Name	Type	Description	Default
`store_path`	`Path or str`	Path to the VOD Icechunk store directory.	required

`init(store_path)` ¶

Initialize the storage manager.

Parameters:

Name	Type	Description	Default
`store_path`	`Path \| str`	Path to the VOD Icechunk store directory.	required

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`store_weights(dataset_name, grid_name, weights, weight_params=None, overwrite=False)` ¶

Store per-cell weight arrays.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier (e.g. `'reference_01_canopy_01'`).	required
`grid_name`	`str`	Grid identifier (e.g. `'equal_area_2deg'`).	required
`weights`	`dict`	`{name: array}` – all arrays must have shape `(ncells,)`.	required
`weight_params`	`dict`	Parameters used to compute each weight type.	`None`
`overwrite`	`bool`	Overwrite existing weights.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_weights(dataset_name, grid_name, weight_type=None)` ¶

Load stored weight arrays.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`weight_type`	`str`	Load only this weight. `None` loads all.	`None`

Returns:

Type	Description
`dict`	`{name: ndarray}` of loaded weights.

`has_weights(dataset_name, grid_name)` ¶

Return True if weights exist for the dataset+grid pair.

`store_filter_masks(dataset_name, grid_name, masks, filter_params=None, overwrite=False)` ¶

Store per-observation filter masks at native resolution.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`masks`	`dict`	`{filter_name: DataArray}` – all must share the same `(epoch, sid)` shape.	required
`filter_params`	`dict`	Parameters used for each filter.	`None`
`overwrite`	`bool`	Overwrite existing masks.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_filter_mask(dataset_name, grid_name, filter_type, attach_coords=True)` ¶

Load a single filter mask.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`filter_type`	`str`	Filter name (e.g. `'iqr'`, `'zscore'`).	required
`attach_coords`	`bool`	Re-attach `epoch` / `sid` coordinates from the source dataset group. Set `False` for faster loading when coordinates are not needed.	`True`

Returns:

Type	Description
`DataArray`	Boolean mask with shape `(epoch, sid)`.

`load_all_filter_masks(dataset_name, grid_name)` ¶

Load all stored filter masks.

Returns:

Type	Description
`dict`	`{filter_name: DataArray}` – boolean masks.

`has_filter_masks(dataset_name, grid_name)` ¶

Return True if filter masks exist for the dataset+grid pair.

`store_spatial_masks(dataset_name, grid_name, masks, mask_descriptions=None, overwrite=False)` ¶

Store per-cell geometric selection masks.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`masks`	`dict`	`{name: bool_array}` – all arrays must have shape `(ncells,)` and boolean dtype.	required
`mask_descriptions`	`dict`	Human-readable description for each mask (stored as variable attrs).	`None`
`overwrite`	`bool`	Overwrite existing masks.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_spatial_mask(dataset_name, grid_name, mask_name)` ¶

Load a single spatial mask.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`mask_name`	`str`	Mask name (e.g. `'north'`, `'high_elevation'`).	required

Returns:

Type	Description
`ndarray`	Boolean array with shape `(ncells,)`.

`load_all_spatial_masks(dataset_name, grid_name)` ¶

Load all spatial masks.

Returns:

Type	Description
`dict`	`{name: bool_ndarray}`.

`has_spatial_masks(dataset_name, grid_name)` ¶

Return True if spatial masks exist for the dataset+grid pair.

`store_statistics(dataset_name, grid_name, stats, overwrite=False)` ¶

Store pre-computed per-cell statistics.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`stats`	`dict`	`{name: array}` – all arrays must have shape `(ncells,)`. Variables whose name ends with `'_count'` or equals `'obs_count'` are stored as `int64`; everything else as `float32`.	required
`overwrite`	`bool`	Overwrite existing statistics.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_statistics(dataset_name, grid_name, stat_name=None)` ¶

Load pre-computed per-cell statistics.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`stat_name`	`str`	Load only this statistic. `None` loads all.	`None`

Returns:

Type	Description
`dict`	`{name: ndarray}`.

`has_statistics(dataset_name, grid_name)` ¶

Return True if statistics exist for the dataset+grid pair.

`list_available_metadata(dataset_name, grid_name)` ¶

Check which metadata categories are stored.

Returns:

Type	Description
`dict`	`{category: bool}` for weights, filter_masks, spatial_masks, statistics.

`get_metadata_summary(dataset_name, grid_name)` ¶

Detailed summary of all stored metadata for a dataset+grid pair.

Returns:

Type	Description
`dict`	Nested summary with availability flags and per-category details.

`delete_weights(dataset_name, grid_name)` ¶

Delete all weights for a dataset+grid pair.

`delete_filter_masks(dataset_name, grid_name)` ¶

Delete all filter masks for a dataset+grid pair.

`delete_spatial_masks(dataset_name, grid_name)` ¶

Delete all spatial masks for a dataset+grid pair.

`delete_statistics(dataset_name, grid_name)` ¶

Delete all statistics for a dataset+grid pair.

`store_percell_timeseries(percell_ds, system, band, code, branch='per_cell', metadata=None)` ¶

Store a per-cell timeseries dataset for a system/band/code combination.

Parameters:

Name	Type	Description	Default
`percell_ds`	`Dataset`	Per-cell timeseries dataset with dims `(cell, time)`.	required
`system`	`str`	GNSS system prefix (e.g. `'G'`, `'E'`).	required
`band`	`str`	Frequency band (e.g. `'L1'`, `'E1'`).	required
`code`	`str`	Tracking code (e.g. `'C'`, `'L'`).	required
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`
`metadata`	`dict`	Additional metadata to attach to the dataset.	`None`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_percell_timeseries(system, band, code, branch='per_cell')` ¶

Load a per-cell timeseries dataset.

Parameters:

Name	Type	Description	Default
`system`	`str`	GNSS system prefix.	required
`band`	`str`	Frequency band.	required
`code`	`str`	Tracking code.	required
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`

Returns:

Type	Description
`Dataset`	Per-cell timeseries dataset.

Raises:

Type	Description
`ValueError`	If the requested group does not exist.

`list_percell_datasets(branch='per_cell')` ¶

List all per-cell timeseries datasets in the store.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`

Returns:

Type	Description
`list of str`	Sorted list of group names in `system_band_code` format.

`has_percell_timeseries(system, band, code, branch='per_cell')` ¶

Return True if a per-cell timeseries exists for the given combination.

`delete_all_metadata(dataset_name, grid_name)` ¶

Delete the entire metadata subtree for a dataset+grid pair.

`delete_specific_weight(dataset_name, grid_name, weight_name)` ¶

Delete a single weight variable from an existing weights group.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`weight_name`	`str`	Weight variable name (e.g. `'observation_count'`).	required

Returns:

Type	Description
`str`	Snapshot ID.

`delete_specific_filter_mask(dataset_name, grid_name, filter_type)` ¶

Delete a single filter mask variable from an existing filter_masks group.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`filter_type`	`str`	Filter type name (e.g. `'iqr'`).	required

Returns:

Type	Description
`str`	Snapshot ID.

`open_gnss_store(store_path)` ¶

Open the GNSS store at store_path; never create it.

A run creates a site's stores when it first writes to them. Reading, branches and tags need no settings file; commits, expiring snapshots and garbage collection do. A store opened without one refuses them with a FileNotFoundError that names canvodpy config init.

Raises:

Type	Description
`FileNotFoundError`	If no store is at `store_path`.

`open_vod_store(store_path)` ¶

Open the VOD store at store_path; never create it.

A run creates a site's stores when it first writes to them. Reading, branches and tags need no settings file; commits, expiring snapshots and garbage collection do. A store opened without one refuses them with a FileNotFoundError that names canvodpy config init.

Raises:

Type	Description
`FileNotFoundError`	If no store is at `store_path`.

`store_grid(grid, store, grid_name)` ¶

Store grid in unified xarray format to Icechunk store.

Converts the grid to an xarray Dataset with vertex information and writes it to the grids/ group in the store.

Parameters:

Name	Type	Description	Default
`grid`	`GridData`	Hemisphere grid instance to store.	required
`store`	`MyIcechunkStore`	Icechunk store to write to.	required
`grid_name`	`str`	Grid identifier for storage path (e.g., 'equal_area_4deg').	required

Returns:

Type	Description
`str`	Snapshot ID from the commit.

Examples:

>>> from canvod.grids import create_hemigrid
>>> from canvod.store import store_grid
>>> grid = create_hemigrid(angular_resolution=4, grid_type='equal_area')
>>> snapshot_id = store_grid(grid, my_store, 'equal_area_4deg')

`load_grid(store, grid_name)` ¶

Load a grid from Icechunk store.

Loads the grid structure from grids/{grid_name} and reconstructs a GridData object.

Parameters:

Name	Type	Description	Default
`store`	`MyIcechunkStore`	Icechunk store to read from.	required
`grid_name`	`str`	Grid identifier (e.g., 'equal_area_4deg').	required

Returns:

Type	Description
`GridData`	Reconstructed grid instance.

Examples:

>>> from canvod.store import load_grid
>>> grid = load_grid(my_store, 'equal_area_4deg')
>>> print(f"Loaded {grid.ncells} cells")

Store manager¶

Research site manager that coordinates RINEX and VOD Icechunk stores.

This module provides the GnssResearchSite class that manages both stores for a research site and provides high-level operations across them.

`GnssResearchSite` ¶

High-level manager for a GNSS research site with dual Icechunk stores.

This class coordinates between RINEX data storage (Level 1) and VOD analysis storage (Level 2), providing a unified interface for site-wide operations.

Architecture: - GNSS Store: Raw/standardized observations per receiver - VOD Store: Analysis products comparing receiver pairs

Features: - Automatic store initialization from config - Receiver management and validation - Analysis workflow coordination - Unified logging and error handling

Parameters:

Name	Type	Description	Default
`site_name`	`str`	Name of the research site (must exist in config).	required

Raises:

Type	Description
`KeyError`	If `site_name` is not found in the RESEARCH_SITES config.

`gnss_store` `cached` `property` ¶

The site's GNSS store, opened on first use.

Raises:

Type	Description
`FileNotFoundError`	If the store doesn't exist yet; a run creates it.

`vod_store` `cached` `property` ¶

The site's VOD store, opened on first use.

Raises:

Type	Description
`FileNotFoundError`	If the store doesn't exist yet; a run creates it.

`site_config` `property` ¶

Get the site configuration as a dictionary.

`receivers` `property` ¶

Get all configured receivers for this site.

`active_receivers` `property` ¶

Get only active receivers for this site.

`vod_analyses` `property` ¶

Get all configured VOD analyses for this site.

Returns auto-derived analyses from scs_from if no explicit vod_analyses are configured.

`active_vod_analyses` `property` ¶

Get only active VOD analyses for this site.

`init(site_name)` ¶

Initialize the site manager.

Parameters:

Name	Type	Description	Default
`site_name`	`str`	Name of the research site.	required

`add_receiver_metadata(dataset, group)` ¶

dataset with the receiver's metadata setting in its attributes.

Attributes the dataset already has are kept; a key of the setting that clashes with one is left out, with a warning. dataset itself is not changed: receivers can share the datasets of one file.

Parameters:

Name	Type	Description	Default
`dataset`	`Dataset`	Data of the receiver.	required
`group`	`str`	The GNSS store group the data go to: a canopy receiver's name, or a reference receiver's group per canopy (`reference_01_canopy_01`), whose setting is `reference_01`'s.	required

`get_receiver_metadata(receiver_name)` ¶

Get freeform metadata for a receiver.

Returns the metadata dict of the receiver in the sites: section of canvod-settings.yaml, or None if not configured.

Parameters:

Name	Type	Description	Default
`receiver_name`	`str`	Name of the receiver.	required

Returns:

Type	Description
`dict or None`	Receiver metadata dict, or None.

`from_gnss_store_path(gnss_store_path)` `classmethod` ¶

Create a GnssResearchSite instance from a GNSS store path.

Parameters:

Name	Type	Description	Default
`gnss_store_path`	`Path`	Path to the RINEX Icechunk store.	required

Returns:

Type	Description
`GnssResearchSite`	Initialized research site manager.

Raises:

Type	Description
`ValueError`	If no matching site is found for the given path.

`get_reference_canopy_pairs()` ¶

Expand scs_from into (reference_name, canopy_name) pairs.

Returns:

Type	Description
`list[tuple[str, str]]`	List of (reference_name, canopy_name) pairs.

`get_auto_vod_analyses()` ¶

Derive VOD analysis pairs from scs_from configuration.

Creates one VOD pair per (canopy, reference_for_that_canopy) combination.

Returns:

Type	Description
`dict[str, VodAnalysisConfig]`	Auto-derived VOD analyses keyed by analysis name.

`validate_site_config()` ¶

Validate that the site configuration is consistent.

Returns:

Type	Description
`bool`	True if configuration is valid.

Raises:

Type	Description
`ValueError`	If configuration is invalid.

`get_receiver_groups()` ¶

Get list of receiver groups that exist in the GNSS store.

Returns:

Type	Description
`list[str]`	Existing receiver group names.

`get_vod_analysis_groups()` ¶

Get list of VOD result groups in the VOD store.

The VOD store nests results as {model}/{analysis_name} (one model per registered VODFactory calculator, disambiguating results from different calculators over the same analysis pair). Top-level list_groups() only returns model names, so this walks one level deeper.

Returns:

Type	Description
`list[str]`	`"{model}/{analysis_name}"` strings for every existing result.

`ensure_store_metadata(store_type, names, source_format)` ¶

Make a store of the site ready for data: the step every run takes first.

Gives a new store its repository metadata from the settings and refuses groups the metadata doesn't list (ADR 0014).

Parameters:

Name	Type	Description	Default
`store_type`	`str`	`"gnss_store"` or `"vod_store"`.	required
`names`	`Sequence[str]`	The groups about to be written: receivers and their pair groups, or VOD analyses.	required
`source_format`	`str \| None`	The reader format (GNSS store) or VOD calculator (VOD store) a new store records; `None` if unknown, which only a store with metadata accepts.	required

Returns:

Type	Description
`dict`	The store's repository metadata.

Raises:

Type	Description
`ValueError`	If the store is new and `source_format` is `None`, the store predates repository metadata (names the migrate command), or a group is not listed in it (names the amend command).

`source_file_hashes_for(group_name, ds)` ¶

Comma-joined hashes of the GNSS files behind ds.

ds is data read from GNSS store group group_name. The hashes come from that group's log book, for every ingested file overlapping ds's epoch range (see MyIcechunkStore.source_file_hashes), so a daily dataset built from many files lists all of them. Falls back to ds.attrs["File Hash"] when the log book has no matching rows.

`preprocessing_records_for(group_name, ds)` ¶

Preprocessing records of the GNSS data behind ds.

ds is data read from GNSS store group group_name; the records come from that group's log book, for the files overlapping ds's epoch range (as :meth:source_file_hashes_for).

`read_receiver_data(receiver_name, time_range=None, drop_empty=True)` ¶

Read data from a specific receiver, its epochs in time order.

Parameters:

Name	Type	Description	Default
`receiver_name`	`str`	Name of the receiver.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the data.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read: the receiver's epochs with data. `False` keeps every slot of a day-block group (see :meth:`MyIcechunkStore.read_group`).	`True`

Returns:

Type	Description
`Dataset`	Dataset containing receiver observations.

Raises:

Type	Description
`ValueError`	If the receiver group does not exist.

`read_vod_analysis(analysis_name, calculator_name, time_range=None, drop_empty=True)` ¶

Read VOD analysis results for one calculator model.

VOD written as day blocks (ADR 0016) has a slot for every epoch of a day at the cadence of its GNSS data; slots without a VOD value hold NaN at their grid times.

Parameters:

Name	Type	Description	Default
`analysis_name`	`str`	Name of the analysis.	required
`calculator_name`	`str`	Registered `VODFactory` name the results were computed with.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the results.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read, after `time_range`: the epochs with VOD. `False` keeps every slot of a day-block group (see :meth:`~canvod.store.store.MyIcechunkStore.read_group`).	`True`

Returns:

Type	Description
`Dataset`	Dataset containing VOD analysis results.

Raises:

Type	Description
`ValueError`	If the analysis group does not exist.

`prepare_vod_input_data(analysis_name, time_range=None)` ¶

Prepare aligned input data for VOD analysis.

Reads data from both receivers specified in the analysis configuration and returns them aligned for VOD processing.

Parameters:

Name	Type	Description	Default
`analysis_name`	`str`	Name of the VOD analysis configuration.	required
`time_range`	`tuple of datetime`	(start_time, end_time) for filtering the data.	`None`

Returns:

Type	Description
`tuple of (xr.Dataset, xr.Dataset)`	Tuple of (canopy_dataset, reference_dataset).

Raises:

Type	Description
`ValueError`	If the analysis is not configured or data is missing.

`get_site_summary()` ¶

Get a comprehensive summary of the research site.

Returns:

Type	Description
`dict`	Dictionary with site statistics, data availability, and store paths.

`is_day_complete(yyyydoy, receiver_types=None, completeness_threshold=0.95)` ¶

Check if a day has complete data coverage for all receiver types.

Parameters:

Name	Type	Description	Default
`yyyydoy`	`str`	Date in YYYYDOY format (e.g., "2024256")	required
`receiver_types`	`List[str]`	Receiver types to check. Defaults to ['canopy', 'reference']	`None`
`completeness_threshold`	`float`	Fraction of expected epochs that must exist (default 0.95 = 95%) Allows for small gaps due to receiver issues	`0.95`

Returns:

Type	Description
`bool`	True if all receiver types have complete data for this day

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`str()` ¶

Return a human-readable summary.

Returns:

Type	Description
`str`	Summary string.

`create_default_site()` ¶

Create a GnssResearchSite instance for the default site.

Returns:

Type	Description
`GnssResearchSite`	Instance for the `DEFAULT_RESEARCH_SITE`.

Data store¶

`StoredOverlapError` ¶

Bases: Exception

A file's epochs fall on epochs a file with another hash already filled.

Deliberately not a ValueError: the run stops on it instead of skipping the file (grilling on two writers, 2026-10-15, Q11). Neither file is preferred silently; gnss_store_strategy: overwrite replaces the stored data on purpose.

`BranchMovedError` ¶

Bases: RuntimeError

A branch moved while work for it was staged on a temporary branch.

Promoting the temporary branch would drop the commits made in the meantime, so canVODpy refuses and keeps the temporary branch.

`MaintenanceOnS3Error` ¶

Bases: RuntimeError

Maintenance on a store on S3 is not supported yet (#357).

Deliberately not a ValueError: nothing records it as a failed file and carries on. A store on S3 may be written by several machines, and maintenance only knows the runs of its own machine (is_pipeline_running), so a maintenance action that changes the store is refused. A garbage-collection dry run and maintenance_due still work: they change nothing.

`VodWriteItem` `dataclass` ¶

One VOD analysis pair's write request, as passed into a batch write.

`VodWritePlan` `dataclass` ¶

A VodWriteItem after dedup-check + normalization, ready to write.

`VodDayWrite` `dataclass` ¶

One analysis pair's VOD of one day, for :meth:MyIcechunkStore.commit_vod_days.

item.cadences holds the one day and its cadence. With in_fork, the day was written into a fork of the commit's session and item.dataset holds only the result's attributes; otherwise it holds the result, which the commit writes (a group's first days: the forks were made before the group existed).

`MyIcechunkStore` ¶

Core Icechunk store manager for GNSS data.

This class encapsulates all operations on a single Icechunk repository, providing a clean interface for GNSS data storage and retrieval with integrated logging and proper resource management.

Features: - Automatic repository creation/connection - Group management with validation - Session management with context managers - Integrated logging with file contexts - Configurable compression and chunking

Note on "metadata"

This class manages two distinct things both historically called "metadata":

File registry ({group}/metadata/table): per-file ingest ledger tracking hashes, temporal ranges, filenames, and paths. Managed by append_metadata(), load_metadata(), backup_metadata_table(), etc.
Store metadata (canvod.store_metadata package): store-level provenance (identity, creator, environment, compliance). Written to Zarr root attrs by the orchestrator. See canvod-store-metadata.

Parameters:

Name	Type	Description	Default
`store_path`	`StoreLocation \| Path \| str`	The store: a local directory or `s3://bucket/prefix`.	required
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").	`"gnss_store"`
`compression_level`	`int \| None`	Override default compression level.	`None`
`compression_algorithm`	`str \| None`	Override default compression algorithm.	`None`

Attributes:

Name	Type	Description
`location`	`StoreLocation`	Where the store lives.
`store_path`	`Path \| str`	The store's local directory, or its `s3://` URL.
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").
`compression_level`	`int`	Compression level (1-9).
`compression_algorithm`	`CompressionAlgorithm`	Compression algorithm enum.
`repo`	`Repository`	The Icechunk repository instance.

`settings` `property` ¶

The settings the store writes with.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings. Reading needs none; every write does.

`repo` `property` ¶

Get the repository instance.

`site_name` `property` ¶

The site this store was written for, from its repository metadata.

Never the name of the directory the store sits in, so moving or renaming the directory doesn't change it.

Raises:

Type	Description
`ValueError`	If the store has no repository metadata.

`source_format` `property` ¶

Return the source_format root attribute, or None.

`tree` `property` ¶

Print hierarchical tree of all branches and groups. See print_tree.

`init(store_path, store_type='gnss_store', compression_level=None, compression_algorithm=None, *, config=None)` ¶

Initialize the Icechunk store manager.

Parameters:

Name	Type	Description	Default
`store_path`	`StoreLocation \| Path \| str`	The store: a local directory, `s3://bucket/prefix` or a :class:`~canvod.store_metadata.location.StoreLocation`.	required
`store_type`	`str`	Type of store ("gnss_store" or "vod_store").	`"gnss_store"`
`compression_level`	`int \| None`	Override default compression level.	`None`
`compression_algorithm`	`str \| None`	Override default compression algorithm.	`None`
`config`	`CanvodConfig \| None`	The settings the store writes with; every commit records them. `None` loads the settings file if there is one. Without one the store can be read but not written.	`None`

`chunk_encoding_for(ds)` ¶

Build {var: {"chunks": (...)}} matching self.chunk_strategy.

Only meaningful at group creation -- Zarr fixes an array's chunk shape the moment it's first written, so this must be passed as encoding= on the mode="w" call, never on an append (which must match whatever shape creation already fixed).

Without this, chunk_strategies config only ever affected read-side chunking (the chunks= hint passed to xr.open_zarr elsewhere in this class) -- nothing applied it at write time, so the physical on-disk chunk shape was left to Zarr's own default, silently disconnected from config. Confirmed by grepping this file and processor.py for any encoding["chunks"] assignment: there wasn't one anywhere (2026-07-18).

A dim with a configured chunk size (e.g. epoch: 17280) gets exactly that size regardless of how much data is in this particular write -- Zarr supports writing less than one full chunk in the first call and filling the rest on later appends, which is the whole point of fixing chunk shape once at creation. A dim configured -1 (or unconfigured) gets one chunk spanning whatever this write's current size is (e.g. sid: -1 -> the full SID universe already present at creation).

`readonly_session(branch='main')` ¶

Thin wrapper over repo.readonly_session() for this CM.

`writable_session(branch='main')` ¶

Thin wrapper over repo.writable_session() for this CM.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings.

`set_root_attrs(attrs, branch='main')` ¶

Set root-level Zarr attributes on the store.

Parameters:

Name	Type	Description	Default
`attrs`	`dict[str, Any]`	Key-value pairs to merge into root attrs.	required
`branch`	`str`	Branch to write to.	`"main"`

Returns:

Type	Description
`str`	Snapshot ID from the commit.

`get_root_attrs(branch='main')` ¶

Read root-level Zarr attributes from the store.

Returns:

Type	Description
`dict[str, Any]`	Root attributes (empty dict if none set).

`get_branch_names()` ¶

List all branches in the store.

Returns:

Type	Description
`list[str]`	List of branch names.

`get_group_names(branch=None)` ¶

List all groups in the store.

Parameters:

Name	Type	Description	Default
`branch`	`str \| None`	Repository branch to examine. Defaults to listing groups from all branches.	`None`

Returns:

Type	Description
`dict[str, list[str]]`	Dictionary mapping branch names to lists of group names.

`list_groups(branch='main')` ¶

List all groups in a branch.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name (default: "main")	`'main'`

Returns:

Type	Description
`list[str]`	List of group names in the branch

`list_data_groups(branch='main')` ¶

List the groups that hold data, at any depth.

In a GNSS store these are the receiver groups, as :meth:list_groups gives them; in a VOD store each analysis sits below its calculator (tau_omega/canopy_01_vs_reference_01).

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name (default: "main")	`'main'`

Returns:

Type	Description
`list[str]`	The paths of the groups, sorted.

`print_tree(max_depth=None)` ¶

Display hierarchical tree of all branches, groups, and subgroups.

Parameters:

Name	Type	Description	Default
`max_depth`	`int \| None`	Maximum depth to display. None for unlimited depth. - 0: Only show branches - 1: Show branches and top-level groups - 2: Show branches, groups, and first level of subgroups/arrays - etc.	`None`

`group_exists(group_name, branch='main')` ¶

Check if a group exists.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to check. May be a nested path (e.g. VOD's `"{calculator}/{analysis_name}"`), not just a top-level name.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Returns:

Type	Description
`bool`	True if the group exists, False otherwise.

Notes

Checks membership directly against the opened root group rather than via :meth:get_group_names/:meth:list_groups -- those only enumerate top-level Zarr group keys (root.group_keys()), so for a nested group_name like a VOD analysis group, membership in that flat list is always False even when the group exists. Confirmed 2026-07-21: this previously caused every VOD write after a group's first to take the mode="w" (full overwrite) branch in :meth:write_or_append_vod_group instead of appending, silently destroying all previously-written VOD data on each subsequent write. Zarr's Group.__contains__ supports slash-separated nested paths directly, so this works for both flat GNSS receiver names (no behavior change there) and nested VOD group names (the actual fix).

`preprocessing_records(group_name, start=None, end=None, branch='main')` ¶

Preprocessing records in group_name's log book, without repeats.

In ingest order; rows written before the record existed give "". Each file row's record includes the steps of the group-wide rows written after it (:meth:add_group_variables): it describes the data as they are now. With start and end, only rows whose data overlap that range, without rows of files skipped as already ingested (as :meth:source_file_hashes). A group without a log book gives none.

`check_preprocessing_matches(group_name, dataset, branch='main')` ¶

Refuse data preprocessed differently from the data in group_name.

Compares the operations and settings of the record dataset carries (:data:~canvod.config.models.PREPROCESSING_ATTR; none counts as no preprocessing) with every record in the group's log book. The results of the operations (e.g. the input sampling) may differ. In a VOD store, the records are those of the GNSS data behind each result (vod_preprocessing_record).

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group about to receive `dataset`.	required
`dataset`	`Dataset`	Data to write.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Raises:

Type	Description
`PreprocessingMismatchError`	If the group holds data preprocessed differently.

`check_settings_match(group_name, settings, source='this run has', branch='main')` ¶

Refuse preprocessing settings unlike those of the data in group_name.

Lets a run compare its configured preprocessing (see canvod.ops.planned_settings) with a group before it processes any data; :meth:check_preprocessing_matches applies the same comparison to a dataset at write time. A group that does not exist yet accepts any settings.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group the data will be written to.	required
`settings`	`list[dict]`	Operations and settings, as `canvod.config.models.preprocessing_settings` gives them.	required
`source`	`str`	Names the new data in the error message.	`"this run has"`
`branch`	`str`	Repository branch to examine.	`"main"`

Raises:

Type	Description
`PreprocessingMismatchError`	If the group holds data preprocessed differently.

`add_group_variables(group_name, variables, step, software, branch='main')` ¶

Add variables computed from the data already in a group.

For example the grid cells of every observation (canvod.ops.assign_grid_cells_in_store). Only the new variables are written; the group's other variables, its attributes, its log book and its other subgroups stay as they are. The write is staged on a temporary branch and moved to branch with :meth:promote_branch.

The log book receives one group-wide row (:data:GROUP_WIDE_ACTION) whose preprocessing record holds step. From then on, the preprocessing of the data already in the group counts as their own plus step (:meth:preprocessing_records), so a later write is accepted only if it was preprocessed with step as well, and thus carries the new variables too.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Store group, e.g. `"canopy_01"` or a VOD analysis group.	required
`variables`	`Dataset`	The new variables, with the group's dimensions and coordinate values as :meth:`read_group` reads them (all of its epochs and signals, in the same order). In a group of day blocks they are written into each day's block, and the padding stays empty.	required
`step`	`dict`	The operation that computed them, as a preprocessing step `{"op", "settings", "result"}` (`canvod.ops.OpResult.step`).	required
`software`	`dict[str, str]`	Version of each package whose code computed them.	required
`branch`	`str`	Branch to change.	`"main"`

Returns:

Type	Description
`str`	The new tip of `branch`.

Raises:

Type	Description
`ValueError`	If the group does not exist, already has one of the variables, or `variables` do not match its dimensions and coordinates; or if the temporary branch already exists.
`BranchMovedError`	If `branch` moved meanwhile; the work stays on the temporary branch.

`read_group(group_name, branch='main', time_slice=None, date=None, chunks=None, drop_empty=False)` ¶

Read data from a group, its epochs in time order.

A group made of day blocks (ADR 0016) has a slot for every epoch of a day at the day's cadence; slots without data hold missing values (NaN, or the _FillValue of integer variables) at their grid times.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to read.	required
`branch`	`str`	Repository branch.	`"main"`
`time_slice`	`slice \| None`	Optional label-based time slice for filtering (passed to `ds.sel`).	`None`
`date`	`str \| None`	YYYYDOY string (e.g. `"2025001"`) selecting a single calendar day, from 00:00 up to, not including, 00:00 of the next day. Converted to a `time_slice`; mutually exclusive with `time_slice`.	`None`
`chunks`	`dict[str, Any] \| None`	Dask chunks for reading. If None, the chunks stored in the group.	`None`
`drop_empty`	`bool`	Drop the empty slots of the epochs read, after `time_slice` (a group written by appending has none; see :func:`~canvod.store.day_blocks.drop_empty_slots`).	`False`

Returns:

Type	Description
`Dataset`	Dataset from the group.

`write_dataset(dataset, group_name, session, mode='a', chunks=None)` ¶

Write a dataset to Icechunk with proper chunking.

Parameters:

Name	Type	Description	Default
`dataset`	`Dataset`	Dataset to write	required
`group_name`	`str`	Group path in store	required
`session`	`Any`	Active writable session or store handle.	required
`mode`	`str`	Write mode: 'w' (overwrite) or 'a' (append)	`'a'`
`chunks`	`dict[str, int] \| None`	Chunking spec. If None, uses store's chunk_strategy. Example: {'epoch': 17280, 'sid': -1}	`None`

`grow_day_blocks(days, *, message, branch='main')` ¶

Give days a block in existing day-block groups, in one commit.

A run calls this once per batch, before reading files, for the days whose Phase 1 succeeded (ADR 0016). A day that has a block keeps it, so a block grown for a day that failed later is reused on the next run. When another commit changed the branch meanwhile and the two conflict, the growth starts again on the new tip: it rereads the epoch axis, reuses the blocks added meanwhile and grows the rest.

Parameters:

Name	Type	Description	Default
`days`	`Mapping[str, Mapping[datetime64, float]]`	For each group, the cadence in seconds of each day to grow.	required
`message`	`str`	Commit message.	required
`branch`	`str`	Branch to commit to.	`"main"`

Returns:

Type	Description
`str or None`	Snapshot ID of the growth commit; `None` if every day already had a block, in which case nothing is committed.

Raises:

Type	Description
`LayoutError`	If a group does not exist (its first write creates it) or is not made of day blocks.
`CadenceError`	If a cadence does not fit the chunks, or a day has a block at a coarser cadence.
`RuntimeError`	If the growth still conflicts after `GROWTH_ATTEMPTS` attempts, with :func:`~canvod.store_metadata.io.wait_after_conflict` between them.

`backup_metadata_table(group_name, session)` ¶

Backup the metadata table to a Polars DataFrame.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name.	required
`session`	`Any`	Active session for reading.	required

Returns:

Type	Description
`DataFrame \| None`	DataFrame with metadata rows, or None if missing.

Raises:

Type	Description
`RuntimeError`	If the table exists but cannot be read.

`restore_metadata_table(group_name, df, session)` ¶

Restore the metadata table from a Polars DataFrame.

This recreates the full Zarr structure for the metadata table.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name.	required
`df`	`DataFrame`	Metadata table to restore.	required
`session`	`Any`	Active session for writing.	required

Returns:

Type	Description
`None`

`rewrite_group(session, group_name, dataset, drop_ranges, metadata_parts=None)` ¶

Rewrite group_name with dataset and keep its metadata/.

A mode="w" write deletes the group's subgroups. The log book comes back unchanged; every other metadata dataset (e.g. sbf_obs) comes back from the session's base snapshot without its epochs inside drop_ranges (the replaced data), merged in epoch order with the datasets in metadata_parts (name -> per-file datasets). A metadata dataset left without epochs is not written back. Writes into session without committing.

Raises:

Type	Description
`LayoutError`	If the group is made of day blocks: its files are written into their day's block instead, and a whole-group write would drop its layout and block index.

`get_group_info(group_name, branch='main')` ¶

Get metadata about a group.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group.	required
`branch`	`str`	Repository branch to examine.	`"main"`

Returns:

Type	Description
`dict[str, Any]`	Group metadata.

Raises:

Type	Description
`ValueError`	If the group does not exist.

`metadata_dataset_exists(group_name, name, branch='main')` ¶

Return True if a metadata dataset name exists for group_name.

`write_metadata_dataset(meta_ds, group_name, name, branch='main')` ¶

Write a pre-concatenated metadata dataset to {group_name}/metadata/{name}.

Always writes with mode="w" (full overwrite for the day).

Parameters:

Name	Type	Description	Default
`meta_ds`	`Dataset`	Pre-concatenated `(epoch, sid)` metadata dataset.	required
`group_name`	`str`	Target group (receiver name).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch to write to.	`"main"`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`write_metadata_parts(parts, group_name, name, session)` ¶

Add per-file metadata datasets to {group_name}/metadata/{name}.

Writes into session without committing, so the caller commits the metadata together with the observations it belongs to. Creates the dataset if it does not exist yet, otherwise appends along epoch.

Parameters:

Name	Type	Description	Default
`parts`	`list[Dataset]`	Per-file metadata datasets with an `epoch` dim, one per file whose observations are written in the same session.	required
`group_name`	`str`	Target group (the observations' store group).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`session`	`Session or ForkSession`	Open writable session of the observations' write.	required

Returns:

Type	Description
`int`	Number of epochs written.

`append_metadata_datasets(parts, group_name, name, branch='main')` ¶

Append metadata datasets in their own commit: no in-memory concat.

Creates {group_name}/metadata/{name} if it does not exist yet, otherwise appends along epoch. Prefer :meth:write_metadata_parts inside the session that writes the observations, so both land in one commit.

Parameters:

Name	Type	Description	Default
`parts`	`list[Dataset]`	Individual per-file metadata datasets with an `epoch` dim.	required
`group_name`	`str`	Target group (receiver name).	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch to write to.	`"main"`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`read_metadata_dataset(group_name, name, branch='main', chunks=None)` ¶

Read a metadata dataset name for group_name.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group (receiver) name.	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch.	`"main"`
`chunks`	`dict \| None`	Dask chunks for reading. If None, the chunks stored in the dataset.	`None`

Returns:

Type	Description
`Dataset`	Lazy `(epoch, sid)` metadata dataset.

`get_metadata_dataset_info(group_name, name, branch='main')` ¶

Get info about metadata dataset name for group_name.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group (receiver) name.	required
`name`	`str`	Dataset name under `metadata/` (e.g. `"sbf_obs"`).	required
`branch`	`str`	Repository branch.	`"main"`

Returns:

Type	Description
`dict[str, Any]`	Info dict with the same structure as `get_group_info()`.

Raises:

Type	Description
`ValueError`	If the metadata dataset does not exist.

`sbf_metadata_exists(group_name, branch='main')` ¶

Return True if an SBF metadata dataset exists for group_name.

`write_sbf_metadata(meta_ds, group_name, branch='main')` ¶

Write SBF metadata dataset.

`read_sbf_metadata(group_name, branch='main', chunks=None)` ¶

Read SBF metadata dataset.

`get_sbf_metadata_info(group_name, branch='main')` ¶

Get SBF metadata info.

`rel_path_for_commit(file_path)` ¶

Generate relative path for commit messages.

Parameters:

Name	Type	Description	Default
`file_path`	`Path`	Full file path.	required

Returns:

Type	Description
`str`	Relative path string with log_path_depth parts.

`get_store_stats()` ¶

Get statistics about the store.

Returns:

Type	Description
`dict[str, Any]`	Store statistics.

`append_metadata(group_name, rinex_hash, start, end, snapshot_id, action, commit_msg, dataset_attrs, branch='main', canonical_name=None, physical_path=None)` ¶

Standalone metadata-row append: commits on its own.

For ingest paths (write_initial_group, append_to_group) use _append_metadata_row directly inside the data-write session to avoid an extra commit per file.

`append_metadata_bulk(group_name, rows, session=None)` ¶

Append multiple metadata rows in one commit.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Group name (e.g. "canopy", "reference")	required
`rows`	`list[dict[str, Any]]`	List of metadata records matching the schema used in append_metadata().	required
`session`	`Session or ForkSession`	If provided, rows are written into this session (caller commits later). If None, this method opens its own writable session and commits once.	`None`

`should_skip_vod_write(group_name, source_file_hashes, time_start, time_end, branch='main')` ¶

Check whether a VOD write should be skipped as a duplicate.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	`{model}/{analysis_name}` group path.	required
`source_file_hashes`	`dict[str, str]`	`{receiver_name: "File Hash"}` for every receiver the computation read from.	required
`time_start`	`datetime64`	Epoch range of the incoming VOD dataset.	required
`time_end`	`datetime64`	Epoch range of the incoming VOD dataset.	required
`branch`	`str`		`"main"`

Returns:

Type	Description
`tuple[bool, str]`	`(True, "hash_match")`, `(True, "temporal_overlap")`, or `(False, "")`.

`commit_vod_days(session, forks, days, branch='main')` ¶

Merge the forks of a unit of VOD days and commit them once (ADR 0021).

forks come from session, which had no changes of its own when they were made; they hold the days the worker pool wrote (:mod:canvodpy.orchestrator.vod_phase). Under the store's commit lock: merge them, rebase onto commits made since the session was opened, write the days not in a fork (VodDayWrite.in_fork), add a log-book row per day and commit.

Returns:

Type	Description
`str or None`	Snapshot ID of the commit; `None` without days.

Raises:

Type	Description
`PreprocessingMismatchError`	If a group holds results of data preprocessed differently.
`ValueError`	If a group holds results of another calculator, or other calculator parameters.
`LayoutError`	If a day's result covers more than its one day.
`(RebaseFailedError, ConflictError)`	If another writer changed the same chunks first; nothing is committed.

`write_or_append_vod_groups_batch(items, branch='main', dedup=True)` ¶

Write/append multiple VOD analysis groups in one fork/merge batch.

The same cross-group parallelization RinexDataProcessor uses for GNSS receiver groups (_write_receiver_batch_forked), applied to VOD analysis groups: any brand-new analysis group is created directly on its own clean session that commits alone (pre-pass) -- verified empirically (see _write_receiver_batch_forked's docstring) that Session.merge() silently drops a fork's data if the base session it merges into had its own pending uncommitted writes before .fork() was called, so new-group creation and forked appends can never safely share one session. Only groups that already exist go through the fork/merge phase; a brand-new group's pre-pass write is its entire contribution to this batch (a VOD write is exactly one dataset per group per call, unlike GNSS's many-files-per-group-per- day case, so there is never leftover data for a newly-created group to append afterward).

Fail-fast: if any group's forked write raises, no merge or commit happens for the fork/merge phase -- a partial merge would silently drop a group's data from an apparently-successful commit. A pre-pass commit that already landed before a later fork failure is safe to leave in place: the next retry's dedup check sees that group's data as already-committed and won't re-attempt it.

`load_metadata(store, group_name)` ¶

Load metadata directly from Zarr into a Polars DataFrame.

Parameters:

Name	Type	Description	Default
`store`	`Any`	Zarr store or session store handle.	required
`group_name`	`str`	Group name.	required

Returns:

Type	Description
`DataFrame`	Metadata table.

`read_metadata_table(session, group_name)` ¶

Read the metadata table from a session.

Parameters:

Name	Type	Description	Default
`session`	`Any`	Active session for reading.	required
`group_name`	`str`	Group name.	required

Returns:

Type	Description
`DataFrame`	Metadata table.

`metadata_row_exists(group_name, rinex_hash, start, end, branch='main')` ¶

Check whether a file already exists or temporally overlaps existing data.

Performs two checks in order:

Hash match — if rinex_hash already appears in the metadata table, the file was previously ingested (exact duplicate).
Temporal overlap — if the incoming [start, end] interval overlaps any existing metadata interval, the file covers a time range that is already (partially) present in the store. This catches cases like a daily concatenation file coexisting with the sub-daily files it was built from.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`rinex_hash`	`str`	Hash of the current GNSS dataset.	required
`start`	`datetime64`	Start epoch of the incoming file.	required
`end`	`datetime64`	End epoch of the incoming file.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`

Returns:

Type	Description
`tuple[bool, DataFrame]`	`(True, overlapping_rows)` when the file should be skipped, `(False, empty_df)` when it is safe to ingest.

`should_skip_file(group_name, file_hash, time_start, time_end, branch='main')` ¶

Check whether a file should be skipped before processing and writing.

Thin public wrapper around :meth:metadata_row_exists that returns a simple (skip, reason) pair instead of a DataFrame, suitable for use in Airflow task functions as an early-exit optimization.

This method is an early-exit optimization: it avoids opening a write session for files that are already present. Pass dedup=True to :meth:write_or_append_group to make the store the authoritative final gate (runs the same check again before writing).

Checks performed (in order):

Hash match — file was previously ingested (exact duplicate).
Temporal overlap — incoming time range overlaps existing epochs.

Layer 3 (intra-batch overlap) is not applicable here because task functions process files one at a time.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`file_hash`	`str or None`	Hash from `dataset.attrs["File Hash"]`. When `None` the check is skipped and `(False, "")` is returned.	required
`time_start`	`datetime64`	First epoch of the incoming dataset.	required
`time_end`	`datetime64`	Last epoch of the incoming dataset.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`

Returns:

Type	Description
`tuple[bool, str]`	`(True, "hash_match")`: exact duplicate, skip. `(True, "temporal_overlap")`: overlapping time range, skip. `(False, "")`: safe to ingest.

`dir_entry_counts()` ¶

Cheap on-disk entry counts for manifests/snapshots/transactions.

A directional proxy for Icechunk-internal growth (manifest count, snapshot count) without walking the full store. Deliberately excludes chunks/ (can reach millions of entries -- not cheap) and only applies to local-filesystem-backed stores; returns {} for anything else (e.g. S3/GCS) since os.scandir doesn't apply.

`metadata_row_count(group_name, branch='main')` ¶

Row count of {group_name}/metadata/table, or 0 if absent.

Reads only the index array's shape (no full table load) -- cheap enough to log alongside every dedup check, so growth in batch_check/check_temporal_overlaps duration can be directly correlated against store size instead of assumed.

`source_file_hashes(group_name, start, end, branch='main')` ¶

Hashes of the ingested files whose data overlaps [start, end].

Read from group_name's log book, in ingest order and without duplicates. Rows recording a file that was skipped as already ingested (exists is "True" and nothing was written) are ignored, so each file counts once however often it was re-run.

Used as the provenance of a dataset read back from the store for that range, e.g. the daily inputs of a VOD computation.

`load_metadata_for_dedup(group_name, branch='main')` ¶

{group_name}'s metadata table at the tip of branch.

Read whole the first time, afterwards only the rows committed since (:class:~canvod.store_metadata.log_book.LogBooks): a run asks for it every day, and a whole read grows with the store (ADR 0021).

batch_check_existing/check_temporal_overlaps each used to call load_metadata() independently -- two full-table reads per batch, growing with store size (confirmed against a real backfill: batch_check duration tracked metadata_rows almost exactly, ~5x over 41,597 rows). Callers needing both checks should load once here and pass the result to both via their metadata_df parameter. Returns None (not an empty DataFrame) when the group/metadata doesn't exist yet, so callers can distinguish "nothing to check against" from "failed to load" without re-attempting the load.

`batch_check_existing(group_name, file_hashes, metadata_df=None)` ¶

Check which file hashes already exist in metadata.

Parameters:

Name	Type	Description	Default
`metadata_df`	`DataFrame`	Pre-loaded metadata table (see `load_metadata_for_dedup`) to skip this call's own full-table read. Loads internally (original behavior) when not provided.	`None`

`check_temporal_overlaps(group_name, file_intervals, branch='main', metadata_df=None)` ¶

Check which files temporally overlap existing metadata intervals.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Icechunk group name.	required
`file_intervals`	`list[tuple[str, datetime64, datetime64]]`	List of `(rinex_hash, start, end)` tuples for incoming files.	required
`branch`	`str`	Branch name in the Icechunk repository.	`"main"`
`metadata_df`	`DataFrame`	Pre-loaded metadata table (see `load_metadata_for_dedup`) to skip this call's own full-table read. Loads internally (original behavior) when not provided.	`None`

Returns:

Type	Description
`set[str]`	Hashes of files whose `[start, end]` overlaps any existing metadata interval. Files whose hash already exists in the store are NOT included (use `batch_check_existing` for those).

`expire_old_snapshots(days=None, branch='main', delete_expired_branches=False, delete_expired_tags=False)` ¶

Expire snapshots older than the given retention period.

Metadata-only "soft delete": rewrites ancestry so expired snapshots become unreachable from any surviving branch/tag. No objects (manifests, chunks, snapshot files) are deleted at this stage — call garbage_collect() separately to reclaim storage. Keeping this expire-only (previously it also ran GC internally) is what lets a caller run GC exactly once via maintenance() instead of twice per call.

Never invoke this automatically inside the pipeline's write path: it's an administrative operation, not safe to run casually alongside active writes. Use a generous days margin (weeks to months, not hours) so it can never touch anything from an in-flight write session.

Parameters:

Name	Type	Description	Default
`days`	`int \| None`	Number of days to retain snapshots. Defaults to 90 (weeks-to- months cadence).	`None`
`branch`	`str`	Branch to apply expiration on.	`"main"`
`delete_expired_branches`	`bool`	Whether to delete branches pointing to expired snapshots. Matches Icechunk's own safe default: flip only deliberately.	`False`
`delete_expired_tags`	`bool`	Whether to delete tags pointing to expired snapshots. Do not set True if using keeper tags (`create_keeper_tag`) — it would delete them along with any other expired-tip tag.	`False`

Returns:

Type	Description
`set[str]`	Expired snapshot IDs.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings.

`garbage_collect(days=90, *, dry_run=False)` ¶

Physically delete objects no longer reachable from any branch/tag.

Only removes what a prior expire_old_snapshots() (or manual ref deletion) already made unreachable — this does not itself expire anything. Always run with dry_run=True first on a store that hasn't been GC'd before, especially one with a large accumulated backlog: there's no documented cost model for GC time/memory at scale, so a first-ever run's duration is unknown until measured.

Not safe to run casually alongside active writes — the cutoff must predate the start of any concurrently-running write session, and only one expire/GC operation should run at a time. Icechunk does have self-protection against races with new commits arriving mid-run (it restarts, bounded retries, rather than deleting something a concurrent writer needs), but that's not a substitute for a generous cutoff margin.

Parameters:

Name	Type	Description	Default
`days`	`int`	Objects older than this (by the same write-time semantics as `expire_old_snapshots`) are eligible for deletion.	`90`
`dry_run`	`bool`	Report what would be deleted without deleting anything.	`False`

Returns:

Type	Description
`dict`	Summary of (would-be-)deleted objects.

Raises:

Type	Description
`FileNotFoundError`	If the store was opened without settings, also for a dry run, so maintenance behaves one way.

`get_history(branch='main', limit=None)` ¶

Return commit ancestry (history) for a branch.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch name.	`"main"`
`limit`	`int \| None`	Maximum number of commits to return.	`None`

Returns:

Type	Description
`list[dict]`	Commit info dictionaries (id, message, written_at, parent_ids).

`print_history(branch='main', limit=100)` ¶

Pretty-print the ancestry for quick inspection.

`get_ops_log(limit=None)` ¶

Return the repository operations log (newest first).

The ops log is an immutable audit trail of every structural mutation: commits, branch and tag operations, GC runs, config changes, etc. Unlike :meth:get_history, this is repo-wide (not per-branch) and covers all event types, not only commits.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	Maximum number of entries to return. `None` returns all.	`None`

Returns:

Type	Description
`list[dict]`	Each entry contains `kind` (str), `updated_at` (datetime), and `backup_path` (str).

`print_ops_log(limit=50)` ¶

Pretty-print the repository operations log for quick inspection.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	Maximum number of entries to display (default: 50).	`50`

`maintenance_due(expire_interval_days, gc_delay_days, manifest_count_threshold=None)` ¶

Determine whether scheduled expiration/GC/manifest compaction are due.

Used by canvodpy store maintain-due (a cron-safe, unattended entry point -- see processing.storage.maintenance config). The repo's own ops log (:meth:get_ops_log) is the source of truth for "when did maintenance last actually run": it is written atomically by Icechunk itself as part of the same operation, so it cannot drift the way a separate marker file could (deleted, stale-copied across hosts on a shared mount, or never written if a prior run crashed after the real operation completed but before a marker write). Confirmed empirically (2026-07-20, throwaway store): a real expire_old_snapshots() call always writes an ExpirationRan entry (even when zero snapshots were actually expired -- root/tip snapshots are protected); garbage_collect(dry_run=True) does not write a GCRan entry, only a real (non-dry-run) GC does -- so a dry run correctly never counts as "GC happened" here.

GC is only ever considered due after a qualifying expiration, and never again for the same expiration -- physical deletion should lag the (reversible-in-effect) soft-delete by gc_delay_days, per the skill doc's "expire every 1-2 months, GC every 15-30 days" guidance.

Parameters:

Name	Type	Description	Default
`expire_interval_days`	`int`	Re-run expiration if this many days have passed since the most recent `ExpirationRan` entry (or if there is none yet).	required
`gc_delay_days`	`int`	Run GC this many days after the most recent expiration, once per expiration (skipped if GC already ran since that expiration).	required
`manifest_count_threshold`	`int \| None`	Trigger `rewrite_manifests()` (manifest compaction) when :meth:`dir_entry_counts`'s `manifests` count reaches this. Unlike expiration/GC, `rewrite_manifests()` writes no distinguishing ops-log entry -- it logs as an ordinary `NewCommit`, indistinguishable from a data-ingest commit -- so due-ness is measured directly from current fragmentation rather than time-since-last-ops-log-entry. `None` (default) always reports `manifests_due=False` (caller opted out via `manifests_enabled=False`). Returns `False` (not an error) if the store isn't local-filesystem-backed (:meth:`dir_entry_counts` returns `{}` for S3/GCS).	`None`

Returns:

Type	Description
`dict[str, Any]`	`expire_due` (bool), `gc_due` (bool), `manifests_due` (bool), `last_expire` (datetime \| None), `last_gc` (datetime \| None), `manifest_count` (int \| None).

`compact_manifests(branch='main')` ¶

Consolidate all of a branch's fragmented manifests in one commit.

Thin wrapper over repo.rewrite_manifests(): canvodpy's write pattern (one to_icechunk() call per file within a batch, e.g. 96 fifteen-minute files/receiver-day) creates one new manifest per flush, not per commit -- a single day's ingest for one receiver can create ~96 manifests. This rewrites them back to the currently-configured splitting boundaries.

Uses commit_method="new_commit" deliberately, not "amend": amend replaces the immediately-preceding commit's own message/ metadata rather than merging with it, and _append_to_icechunk already attaches structured provenance (receiver, date, file hashes) to every ingest commit -- amending could silently clobber that. An explicit, separate, auditable commit matches the precedent expire/GC already set. Like every commit, it records software, environment and settings as commit metadata (:func:canvod.store_metadata.rewrite_manifests).

Like expire_old_snapshots()/garbage_collect(), this is an administrative operation -- never call it while a pipeline run is active against this branch (ordinary commit conflict-retry semantics apply, but a still-draining worker pool committing per-file would just make every retry lose, burning cycles for nothing).

Does not shrink on-disk manifest count by itself -- confirmed empirically (2026-07-20, throwaway store, 300 commits): compaction adds a new, consolidated manifest set but leaves the old fragmented ones on disk (399 -> 401 manifests immediately after compacting); only a subsequent expire_old_snapshots() + garbage_collect() actually reclaims them (399 -> 2 in the same test). This mirrors GC's own relationship to expiration -- don't expect dir_entry_counts()['manifests'] to drop right after calling this; it drops after the next expire/GC cycle instead.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Branch to compact.	`"main"`

Returns:

Type	Description
`str`	Snapshot ID of the compaction commit.

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`str()` ¶

Return a human-readable summary.

Returns:

Type	Description
`str`	Summary string.

`rechunk_group(group_name, chunks, source_branch='main', temp_branch=None, promote_to_main=True, delete_temp_branch=True)` ¶

Rechunk a group with optimal chunk sizes.

Zarr arrays have a fixed chunk shape once created — writing with mode="r+" onto an existing array cannot change chunk boundaries, it can only update values within the existing grid (verified 2026-07-08: a previous version of this method used mode="r+", which silently failed to change on-disk chunking whenever the request didn't align with the current chunks). Actually changing chunk size requires a destructive mode="w" rewrite of the group, which also wipes the group's nested subgroups (metadata/ table, metadata/sbf_obs) — so this method copies that subtree back afterward via a raw recursive Zarr copy (see _copy_zarr_subtree), all on a temporary branch, before main is ever touched. The copy handles canvod-store's two-level-deep metadata/{table,sbf_obs} nesting.

Both the data read and the subtree copy are pinned to the exact snapshot captured at the start of this call. If source_branch moves before the promotion, the promotion is refused with BranchMovedError (the new commits would be dropped) and the rechunked data stays on temp_branch.

Parameters:

Name	Type	Description	Default
`group_name`	`str`	Name of the group to rechunk	required
`chunks`	`dict[str, int]`	Chunking specification, e.g. {'epoch': 180, 'sid': -1}. A group stored in day blocks needs an epoch chunk that divides every day into whole chunks.	required
`source_branch`	`str`	Branch to read original data from (default: "main")	`'main'`
`temp_branch`	`str \| None`	Temporary branch name for rechunked data. If None, uses "{group_name}_rechunked".	`None`
`promote_to_main`	`bool`	If True, move `source_branch` to the rechunked snapshot after writing (see `promote_branch`).	`True`
`delete_temp_branch`	`bool`	If True, delete temporary branch after promotion (only if promote_to_main=True).	`True`

Returns:

Type	Description
`str`	Snapshot ID of the rechunked data

Raises:

Type	Description
`ValueError`	If `temp_branch` already exists, or a day-block group gets no positive epoch chunk.
`CadenceError`	If the epoch chunks would split a day of a day-block group.
`BranchMovedError`	If `source_branch` moved during the rechunk.

`promote_branch(temp_branch, target_branch, expected_tip)` ¶

Move target_branch to the tip of temp_branch.

Only if target_branch still points at expected_tip, the snapshot temp_branch was created from. Otherwise other commits landed on target_branch meanwhile and moving it would drop them.

Parameters:

Name	Type	Description	Default
`temp_branch`	`str`	Branch holding the staged work.	required
`target_branch`	`str`	Branch to move (usually `"main"`).	required
`expected_tip`	`str`	Snapshot ID `target_branch` pointed at when `temp_branch` was created.	required

Returns:

Type	Description
`str`	The new tip of `target_branch`.

Raises:

Type	Description
`BranchMovedError`	If `target_branch` moved. `temp_branch` is left in place.

`create_branch(branch_name, snapshot_id)` ¶

Create a branch pointing at a specific snapshot.

Accepts the full snapshot ID or a prefix of it, such as the 8 characters plot_commit_graph() shows (case-insensitive). The snapshot can be on any branch or tag.

Parameters:

Name	Type	Description	Default
`branch_name`	`str`	Name of the new branch (e.g. `"explore/my-analysis"`).	required
`snapshot_id`	`str`	Full snapshot ID or a prefix of it.	required

Raises:

Type	Description
`ValueError`	If the ID is empty, no snapshot on a branch or tag has this ID or prefix, or the prefix fits several.

`create_release_tag(tag_name, snapshot_id=None)` ¶

Create an immutable tag for an important version.

Parameters:

Name	Type	Description	Default
`tag_name`	`str`	Name for the tag (e.g., "v2024_complete", "before_reprocess")	required
`snapshot_id`	`str \| None`	Snapshot to tag. If None, uses current tip of main branch.	`None`

`list_tags()` ¶

List all tags in the repository.

`delete_tag(tag_name)` ¶

Delete a tag (use with caution - tags are meant to be permanent).

`create_keeper_tag(receiver, yyyydoy, snapshot_id)` ¶

Tag a per-(receiver, date) commit as a permanent retention keeper.

Part of the tag-based retention scheme: tagging keep/{receiver}/{yyyydoy} after each commit means a later expire_old_snapshots(delete_expired_tags=False) run can safely prune everything else while these specific snapshots stay time-travelable forever. Additive and inert on its own — creating a tag doesn't expire or delete anything.

Never raises: a tag name collision (already exists, or was previously deleted — Icechunk tag names are never reusable) is logged and skipped rather than failing the pipeline write that called this.

Parameters:

Name	Type	Description	Default
`receiver`	`str`	Receiver/group name (e.g. `"canopy_01"`).	required
`yyyydoy`	`str`	Date key (e.g. `"2025001"`).	required
`snapshot_id`	`str`	Snapshot to tag — typically the just-committed snapshot.	required

Returns:

Type	Description
`bool`	True if the tag was created, False if skipped.

`plot_commit_graph(branch=None, *, plain=False)` ¶

Visualize commit history as a graph.

Delegates to icechunk's native repo.ancestry_graph(). Renders as an SVG diagram in Jupyter/marimo notebooks, or as colored text in a terminal (print(store.plot_commit_graph())).

Parameters:

Name	Type	Description	Default
`branch`	`str`	Restrict the graph to this branch's linear history. When omitted, all branches are shown as a tree.	`None`
`plain`	`bool`	Render without ANSI colors or SVG fill colors (useful for CI logs or piping to a file).	`False`

Returns:

Type	Description
`AncestryGraph`	Displayable object with `_repr_svg_()` for notebooks.

`cleanup_stale_branches(keep_patterns=None)` ¶

Delete stale temporary branches (e.g., from failed rechunking).

Parameters:

Name	Type	Description	Default
`keep_patterns`	`list[str] \| None`	Patterns to preserve. Default: ["main", "dev"]	`None`

Returns:

Type	Description
`list[str]`	Names of deleted branches

`delete_branch(branch_name)` ¶

Delete a branch.

`get_snapshot_info(snapshot_id)` ¶

Get detailed information about a specific snapshot.

Parameters:

Name	Type	Description	Default
`snapshot_id`	`str`	Snapshot ID to inspect	required

Returns:

Type	Description
`dict`	Snapshot metadata and statistics

`compare_snapshots(snapshot_id_1, snapshot_id_2)` ¶

Compare two snapshots to see what changed.

Parameters:

Name	Type	Description	Default
`snapshot_id_1`	`str`	First snapshot (older)	required
`snapshot_id_2`	`str`	Second snapshot (newer)	required

Returns:

Type	Description
`dict`	Comparison results showing added/removed/modified groups

`maintenance(expire_days=90, cleanup_branches=True, run_gc=True, *, dry_run_gc=False, delete_expired_branches=False, delete_expired_tags=False)` ¶

Run full maintenance on the store: expire, then (optionally) GC.

Must be invoked manually or from a separately scheduled maintenance job: never from the pipeline's own run path, and never on an automatic schedule until garbage_collect(dry_run=True) has been checked against the real store and, if using keeper tags, create_tag() cost at scale has been confirmed. Not safe during active writes — see expire_old_snapshots/ garbage_collect docstrings for the concurrency caveats.

Parameters:

Name	Type	Description	Default
`expire_days`	`int`	Days of snapshot history to keep. Default 90 (weeks-to-months cadence, not the previous 7-day default — 7 days is far too aggressive for a store under sustained daily writes).	`90`
`cleanup_branches`	`bool`	Remove stale temporary branches.	`True`
`run_gc`	`bool`	Run garbage collection after expiration.	`True`
`dry_run_gc`	`bool`	Pass through to `garbage_collect` — report without deleting.	`False`
`delete_expired_branches`	`bool`	Pass through to `expire_old_snapshots`.	`False`
`delete_expired_tags`	`bool`	Pass through to `expire_old_snapshots`. Do not set True if using keeper tags — see `create_keeper_tag`.	`False`

Returns:

Type	Description
`dict`	Summary of maintenance actions.

`normalize_encodings(ds)` ¶

Normalize dataset encodings for Icechunk.

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset to normalize.	required

Returns:

Type	Description
`Dataset`	Dataset with normalized encodings.

`cleanse_dataset_attrs(dataset)` ¶

Remove the attributes that must not be stored (in place; returns dataset).

`settings_file()` ¶

The settings file a store loads when it is given no settings.

`open_gnss_store(store_path)` ¶

Open the GNSS store at store_path; never create it.

A run creates a site's stores when it first writes to them. Reading, branches and tags need no settings file; commits, expiring snapshots and garbage collection do. A store opened without one refuses them with a FileNotFoundError that names canvodpy config init.

Raises:

Type	Description
`FileNotFoundError`	If no store is at `store_path`.

`open_vod_store(store_path)` ¶

Open the VOD store at store_path; never create it.

A run creates a site's stores when it first writes to them. Reading, branches and tags need no settings file; commits, expiring snapshots and garbage collection do. A store opened without one refuses them with a FileNotFoundError that names canvodpy config init.

Raises:

Type	Description
`FileNotFoundError`	If no store is at `store_path`.

Grid storage¶

Store hemisphere grids in an Icechunk store and read them back.

A grid is written to the grids/{grid_name} group of the store as an xarray Dataset with the cell centers, vertices and solid angles (canvod.grids.grid_to_dataset).

`store_grid(grid, store, grid_name)` ¶

Store grid in unified xarray format to Icechunk store.

Converts the grid to an xarray Dataset with vertex information and writes it to the grids/ group in the store.

Parameters:

Name	Type	Description	Default
`grid`	`GridData`	Hemisphere grid instance to store.	required
`store`	`MyIcechunkStore`	Icechunk store to write to.	required
`grid_name`	`str`	Grid identifier for storage path (e.g., 'equal_area_4deg').	required

Returns:

Type	Description
`str`	Snapshot ID from the commit.

Examples:

>>> from canvod.grids import create_hemigrid
>>> from canvod.store import store_grid
>>> grid = create_hemigrid(angular_resolution=4, grid_type='equal_area')
>>> snapshot_id = store_grid(grid, my_store, 'equal_area_4deg')

`load_grid(store, grid_name)` ¶

Load a grid from Icechunk store.

Loads the grid structure from grids/{grid_name} and reconstructs a GridData object.

Parameters:

Name	Type	Description	Default
`store`	`MyIcechunkStore`	Icechunk store to read from.	required
`grid_name`	`str`	Grid identifier (e.g., 'equal_area_4deg').	required

Returns:

Type	Description
`GridData`	Reconstructed grid instance.

Examples:

>>> from canvod.store import load_grid
>>> grid = load_grid(my_store, 'equal_area_4deg')
>>> print(f"Loaded {grid.ncells} cells")

Analysis storage¶

Persistent storage for precomputed analysis results via Icechunk.

Stores dataset+grid-specific analysis outputs in an Icechunk repository under metadata/{dataset_name}/{grid_name}/:

weights – per-cell weight arrays (ncells,).
filter_masks – per-observation statistical masks (epoch × sid).
spatial_masks – per-cell geometric selection masks (ncells,).
statistics – per-cell aggregated statistics (ncells,).

Classes:

Name	Description
``AnalysisStorage``	Read / write / delete analysis metadata for a single Icechunk store.

`AnalysisStorage` ¶

Manage persistent storage of analysis results for dataset+grid pairs.

Storage layout inside the Icechunk repository::

metadata/{dataset_name}/{grid_name}/
├── weights/              # (ncells,)
│   ├── observation_count
│   ├── solid_angle
│   └── combined
├── filter_masks/         # (epoch, sid)
│   ├── mask_iqr
│   └── mask_zscore
├── spatial_masks/        # (ncells,)
│   ├── mask_north
│   └── mask_high_elevation
└── statistics/           # (ncells,)
    ├── obs_count
    ├── mean_vod
    └── std_vod

Parameters:

Name	Type	Description	Default
`store_path`	`Path or str`	Path to the VOD Icechunk store directory.	required

`init(store_path)` ¶

Initialize the storage manager.

Parameters:

Name	Type	Description	Default
`store_path`	`Path \| str`	Path to the VOD Icechunk store directory.	required

`repr()` ¶

Return the developer-facing representation.

Returns:

Type	Description
`str`	Representation string.

`store_weights(dataset_name, grid_name, weights, weight_params=None, overwrite=False)` ¶

Store per-cell weight arrays.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier (e.g. `'reference_01_canopy_01'`).	required
`grid_name`	`str`	Grid identifier (e.g. `'equal_area_2deg'`).	required
`weights`	`dict`	`{name: array}` – all arrays must have shape `(ncells,)`.	required
`weight_params`	`dict`	Parameters used to compute each weight type.	`None`
`overwrite`	`bool`	Overwrite existing weights.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_weights(dataset_name, grid_name, weight_type=None)` ¶

Load stored weight arrays.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`weight_type`	`str`	Load only this weight. `None` loads all.	`None`

Returns:

Type	Description
`dict`	`{name: ndarray}` of loaded weights.

`has_weights(dataset_name, grid_name)` ¶

Return True if weights exist for the dataset+grid pair.

`store_filter_masks(dataset_name, grid_name, masks, filter_params=None, overwrite=False)` ¶

Store per-observation filter masks at native resolution.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`masks`	`dict`	`{filter_name: DataArray}` – all must share the same `(epoch, sid)` shape.	required
`filter_params`	`dict`	Parameters used for each filter.	`None`
`overwrite`	`bool`	Overwrite existing masks.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_filter_mask(dataset_name, grid_name, filter_type, attach_coords=True)` ¶

Load a single filter mask.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`filter_type`	`str`	Filter name (e.g. `'iqr'`, `'zscore'`).	required
`attach_coords`	`bool`	Re-attach `epoch` / `sid` coordinates from the source dataset group. Set `False` for faster loading when coordinates are not needed.	`True`

Returns:

Type	Description
`DataArray`	Boolean mask with shape `(epoch, sid)`.

`load_all_filter_masks(dataset_name, grid_name)` ¶

Load all stored filter masks.

Returns:

Type	Description
`dict`	`{filter_name: DataArray}` – boolean masks.

`has_filter_masks(dataset_name, grid_name)` ¶

Return True if filter masks exist for the dataset+grid pair.

`store_spatial_masks(dataset_name, grid_name, masks, mask_descriptions=None, overwrite=False)` ¶

Store per-cell geometric selection masks.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`masks`	`dict`	`{name: bool_array}` – all arrays must have shape `(ncells,)` and boolean dtype.	required
`mask_descriptions`	`dict`	Human-readable description for each mask (stored as variable attrs).	`None`
`overwrite`	`bool`	Overwrite existing masks.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_spatial_mask(dataset_name, grid_name, mask_name)` ¶

Load a single spatial mask.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`mask_name`	`str`	Mask name (e.g. `'north'`, `'high_elevation'`).	required

Returns:

Type	Description
`ndarray`	Boolean array with shape `(ncells,)`.

`load_all_spatial_masks(dataset_name, grid_name)` ¶

Load all spatial masks.

Returns:

Type	Description
`dict`	`{name: bool_ndarray}`.

`has_spatial_masks(dataset_name, grid_name)` ¶

Return True if spatial masks exist for the dataset+grid pair.

`store_statistics(dataset_name, grid_name, stats, overwrite=False)` ¶

Store pre-computed per-cell statistics.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`stats`	`dict`	`{name: array}` – all arrays must have shape `(ncells,)`. Variables whose name ends with `'_count'` or equals `'obs_count'` are stored as `int64`; everything else as `float32`.	required
`overwrite`	`bool`	Overwrite existing statistics.	`False`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_statistics(dataset_name, grid_name, stat_name=None)` ¶

Load pre-computed per-cell statistics.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`stat_name`	`str`	Load only this statistic. `None` loads all.	`None`

Returns:

Type	Description
`dict`	`{name: ndarray}`.

`has_statistics(dataset_name, grid_name)` ¶

Return True if statistics exist for the dataset+grid pair.

`list_available_metadata(dataset_name, grid_name)` ¶

Check which metadata categories are stored.

Returns:

Type	Description
`dict`	`{category: bool}` for weights, filter_masks, spatial_masks, statistics.

`get_metadata_summary(dataset_name, grid_name)` ¶

Detailed summary of all stored metadata for a dataset+grid pair.

Returns:

Type	Description
`dict`	Nested summary with availability flags and per-category details.

`delete_weights(dataset_name, grid_name)` ¶

Delete all weights for a dataset+grid pair.

`delete_filter_masks(dataset_name, grid_name)` ¶

Delete all filter masks for a dataset+grid pair.

`delete_spatial_masks(dataset_name, grid_name)` ¶

Delete all spatial masks for a dataset+grid pair.

`delete_statistics(dataset_name, grid_name)` ¶

Delete all statistics for a dataset+grid pair.

`store_percell_timeseries(percell_ds, system, band, code, branch='per_cell', metadata=None)` ¶

Store a per-cell timeseries dataset for a system/band/code combination.

Parameters:

Name	Type	Description	Default
`percell_ds`	`Dataset`	Per-cell timeseries dataset with dims `(cell, time)`.	required
`system`	`str`	GNSS system prefix (e.g. `'G'`, `'E'`).	required
`band`	`str`	Frequency band (e.g. `'L1'`, `'E1'`).	required
`code`	`str`	Tracking code (e.g. `'C'`, `'L'`).	required
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`
`metadata`	`dict`	Additional metadata to attach to the dataset.	`None`

Returns:

Type	Description
`str`	Icechunk snapshot ID.

`load_percell_timeseries(system, band, code, branch='per_cell')` ¶

Load a per-cell timeseries dataset.

Parameters:

Name	Type	Description	Default
`system`	`str`	GNSS system prefix.	required
`band`	`str`	Frequency band.	required
`code`	`str`	Tracking code.	required
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`

Returns:

Type	Description
`Dataset`	Per-cell timeseries dataset.

Raises:

Type	Description
`ValueError`	If the requested group does not exist.

`list_percell_datasets(branch='per_cell')` ¶

List all per-cell timeseries datasets in the store.

Parameters:

Name	Type	Description	Default
`branch`	`str`	Icechunk branch name (default: `'per_cell'`).	`'per_cell'`

Returns:

Type	Description
`list of str`	Sorted list of group names in `system_band_code` format.

`has_percell_timeseries(system, band, code, branch='per_cell')` ¶

Return True if a per-cell timeseries exists for the given combination.

`delete_all_metadata(dataset_name, grid_name)` ¶

Delete the entire metadata subtree for a dataset+grid pair.

`delete_specific_weight(dataset_name, grid_name, weight_name)` ¶

Delete a single weight variable from an existing weights group.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`weight_name`	`str`	Weight variable name (e.g. `'observation_count'`).	required

Returns:

Type	Description
`str`	Snapshot ID.

`delete_specific_filter_mask(dataset_name, grid_name, filter_type)` ¶

Delete a single filter mask variable from an existing filter_masks group.

Parameters:

Name	Type	Description	Default
`dataset_name`	`str`	Dataset identifier.	required
`grid_name`	`str`	Grid identifier.	required
`filter_type`	`str`	Filter type name (e.g. `'iqr'`).	required

Returns:

Type	Description
`str`	Snapshot ID.

Preprocessing¶

Preprocessing wrapper for Icechunk storage.

This module provides the IcechunkPreprocessor class which wraps preprocessing functions from canvod.readers.preprocessing and canvod.auxiliary.preprocessing.

`IcechunkPreprocessor` ¶

Handles preprocessing of RINEX-converted datasets before writing to Icechunk.

This class wraps functions from canvod.readers.preprocessing and canvod.auxiliary.preprocessing.

Note: All methods now delegate to canvod.readers.preprocessing / canvod.auxiliary.preprocessing functions. The aggregate_glonass_fdma parameter should be passed from configuration.

`map_aux_sv_to_sid(aux_ds, fill_value=None, aggregate_glonass_fdma=True)` `staticmethod` ¶

Transform auxiliary dataset from sv → sid dimension.

Delegates to canvod.readers.preprocessing.map_aux_sv_to_sid()

Parameters:

Name	Type	Description	Default
`aux_ds`	`Dataset`	Dataset with 'sv' dimension	required
`fill_value`	`float`	Fill value for missing entries (default: np.nan)	`None`
`aggregate_glonass_fdma`	`bool`	Whether to aggregate GLONASS FDMA bands	`True`

Returns:

Type	Description
`Dataset`	Dataset with 'sid' dimension replacing 'sv'

`pad_to_global_sid(ds, keep_sids=None, aggregate_glonass_fdma=True)` `staticmethod` ¶

Pad dataset so it has all possible SIDs across all constellations.

Delegates to canvod.readers.preprocessing.pad_to_global_sid()

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset with 'sid' dimension	required
`keep_sids`	`list[str] \| None`	Optional list of specific SIDs to keep	`None`
`aggregate_glonass_fdma`	`bool`	Whether to aggregate GLONASS FDMA bands	`True`

Returns:

Type	Description
`Dataset`	Dataset padded with NaN for missing SIDs

`normalize_sid_dtype(ds)` `staticmethod` ¶

Ensure sid coordinate uses object dtype.

Delegates to canvod.readers.preprocessing.normalize_sid_dtype()

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset with 'sid' coordinate	required

Returns:

Type	Description
`Dataset`	Dataset with sid as object dtype

`strip_fillvalue(ds)` `staticmethod` ¶

Remove _FillValue attrs/encodings.

Delegates to canvod.readers.preprocessing.strip_fillvalue()

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset to clean	required

Returns:

Type	Description
`Dataset`	Dataset with _FillValue attributes removed

`add_future_datavars(ds, var_config)` `staticmethod` ¶

Add placeholder data variables from configuration.

Delegates to canvod.readers.preprocessing.add_future_datavars()

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset to add variables to	required
`var_config`	`dict[str, dict[str, Any]]`	Variable configuration dictionary	required

Returns:

Type	Description
`Dataset`	Dataset with new variables added

`prep_aux_ds(aux_ds, fill_value=None, aggregate_glonass_fdma=True, keep_sids=None)` `staticmethod` ¶

Preprocess auxiliary dataset before writing to Icechunk.

Delegates to canvod.auxiliary.preprocessing.prep_aux_ds()

Performs complete 4-step preprocessing: 1. Convert sv → sid dimension 2. Pad to global sid list or filter to keep_sids 3. Normalize sid dtype to object 4. Strip _FillValue attributes

Parameters:

Name	Type	Description	Default
`aux_ds`	`Dataset`	Dataset with 'sv' dimension	required
`fill_value`	`float`	Fill value for missing entries (default: np.nan)	`None`
`aggregate_glonass_fdma`	`bool`	Whether to aggregate GLONASS FDMA bands	`True`
`keep_sids`	`list[str] \| None`	List of specific SIDs to keep. If None, keeps all possible SIDs.	`None`

Returns:

Type	Description
`Dataset`	Fully preprocessed dataset ready for Icechunk

canvod.store API reference¶

Package¶

MyIcechunkStore ¶

settings property ¶

repo property ¶

site_name property ¶

source_format property ¶

tree property ¶

__init__(store_path, store_type='gnss_store', compression_level=None, compression_algorithm=None, *, config=None) ¶

chunk_encoding_for(ds) ¶

readonly_session(branch='main') ¶

writable_session(branch='main') ¶

set_root_attrs(attrs, branch='main') ¶

get_root_attrs(branch='main') ¶

get_branch_names() ¶

get_group_names(branch=None) ¶

list_groups(branch='main') ¶

list_data_groups(branch='main') ¶

print_tree(max_depth=None) ¶

group_exists(group_name, branch='main') ¶

preprocessing_records(group_name, start=None, end=None, branch='main') ¶

check_preprocessing_matches(group_name, dataset, branch='main') ¶

check_settings_match(group_name, settings, source='this run has', branch='main') ¶

add_group_variables(group_name, variables, step, software, branch='main') ¶

read_group(group_name, branch='main', time_slice=None, date=None, chunks=None, drop_empty=False) ¶

write_dataset(dataset, group_name, session, mode='a', chunks=None) ¶

grow_day_blocks(days, *, message, branch='main') ¶

backup_metadata_table(group_name, session) ¶

restore_metadata_table(group_name, df, session) ¶

rewrite_group(session, group_name, dataset, drop_ranges, metadata_parts=None) ¶

get_group_info(group_name, branch='main') ¶

metadata_dataset_exists(group_name, name, branch='main') ¶

write_metadata_dataset(meta_ds, group_name, name, branch='main') ¶

write_metadata_parts(parts, group_name, name, session) ¶

append_metadata_datasets(parts, group_name, name, branch='main') ¶

read_metadata_dataset(group_name, name, branch='main', chunks=None) ¶

get_metadata_dataset_info(group_name, name, branch='main') ¶

sbf_metadata_exists(group_name, branch='main') ¶

write_sbf_metadata(meta_ds, group_name, branch='main') ¶

read_sbf_metadata(group_name, branch='main', chunks=None) ¶

get_sbf_metadata_info(group_name, branch='main') ¶

rel_path_for_commit(file_path) ¶

get_store_stats() ¶

append_metadata(group_name, rinex_hash, start, end, snapshot_id, action, commit_msg, dataset_attrs, branch='main', canonical_name=None, physical_path=None) ¶

append_metadata_bulk(group_name, rows, session=None) ¶

should_skip_vod_write(group_name, source_file_hashes, time_start, time_end, branch='main') ¶

commit_vod_days(session, forks, days, branch='main') ¶

write_or_append_vod_groups_batch(items, branch='main', dedup=True) ¶

load_metadata(store, group_name) ¶

read_metadata_table(session, group_name) ¶

metadata_row_exists(group_name, rinex_hash, start, end, branch='main') ¶

should_skip_file(group_name, file_hash, time_start, time_end, branch='main') ¶

dir_entry_counts() ¶

metadata_row_count(group_name, branch='main') ¶

source_file_hashes(group_name, start, end, branch='main') ¶

load_metadata_for_dedup(group_name, branch='main') ¶

batch_check_existing(group_name, file_hashes, metadata_df=None) ¶

check_temporal_overlaps(group_name, file_intervals, branch='main', metadata_df=None) ¶

expire_old_snapshots(days=None, branch='main', delete_expired_branches=False, delete_expired_tags=False) ¶

garbage_collect(days=90, *, dry_run=False) ¶

get_history(branch='main', limit=None) ¶

print_history(branch='main', limit=100) ¶

get_ops_log(limit=None) ¶

print_ops_log(limit=50) ¶

maintenance_due(expire_interval_days, gc_delay_days, manifest_count_threshold=None) ¶

compact_manifests(branch='main') ¶

__repr__() ¶

__str__() ¶

rechunk_group(group_name, chunks, source_branch='main', temp_branch=None, promote_to_main=True, delete_temp_branch=True) ¶

promote_branch(temp_branch, target_branch, expected_tip) ¶

create_branch(branch_name, snapshot_id) ¶

create_release_tag(tag_name, snapshot_id=None) ¶

list_tags() ¶

delete_tag(tag_name) ¶

create_keeper_tag(receiver, yyyydoy, snapshot_id) ¶

plot_commit_graph(branch=None, *, plain=False) ¶

cleanup_stale_branches(keep_patterns=None) ¶

delete_branch(branch_name) ¶

get_snapshot_info(snapshot_id) ¶

compare_snapshots(snapshot_id_1, snapshot_id_2) ¶

`canvod.store` API reference¶

`MyIcechunkStore` ¶

`settings` `property` ¶

`repo` `property` ¶

`site_name` `property` ¶

`source_format` `property` ¶

`tree` `property` ¶

`init(store_path, store_type='gnss_store', compression_level=None, compression_algorithm=None, *, config=None)` ¶

`chunk_encoding_for(ds)` ¶

`readonly_session(branch='main')` ¶

`writable_session(branch='main')` ¶

`set_root_attrs(attrs, branch='main')` ¶

`get_root_attrs(branch='main')` ¶

`get_branch_names()` ¶

`get_group_names(branch=None)` ¶

`list_groups(branch='main')` ¶

`list_data_groups(branch='main')` ¶

`print_tree(max_depth=None)` ¶

`group_exists(group_name, branch='main')` ¶

`preprocessing_records(group_name, start=None, end=None, branch='main')` ¶

`check_preprocessing_matches(group_name, dataset, branch='main')` ¶

`check_settings_match(group_name, settings, source='this run has', branch='main')` ¶

`add_group_variables(group_name, variables, step, software, branch='main')` ¶

`read_group(group_name, branch='main', time_slice=None, date=None, chunks=None, drop_empty=False)` ¶

`write_dataset(dataset, group_name, session, mode='a', chunks=None)` ¶

`grow_day_blocks(days, *, message, branch='main')` ¶

`backup_metadata_table(group_name, session)` ¶

`restore_metadata_table(group_name, df, session)` ¶

`rewrite_group(session, group_name, dataset, drop_ranges, metadata_parts=None)` ¶

`get_group_info(group_name, branch='main')` ¶

`metadata_dataset_exists(group_name, name, branch='main')` ¶

`write_metadata_dataset(meta_ds, group_name, name, branch='main')` ¶

`write_metadata_parts(parts, group_name, name, session)` ¶

`append_metadata_datasets(parts, group_name, name, branch='main')` ¶

`read_metadata_dataset(group_name, name, branch='main', chunks=None)` ¶

`get_metadata_dataset_info(group_name, name, branch='main')` ¶

`sbf_metadata_exists(group_name, branch='main')` ¶

`write_sbf_metadata(meta_ds, group_name, branch='main')` ¶

`read_sbf_metadata(group_name, branch='main', chunks=None)` ¶

`get_sbf_metadata_info(group_name, branch='main')` ¶

`rel_path_for_commit(file_path)` ¶

`get_store_stats()` ¶

`append_metadata(group_name, rinex_hash, start, end, snapshot_id, action, commit_msg, dataset_attrs, branch='main', canonical_name=None, physical_path=None)` ¶

`append_metadata_bulk(group_name, rows, session=None)` ¶

`should_skip_vod_write(group_name, source_file_hashes, time_start, time_end, branch='main')` ¶

`commit_vod_days(session, forks, days, branch='main')` ¶

`write_or_append_vod_groups_batch(items, branch='main', dedup=True)` ¶

`load_metadata(store, group_name)` ¶

`read_metadata_table(session, group_name)` ¶

`metadata_row_exists(group_name, rinex_hash, start, end, branch='main')` ¶

`should_skip_file(group_name, file_hash, time_start, time_end, branch='main')` ¶

`dir_entry_counts()` ¶

`metadata_row_count(group_name, branch='main')` ¶

`source_file_hashes(group_name, start, end, branch='main')` ¶

`load_metadata_for_dedup(group_name, branch='main')` ¶

`batch_check_existing(group_name, file_hashes, metadata_df=None)` ¶

`check_temporal_overlaps(group_name, file_intervals, branch='main', metadata_df=None)` ¶

`expire_old_snapshots(days=None, branch='main', delete_expired_branches=False, delete_expired_tags=False)` ¶

`garbage_collect(days=90, *, dry_run=False)` ¶

`get_history(branch='main', limit=None)` ¶

`print_history(branch='main', limit=100)` ¶

`get_ops_log(limit=None)` ¶

`print_ops_log(limit=50)` ¶

`maintenance_due(expire_interval_days, gc_delay_days, manifest_count_threshold=None)` ¶

`compact_manifests(branch='main')` ¶

`repr()` ¶

`str()` ¶

`rechunk_group(group_name, chunks, source_branch='main', temp_branch=None, promote_to_main=True, delete_temp_branch=True)` ¶

`promote_branch(temp_branch, target_branch, expected_tip)` ¶

`create_branch(branch_name, snapshot_id)` ¶

`create_release_tag(tag_name, snapshot_id=None)` ¶

`list_tags()` ¶

`delete_tag(tag_name)` ¶

`create_keeper_tag(receiver, yyyydoy, snapshot_id)` ¶

`plot_commit_graph(branch=None, *, plain=False)` ¶

`cleanup_stale_branches(keep_patterns=None)` ¶

`delete_branch(branch_name)` ¶

`get_snapshot_info(snapshot_id)` ¶

`compare_snapshots(snapshot_id_1, snapshot_id_2)` ¶

`maintenance(expire_days=90, cleanup_branches=True, run_gc=True, *, dry_run_gc=False, delete_expired_branches=False, delete_expired_tags=False)` ¶