STAC¶

def create_base_models_collection() -> pystac.Collection:
    """base-models: model blueprints contributed via PR."""
    return pystac.Collection(
        id=BASE_MODELS_COLLECTION,
        description="Model blueprints contributed via PR. Each item is a complete model card.",
        extent=pystac.Extent(
            spatial=pystac.SpatialExtent(bboxes=[[-180, -90, 180, 90]]),
            temporal=pystac.TemporalExtent(intervals=[[datetime(2026, 1, 1, tzinfo=UTC), None]]),
        ),
        license="various",
        stac_extensions=BASE_MODEL_EXTENSIONS,
    )

def create_local_models_collection() -> pystac.Collection:
    """local-models: finetuned models, only promoted versions."""
    return pystac.Collection(
        id=LOCAL_MODELS_COLLECTION,
        description="Finetuned models produced by ZenML pipelines. Only promoted versions appear here.",
        extent=pystac.Extent(
            spatial=pystac.SpatialExtent(bboxes=[[-180, -90, 180, 90]]),
            temporal=pystac.TemporalExtent(intervals=[[datetime(2026, 1, 1, tzinfo=UTC), None]]),
        ),
        license="various",
        stac_extensions=LOCAL_MODEL_EXTENSIONS,
    )

def create_datasets_collection() -> pystac.Collection:
    """datasets: training data registered via fAIr UI/backend."""
    return pystac.Collection(
        id=DATASETS_COLLECTION,
        description="Training data registered via fAIr UI/backend.",
        extent=pystac.Extent(
            spatial=pystac.SpatialExtent(bboxes=[[-180, -90, 180, 90]]),
            temporal=pystac.TemporalExtent(intervals=[[datetime(2026, 1, 1, tzinfo=UTC), None]]),
        ),
        license="various",
        stac_extensions=DATASET_EXTENSIONS,
    )

def initialize_catalog(catalog_path: str) -> pystac.Catalog:
    """Create catalog.json + 3 empty collections. Saves to disk.

    returns existing catalog if already present.
    """
    if os.path.exists(catalog_path):
        return pystac.Catalog.from_file(catalog_path)

    catalog = pystac.Catalog(
        id="fair-models",
        description="fAIr model registry and dataset catalog",
    )

    catalog.add_child(create_base_models_collection())
    catalog.add_child(create_local_models_collection())
    catalog.add_child(create_datasets_collection())

    catalog_dir = os.path.dirname(catalog_path) or "."
    catalog.normalize_hrefs(catalog_dir)
    catalog.save(catalog_type=CatalogType.SELF_CONTAINED)

    return catalog

def normalize_version_link_hrefs(
    item: pystac.Item,
    item_href_factory,
    default_collection_id: str,
) -> None:
    """Rewrite version-link relative hrefs (`../../coll/id/id.json`) to absolute API URLs.

    pystac's local-catalog write produces relative hrefs; remote backends need the same
    links pointing at the API URL the catalog will live at. Use raw `link.target` so we
    don't trigger pystac's root resolution (which makes HTTP calls).
    """
    for link in item.links:
        if link.rel not in _VERSION_RELS:
            continue
        target = link.target if isinstance(link.target, str) else ""
        if target.startswith(("http://", "https://")):
            continue
        parts = target.replace("\\", "/").split("/")
        json_parts = [p for p in parts if p.endswith(".json")]
        if not json_parts:
            continue
        target_item_id = json_parts[-1].removesuffix(".json")
        target_coll = parts[-3] if len(parts) >= 3 else default_collection_id
        link.target = item_href_factory(target_coll, target_item_id)

def archive_previous_version(
    backend: StacBackend,
    collection_id: str,
    old_item: pystac.Item,
    new_item_href: str,
) -> pystac.Item:
    """Copy the old item to a version-suffixed ID and deprecate it.

    Preserves the old version as a discoverable, deprecated STAC item
    per the STAC Version Extension v1.2.0 spec.
    """
    old_version = old_item.properties.get("version", "1")
    archived_id = f"{old_item.id}-v{old_version}"

    archived = old_item.clone()
    archived.id = archived_id
    archived.properties["deprecated"] = True
    archived.links = [lnk for lnk in archived.links if lnk.rel not in {"latest-version", "self"}]
    archived.add_link(pystac.Link(rel="successor-version", target=new_item_href))

    archived_href = backend.item_href(collection_id, archived_id)
    archived.add_link(pystac.Link(rel="self", target=archived_href, media_type="application/geo+json"))

    published = backend.publish_item(collection_id, archived)
    log.info("Archived %s/%s as deprecated", collection_id, archived_id)
    return published