canvod.store_metadata API Reference¶

def collect_metadata(
    *,
    config: Any,
    site_name: str,
    site_config: Any,
    store_type: str,
    source_format: str,
    store_path: Path | None = None,
    dask_workers: int | None = None,
    dask_threads_per_worker: int | None = None,
) -> StoreMetadata:
    """Collect all metadata sections into a StoreMetadata object.

    This is the main entry point for metadata collection.
    """
    now = datetime.now(UTC).isoformat()
    meta_cfg = config.processing.metadata

    return StoreMetadata(
        identity=StoreIdentity(
            id=f"{site_name}/{store_type}",
            title=f"{site_name} {store_type.replace('_', ' ').title()}",
            store_type=store_type,
            source_format=source_format,
            keywords=["GNSS", "VOD", site_name, source_format],
            naming_authority=getattr(meta_cfg, "naming_authority", None),
        ),
        creator=collect_creator(meta_cfg),
        publisher=collect_publisher(meta_cfg),
        temporal=TemporalExtent(created=now, updated=now),
        spatial=collect_spatial(site_config, site_name),
        instruments=collect_instruments(site_config),
        processing=collect_processing_provenance(store_type, source_format),
        environment=collect_environment(
            store_path, dask_workers, dask_threads_per_worker
        ),
        config=collect_config_snapshot(config),
        references=collect_references(config),
        summaries=Summaries(
            history=[f"{now}: Store created ({source_format})"],
        ),
    )

def write_metadata(
    store_path: Path,
    metadata: StoreMetadata,
    branch: str = "main",
) -> str:
    """Write metadata to Icechunk store root attrs.

    Returns
    -------
    str
        Snapshot ID from the commit.
    """
    repo = _open_repo(store_path)
    session = repo.writable_session(branch)
    store = session.store
    try:
        root = zarr.open_group(store, mode="r+")
    except zarr_errors.GroupNotFoundError:
        root = zarr.open_group(store, mode="w")

    root.attrs.update(metadata.to_root_attrs())
    return session.commit("Write store metadata")

def read_metadata(
    store_path: Path,
    branch: str = "main",
) -> StoreMetadata:
    """Read metadata from Icechunk store root attrs.

    Raises
    ------
    KeyError
        If no metadata found in store.
    """
    repo = _open_repo(store_path)
    session = repo.readonly_session(branch=branch)
    store = session.store
    root = zarr.open_group(store, mode="r")
    attrs = dict(root.attrs)
    return StoreMetadata.from_root_attrs(attrs)

def update_metadata(
    store_path: Path,
    updates: dict[str, Any],
    branch: str = "main",
) -> str:
    """Merge updates into existing metadata.

    Supports dotted keys for nested updates (e.g. "temporal.updated").

    Returns
    -------
    str
        Snapshot ID from the commit.
    """
    existing = read_metadata(store_path, branch)
    data = existing.model_dump(mode="json")

    for key, value in updates.items():
        parts = key.split(".")
        target = data
        for part in parts[:-1]:
            target = target[part]
        target[parts[-1]] = value

    updated = StoreMetadata.model_validate(data)

    repo = _open_repo(store_path)
    session = repo.writable_session(branch)
    store = session.store
    try:
        root = zarr.open_group(store, mode="r+")
    except zarr_errors.GroupNotFoundError:
        root = zarr.open_group(store, mode="w")

    root.attrs.update(updated.to_root_attrs())
    return session.commit("Update store metadata")

def metadata_exists(store_path: Path, branch: str = "main") -> bool:
    """Check if metadata exists in the store."""
    try:
        repo = _open_repo(store_path)
        session = repo.readonly_session(branch=branch)
        store = session.store
        root = zarr.open_group(store, mode="r")
        return _METADATA_KEY in root.attrs
    except Exception:
        return False

def validate_all(metadata: StoreMetadata) -> dict[str, list[str]]:
    """Run all validators, return issues grouped by standard."""
    return {
        "fair": validate_fair(metadata),
        "datacite": validate_datacite(metadata),
        "acdd": validate_acdd(metadata),
        "stac": validate_stac(metadata),
    }

def scan_stores(root_dir: Path, recursive: bool = True) -> pl.DataFrame:
    """Walk directories, find Icechunk stores, build a catalog.

    Returns
    -------
    pl.DataFrame
        One row per store with metadata columns.
    """
    store_paths = _find_icechunk_stores(root_dir, recursive)
    rows: list[dict[str, Any]] = []

    for sp in store_paths:
        if metadata_exists(sp):
            try:
                meta = read_metadata(sp)
                rows.append(_metadata_to_row(sp, meta))
            except Exception:
                rows.append(_empty_row(sp))
        else:
            rows.append(_empty_row(sp))

    if not rows:
        return pl.DataFrame(schema=_SCHEMA)

    return pl.DataFrame(rows)

def collect_software_versions() -> dict[str, str]:
    """Collect versions of key packages via importlib.metadata."""
    import importlib.metadata

    packages = [
        "canvodpy",
        "canvod-readers",
        "canvod-store",
        "canvod-store-metadata",
        "canvod-utils",
        "icechunk",
        "zarr",
        "xarray",
        "dask",
        "numpy",
        "polars",
        "pydantic",
    ]
    versions: dict[str, str] = {}
    for pkg in packages:
        try:
            versions[pkg] = importlib.metadata.version(pkg)
        except importlib.metadata.PackageNotFoundError:
            pass
    return versions

def collect_python_info() -> str:
    """Return Python version + implementation."""
    return f"{sys.version} ({platform.python_implementation()})"

def collect_uv_version() -> str | None:
    """Return uv version or None if not installed."""
    try:
        result = subprocess.run(
            ["uv", "--version"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except FileNotFoundError, subprocess.TimeoutExpired:
        pass
    return None

def collect_environment(
    store_path: Path | None = None,
    dask_workers: int | None = None,
    dask_threads_per_worker: int | None = None,
) -> Environment:
    """Collect runtime environment information."""
    memory_gb = None
    try:
        import psutil

        memory_gb = round(psutil.virtual_memory().total / (1024**3), 1)
    except ImportError:
        pass

    disk_free_gb = None
    if store_path is not None:
        check_path = store_path if store_path.exists() else store_path.parent
        if check_path.exists():
            usage = shutil.disk_usage(check_path)
            disk_free_gb = round(usage.free / (1024**3), 1)

    # Read raw pyproject.toml + uv.lock from monorepo root
    pyproject_text = None
    uv_lock_text = None
    uv_lock_hash = None
    root = _find_monorepo_root()
    if root is not None:
        pyproject_text = _read_file_text(root / "pyproject.toml")
        uv_lock_text = _read_file_text(root / "uv.lock")
        uv_lock_hash = _read_uv_lock_hash(root)

    return Environment(
        hostname=socket.gethostname(),
        os=platform.platform(),
        arch=platform.machine(),
        cpu_count=os.cpu_count(),
        memory_gb=memory_gb,
        disk_free_gb=disk_free_gb,
        dask_workers=dask_workers,
        dask_threads_per_worker=dask_threads_per_worker,
        uv_lock_hash=uv_lock_hash,
        pyproject_toml_text=pyproject_text,
        uv_lock_text=uv_lock_text,
    )

def collect_config_snapshot(config: Any) -> ConfigSnapshot:
    """Serialize config sections + compute SHA256 hash."""
    sections: dict[str, Any] = {}
    keys = (
        "processing",
        "preprocessing",
        "aux_data",
        "compression",
        "icechunk",
        "sids",
    )
    for key in keys:
        if isinstance(config, dict):
            obj = config.get(key)
        else:
            obj = getattr(config, key, None)
        if obj is None:
            proc = getattr(config, "processing", None)
            if proc is not None:
                obj = getattr(proc, key, None)
        if obj is not None:
            if hasattr(obj, "model_dump"):
                sections[key] = obj.model_dump(mode="json")
            elif isinstance(obj, dict):
                sections[key] = obj
            else:
                sections[key] = str(obj)

    config_str = json.dumps(sections, sort_keys=True, default=str)
    config_hash = hashlib.sha256(config_str.encode()).hexdigest()

    return ConfigSnapshot(
        processing=sections.get("processing"),
        preprocessing=sections.get("preprocessing"),
        aux_data=sections.get("aux_data"),
        compression=sections.get("compression"),
        icechunk=sections.get("icechunk"),
        sids=sections.get("sids"),
        config_hash=config_hash,
    )

def collect_creator(metadata_config: Any) -> Creator:
    """Build Creator from MetadataConfig."""
    return Creator(
        name=metadata_config.author,
        email=str(metadata_config.email),
        orcid=getattr(metadata_config, "orcid", None),
        institution=metadata_config.institution,
        institution_ror=getattr(metadata_config, "institution_ror", None),
        department=getattr(metadata_config, "department", None),
        research_group=getattr(metadata_config, "research_group", None),
        website=getattr(metadata_config, "website", None),
    )

def collect_publisher(metadata_config: Any) -> Publisher:
    """Build Publisher from MetadataConfig."""
    return Publisher(
        name=getattr(metadata_config, "publisher", None),
        url=getattr(metadata_config, "publisher_url", None),
        license=getattr(metadata_config, "license", None),
    )

def collect_spatial(site_config: Any, site_name: str) -> SpatialExtent:
    """Build SpatialExtent from SiteConfig."""
    lat = getattr(site_config, "latitude", None)
    lon = getattr(site_config, "longitude", None)
    alt = getattr(site_config, "altitude_m", None)

    bbox = None
    if lat is not None and lon is not None:
        bbox = [lon, lat, lon, lat]

    return SpatialExtent(
        site=SiteInfo(
            name=site_name,
            description=getattr(site_config, "description", None),
            country=getattr(site_config, "country", None),
        ),
        geospatial_lat=lat,
        geospatial_lon=lon,
        geospatial_alt_m=alt,
        geospatial_lat_min=lat,
        geospatial_lat_max=lat,
        geospatial_lon_min=lon,
        geospatial_lon_max=lon,
        bbox=bbox,
    )

def collect_instruments(site_config: Any) -> Instruments:
    """Build Instruments from SiteConfig receivers."""
    receivers: dict[str, ReceiverInfo] = {}
    for name, rcv in site_config.receivers.items():
        receivers[name] = ReceiverInfo(
            type=rcv.type,
            directory=rcv.directory,
            reader_format=getattr(rcv, "reader_format", "auto"),
            description=getattr(rcv, "description", None),
            recipe=getattr(rcv, "recipe", None),
            metadata=getattr(rcv, "metadata", None),
        )
    return Instruments(receivers=receivers)

def collect_processing_provenance(
    store_type: str,
    source_format: str,
) -> ProcessingProvenance:
    """Build ProcessingProvenance."""
    now = datetime.now(UTC).isoformat()
    return ProcessingProvenance(
        software=collect_software_versions(),
        python=collect_python_info(),
        uv_version=collect_uv_version(),
        level="L1" if store_type == "rinex_store" else "L2",
        lineage=f"Raw {source_format} data ingested into Icechunk store",
        facility=socket.gethostname(),
        datetime=now,
    )

def collect_references(config: Any) -> References:
    """Build References from config if available."""
    from .schema import FundingRef, PublicationRef

    refs_config = None
    proc = getattr(config, "processing", None)
    if proc is not None:
        refs_config = getattr(proc, "references", None)

    if refs_config is None:
        return References()

    pubs = [
        PublicationRef(doi=p.doi, citation=getattr(p, "citation", None))
        for p in getattr(refs_config, "publications", [])
    ]
    funding = [
        FundingRef(
            funder=f.funder,
            funder_ror=getattr(f, "funder_ror", None),
            grant_number=getattr(f, "grant_number", None),
            award_title=getattr(f, "award_title", None),
        )
        for f in getattr(refs_config, "funding", [])
    ]
    return References(publications=pubs, funding=funding)

def collect_metadata(
    *,
    config: Any,
    site_name: str,
    site_config: Any,
    store_type: str,
    source_format: str,
    store_path: Path | None = None,
    dask_workers: int | None = None,
    dask_threads_per_worker: int | None = None,
) -> StoreMetadata:
    """Collect all metadata sections into a StoreMetadata object.

    This is the main entry point for metadata collection.
    """
    now = datetime.now(UTC).isoformat()
    meta_cfg = config.processing.metadata

    return StoreMetadata(
        identity=StoreIdentity(
            id=f"{site_name}/{store_type}",
            title=f"{site_name} {store_type.replace('_', ' ').title()}",
            store_type=store_type,
            source_format=source_format,
            keywords=["GNSS", "VOD", site_name, source_format],
            naming_authority=getattr(meta_cfg, "naming_authority", None),
        ),
        creator=collect_creator(meta_cfg),
        publisher=collect_publisher(meta_cfg),
        temporal=TemporalExtent(created=now, updated=now),
        spatial=collect_spatial(site_config, site_name),
        instruments=collect_instruments(site_config),
        processing=collect_processing_provenance(store_type, source_format),
        environment=collect_environment(
            store_path, dask_workers, dask_threads_per_worker
        ),
        config=collect_config_snapshot(config),
        references=collect_references(config),
        summaries=Summaries(
            history=[f"{now}: Store created ({source_format})"],
        ),
    )

def write_metadata(
    store_path: Path,
    metadata: StoreMetadata,
    branch: str = "main",
) -> str:
    """Write metadata to Icechunk store root attrs.

    Returns
    -------
    str
        Snapshot ID from the commit.
    """
    repo = _open_repo(store_path)
    session = repo.writable_session(branch)
    store = session.store
    try:
        root = zarr.open_group(store, mode="r+")
    except zarr_errors.GroupNotFoundError:
        root = zarr.open_group(store, mode="w")

    root.attrs.update(metadata.to_root_attrs())
    return session.commit("Write store metadata")

def read_metadata(
    store_path: Path,
    branch: str = "main",
) -> StoreMetadata:
    """Read metadata from Icechunk store root attrs.

    Raises
    ------
    KeyError
        If no metadata found in store.
    """
    repo = _open_repo(store_path)
    session = repo.readonly_session(branch=branch)
    store = session.store
    root = zarr.open_group(store, mode="r")
    attrs = dict(root.attrs)
    return StoreMetadata.from_root_attrs(attrs)

def update_metadata(
    store_path: Path,
    updates: dict[str, Any],
    branch: str = "main",
) -> str:
    """Merge updates into existing metadata.

    Supports dotted keys for nested updates (e.g. "temporal.updated").

    Returns
    -------
    str
        Snapshot ID from the commit.
    """
    existing = read_metadata(store_path, branch)
    data = existing.model_dump(mode="json")

    for key, value in updates.items():
        parts = key.split(".")
        target = data
        for part in parts[:-1]:
            target = target[part]
        target[parts[-1]] = value

    updated = StoreMetadata.model_validate(data)

    repo = _open_repo(store_path)
    session = repo.writable_session(branch)
    store = session.store
    try:
        root = zarr.open_group(store, mode="r+")
    except zarr_errors.GroupNotFoundError:
        root = zarr.open_group(store, mode="w")

    root.attrs.update(updated.to_root_attrs())
    return session.commit("Update store metadata")

def metadata_exists(store_path: Path, branch: str = "main") -> bool:
    """Check if metadata exists in the store."""
    try:
        repo = _open_repo(store_path)
        session = repo.readonly_session(branch=branch)
        store = session.store
        root = zarr.open_group(store, mode="r")
        return _METADATA_KEY in root.attrs
    except Exception:
        return False

def validate_datacite(metadata: StoreMetadata) -> list[str]:
    """Check DataCite 4.5 mandatory fields."""
    issues: list[str] = []

    checks = [
        (metadata.identity, "id", "identity"),
        (metadata.identity, "title", "identity"),
        (metadata.creator, "name", "creator"),
        (metadata.creator, "institution", "creator"),
        (metadata.temporal, "created", "temporal"),
        (metadata.identity, "store_type", "identity"),
    ]
    for obj, field, label in checks:
        issue = _check_field(obj, field, label)
        if issue:
            issues.append(issue)

    if metadata.publisher.name is None:
        issues.append("publisher.name is missing (DataCite mandatory)")

    return issues

def validate_acdd(metadata: StoreMetadata) -> list[str]:
    """Check ACDD 1.3 highly recommended + recommended fields."""
    issues: list[str] = []

    hr_checks = [
        (metadata.identity, "title", "identity"),
        (metadata.identity, "id", "identity"),
        (metadata.identity, "conventions", "identity"),
        (metadata.creator, "name", "creator"),
        (metadata.creator, "email", "creator"),
        (metadata.creator, "institution", "creator"),
        (metadata.temporal, "created", "temporal"),
    ]
    for obj, field, label in hr_checks:
        issue = _check_field(obj, field, label)
        if issue:
            issues.append(f"[highly recommended] {issue}")

    rec_checks = [
        (metadata.identity, "keywords", "identity"),
        (metadata.identity, "description", "identity"),
        (metadata.spatial, "geospatial_lat", "spatial"),
        (metadata.spatial, "geospatial_lon", "spatial"),
    ]
    for obj, field, label in rec_checks:
        issue = _check_field(obj, field, label)
        if issue:
            issues.append(f"[recommended] {issue}")

    return issues

def validate_stac(metadata: StoreMetadata) -> list[str]:
    """Check STAC 1.1 Collection required fields."""
    issues: list[str] = []

    if not metadata.identity.id:
        issues.append("identity.id required for STAC Collection")
    if not metadata.identity.title:
        issues.append("identity.title required for STAC Collection")
    if not metadata.identity.description:
        issues.append("identity.description recommended for STAC Collection")
    if metadata.spatial.bbox is None:
        issues.append("spatial.bbox required for STAC spatial extent")
    if metadata.spatial.extent_temporal_interval is None:
        issues.append(
            "spatial.extent_temporal_interval required for STAC temporal extent"
        )
    if metadata.publisher.license is None:
        issues.append("publisher.license required for STAC Collection")

    return issues

def validate_fair(metadata: StoreMetadata) -> list[str]:
    """Check FAIR data principles compliance.

    Checks each sub-principle (F1–F4, A1–A2, I1–I3, R1.1–R1.3)
    and returns actionable issues for anything missing.
    """
    issues: list[str] = []

    # F1 — Globally unique, persistent identifier
    if not metadata.identity.id:
        issues.append(
            "[F1] identity.id is missing — data must have a unique identifier"
        )
    if not metadata.identity.persistent_identifier:
        issues.append(
            "[F1] identity.persistent_identifier is missing — "
            "assign a DOI or other persistent identifier for long-term findability"
        )

    # F2 — Rich metadata
    rich_fields = [
        (metadata.identity, "title", "identity"),
        (metadata.identity, "description", "identity"),
        (metadata.identity, "keywords", "identity"),
        (metadata.creator, "name", "creator"),
        (metadata.creator, "institution", "creator"),
        (metadata.temporal, "collected_start", "temporal"),
        (metadata.spatial, "geospatial_lat", "spatial"),
    ]
    missing_rich = sum(
        1 for obj, field, _ in rich_fields if _check_field(obj, field, "") is not None
    )
    if missing_rich > 2:
        issues.append(
            f"[F2] {missing_rich} of {len(rich_fields)} recommended descriptive "
            "fields are missing — rich metadata improves findability"
        )

    # F3 — Metadata clearly includes identifier of the data
    # Satisfied by design: identity.id is part of the metadata blob

    # F4 — Registered in searchable resource
    # We can't verify external registration, but we can check STAC readiness
    if metadata.spatial.bbox is None:
        issues.append(
            "[F4] spatial.bbox is missing — needed for STAC catalog registration"
        )

    # A1 — Retrievable by identifier using standardised protocol
    if not metadata.references.access_url:
        issues.append(
            "[A1] references.access_url is missing — "
            "provide a URL (S3, HTTP, or local) where data can be accessed"
        )

    # A2 — Metadata accessible even when data is no longer available
    # This is an architectural concern; we note it as advice
    # (metadata is coupled to data in root attrs)

    # I1 — Formal, shared knowledge representation
    # Satisfied: JSON in Zarr root attrs with Pydantic schema

    # I2 — Use FAIR-compliant vocabularies
    if not metadata.identity.conventions:
        issues.append(
            "[I2] identity.conventions is missing — "
            "declare metadata conventions (e.g. ACDD-1.3)"
        )

    # I3 — Qualified references to other metadata
    if not metadata.references.related_stores and not metadata.references.publications:
        issues.append(
            "[I3] No related_stores or publications — "
            "cross-references improve interoperability"
        )

    # R1.1 — Clear, accessible data usage license
    if not metadata.publisher.license:
        issues.append(
            "[R1.1] publisher.license is missing — "
            "specify an SPDX license (e.g. CC-BY-4.0) for reusability"
        )

    # R1.2 — Detailed provenance
    if not metadata.processing.software:
        issues.append(
            "[R1.2] processing.software is empty — "
            "record software versions for provenance"
        )
    if metadata.environment.uv_lock_hash is None:
        issues.append(
            "[R1.2] environment.uv_lock_hash is missing — "
            "store the lock file for environment reproducibility"
        )

    # R1.3 — Meet domain-relevant community standards
    # Checked by the other validators (DataCite, ACDD, STAC)

    return issues

def validate_all(metadata: StoreMetadata) -> dict[str, list[str]]:
    """Run all validators, return issues grouped by standard."""
    return {
        "fair": validate_fair(metadata),
        "datacite": validate_datacite(metadata),
        "acdd": validate_acdd(metadata),
        "stac": validate_stac(metadata),
    }

def scan_stores(root_dir: Path, recursive: bool = True) -> pl.DataFrame:
    """Walk directories, find Icechunk stores, build a catalog.

    Returns
    -------
    pl.DataFrame
        One row per store with metadata columns.
    """
    store_paths = _find_icechunk_stores(root_dir, recursive)
    rows: list[dict[str, Any]] = []

    for sp in store_paths:
        if metadata_exists(sp):
            try:
                meta = read_metadata(sp)
                rows.append(_metadata_to_row(sp, meta))
            except Exception:
                rows.append(_empty_row(sp))
        else:
            rows.append(_empty_row(sp))

    if not rows:
        return pl.DataFrame(schema=_SCHEMA)

    return pl.DataFrame(rows)

def scan_stores_as_stac(root_dir: Path) -> dict[str, Any]:
    """Build a STAC Catalog JSON dict from stores."""
    store_paths = _find_icechunk_stores(root_dir)
    collections: list[dict[str, Any]] = []

    for sp in store_paths:
        if not metadata_exists(sp):
            continue
        try:
            meta = read_metadata(sp)
        except Exception:
            continue

        collection: dict[str, Any] = {
            "type": "Collection",
            "stac_version": "1.1.0",
            "id": meta.identity.id,
            "title": meta.identity.title,
            "description": meta.identity.description or "",
            "license": meta.publisher.license or "proprietary",
            "extent": {
                "spatial": {
                    "bbox": ([meta.spatial.bbox] if meta.spatial.bbox else [[]])
                },
                "temporal": {
                    "interval": (
                        meta.spatial.extent_temporal_interval
                        or [
                            [
                                meta.temporal.collected_start,
                                meta.temporal.collected_end,
                            ]
                        ]
                    )
                },
            },
            "links": [],
        }
        if meta.identity.keywords:
            collection["keywords"] = meta.identity.keywords
        collections.append(collection)

    return {
        "type": "Catalog",
        "stac_version": "1.1.0",
        "id": "canvod-catalog",
        "description": "canVOD Icechunk Store Catalog",
        "links": [{"rel": "child", "href": f"#{c['id']}"} for c in collections],
        "collections": collections,
    }

def write_stac_collection(
    store_path: Path,
    output_path: Path | None = None,
    branch: str = "main",
) -> Path:
    """Write a STAC Collection JSON file for a single store.

    Parameters
    ----------
    store_path : Path
        Path to the Icechunk store.
    output_path : Path | None
        Output JSON file path. Defaults to ``store_path / "collection.json"``.
    branch : str
        Store branch to read metadata from.

    Returns
    -------
    Path
        The written JSON file path.
    """
    meta = read_metadata(store_path, branch)
    collection = _metadata_to_stac_collection(meta)

    if output_path is None:
        output_path = store_path / "collection.json"

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(collection, indent=2, default=str))
    return output_path

def write_stac_catalog(
    root_dir: Path,
    output_path: Path | None = None,
    write_collections: bool = True,
) -> Path:
    """Write a STAC Catalog JSON and optional per-store Collection JSONs.

    Parameters
    ----------
    root_dir : Path
        Root directory to scan for Icechunk stores.
    output_path : Path | None
        Output catalog JSON path. Defaults to ``root_dir / "catalog.json"``.
    write_collections : bool
        If True, also write a ``collection.json`` next to each store.

    Returns
    -------
    Path
        The written catalog JSON file path.
    """
    catalog = scan_stores_as_stac(root_dir)

    if output_path is None:
        output_path = root_dir / "catalog.json"

    # Rewrite links to point to actual collection.json files
    if write_collections:
        store_paths = _find_icechunk_stores(root_dir)
        for sp in store_paths:
            if not metadata_exists(sp):
                continue
            try:
                coll_path = write_stac_collection(sp)
                # Update catalog link to relative path
                rel = coll_path.relative_to(output_path.parent)
                for link in catalog["links"]:
                    meta = read_metadata(sp)
                    if link.get("href") == f"#{meta.identity.id}":
                        link["href"] = str(rel)
            except Exception:
                continue

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(catalog, indent=2, default=str))
    return output_path

def format_metadata(
    meta: StoreMetadata,
    section: str | None = None,
) -> str:
    """Format metadata as a human-readable string.

    Parameters
    ----------
    meta : StoreMetadata
        The metadata to format.
    section : str | None
        If given, show only this section. Special values:
        - "uv" / "uv.lock": dump raw uv.lock content
        - "toml" / "pyproject": dump raw pyproject.toml content
        - "env-reproduce": print instructions to reproduce env
        - Any key from _SECTIONS: show that section only
        If None, show full report.
    """
    if section in ("uv", "uv.lock"):
        if meta.environment.uv_lock_text:
            return meta.environment.uv_lock_text
        return "(uv.lock not stored in this metadata)"

    if section in ("toml", "pyproject", "pyproject.toml"):
        if meta.environment.pyproject_toml_text:
            return meta.environment.pyproject_toml_text
        return "(pyproject.toml not stored in this metadata)"

    if section in ("env-reproduce", "reproduce"):
        return _format_reproduce_instructions(meta)

    if section is not None:
        formatter = _SECTIONS.get(section)
        if formatter is None:
            available = ", ".join(list(_SECTIONS.keys()) + ["uv", "toml", "reproduce"])
            return f"Unknown section '{section}'. Available: {available}"
        return formatter(meta)

    # Full report
    parts = [
        f"{'=' * 72}",
        "  canvod Store Metadata Report",
        f"  Schema version: {meta.metadata_version}",
        f"{'=' * 72}",
    ]
    for formatter in _SECTIONS.values():
        parts.append("")
        parts.append(formatter(meta))
    parts.append("")
    parts.append(_hr("="))
    return "\n".join(parts)

def extract_env(
    store_path: Path,
    output_dir: Path,
    branch: str = "main",
) -> Path:
    """Extract pyproject.toml + uv.lock from a store for reproduction.

    Parameters
    ----------
    store_path : Path
        Path to the Icechunk store.
    output_dir : Path
        Directory to write the files into.
    branch : str
        Store branch.

    Returns
    -------
    Path
        The output directory (ready for ``uv sync --frozen``).
    """
    meta = read_metadata(store_path, branch)
    output_dir.mkdir(parents=True, exist_ok=True)

    if meta.environment.pyproject_toml_text is None:
        msg = "pyproject.toml not stored in this metadata"
        raise ValueError(msg)
    if meta.environment.uv_lock_text is None:
        msg = "uv.lock not stored in this metadata"
        raise ValueError(msg)

    (output_dir / "pyproject.toml").write_text(meta.environment.pyproject_toml_text)
    (output_dir / "uv.lock").write_text(meta.environment.uv_lock_text)
    return output_dir

def show_metadata(
    store_path: str | Path,
    section: str | None = None,
    branch: str = "main",
) -> None:
    """Print store metadata to stdout.

    Parameters
    ----------
    store_path : str | Path
        Path to an Icechunk store.
    section : str | None
        Section to show (None = full report).
    branch : str
        Store branch.
    """
    path = Path(store_path)
    if not metadata_exists(path, branch):
        print(f"No metadata found in {path}")
        return
    meta = read_metadata(path, branch)
    print(format_metadata(meta, section))