gnss-product-management API

YAML-driven GNSS product discovery, query expansion, and dependency resolution.

Client

GNSSClient — high-level entry point for searching and downloading GNSS products.

class gnss_product_management.client.gnss_client.GNSSClient(product_registry, workspace, *, max_connections=4)[source]

Bases: object

Single entry point for IGS product search, download, and dependency resolution.

Wraps ProductQuery, DownloadPipeline, and ResolvePipeline. All user-facing operations go through this class.

Use from_defaults() to build from the bundled IGS center and product specs. Pass a base_dir to enable downloads and local-first resolution:

# Search only (no local sink)
client = GNSSClient.from_defaults()
results = client.search(date, product="ORBIT", parameters={"TTT": "FIN"})

# Search and download to a local directory
client = GNSSClient.from_defaults(base_dir="/data/gnss")
paths = client.download(results[:1], sink_id="local")

# Search and download to S3
client = GNSSClient.from_defaults(base_dir="s3://my-bucket/gnss")
paths = client.download(results[:1], sink_id="local")

# Full dependency resolution (orbit + clock + bias + ERP + ...)
resolution, lockfile = client.resolve_dependencies("spec.yaml", date, sink_id="local")

max_connections sets the per-host FTP/HTTPS connection pool size. CDDIS (NASA FTPS) enforces strict anonymous connection limits; keep it at 2–4 for CDDIS-heavy queries. FTP centers (COD, WUM, GFZ) generally tolerate 6–8.

Parameters:

product_registry (ProductRegistry) – Built ProductRegistry with loaded product and center specs.
workspace (WorkSpace) – WorkSpace with registered local directories.
max_connections (int) – Maximum concurrent connections per host (default 4).

display()[source]

Display the client’s loaded product registry and workspace specs.

Return type:: None

download(results, *, sink_id)[source]

Download product candidates to the local sink.

Pass a pre-sliced list to limit how many files are fetched, e.g. client.download(results[:1], sink_id="local").

Parameters:

results (list[FoundResource]) – Ranked FoundResource list from search() or ProductQuery.search(). Each result must carry a date (set automatically by the query builders).
sink_id (str) – Local resource identifier (alias registered in the workspace, e.g. "local").

Returns:

Paths to successfully downloaded (and decompressed) files.

Return type:

list[Path]

classmethod from_defaults(base_dir=None, *, local_alias='local', max_connections=4)[source]

Construct a client from the bundled default specs.

Loads the pre-built DefaultProductEnvironment and creates a fresh WorkSpace from the bundled local layout specs. If base_dir is provided, all local specs are registered against that directory, enabling downloads and local-first resolution.

base_dir may be a local path or a cloud storage URI supported by cloudpathlib:

# Local
client = GNSSClient.from_defaults(base_dir="/data/gnss")
# Amazon S3
client = GNSSClient.from_defaults(base_dir="s3://my-bucket/gnss")
# Google Cloud Storage
client = GNSSClient.from_defaults(base_dir="gs://my-bucket/gnss")

Parameters:

base_dir (Path | str | None) – Root directory for local product storage. If None, the client operates in search-only mode (no local sink).
local_alias (str) – Alias for the registered local resource (default "local"). Used as the sink_id in download() and resolve_dependencies().
max_connections (int) – Maximum concurrent connections per host. Keep at 2–4 for CDDIS; 6–8 for COD, WUM, GFZ.

Returns:

A configured GNSSClient instance.

Return type:

GNSSClient

query()[source]

Return a fluent ProductQuery builder.

This is the preferred entry point for building searches. Chain calls to narrow the query, then call ProductQuery.search() or ProductQuery.download() to execute:

results = (
    client.query()
    .for_product("ORBIT")
    .on(date)
    .where(TTT="FIN")
    .sources("COD", "ESA")
    .search()
)

Returns:: A ProductQuery bound to this client.
Return type:: ProductQuery

resolve_dependencies(dep_spec, date, *, sink_id)[source]

Resolve all dependencies in a spec for the given date.

Accepts a DependencySpec object or a path to a YAML file. Checks local disk first, then downloads any missing files. If a DependencyLockFile already exists for (package, task, date, version), resolution returns immediately without any network calls.

Parameters:

dep_spec (DependencySpec | Path | str) – Dependency specification — a DependencySpec instance, or a path (str or Path) to a YAML file. The spec encodes which products are required, the center/ timeliness preference cascade, and per-product constraints.
date (datetime) – Target date (timezone-aware datetime, midnight UTC).
sink_id (str) – Local resource alias for storing resolved files (e.g. "local").

Returns:

A (DependencyResolution, lockfile_path) tuple. DependencyResolution exposes .summary(), .table(), .product_paths(), .missing, and .all_required_fulfilled.

Return type:

tuple[DependencyResolution, Path | CloudPath | None]

search(date, product, *, parameters=None, preferences=None, local_resources=None, remote_resources=None)[source]

Search for IGS products matching the given criteria.

Builds SearchTarget objects from the product catalog, lists remote directories via the connection pool, matches filenames by regex, and ranks results by preference and protocol.

Key parameters keys follow the IGS long filename convention:

TTT — solution timeliness: "FIN" (final, ≥13 d), "RAP" (rapid, ≤17 h), "ULT" (ultra-rapid, ≤3 h)
AAA — analysis center: "COD", "ESA", "GFZ", "WUM", "IGS", …

For the full fluent interface (including date-range searches) use query() instead.

Parameters:

date (datetime) – Target date (timezone-aware datetime).
product (str | dict) – Product name (e.g. "ORBIT", "CLOCK", "BIA") or a dict with name, and optionally version / variant.
parameters (dict | None) – Parameter constraints, e.g. {"TTT": "FIN", "AAA": "WUM"}.
preferences (list[SearchPreference] | None) – Preference cascade for ranking. Each SearchPreference names a parameter and an ordered list of preferred values, e.g. SearchPreference(parameter="TTT", sorting=["FIN","RAP"]).
local_resources (list[str] | None) – Restrict to these local resource aliases.
remote_resources (list[str] | None) – Restrict to these remote center IDs.

Returns:

//) results precede remote ones; within each protocol tier results are ordered by preferences.

Return type:

Ranked list of FoundResource objects. Local (file

ProductQuery — fluent builder for GNSS product search and download.

class gnss_product_management.client.product_query.ProductQuery(wormhole, search_planner)[source]

Bases: object

Fluent builder for constructing and executing a GNSS product search.

Constructed via GNSSClient.query() — do not instantiate directly.

Chain method calls to build the query, then call search() or download() to execute:

results = (
    client.query("ORBIT")
    .on(date)
    .where(TTT="FIN")
    .sources("COD", "ESA")
    .prefer(TTT=["FIN", "RAP", "ULT"])
    .search()
)

paths = (
    client.query("CLOCK")
    .on(date)
    .where(TTT="FIN")
    .sources("local", "COD")
    .download(sink_id="local")
)

Parameters:

fetcher – WormHole used for directory listing and file download.
query_factory – SearchPlanner used to build SearchTarget objects from product specs.
product – Product name (e.g. "ORBIT") or dict with name, and optionally version / variant.
wormhole (WormHole)
search_planner (SearchPlanner)

download(sink_id, *, limit=None)[source]

Search and download results in one call.

Parameters:

sink_id (str) – Local resource alias to download into (e.g. "local").
limit (int | None) – Maximum number of files to download. None downloads all results.

Returns:

Paths to successfully downloaded files.

Raises:

ValueError – If for_product() or on() have not been called.

Return type:

list[Path]

for_product(product)[source]

Set the target product for the query.

Parameters:: product (str | dict) – Product name (e.g. "ORBIT") or dict with name, and optionally version / variant.
Returns:: self for chaining.
Return type:: ProductQuery

on(date)[source]

Set the target date for the query.

Parameters:: date (datetime) – Timezone-aware datetime.
Returns:: self for chaining.
Return type:: ProductQuery

on_range(start, end, *, step=datetime.timedelta(days=1))[source]

Set a date range for the query.

Searches are run for every date from start to end (inclusive) with the given step (default: 1 day). Results from all dates are merged into a single flat list from search().

Parameters:

start (datetime) – First date to query (inclusive).
end (datetime) – Last date to query (inclusive).
step (timedelta) – Interval between consecutive dates (default: 1 day).

Returns:

self for chaining.

Raises:

ValueError – If start is after end.

Return type:

ProductQuery

prefer(**kwargs)[source]

Rank results by a preference cascade without hard-filtering.

Unlike where(), prefer keeps all matching products in the result list but sorts them so the most-preferred appear first. The standard IGS timeliness cascade is:

.prefer(TTT=["FIN", "RAP", "ULT"])

Center preference can be layered on top:

.prefer(TTT=["FIN", "RAP", "ULT"], AAA=["WUM", "COD", "GFZ"])

Multiple prefer() calls accumulate in call order; later calls do not override earlier ones.

Parameters:: **kwargs – IGS field name → ordered list of preferred values, most preferred first.
Returns:: self for chaining.
Return type:: ProductQuery

search()[source]

Execute the query and return ranked results.

Returns:: Ranked list of FoundResource objects, best first. Local/file results precede remote ones; within each protocol tier results are ordered by preferences. When on_range() was used, results from all dates are returned as a flat combined list.
Raises:: ValueError – If neither on() nor on_range() has been called.
Return type:: list[FoundResource]

sources(*ids)[source]

Restrict the search to specific local or remote sources.

Pass any mix of registered local aliases and remote center IDs. Each ID is resolved at search() time — local aliases are routed to local disk, everything else is treated as a remote center ID.

Calling sources() with no arguments is an error; omit the call entirely to search all available sources.

Parameters:: *ids (str) – One or more source identifiers.
Returns:: self for chaining.
Raises:: ValueError – If no IDs are provided.
Return type:: ProductQuery

where(**parameters)[source]

Constrain product parameters (hard filter).

Keyword arguments use IGS long filename field codes as keys. Common constraints:

TTT="FIN" — final solutions only (≥13 days latency)
TTT="RAP" — rapid solutions only (≤17 hours latency)
TTT="ULT" — ultra-rapid solutions only (≤3 hours latency)
AAA="WUM" — Wuhan University products only
AAA=["WUM", "COD"] — WUM or COD only

Use prefer() instead of where() when you want to rank results without excluding alternatives.

Parameters:: **parameters – IGS field name → required value or list of values.
Returns:: self for chaining.
Return type:: ProductQuery

Environments

ProductRegistry — builds the full catalog chain and manages remote resources.

Loads parameter, format, product, and resource specification YAMLs, then builds derived catalogs (ParameterCatalog → FormatCatalog → ProductCatalog). Also registers remote ResourceCatalog objects and provides classify() for parsing product filenames back into structured metadata.

class gnss_product_management.environments.environment.LoadedSpecs(*, filename, built)[source]

Bases: BaseModel

Record of a loaded specification file and its parsed result.

Parameters:

filename (Path | str)
built (Any)

built: Any

filename: Path | str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.environments.environment.ProductRegistry[source]

Bases: object

Unified container for the specification / factory layer.

Incrementally loads YAML specs via add_*() methods, then calls build() to derive the full catalog chain:

ParameterCatalog → FormatCatalog → ProductCatalog

Remote resource specs loaded via add_resource_spec() are built into ResourceCatalog objects that support source_product() and sink_product() for remote query resolution.

After building, classify() parses a product filename into structured metadata (product name, format, version, variant, parameters).

_parameter_catalog: Built parameter catalog (available after build()).

_format_catalog: Built format catalog (available after build()).

_product_catalog: Built product catalog (available after build()).

_catalogs: Built remote resource catalogs (available after build()).

__init__()[source]

Initialise an empty registry with no specs loaded.

Return type:: None

add_format_spec(path, id='default')[source]

Load and register a format specification YAML file.

Parameters:

path (Path | str) – Filesystem path to the YAML file.
id (str) – Unique identifier for this spec (default 'default').

Return type:

None

add_parameter_spec(path, id='default')[source]

Load and register a parameter specification YAML file.

Parameters:

path (Path | str) – Filesystem path to the YAML file.
id (str) – Unique identifier for this spec (default 'default').

Return type:

None

add_product_spec(path, id='default')[source]

Load and register a product specification YAML file.

Parameters:

path (Path | str) – Filesystem path to the YAML file.
id (str) – Unique identifier for this spec (default 'default').

Return type:

None

add_resource_spec(path)[source]

Load and register a remote resource specification YAML file.

Parameters:: path (Path | str) – Filesystem path to the YAML file.
Return type:: None

property all_queries: list[SearchTarget]: Flattened list of every search target across all remote centers.

build()[source]

Build the full catalog chain from loaded specs.

Must be called after all add_*() methods. Builds: ParameterCatalog → FormatCatalog → ProductCatalog → remote ResourceCatalog objects.

Return type:: None

property catalogs: list[ResourceCatalog]: All registered remote resource catalogs.

property centers: list[str]: Alias for resource_ids.

classify(filename, parameters=None)[source]

Parse a product filename and return its metadata.

Parameters:

filename (str) – A product filename, optionally including a directory path and/or compression extension.
parameters (list[Parameter] | None) – Optional hard constraints. Products whose fixed parameters conflict with a supplied value are skipped.

Returns:

A dict with keys product, format, version, variant, and parameters on match, or None if no product template matches.

Return type:

dict[str, str] | None

display()[source]

Print a rich summary of loaded products and registered remote centers.

Prints two tables to the terminal:

Products — every product name with its versions and variants.
Remote Centers — every registered data center with its available products, protocols, and hostnames.

Requires the rich package (bundled as a project dependency).

Return type:: None

get(center_id)[source]

Retrieve a remote resource catalog by center identifier.

Parameters:: center_id (str) – Data center identifier.
Returns:: The matching ResourceCatalog.
Return type:: ResourceCatalog

static match_pinned_query(found, incoming)[source]

Check if a found query matches an incoming product based on pinned parameters.

Parameters:

found (Product) – Product from the resource catalog.
incoming (Product) – Product being searched for.

Returns:

The incoming product with matched values filled in, or None if pinned parameters conflict.

Return type:

Product | None

property resource_ids: list[str]: Identifiers for all registered remote resource centers.

sink_product(product, resource_id, date)[source]

Resolve the remote directory/filename for uploading product.

Parameters:

product (Product) – Product to upload.
resource_id (str) – Remote resource identifier.
date (datetime) – Target date for computed fields.

Returns:

A SearchTarget with resolved paths.

Raises:

KeyError – If no matching entry exists.

Return type:

SearchTarget

source_product(product, resource_id)[source]

Resolve a product into all matching SearchTargets for a remote resource.

Parameters:

product (Product) – Product to resolve.
resource_id (str) – Remote resource identifier.

Returns:

A list of SearchTarget objects.

Raises:

KeyError – If resource_id or product.name is not found.

Return type:

list[SearchTarget]

Workspace management for local GNSS product storage.

Maps LocalResourceSpec definitions (loaded from YAML) to concrete base directories on disk so that local resources can be queried through the same ResourceQuery interface used for remote servers.

Base directories may be local filesystem paths (/data/gnss) or cloud URIs (s3://bucket/prefix). Path operations are dispatched through as_path() so that all filesystem interactions work uniformly regardless of backend.

class gnss_product_management.environments.workspace.RegisteredLocalResource(*, name, base_dir, spec, item_to_dir, server)[source]

Bases: BaseModel

A local resource spec that has been bound to a base directory.

base_dir is stored as a URI string so that it can represent both local paths (/data/gnss) and cloud locations (s3://bucket/prefix). Use the base_path property to obtain the appropriate Path or CloudPath object for filesystem operations.

Parameters:

name (str)
base_dir (str)
spec (LocalResourceSpec)
item_to_dir (dict[str, str])
server (Server)

name

Human-readable identifier for this resource.

Type:: str

base_dir

Base directory URI (local path or cloud URI).

Type:: str

spec

The underlying local resource specification.

Type:: LocalResourceSpec

item_to_dir

Mapping of item names to their subdirectory.

Type:: dict[str, str]

server

A file-protocol Server wrapping base_dir.

Type:: Server

base_dir: str

property base_path: Path | CloudPath: The base directory as a Path or cloud path.

item_to_dir: dict[str, str]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

server: Server

spec: LocalResourceSpec

class gnss_product_management.environments.workspace.WorkSpace[source]

Bases: object

Registry of local storage directories and their layout specifications.

Manages the mapping between LocalResourceSpec definitions (loaded from YAML) and concrete base directories on disk or in cloud storage. Each registered spec gets a Server(protocol='file') so that local resources can be queried with the same ResourceQuery interface used for remote servers.

Also provides product-resolution methods (source_product, sink_product, lockfile_dir, find_local_files) once bound to a ProductRegistry via bind().

_registered_specs: Mapping of spec names to registered resources.

_alias_map: Mapping of aliases to canonical spec names.

_resource_specs: Loaded but not-yet-registered spec objects.

_product_catalog: Product catalog (available after bind()).

_parameter_catalog: Parameter catalog (available after bind()).

Usage:

ws = WorkSpace()
ws.add_resource_spec('local_config.yaml')
ws.register_spec(base_dir='/data/gnss', spec_ids=['local_config'], alias='local')

__init__()[source]: Initialise an empty workspace with no specs loaded.

add_resource_spec(path, id=None)[source]

Load a LocalResourceSpec from a YAML file.

Parameters:

path (Path | str) – Path to the YAML specification file.
id (str | None) – Optional override for the spec name. Defaults to the name declared inside the YAML file.

Raises:

AssertionError – If path does not exist or a spec with the same name is already registered.

Return type:

None

bind(product_registry)[source]

Inject catalog references from a built ProductRegistry.

Must be called before using source_product(), sink_product(), or find_local_files().

Parameters:: product_registry (ProductRegistry) – A fully built registry with catalogs.
Return type:: None

display()[source]

Print a rich summary of loaded specs and registered local resources.

Prints two tables:

Loaded Specs — every spec loaded via add_resource_spec(), with its name and product list.
Registered Resources — every resource bound to a base directory via register_spec(), with its alias(es), base directory, and the products it covers.

Requires the rich package (bundled as a project dependency).

Return type:: None

find_local_files(query, date=None)[source]

Search local or cloud storage for files matching a query.

Works identically for local Path and cloud CloudPath base directories.

Parameters:

query (SearchTarget) – SearchTarget with directory and filename patterns.
date (date | None) – Optional date for interpolating computed fields.

Returns:

Sorted list of matching paths (local or cloud).

Raises:

KeyError – If the query’s server is not registered.

Return type:

list[Path | CloudPath]

lockfile_dir(resource_id)[source]

Return the dependency lockfile directory for a resource.

For local resources this is a subdirectory of the base directory. For cloud resources (e.g. s3://bucket/prefix) it is an equivalent cloud path, enabling distributed workers to share lockfile state via cloud storage.

Parameters:: resource_id (str) – Local resource identifier.
Returns:: Path to the lockfile directory (created if needed).
Return type:: Path | CloudPath

register(spec, base_dir, alias=None)[source]

Register a local resource specification in one step.

Convenience alternative to add_resource_spec() + register_spec(). Accepts a LocalResourceSpec object or a path to a YAML file.

Parameters:

spec (LocalResourceSpec | Path | str) – A LocalResourceSpec, or a path to a YAML file.
base_dir (Path | str) – Root directory on disk for the resource.
alias (str | None) – Optional alias that also maps to this resource.

Raises:

ValueError – If base_dir overlaps with an existing registration or alias is already taken.

Return type:

None

register_spec(base_dir, spec_ids, alias=None)[source]

Bind loaded spec(s) to a base directory and register the result.

base_dir may be a local filesystem path or a cloud URI such as s3://bucket/prefix. When multiple spec_ids are given they are merged into a single LocalResourceSpec.

Parameters:

base_dir (Path | CloudPath | str) – Root directory for the resource (local path or cloud URI).
spec_ids (list[str]) – One or more previously loaded spec identifiers.
alias (str | None) – Optional alias that also maps to this resource.

Raises:

AssertionError – If base_dir does not exist or any spec_id has not been loaded.
ValueError – If alias is already in use or base_dir overlaps with an existing registration.

Return type:

None

property resource_ids: list[str]: Identifiers for all registered local resources.

sink_product(product, resource_id, date)[source]

Resolve the local directory for a product on a given date.

Parameters:

product (Product) – Product to locate.
resource_id (str) – Local resource identifier.
date (date) – Target date for computed template fields.

Returns:

A SearchTarget with resolved directory path.

Raises:

KeyError – If resource_id or product.name is not found.

Return type:

SearchTarget

source_product(product, resource_id)[source]

Resolve a product into SearchTarget objects for a local resource.

Parameters:

product (Product) – Product to resolve.
resource_id (str) – Local resource identifier.

Returns:

A list of SearchTarget objects.

Raises:

KeyError – If resource_id or product.name is not found.

Return type:

list[SearchTarget]

gnss_product_management.environments.workspace.paths_overlap(p1, p2)[source]

Check whether two paths share a common ancestor-descendant relationship.

For local paths, checks the resolved filesystem hierarchy. For cloud URIs, falls back to string prefix comparison (cloud paths have no symlinks to resolve).

Parameters:

p1 (Path | CloudPath | str) – First path or URI.
p2 (Path | CloudPath | str) – Second path or URI.

Returns:

True if either path is a parent of (or equal to) the other.

Return type:

bool

Specifications

Parameter model and ParameterCatalog — replaces MetadataField + MetadataCatalog.

class gnss_product_management.specifications.parameters.parameter.DerivationMethod(*values)[source]

Bases: str, Enum

How a parameter value is obtained.

COMPUTED = 'computed'

ENUM = 'enum'

class gnss_product_management.specifications.parameters.parameter.Parameter(*, name, value=None, pattern=None, description=None, derivation=DerivationMethod.ENUM, compute=None)[source]

Bases: BaseModel

A single metadata parameter with optional regex pattern and compute function.

Parameters:

name (str)
value (str | None)
pattern (str | None)
description (str | None)
derivation (DerivationMethod | None)
compute (Callable[[datetime], str] | None)

class Config[source]

Bases: object

arbitrary_types_allowed = True

compute: Callable[[datetime], str] | None

derivation: DerivationMethod | None

description: str | None

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

pattern: str | None

value: str | None

class gnss_product_management.specifications.parameters.parameter.ParameterCatalog(parameters)[source]

Bases: object

Registry of parameters with pattern defaults and computed-field support.

Replaces MetadataCatalog. Compatible with register_computed_fields().

Parameters:: parameters (list[Parameter])

parameters: Mapping of parameter names to Parameter objects.

computed(name, pattern=None, *, description=None)[source]

Decorator that registers a computed parameter field.

Parameters:

name (str) – Parameter name.
pattern (str | None) – Regex pattern for the parameter value.
description (str | None) – Human-readable description.

Returns:

A decorator that wraps the compute function.

defaults()[source]

Return {name: pattern} for every parameter with a pattern.

Returns:: Mapping of parameter names to their regex patterns.
Return type:: dict[str, str]

classmethod from_yaml(yaml_path)[source]

Load parameter definitions from a meta-spec YAML file.

Does not register computed fields — call register_computed_fields() separately after loading.

Parameters:: yaml_path (str | Path)
Return type:: ParameterCatalog

get(name, default=None)[source]

Retrieve a parameter by name.

Parameters:

name (str) – Parameter name.
default – Value returned when name is not found.

Returns:

The Parameter or default.

Return type:

Parameter | None

interpolate(template, date, *, computed_only=False)[source]

Substitute {NAME} placeholders in template.

Parameters:

template (str) – String containing {NAME}-style placeholders.
date (datetime) – Reference datetime for computed fields.
computed_only (bool) – If True, only replace computed parameters.

Returns:

The interpolated string.

Return type:

str

merge(other)[source]

Merge another catalog into this one.

Duplicate names are overwritten by other with a warning.

Parameters:: other (ParameterCatalog) – Catalog to merge.
Returns:: A new ParameterCatalog with combined parameters.
Return type:: ParameterCatalog

register(name, pattern=None, *, compute=None, description=None)[source]

Register or update a parameter, optionally adding a compute function.

Parameters:

name (str) – Parameter name.
pattern (str | None) – Regex pattern for the parameter value.
compute (Callable[[datetime], str] | None) – Callable that derives the value from a datetime.
description (str | None) – Human-readable description.

Returns:

The newly created or updated Parameter.

Return type:

Parameter

resolve_params(params, date)[source]

Set .value on computed parameters from date.

Parameters:

params (list[Any]) – List of Parameter-like objects.
date (datetime) – Reference datetime for computed fields.

Returns:

The same params list with computed values filled in.

Return type:

Any

Raw Pydantic models and registry for format specifications.

Two distinct types live here:

FormatSpec / FormatVersionSpec / FormatFieldDef / FormatSpecCollection: Describe the shape of a file format as a reusable library entry — e.g. “RINEX v3 has these metadata fields and these filename templates”. Loaded from the formats: section of the product spec YAML.
FormatRegistry: Validates and indexes a FormatSpecCollection against a ParameterCatalog, providing get_format() / get_version() look-ups.

See format_spec for the product-facing catalog that resolves format-variant bindings into Product objects.

class gnss_product_management.specifications.format.spec.FormatFieldDef(*, pattern=None, default=None, description=None)[source]

Bases: BaseModel

Properties of a metadata field declared inside a format version.

Parameters:

pattern (str | None)
default (str | None)
description (str | None)

default: str | None

description: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: str | None

class gnss_product_management.specifications.format.spec.FormatRegistry(*, formats=<factory>)[source]

Bases: BaseModel

Read-only registry of reusable format definitions.

Built from a FormatSpecCollection by resolving every metadata field’s default value against a ParameterCatalog.

Parameters:: formats (dict[str, FormatSpec])

formats

Mapping of format name to resolved FormatSpec.

Type:: dict[str, FormatSpec]

classmethod build(format_spec, metadata_catalog)[source]

Build a FormatRegistry by resolving metadata field defaults.

Verifies that every metadata field referenced in a format version has either a pattern value or an entry in metadata_catalog.

Parameters:

format_spec (FormatSpecCollection) – Raw format specification collection.
metadata_catalog (ParameterCatalog) – Global parameter catalog for field defaults.

Returns:

A FormatRegistry with all fields resolved.

Raises:

AssertionError – If a field is missing both a pattern and a catalog entry.
ValueError – If a file template placeholder has no corresponding metadata field.

Return type:

FormatRegistry

formats: dict[str, FormatSpec]

get_format(name)[source]

Retrieve a format by name.

Parameters:: name (str) – Format name.
Returns:: The matching FormatSpec.
Raises:: KeyError – If name is not registered.
Return type:: FormatSpec

get_version(format_name, version)[source]

Retrieve a specific version of a format.

Parameters:

format_name (str) – Format name.
version (str) – Version identifier.

Returns:

The matching FormatVersionSpec.

Raises:

KeyError – If the format or version is not found.

Return type:

FormatVersionSpec

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.specifications.format.spec.FormatSpec(*, description='', versions=<factory>, compression=<factory>)[source]

Bases: BaseModel

A top-level format definition (e.g. RINEX, PRODUCT, TABLE).

Contains the format description and a mapping of version → FormatVersionSpec. Each version in turn maps variant names to filename templates and metadata field definitions.

Note

This is the format-library model. For the product-facing model that binds a specific format+version+variant to parameter lists and filename templates, see FormatVariantSpec.

Parameters:

description (str)
versions (dict[str, FormatVersionSpec])
compression (list[str])

compression: list[str]

description: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

versions: dict[str, FormatVersionSpec]

class gnss_product_management.specifications.format.spec.FormatSpecCollection(*, formats=<factory>)[source]

Bases: BaseModel

Collection of format specifications from the formats: YAML key.

Parameters:: formats (dict[str, FormatSpec])

formats: dict[str, FormatSpec]

classmethod from_yaml(path)[source]

Load from a YAML file.

Accepts two layouts:

Wrapped — a top-level formats: key whose value is a mapping of format name → FormatSpec-compatible dict.
Flat (format_spec.yaml convention) — format names are top-level keys; each entry has versions → variants → {parameters, filename} which are converted to the metadata / file_templates expected by FormatVersionSpec.

Parameters:: path (str | Path) – Path to the YAML file.
Returns:: A FormatSpecCollection instance.
Return type:: FormatSpecCollection

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.specifications.format.spec.FormatVersionSpec(*, description=None, notes=None, metadata=<factory>, file_templates=<factory>, compression=<factory>, **extra_data)[source]

Bases: BaseModel

A single version of a format (e.g. RINEX v3, PRODUCT v1).

Parameters:

description (str | None)
notes (str | None)
metadata (dict[str, FormatFieldDef | None])
file_templates (dict[str, str])
compression (list[str])
extra_data (Any)

compression: list[str]

description: str | None

file_templates: dict[str, str]

metadata: dict[str, FormatFieldDef | None]

model_config = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

notes: str | None

Format-variant specifications and FormatCatalog — resolves FormatVariantSpec → Product.

class gnss_product_management.specifications.format.format_spec.FormatCatalog(*, formats)[source]

Bases: Catalog

Resolved format catalog — maps format names to VersionCatalog[VariantCatalog[Product]].

Parameters:: formats (dict[str, VersionCatalog[Product]])

formats

Mapping of format name to version/variant product hierarchy.

Type:: dict[str, gnss_product_management.specifications.products.product.VersionCatalog[Product]]

classmethod build(format_spec_catalog, parameter_catalog)[source]

Build concrete Products from abstract format specs and a ParameterCatalog.

Parameters:

format_spec_catalog (FormatSpecCatalog) – The raw format spec definitions.
parameter_catalog (ParameterCatalog) – Global parameter catalog.

Returns:

A FormatCatalog mapping format names to resolved products.

Return type:

FormatCatalog

formats: dict[str, VersionCatalog[Product]]

merge(other)[source]

Merge another catalog into this one.

Duplicate entries are overwritten by other with a warning.

Parameters:: other (FormatCatalog) – Catalog to merge.
Returns:: A new FormatCatalog with combined entries.
Return type:: FormatCatalog

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.specifications.format.format_spec.FormatSpecCatalog(*, formats)[source]

Bases: BaseModel

Collection of raw format-variant specifications loaded from YAML.

Parameters:: formats (dict[str, VersionCatalog[FormatVariantSpec]])

formats: dict[str, VersionCatalog[FormatVariantSpec]]

classmethod from_yaml(path)[source]

Load from a YAML file containing pre-built format specs.

Expected YAML layout:

FORMAT_NAME:
  name: FORMAT_NAME
  versions:
    "1":
      name: "1"
      variants:
        variant_name:
          name: FORMAT_NAME
          version: "1"
          variant: variant_name
          parameters:
            - name: PARAM
          filename: "{PARAM}..."

Parameters:: path (Path)
Return type:: FormatSpecCatalog

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.specifications.format.format_spec.FormatVariantSpec(*, name, version=None, variant=None, parameters=<factory>, filename=None)[source]

Bases: BaseModel

A single file-format variant binding: name × version × variant → parameters + filename.

This is a resolved “leaf” entry in the format spec YAML — it names which format (e.g. RINEX), which version ("3"), which variant (observation), and lists the parameters and filename template that together define a concrete file shape.

Parameters:

name (str)
version (str | None)
variant (str | None)
parameters (list[dict] | None)
filename (str | None)

filename: str | None

materialize(parameter_catalog)[source]

Materialize against a ParameterCatalog to produce a Product.

Parameters:: parameter_catalog (ParameterCatalog) – Global parameter catalog for defaults.
Returns:: A Product with resolved parameters.
Return type:: Product

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

parameters: list[dict] | None

variant: str | None

version: str | None

Core product models — PathTemplate, Product, and catalog hierarchies.

class gnss_product_management.specifications.products.product.PathTemplate(*, pattern, value=None, description=None)[source]

Bases: BaseModel

A template pattern with {NAME}-style placeholders, resolved via derive().

Parameters:

pattern (str)
value (str | None)
description (str | None)

derive(parameters)[source]

Replace {PARAM} placeholders in pattern with parameter values.

Parameters:: parameters (list[Parameter]) – List of parameters to substitute.
Return type:: None

description: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: str

to_regex(parameter_catalog)[source]

Convert the template pattern into a regex with named capture groups.

Each {PARAM} placeholder is replaced with (?P<PARAM>pattern) using the parameter’s regex from parameter_catalog. Literal characters outside placeholders are escaped.

Parameters:: parameter_catalog (ParameterCatalog) – Catalog supplying regex patterns for each parameter name.
Returns:: A regex string suitable for re.fullmatch().
Return type:: str

value: str | None

class gnss_product_management.specifications.products.product.Product(*, name, parameters, directory=None, filename=None)[source]

Bases: BaseModel

A resolved product with its parameters and file/directory templates.

Parameters:

name (str)
parameters (list[Parameter])
directory (PathTemplate | None)
filename (PathTemplate | None)

directory: PathTemplate | None

filename: PathTemplate | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

parameters: list[Parameter]

class gnss_product_management.specifications.products.product.VariantCatalog(*, variants)[source]

Bases: BaseModel, Generic[T]

Collection of named variants for a single version.

Parameters:: variants (dict[str, T])

variants

Mapping of variant name to product instance.

Type:: dict[str, gnss_product_management.specifications.products.product.T]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

variants: dict[str, T]

class gnss_product_management.specifications.products.product.VersionCatalog(*, versions)[source]

Bases: BaseModel, Generic[T]

Collection of named versions, each containing variants.

Parameters:: versions (dict[str, VariantCatalog])

versions

Mapping of version name to VariantCatalog.

Type:: dict[str, gnss_product_management.specifications.products.product.VariantCatalog]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

versions: dict[str, VariantCatalog]

gnss_product_management.specifications.products.product.infer_from_regex(regex, filename, parameters)[source]

Infer parameter values from filename using a pre-built regex.

After derive() and the query-factory’s “fill in patterns” step, each parameter’s .value is either a concrete literal or its regex pattern. This function reconstructs a named-group regex by replacing each parameter’s contribution with (?P<name>pattern), then matches filename and updates every parameter’s .value.

Parameters:

regex (str) – Pre-built regex string (values already substituted).
filename (str) – Product filename to match.
parameters (list[Parameter]) – Ordered list of parameters (template order, left-to-right).

Returns:

The updated parameter list on match, or None.

Return type:

list[Parameter] | None

ProductSpec and ProductCatalog — resolve product specs against FormatCatalog.

class gnss_product_management.specifications.products.catalog.ProductCatalog(*, products)[source]

Bases: Catalog

Resolved product catalog — maps product names to VersionCatalog[VariantCatalog[Product]].

Parameters:: products (dict[str, VersionCatalog[Product]])

products

Mapping of product name to version/variant product hierarchy.

Type:: dict[str, gnss_product_management.specifications.products.product.VersionCatalog[Product]]

classmethod build(product_spec_catalog, format_catalog)[source]

Build concrete Products from abstract product specs and a FormatCatalog.

Parameters:

product_spec_catalog (ProductSpecCatalog) – Raw product spec definitions.
format_catalog (FormatCatalog) – Resolved format catalog.

Returns:

A ProductCatalog with all products materialized.

Return type:

ProductCatalog

merge(other)[source]

Merge another catalog into this one.

Duplicate entries are overwritten by other with a warning.

Parameters:: other (ProductCatalog) – Catalog to merge.
Returns:: A new ProductCatalog with combined entries.
Return type:: ProductCatalog

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

products: dict[str, VersionCatalog[Product]]

class gnss_product_management.specifications.products.catalog.ProductSpec(*, name, format, version, variant, parameters=<factory>, filename=None)[source]

Bases: BaseModel

A product specification that binds to a format variant with parameter overrides.

Parameters:

name (str)
format (str)
version (str)
variant (str)
parameters (list[Parameter] | None)
filename (str | None)

filename: str | None

format: str

materialize(format_catalog)[source]

Materialize against a FormatCatalog to produce a fully-merged Product.

Parameters:: format_catalog (FormatCatalog) – Resolved format catalog.
Returns:: A Product with format-level parameters overlaid by product-spec overrides.
Return type:: Product

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

parameters: list[Parameter] | None

variant: str

version: str

class gnss_product_management.specifications.products.catalog.ProductSpecCatalog(*, products)[source]

Bases: BaseModel

Collection of raw product specifications loaded from YAML.

Parameters:: products (dict[str, VersionCatalog[ProductSpec]])

products

Mapping of product name to version/variant spec hierarchy.

Type:: dict[str, gnss_product_management.specifications.products.product.VersionCatalog[ProductSpec]]

classmethod from_yaml(path)[source]

Load product specs from a YAML file, extracting the products: section.

Transforms the YAML structure into nested Pydantic models by injecting name from dict keys and converting formats lists into version/variant dicts.

Return type:: ProductSpecCatalog

merge(other)[source]

Merge another catalog into this one.

Duplicate entries are overwritten by other with a warning.

Parameters:: other (ProductSpecCatalog) – Catalog to merge.
Returns:: A new ProductSpecCatalog with combined entries.
Return type:: ProductSpecCatalog

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

products: dict[str, VersionCatalog[ProductSpec]]

Pure Pydantic models for local storage specifications.

class gnss_product_management.specifications.local.local.LocalCollection(*, directory, description=None, items=<factory>)[source]

Bases: BaseModel

A group of product specs sharing a directory template.

Parameters:

directory (str)
description (str | None)
items (list)

description: str | None

directory: str

items: list

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gnss_product_management.specifications.local.local.LocalResourceSpec(*, name='default', description=None, collections=<factory>, source_file=None)[source]

Bases: BaseModel

Root model for a local storage layout.

A single spec maps collection names to LocalCollection objects. Multiple specs can be merged via merge() so that different YAML files (e.g. per-project or per-workflow) combine into one unified layout.

Parameters:

name (str)
description (str | None)
collections (dict[str, LocalCollection])
source_file (Path | None)

collections: dict[str, LocalCollection]

description: str | None

classmethod from_yaml(path)[source]

Load from a YAML file.

Accepts either a top-level local: wrapper or a flat file whose top-level key is collections:.

Parameters:: path (str | Path) – Path to the YAML file.
Returns:: A LocalResourceSpec instance.
Return type:: LocalResourceSpec

classmethod merge(specs)[source]

Merge multiple local storage specs into one.

Later specs override collections with the same name. Items within identically-named collections are combined (union).

Parameters:: specs (Sequence[LocalResourceSpec]) – Sequence of specs to merge.
Returns:: A new LocalResourceSpec with combined collections.
Return type:: LocalResourceSpec

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

source_file: Path | None

Server, ResourceSpec, SearchTarget — remote resource models (Layer 1).

class gnss_product_management.specifications.remote.resource.ResourceProductSpec(*, id, server_id, available=True, product_name, product_version=None, description=None, parameters, directory)[source]

Bases: BaseModel

A product offering within a resource/center.

Maps a catalog product to a server with parameter overrides.

Parameters:

id (str)
server_id (str)
available (bool)
product_name (str)
product_version (list[str] | str | None)
description (str | None)
parameters (list[Parameter])
directory (PathTemplate)

id

Unique identifier for the product offering.

Type:: str

server_id

Server that hosts this product.

Type:: str

available

Whether the product is currently available.

Type:: bool

product_name

Catalog product name (e.g. 'ORBIT').

Type:: str

product_version

Version filter(s) or None for all.

Type:: list[str] | str | None

description

Human-readable description.

Type:: str | None

parameters

Parameter overrides (values pinned by this center).

Type:: list[gnss_product_management.specifications.parameters.parameter.Parameter]

directory

Directory template for this product.

Type:: gnss_product_management.specifications.products.product.PathTemplate

available: bool

description: str | None

directory: PathTemplate

id: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: list[Parameter]

product_name: str

product_version: list[str] | str | None

server_id: str

class gnss_product_management.specifications.remote.resource.ResourceSpec(*, id, name, description=None, website=None, servers=[], products=[])[source]

Bases: BaseModel

Root resource specification for a data center.

Parameters:

id (str)
name (str)
description (str | None)
website (str | None)
servers (list[Server])
products (list[ResourceProductSpec])

id

Unique center identifier.

Type:: str

name

Display name for the center.

Type:: str

description

Human-readable description.

Type:: str | None

website

Center website URL.

Type:: str | None

servers

Server endpoints for this center.

Type:: list[gnss_product_management.specifications.remote.resource.Server]

products

Product offerings hosted by this center.

Type:: list[gnss_product_management.specifications.remote.resource.ResourceProductSpec]

description: str | None

classmethod from_yaml(path)[source]

Load a resource specification from a YAML file.

Parameters:: path (str | Path) – Path to the YAML file.
Returns:: A ResourceSpec instance.
Return type:: ResourceSpec

id: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

products: list[ResourceProductSpec]

servers: list[Server]

website: str | None

class gnss_product_management.specifications.remote.resource.SearchTarget(*, product, server, directory)[source]

Bases: BaseModel

A single concrete query target — one combination of parameter values.

Parameters:

product (Product)
server (Server)
directory (PathTemplate)

product

The product being queried.

Type:: gnss_product_management.specifications.products.product.Product

server

The server endpoint to query.

Type:: gnss_product_management.specifications.remote.resource.Server

directory

Directory path template for the product.

Type:: gnss_product_management.specifications.products.product.PathTemplate

directory: PathTemplate

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

narrow()[source]

Substitute already-known parameter values into directory/filename patterns.

Returns:: self, mutated in place.
Return type:: SearchTarget

product: Product

server: Server

class gnss_product_management.specifications.remote.resource.Server(*, id, hostname, protocol=None, auth_required=False, description=None)[source]

Bases: BaseModel

A remote or local server endpoint.

Parameters:

id (str)
hostname (str)
protocol (str | None)
auth_required (bool | None)
description (str | None)

id

Unique identifier for the server.

Type:: str

hostname

Server hostname or URL.

Type:: str

protocol

Protocol ('ftp', 'http', 'https', etc.).

Type:: str | None

auth_required

Whether authentication is needed.

Type:: bool | None

description

Human-readable server description.

Type:: str | None

auth_required: bool | None

description: str | None

hostname: str

id: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

protocol: str | None

ResourceCatalog — resolve a ResourceSpec against a ProductCatalog into queryable products.

class gnss_product_management.specifications.remote.resource_catalog.ResourceCatalog(*, id, name, description=None, website=None, servers, queries)[source]

Bases: Catalog

Resolves a ResourceSpec against a ProductCatalog into queryable products.

Parameters:

id (str)
name (str)
description (str | None)
website (str | None)
servers (list[Server])
queries (list[SearchTarget])

id

Center identifier.

Type:: str

name

Display name.

Type:: str

description

Human-readable description.

Type:: str | None

website

Center website URL.

Type:: str | None

servers

Server endpoints for this center.

Type:: list[gnss_product_management.specifications.remote.resource.Server]

queries

Expanded SearchTarget objects.

Type:: list[gnss_product_management.specifications.remote.resource.SearchTarget]

classmethod build(resource_spec, product_catalog)[source]

Build concrete queries by expanding a ResourceSpec against a ProductCatalog.

Parameters:

resource_spec (ResourceSpec) – Raw resource specification for a data center.
product_catalog – Resolved product catalog.

Returns:

A ResourceCatalog containing all expanded queries.

Return type:

ResourceCatalog

description: str | None

id: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

queries: list[SearchTarget]

servers: list[Server]

website: str | None

Pure Pydantic models and result types for dependency specifications.

class gnss_product_management.specifications.dependencies.dependencies.Dependency(*, spec, required=True, description='', constraints=<factory>)[source]

Bases: BaseModel

A single product dependency.

Parameters:

spec (str)
required (bool)
description (str)
constraints (dict[str, str])

constraints: dict[str, str]

description: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

required: bool

spec: str

class gnss_product_management.specifications.dependencies.dependencies.DependencyResolution(spec_name, resolved=<factory>)[source]

Bases: object

Aggregated resolution result for all dependencies in a spec.

Parameters:

spec_name (str)
resolved (list[ResolvedDependency])

spec_name

Name of the dependency specification.

Type:: str

resolved

List of ResolvedDependency results.

Type:: list[gnss_product_management.specifications.dependencies.dependencies.ResolvedDependency]

property all_required_fulfilled: bool: True if every required dependency has been resolved.

property fulfilled: list[ResolvedDependency]: Dependencies that have been resolved (not missing).

property missing: list[ResolvedDependency]: Dependencies that could not be resolved.

product_paths()[source]

Return a {spec: uri} mapping for resolved local files.

Values are URI strings that work for both local paths and cloud locations. Pass them through gnss_product_management.utilities.paths.as_path() to get a path object suitable for filesystem operations.

Returns:: Dict mapping spec names to their local-path or cloud URIs.
Return type:: dict[str, str]

resolved: list[ResolvedDependency]

spec_name: str

summary()[source]

Return a one-line summary of resolution counts.

Returns:: Human-readable summary string.
Return type:: str

table()[source]

Return a formatted table of all resolved dependencies.

Returns:: Multi-line string with columns for spec, required, status, and path.
Return type:: str

class gnss_product_management.specifications.dependencies.dependencies.DependencySpec(*, name, description='', preferences=<factory>, dependencies=<factory>, package, task)[source]

Bases: BaseModel

Full dependency specification for a processing task.

Parameters:

name (str)
description (str)
preferences (list[SearchPreference])
dependencies (list[Dependency])
package (str)
task (str)

dependencies: list[Dependency]

description: str

classmethod from_yaml(path)[source]

Load a dependency specification from a YAML file.

Parameters:: path (str | Path) – Path to the YAML file.
Returns:: A DependencySpec instance.
Return type:: DependencySpec

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

package: str

preferences: list[SearchPreference]

task: str

class gnss_product_management.specifications.dependencies.dependencies.ResolvedDependency(*, spec, required, status, local_path=None, remote_url=None)[source]

Bases: BaseModel

Resolution result for one dependency.

Parameters:

spec (str)
required (bool)
status (str)
local_path (str | None)
remote_url (str | None)

local_path: str | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remote_url: str | None

required: bool

spec: str

status: str

class gnss_product_management.specifications.dependencies.dependencies.SearchPreference(*, parameter, sorting=<factory>, description='')[source]

Bases: BaseModel

One slot in the preference cascade.

Parameters:

parameter (str)
sorting (list[str])
description (str)

description: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameter: str

sorting: list[str]

Base class for Layer 2 catalog objects.

All catalog classes must implement build() as a classmethod that constructs a fully-built instance from lower-layer specifications.

class gnss_product_management.specifications.catalog.Catalog[source]

Bases: BaseModel

Abstract base for built catalogs.

Subclasses must implement:

@classmethod
def build(cls, ...) -> Self:
    ...

The build classmethod is the only way to construct a catalog in production code. Direct __init__ is still available for testing or deserialization.

abstractmethod classmethod build(*args, **kwargs)[source]

Build a concrete catalog instance from lower-layer specs.

Parameters:

*args (Any) – Positional arguments specific to each subclass.
**kwargs (Any) – Keyword arguments specific to each subclass.

Returns:

A fully built catalog instance.

Return type:

Catalog

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Factories

Public return types and exceptions for the ProductRegistry API.

class gnss_product_management.factories.models.DiscoveryEntry(*, product, center='', quality='', source='', uri='')[source]

Bases: BaseModel

A single entry in a discovery report.

Parameters:

product (str)
center (str)
quality (str)
source (str)
uri (str)

center: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

product: str

quality: str

source: str

uri: str

class gnss_product_management.factories.models.DiscoveryReport(*, entries=<factory>)[source]

Bases: BaseModel

Structured summary of available products for a date.

Parameters:: entries (list[DiscoveryEntry])

property centers: list[str]: Sorted list of unique center identifiers in this report.

entries: list[DiscoveryEntry]

filter(product=None, center=None)[source]

Filter entries by product name and/or center.

Parameters:

product (str | None) – Product name filter.
center (str | None) – Center identifier filter.

Returns:

Matching DiscoveryEntry instances.

Return type:

list[DiscoveryEntry]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property products: list[str]: Sorted list of unique product names in this report.

class gnss_product_management.factories.models.FoundResource(*, product, source, uri, parameters=<factory>, date=None, local_path=None)[source]

Bases: BaseModel

A discovered IGS product — either a local file or a remote URI.

Returned by GNSSClient.search() and ProductQuery.search(). The most useful properties for geodetic workflows:

r.center — analysis center code (AAA field, e.g. "WUM")
r.quality — timeliness code (TTT field: "FIN", "RAP", "ULT")
r.filename — bare IGS long filename
r.uri — full remote URL (ftp://...) or local path
r.is_local — True if already on disk
r.downloaded — True after a successful download

Parameters:

product (str)
source (str)
uri (str)
parameters (dict[str, str])
date (datetime | None)
local_path (Path | CloudPath | None)

property center: str: Analysis center identifier (e.g. 'WUM'), or '' if not applicable.

date: datetime.datetime | None

property directory: str: Parent directory of the resource file.

property downloaded: bool: True if the file has been downloaded and exists on disk.

property filename: str: Filename (basename) of the resource.

property hostname: str: Server hostname, or '' for local resources.

property is_local: bool: True if this resource was found on the local filesystem.

local_path: AnyPath | None

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self (BaseModel) – The BaseModel instance.
context (Any) – The context.

Return type:

None

parameters: dict[str, str]

property path: Path | None: Return the local Path if this is a local resource, else None.

product: str

property protocol: str: Transport protocol (e.g. 'ftp', 'https', 'file').

property quality: str: Solution quality/type (e.g. 'FIN', 'RAP'), or '' if not applicable.

source: str

uri: str

exception gnss_product_management.factories.models.MissingProductError(missing, task='')[source]

Bases: Exception

Raised when a required product cannot be found during resolve().

Parameters:

missing (list[str])
task (str)

__init__(missing, task='')[source]

Initialise with the list of missing product names.

Parameters:

missing (list[str]) – Product names that could not be found.
task (str) – Optional task identifier for the error message.

class gnss_product_management.factories.models.Resolution(*, task, paths=<factory>, lockfile=None)[source]

Bases: BaseModel

Result of resolving all dependencies for a task.

Parameters:

task (str)
paths (list[Path])
lockfile (DependencyLockFile | None)

lockfile: DependencyLockFile | None

model_config = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

paths: list[Path]

task: str

SearchPlanner — lazy narrowing query builder.

class gnss_product_management.factories.search_planner.SearchPlanner(product_registry, workspace)[source]

Bases: object

Lazy search planner — narrows parameter ranges, resolves on demand.

Uses ProductRegistry for remote resource catalogs, WorkSpace for local resource resolution, ProductCatalog (nested version→variant→Product hierarchy), and ParameterCatalog for fallback regex patterns and computed date-field resolution.

Parameters:

product_registry (ProductRegistry)
workspace (WorkSpace)

_env: The product registry backing this planner.

_workspace: Workspace with registered local resources.

Usage:

sp = SearchPlanner(
    product_registry=registry,
    workspace=workspace,
)

results = sp.get(
    datetime.date(2024, 1, 1),
    product={"name": "ORBIT"},
    parameters={"AAA": ["WUM", "COD"]},
)

__init__(product_registry, workspace)[source]

Initialise the search planner.

Parameters:

product_registry (ProductRegistry) – Built ProductRegistry with catalogs and remote resource catalogs ready.
workspace (WorkSpace) – WorkSpace with registered local resources.

static build_queries_from_planner(templates, date, query_planner, resource_selection=None)[source]

Build search queries from a given planner and resource selection.

Parameters:

templates (list[Product]) – List of product templates to build queries from.
date (datetime) – Target date for resolving computed directory fields.
query_planner (WorkSpace | ProductRegistry) – A WorkSpace (local) or ProductRegistry (remote) with resource_ids and source_product interface.
resource_selection (list[str] | None) – Optional list of resource IDs to restrict to.

Returns:

A list of SearchTarget objects built from the planner.

Return type:

list[SearchTarget]

get(date, product, parameters=None, local_resources=None, remote_resources=None)[source]

Narrow parameter ranges and return searchable resources.

Parameters:

date (datetime) – Target date for computed metadata fields.
product (dict[str, str | list[str]]) – Product query dict with name, optionally version and variant.
parameters (dict[str, str | list[str]] | None) – User constraints on metadata fields. Unset fields remain as wildcard regex patterns.
local_resources (list[str] | None) – If given, restrict to these local collection IDs.
remote_resources (list[str] | None) – If given, restrict to these remote center IDs.

Returns:

A list of SearchTarget objects.

Raises:

ValueError – If the product, version, or variant is not found.

Return type:

list[SearchTarget]

SourcePlanner — common interface for local and remote search planners.

class gnss_product_management.factories.source_planner.SourcePlanner(*args, **kwargs)[source]

Bases: Protocol

Shared query-side interface for search planners.

Both RemoteSearchPlanner and LocalSearchPlanner satisfy this protocol, allowing SearchPlanner and other consumers to treat local and remote resources uniformly.

Registration (register()) is intentionally not part of this protocol — each planner accepts different spec types at setup time.

register(spec, **args)[source]

Register a resource specification with the planner.

Parameters:

spec – The resource specification to register.
**args – Additional keyword arguments.

Return type:

None

property resource_ids: list[str]: Return identifiers for all registered resources.

sink_product(product, resource_id, **args)[source]

Return the target path for product on resource_id.

Parameters:

product (Product) – Product to resolve.
resource_id (str) – Identifier of the target resource.
**args – Additional keyword arguments.

Returns:

A SearchTarget with resolved path templates.

Return type:

SearchTarget

source_product(product, resource_id, **args)[source]

Resolve product against the resource identified by resource_id.

Parameters:

product (Product) – Product to resolve.
resource_id (str) – Identifier of the target resource.
**args – Additional keyword arguments.

Returns:

A list of SearchTarget objects.

Return type:

list[SearchTarget]

ranking — module-level helpers for sorting SearchTarget results.

gnss_product_management.factories.ranking.sort_by_preferences(targets, preferences)[source]

Sort search targets according to a preference cascade.

Iterates preferences in reverse order (lowest priority first) so that the highest-priority preference ends up as the primary sort key.

Parameters:

targets (list[SearchTarget]) – SearchTarget objects to sort.
preferences (list[SearchPreference]) – Ordered list of SearchPreference objects defining the desired sort cascade.

Returns:

Sorted list of targets.

Return type:

list[SearchTarget]

gnss_product_management.factories.ranking.sort_by_protocol(targets)[source]

Sort search targets by server protocol, preferring local/file over remote.

Parameters:: targets (list[SearchTarget]) – SearchTarget objects to sort.
Returns:: Sorted list of targets, with FILE / LOCAL first, then FTP / FTPS, then HTTP / HTTPS.
Return type:: list[SearchTarget]

WormHole — protocol-agnostic file search and download.

class gnss_product_management.factories.remote_transport.WormHole(*, max_connections=4, product_registry=None)[source]

Bases: object

Search for files described by SearchTarget objects.

For each target, lists the remote (FTP/HTTP) or local directory, matches product.filename.pattern against the listing, and populates directory.value and filename.value on the target.

Parameters:: max_connections (int)

_connection_pool_factory: Factory managing per-host connection pools.

_env: Optional product registry for parameter back-filling.

Usage:

targets = sp.get(date=..., product=..., parameters=...)
transport = WormHole()
results = transport.search(targets)

for st in results:
    print(st.server.hostname, st.product.filename.value)

__init__(*, max_connections=4, product_registry=None)[source]

Initialise the transport.

Parameters:

max_connections (int) – Maximum connections per host pool.
product_registry – Optional ProductRegistry for parameter back-filling after a filename match.

Return type:

None

download_one(query, local_resource_id, local_factory, date)[source]

Synchronously download matched files for one search target.

Skips the download if the destination file already exists and is non-empty.

Parameters:

query (SearchTarget) – The resolved search target with filename value.
local_resource_id (str) – Target local resource identifier.
local_factory (WorkSpace) – Planner for resolving local sink paths.
date (datetime) – Target date for computing sink directory.

Returns:

Path (local or cloud) to the downloaded file, or None on failure.

Return type:

Path | CloudPath | None

search(targets)[source]

Search every target’s server/directory for matching files.

Targets are grouped by (hostname, directory) so each unique remote directory is listed exactly once. Pattern matching for every target in the group runs against the shared listing. The filename.value field is set on each returned target.

Parameters:: targets (list[SearchTarget]) – SearchTarget objects to search.
Returns:: A list of SearchTarget objects with filename.value already populated — one per matched filename.
Return type:: list[SearchTarget]

Connection pool — thread-safe fsspec filesystem instances per host.

class gnss_product_management.factories.connection_pool.ConnectionPool(hostname, max_connections=4)[source]

Bases: object

Thread-safe pool of fsspec filesystem instances for a single host.

Parameters:

hostname (str)
max_connections (int)

hostname: Server address (URL or local path).

protocol: Inferred protocol ('ftp', 'http', 'file', etc.).

max_connections: Maximum number of concurrent connections.

__init__(hostname, max_connections=4)[source]

Initialise a connection pool for hostname.

Parameters:

hostname (str) – Server address or local path.
max_connections (int) – Maximum number of concurrent connections.

full_path(directory)[source]

Combine hostname base with a relative directory.

Behaviour differs by protocol: file uses os.path.join(), http/https concatenates with forward slashes, and ftp treats the directory as an already-absolute server path.

Parameters:: directory (str) – Relative directory path (or absolute for FTP).
Returns:: Full path string suitable for passing to fsspec operations.
Return type:: str

get_connection()[source]

Acquire a connection from the pool.

Yields:: An fsspec.AbstractFileSystem instance.
Raises:: ConnectionError – If the pool failed to initialise.

replace_connection(dead)[source]

Swap a dead connection for a fresh one in the pool.

The caller must already hold a semaphore slot (i.e. be inside get_connection).

Parameters:: dead (AbstractFileSystem) – The stale filesystem instance to remove and replace.
Returns:: A fresh fsspec.AbstractFileSystem, or None if reconnection fails (the pool will be one slot smaller).
Return type:: AbstractFileSystem | None

class gnss_product_management.factories.connection_pool.ConnectionPoolFactory(max_connections=4)[source]

Bases: object

Manage per-host connection pools with a shared directory listing cache.

Parameters:: max_connections (int)

max_connections: Default maximum connections per host.

__init__(max_connections=4)[source]

Initialise the factory.

Parameters:: max_connections (int) – Default maximum connections per host pool.

add_connection(hostname)[source]

Ensure a connection pool exists for hostname.

Parameters:: hostname (str) – Server address to pool.

download_file(hostname, remote_path, target_dir)[source]

Download a file from a remote or local host.

Retries once with a fresh connection on broken-pipe errors.

Parameters:

hostname (str) – Server address.
remote_path (str) – Relative path on the remote host.
target_dir (str) – Local directory to write the file into.

Returns:

Path to the downloaded file, or None on failure.

Raises:

ValueError – If no pool exists for hostname.

Return type:

Path | None

get_connection(hostname)[source]

Acquire a connection for hostname.

Parameters:: hostname (str) – Server address.
Yields:: An fsspec.AbstractFileSystem instance.
Raises:: ValueError – If no pool exists for hostname.

get_file_size(hostname, remote_path)[source]

Return the size of a remote file in bytes, or None if unavailable.

Uses fsspec.info() which is a metadata-only call — no data is transferred. Returns None on any error so callers can treat an unknown size as “proceed with download”.

Parameters:

hostname (str) – Server address.
remote_path (str) – Relative path on the remote host.

Returns:

File size in bytes, or None if the size cannot be determined.

Return type:

int | None

list_directory(hostname, directory)[source]

List a remote or local directory with caching.

Results (including empty lists for failed lookups) are cached to avoid redundant network calls.

Parameters:

hostname (str) – Server address.
directory (str) – Directory path.

Returns:

A list of filenames.

Raises:

ValueError – If no pool exists for hostname.

Return type:

list[str]

ResolvePipeline — Find + Download + LockfileWriter in one call.

High-level composition that resolves a DependencySpec for a given date: finds resources, optionally downloads them, writes per-file sidecar lockfiles, and persists an aggregate lockfile.

Fast path: if an aggregate lockfile already exists for (package, task, date, version) the pipeline returns immediately without searching or downloading.

class gnss_product_management.factories.pipelines.resolve.ResolvePipeline(env, workspace, *, max_connections=4, transport=None)[source]

Bases: object

Find → Download → Lockfile for every dependency in a spec.

Uses ProductQuery, DownloadPipeline, and LockfileWriter internally. All dependencies are resolved in parallel via a ThreadPoolExecutor.

Fast path: if an aggregate lockfile already exists for the (package, task, date, version) identity, returns immediately without searching or downloading.

Parameters:

env (ProductRegistry) – The product registry with built catalogs.
workspace (WorkSpace) – Workspace with registered local resources.
max_connections (int) – Maximum concurrent connections per host.
transport (WormHole | None) – Optional shared WormHole instance. If provided, the pipeline reuses it instead of creating a new one — useful when GNSSClient already holds a pool.

run(spec, date, *, sink_id='local_config', centers=None, download=True)[source]

Resolve all dependencies in spec for date.

Parameters:

spec (DependencySpec) – The dependency specification.
date (datetime) – Target date (timezone-aware datetime).
sink_id (str) – Local resource alias for download destination and lockfile storage.
centers (list[str] | None) – Restrict remote search to these center IDs.
download (bool) – If True (default), download remote resources.

Returns:

A tuple of (DependencyResolution, lockfile path or None if nothing was resolved).

Return type:

tuple[DependencyResolution, Path | CloudPath | None]

DownloadPipeline — found resource → local path.

Fetches remote FoundResource objects to the local workspace using WormHole and SearchPlanner for sink path resolution. Writes a per-file sidecar lockfile after every successful fetch (local or remote) so that callers can verify integrity later.

class gnss_product_management.factories.pipelines.download.DownloadPipeline(env, workspace, *, transport=None, max_connections=4)[source]

Bases: object

Download FoundResource objects to the local workspace.

Already-local resources return immediately with their existing path. Remote resources are downloaded via WormHole.

A per-file sidecar <filename>_lock.json is written alongside every successfully resolved file if one does not already exist.

Parameters:

env (ProductRegistry) – The product registry with built catalogs.
workspace (WorkSpace) – Workspace with registered local resources.
max_connections (int) – Maximum concurrent connections per host.
transport (WormHole | None) – Optional shared WormHole instance.

run(resources, date, *, sink_id='local_config')[source]

Download found resources to the workspace.

Parameters:

resources (FoundResource | list[FoundResource]) – A single FoundResource or a list of them.
date (datetime) – Target date for computing sink directory.
sink_id (str) – Local resource alias to download into.

Returns:

A Path (or None on failure) for a single resource, or a list of paths for multiple resources.

Return type:

Path | None | list[Path | None]

LockfileWriter — write a DependencyResolution to a lockfile.

Thin wrapper around LockfileManager that converts a resolution into a persisted, reproducible manifest.

class gnss_product_management.factories.pipelines.lockfile_writer.LockfileWriter(lockfile_dir, *, package='PRIDE')[source]

Bases: object

Write a DependencyResolution to a JSON lockfile.

Parameters:

lockfile_dir (Path) – Directory where lockfiles are stored.
package (str) – Package name (e.g. 'PRIDE').

write(resolution, date, *, version=None)[source]

Persist resolution as a lockfile.

Parameters:

resolution (DependencyResolution) – The completed dependency resolution.
date (datetime) – Processing date.
version (str | None) – Package version; defaults to the installed version.

Returns:

Path to the written lockfile.

Return type:

Path

Lockfile

Pydantic models for lockfile entries and dependency lockfiles.

Defines the data structures that represent a single resolved product (LockProduct) and a collection of resolved products for one processing day (DependencyLockFile). These models are serialized to JSON sidecar files on disk.

class gnss_product_management.lockfile.models.DependencyLockFile(*, date, package, task, version, requires_date=True, timestamp=<factory>, products=<factory>, metadata=None)[source]

Bases: BaseModel

Top-level lockfile: a fully-resolved, reproducible product manifest.

The lockfile is date-scoped — one lockfile per processing day. Identity key: (task, date, version) where version is the gnss-product-management package version.

Parameters:

date (str)
package (str)
task (str)
version (str)
requires_date (bool)
timestamp (str)
products (list[LockProduct])
metadata (dict | None)

date: str

metadata: dict | None

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

package: str

products: list[LockProduct]

requires_date: bool

task: str

timestamp: str

version: str

class gnss_product_management.lockfile.models.LockProduct(*, name, description='', timestamp=<factory>, url, hash='', size=None, sink='', alternatives=<factory>)[source]

Bases: BaseModel

A single resolved product entry in the lockfile.

Parameters:

name (str)
description (str)
timestamp (str)
url (str)
hash (str)
size (int | None)
sink (str)
alternatives (list[LockProductAlternative])

alternatives: list[LockProductAlternative]

description: str

hash: str

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

sink: str

size: int | None

timestamp: str

url: str

class gnss_product_management.lockfile.models.LockProductAlternative(*, url)[source]

Bases: BaseModel

An alternative (mirror / fallback) source for a locked product.

Parameters:: url (str)

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

url: str

Lockfile manager — single facade for all lockfile operations.

Provides a unified interface for checking, loading, saving, and sharing dependency lockfiles. All callers (ResolvePipeline, PrideProcessor, CLI) should go through LockfileManager rather than calling operations functions directly.

lockfile_dir may be a local filesystem path or a cloud URI (s3://bucket/prefix). All reads and writes are dispatched through CloudPath / Path so the manager is storage-agnostic.

class gnss_product_management.lockfile.manager.LockfileManager(lockfile_dir)[source]

Bases: object

Facade for dependency lockfile lifecycle.

_dir: Directory where aggregate lockfiles are stored.

Parameters:: lockfile_dir (AnyPath | str) – Directory for aggregate lockfile storage. May be a local Path, a URI string, or a CloudPath (e.g. s3://bucket/locks).

__init__(lockfile_dir)[source]

Initialise the manager.

Parameters:: lockfile_dir (Path | CloudPath | str) – Directory for aggregate lockfile storage.
Return type:: None

build_aggregate(products, package, task, date, version=None)[source]

Build a DependencyLockFile from per-file sidecar products.

Parameters:

products (list[LockProduct]) – Lock products collected from sidecar files.
package (str) – Package name (e.g. 'PRIDE').
task (str) – Task name (e.g. 'PPP').
date (datetime | str) – Processing date.
version (str | None) – Package version; defaults to installed version.

Returns:

A new DependencyLockFile ready to be saved.

Return type:

DependencyLockFile

exists(package, task, date, version=None)[source]

Check if a lockfile exists for the given identity.

Parameters:

package (str) – Package name.
task (str) – Task name.
date (datetime) – Processing date.
version (str | None) – Optional package version.

Return type:

bool

export_lockfile(package, task, date, version=None)[source]

Return the path to the aggregate lockfile for sharing.

Raises:

FileNotFoundError – If no lockfile exists for the identity.

Parameters:

package (str)
task (str)
date (datetime)
version (str | None)

Return type:

Path | CloudPath

import_lockfile(path, strict=False)[source]

Import a lockfile from another machine or cloud location.

Validates each product’s hash. In warn mode (default), mismatches are logged but products are kept. In strict mode, invalid products are removed so the caller can re-resolve them.

Parameters:

path (Path | CloudPath | str) – Path or URI to the lockfile JSON to import.
strict (bool) – If True, remove products with hash mismatches.

Returns:

The imported (and possibly pruned) lockfile.

Return type:

DependencyLockFile

load(package, task, date, version=None)[source]

Load an existing lockfile, or None.

Parameters:

package (str) – Package name.
task (str) – Task name.
date (datetime) – Processing date.
version (str | None) – Optional package version.

Return type:

DependencyLockFile | None

property lockfile_dir: Path | CloudPath: The directory where aggregate lockfiles are stored.

static lockfile_name(package, task, date, version=None)[source]

Return the canonical lockfile filename for the given identity.

Parameters:

package (str) – Package name.
task (str) – Task name.
date (datetime) – Processing date.
version (str | None) – Optional package version.

Return type:

str

lockfile_path(package, task, date, version=None)[source]

Return the expected path for a lockfile (may not exist yet).

Parameters:

package (str) – Package name.
task (str) – Task name.
date (datetime) – Processing date.
version (str | None) – Optional package version.

Return type:

Path | CloudPath

save(lockfile)[source]

Write (or overwrite) an aggregate lockfile.

Returns:: Path (local or cloud) to the written file.
Parameters:: lockfile (DependencyLockFile)
Return type:: Path | CloudPath

Read, write, build, and validate lockfile entries.

Provides pure-function helpers for the lockfile lifecycle:

Validate — verify that a lock-product’s sink file still exists and that its hash matches.
Build — construct a LockProduct from a local or cloud file, computing its SHA-256 hash and byte size.
Read / Write — serialize and deserialize individual LockProduct sidecar JSON files and date-scoped DependencyLockFile manifests.

All path arguments accept local Path objects and cloud URIs / CloudPath objects interchangeably via as_path().

class gnss_product_management.lockfile.operations.HashMismatchMode(*values)[source]

Bases: Enum

How to handle hash mismatches during lockfile validation.

STRICT = 'strict'

WARN = 'warn'

gnss_product_management.lockfile.operations.build_lock_product(sink, url, name='', description='', alternative_urls=None)[source]

Build a LockProduct from a local or cloud file.

Computes the SHA-256 hash and byte size of sink and packages them into a new LockProduct model.

Parameters:

sink (Path | CloudPath | str) – Path or URI to the file (local or cloud).
url (str) – Primary remote URL the file was downloaded from.
name (str) – Human-readable product name (e.g. 'ORBIT').
description (str) – Free-text description of the product.
alternative_urls (list[str] | None) – Optional mirror / fallback URLs.

Returns:

A fully-populated LockProduct.

Raises:

FileNotFoundError – If sink does not exist.

Return type:

LockProduct

gnss_product_management.lockfile.operations.get_dependency_lockfile(directory, package, task, date, version=None)[source]

Read a DependencyLockFile from directory.

Parameters:

directory (Path | CloudPath | str) – Folder (local or cloud) containing lockfile JSON files.
package (str) – Processing package name.
task (str) – Processing task name.
date (datetime | str) – Processing date used to derive the filename.
version (str | None) – gnss-product-management package version. Defaults to the installed version.

Returns:

A (lockfile, path) tuple. If the file does not exist, lockfile is None but path is still returned so the caller knows where to write a new one.

Return type:

tuple[DependencyLockFile | None, Path | CloudPath | None]

gnss_product_management.lockfile.operations.get_dependency_lockfile_name(package, task, date, version=None)[source]

Derive the canonical filename for a dependency lockfile.

The filename encodes package, task, year, day-of-year, and version so that each processing day gets its own file. Station is not part of the identity.

Parameters:

package (str) – Processing package name (e.g. 'PRIDE').
task (str) – Processing task name (e.g. 'PPP').
date (str | datetime) – Processing date as a datetime or 'YYYY-MM-DD' string.
version (str | None) – gnss-product-management package version. Defaults to the installed version.

Returns:

A filename string like PRIDE_PPP_2025_015_0.1.0_lock.json.

Raises:

ValueError – If date cannot be parsed.

Return type:

str

gnss_product_management.lockfile.operations.get_lock_product(sink)[source]

Read the LockProduct sidecar JSON for sink.

Parameters:: sink (Path | CloudPath | str) – Path or URI to the downloaded product file.
Returns:: The deserialized LockProduct, or None if the sidecar file does not exist.
Return type:: LockProduct | None

gnss_product_management.lockfile.operations.get_lock_product_path(sink)[source]

Return the _lock.json sidecar path for a sink file.

Works for both local paths and cloud URIs — the sidecar is placed alongside the data file in the same storage backend.

Parameters:: sink (Path | CloudPath | str) – Path or URI to the downloaded product file.
Returns:: A path (local or cloud) pointing to the companion lock JSON (<sink>_lock.json).
Raises:: FileNotFoundError – If sink does not exist.
Return type:: Path | CloudPath

gnss_product_management.lockfile.operations.get_package_version()[source]

Return the installed gnss-product-management version.

Return type:: str

gnss_product_management.lockfile.operations.validate_lock_product(product, mode=HashMismatchMode.WARN)[source]

Check that a lock-product’s sink file exists and its hash matches.

Also accepts the decompressed version of a .gz sink (e.g. foo.SP3 when the lock records foo.SP3.gz).

Parameters:

product (LockProduct) – The lock-product entry to validate.
mode (HashMismatchMode) – How to handle hash mismatches. WARN logs a warning but returns True. STRICT returns False so the caller can re-download.

Returns:

True if the sink file (or its decompressed counterpart) exists and (when a hash is recorded) the file’s current SHA-256 matches the stored hash (or mode is WARN).

Return type:

bool

gnss_product_management.lockfile.operations.write_dependency_lockfile(lockfile, directory, update=False)[source]

Write a DependencyLockFile to directory.

Parameters:

lockfile (DependencyLockFile) – The lockfile model to persist.
directory (Path | CloudPath | str) – Target directory (local path or cloud URI).
update (bool) – If True, overwrite an existing file. Otherwise raise FileExistsError.

Returns:

Path (local or cloud) to the written lockfile.

Raises:

FileExistsError – If the lockfile already exists and update is False.

Return type:

Path | CloudPath

gnss_product_management.lockfile.operations.write_lock_product(lock_product)[source]

Write a LockProduct to its sidecar _lock.json file.

Parameters:: lock_product (LockProduct) – The lock entry to persist.
Returns:: Path (local or cloud) to the written sidecar file.
Return type:: Path | CloudPath

Utilities

Path utilities — unified local/cloud path construction.

as_path is the single entry point for converting URI strings into the appropriate path object. Callers never need to import cloudpathlib directly; this module dispatches to CloudPath for s3://, gs://, and az:// URIs and falls back to Path for everything else.

gnss_product_management.utilities.paths.as_path(uri)[source]

Return a Path or CloudPath.

Detects the scheme from uri and dispatches accordingly:

s3://, gs://, az:// → CloudPath
anything else (including bare POSIX paths) → Path

Parameters:: uri (str | Path | CloudPath) – A URI string, local path string, Path, or CloudPath.
Returns:: The appropriate path object for the given URI.
Return type:: Path | CloudPath

Examples:

>>> as_path("/data/gnss")
PosixPath('/data/gnss')
>>> as_path("s3://my-bucket/gnss-data")
S3Path('s3://my-bucket/gnss-data')
>>> as_path(Path("/data/gnss"))
PosixPath('/data/gnss')

Computed metadata field registrations for a ParameterCatalog.

Defines date-to-metadata transformations (DDD, GPSWEEK, YYYY, REFFRAME, etc.) and register_computed_fields() which wires them onto a catalog so that parameter values can be derived from a processing date at query time.

Usage:

from gnss_product_management.specifications.parameters.parameter import ParameterCatalog
from gnss_product_management.utilities.metadata_funcs import register_computed_fields

cat = ParameterCatalog.from_yaml("meta_spec.yaml")
register_computed_fields(cat)

class gnss_product_management.utilities.metadata_funcs.IGSAntexReferenceFrameType(*values)[source]

Bases: Enum

IGS ANTEX reference frame identifiers.

Maps each IGS reference frame to its canonical lowercase string used in ANTEX filenames (e.g. igs20.atx).

IGS05 = 'igs05'

IGS08 = 'igs08'

IGS14 = 'igs14'

IGS20 = 'igs20'

IGSR3 = 'igsR3'

gnss_product_management.utilities.metadata_funcs.register_computed_fields(registry)[source]

Register all date-derived computed fields onto registry.

Iterates over the built-in _COMPUTED_FIELDS list and calls registry.computed() for each one, wiring the pure date transformation functions into the catalog.

Parameters:: registry – A ParameterCatalog instance to extend.
Return type:: None

Shared helper functions and sentinel types.

Contains low-level utilities used across the package:

hash_file() — SHA-256 file hashing.
_ensure_datetime() — date/datetime normalisation to UTC.
_PassthroughDict — dict that preserves {key} for missing keys.
_listify() — coerce scalars to single-element lists.
expand_dict_combinations() — Cartesian product of dict values.

gnss_product_management.utilities.helpers.decompress_gzip(file_path, dest_dir=None)[source]

Decompress a gzip file and remove the original.

Parameters:

file_path (Path) – Path to the .gz file.
dest_dir (Path | None) – Destination directory for the decompressed file. Defaults to the same directory as file_path.

Returns:

Path to the decompressed file, or None on failure.

Return type:

Path | None

gnss_product_management.utilities.helpers.expand_dict_combinations(d)[source]

Compute the Cartesian product of dict values.

Parameters:: d (dict[str, list[str]]) – Mapping from parameter names to lists of candidate values.
Returns:: A list of dicts, one per combination, with a single value per key.
Return type:: list[dict[str, str]]

Example:

>>> expand_dict_combinations({"A": ["1","2"], "B": ["x","y"]})
[{"A":"1","B":"x"}, {"A":"1","B":"y"}, {"A":"2","B":"x"}, {"A":"2","B":"y"}]

gnss_product_management.utilities.helpers.hash_file(path)[source]

Return the SHA-256 hex digest of a file.

Accepts both local Path and cloud CloudPath objects.

Parameters:: path – Filesystem or cloud path to the file to hash.
Returns:: A string in the form sha256:<hex_digest>.
Return type:: str