Python API¶

class OscType(StringEnum):
    """The supported Open Science Catalog resource types."""

    PROJECT = "project"
    PRODUCT = "product"

class OscStatus(StringEnum):
    """The supported lifecycle states for OSC projects and products."""

    PLANNED = "planned"
    ONGOING = "ongoing"
    COMPLETED = "completed"

class OscExtension(
    Generic[T],
    PropertiesExtension,
    ExtensionManagementMixin[pystac.Catalog | pystac.Collection | pystac.Item],
):
    """The Open Science Catalog Extension for Catalogs, Collections, and Items."""

    name: Literal["osc"] = "osc"

    def apply_project(
        self,
        *,
        status: OscStatus,
        workflows: list[str] | None = None,
    ) -> None:
        """Apply the OSC project fields and remove product-only fields."""
        self.osc_type = OscType.PROJECT
        self.status = status
        self.workflows = workflows

        self.project = None
        self.region = None
        self.variables = None
        self.missions = None
        self.experiment = None

    def apply_product(
        self,
        *,
        status: OscStatus,
        project: str,
        region: str | None = None,
        variables: list[str] | None = None,
        missions: list[str] | None = None,
        experiment: str | None = None,
    ) -> None:
        """Apply the OSC product fields and remove project-only fields."""
        self.osc_type = OscType.PRODUCT
        self.status = status
        self.project = project
        self.region = region
        self.variables = variables
        self.missions = missions
        self.experiment = experiment

        self.workflows = None

    @property
    def osc_type(self) -> OscType | None:
        """The OSC resource type."""
        value = self._get_property(TYPE_PROP, str)
        return OscType(value) if value is not None else None

    @osc_type.setter
    def osc_type(self, value: OscType | None) -> None:
        self._set_property(TYPE_PROP, value.value if value is not None else None)

    @property
    def status(self) -> OscStatus | None:
        """The OSC lifecycle status."""
        value = self._get_property(STATUS_PROP, str)
        return OscStatus(value) if value is not None else None

    @status.setter
    def status(self, value: OscStatus | None) -> None:
        self._set_property(STATUS_PROP, value.value if value is not None else None)

    @property
    def workflows(self) -> list[str] | None:
        """The workflows created by an OSC project."""
        return self._get_property(WORKFLOWS_PROP, list[str])

    @workflows.setter
    def workflows(self, value: list[str] | None) -> None:
        self._set_property(WORKFLOWS_PROP, value)

    @property
    def project(self) -> str | None:
        """The project associated with an OSC product."""
        return self._get_property(PROJECT_PROP, str)

    @project.setter
    def project(self, value: str | None) -> None:
        self._set_property(PROJECT_PROP, value)

    @property
    def region(self) -> str | None:
        """The geographic region associated with an OSC product."""
        return self._get_property(REGION_PROP, str)

    @region.setter
    def region(self, value: str | None) -> None:
        self._set_property(REGION_PROP, value)

    @property
    def variables(self) -> list[str] | None:
        """The variables observed by an OSC product."""
        return self._get_property(VARIABLES_PROP, list[str])

    @variables.setter
    def variables(self, value: list[str] | None) -> None:
        self._set_property(VARIABLES_PROP, value)

    @property
    def missions(self) -> list[str] | None:
        """The missions providing input to an OSC product."""
        return self._get_property(MISSIONS_PROP, list[str])

    @missions.setter
    def missions(self, value: list[str] | None) -> None:
        self._set_property(MISSIONS_PROP, value)

    @property
    def experiment(self) -> str | None:
        """The experiment that created an OSC product."""
        return self._get_property(EXPERIMENT_PROP, str)

    @experiment.setter
    def experiment(self, value: str | None) -> None:
        self._set_property(EXPERIMENT_PROP, value)

    @classmethod
    def get_schema_uri(cls) -> str:
        """Return the supported OSC extension schema identifier."""
        return SCHEMA_URI

    @classmethod
    def ext(cls, obj: T, add_if_missing: bool = False) -> OscExtension[T]:
        """Wrap a Catalog, Collection, or Item with live OSC accessors.

        Args:
            obj: Object whose OSC fields will be read or modified.
            add_if_missing: Add the schema declaration when absent.

        Raises:
            pystac.ExtensionNotImplemented: If the declaration is missing and
                adding it was not requested.
            pystac.ExtensionTypeError: If the object type is unsupported.
        """
        if isinstance(obj, pystac.Collection):
            cls.ensure_has_extension(obj, add_if_missing)
            return cast("OscExtension[T]", CollectionOscExtension(obj))
        if isinstance(obj, pystac.Catalog):
            cls.ensure_has_extension(obj, add_if_missing)
            return cast("OscExtension[T]", CatalogOscExtension(obj))
        if isinstance(obj, pystac.Item):
            cls.ensure_has_extension(obj, add_if_missing)
            return cast("OscExtension[T]", ItemOscExtension(obj))
        raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

class CatalogOscExtension(OscExtension[pystac.Catalog]):
    """Expose OSC fields on a PySTAC catalog."""

    catalog: pystac.Catalog
    properties: dict[str, Any]

    def __init__(self, catalog: pystac.Catalog) -> None:
        self.catalog = catalog
        self.properties = catalog.extra_fields

    def __repr__(self) -> str:
        return f"<CatalogOscExtension Catalog id={self.catalog.id}>"

class CollectionOscExtension(OscExtension[pystac.Collection]):
    """Expose OSC fields on a PySTAC collection."""

    collection: pystac.Collection
    properties: dict[str, Any]

    def __init__(self, collection: pystac.Collection) -> None:
        self.collection = collection
        self.properties = collection.extra_fields

    def __repr__(self) -> str:
        return f"<CollectionOscExtension Collection id={self.collection.id}>"

class ItemOscExtension(OscExtension[pystac.Item]):
    """Expose OSC fields on a PySTAC item."""

    item: pystac.Item
    properties: dict[str, Any]

    def __init__(self, item: pystac.Item) -> None:
        self.item = item
        self.properties = item.properties

    def __repr__(self) -> str:
        return f"<ItemOscExtension Item id={self.item.id}>"

class OscExtensionHooks(ExtensionHooks):
    """Describe OSC schema membership for PySTAC migration hooks."""

    schema_uri: str = SCHEMA_URI
    prev_extension_ids: ClassVar[set[str]] = set()
    stac_object_types: ClassVar[set[pystac.STACObjectType]] = {
        pystac.STACObjectType.CATALOG,
        pystac.STACObjectType.COLLECTION,
        pystac.STACObjectType.ITEM,
    }

class OGCRecord(RecordMetadataMixin, pystac.Item):
    """Item-compatible Record, with OGC serialization as the default.

    Initializes STACObject directly to avoid Item's mandatory temporal extent.
    This compatibility seam must be reviewed for future PySTAC releases.
    `time` and STAC datetime fields are independent: no implicit conversion.
    """

    SCHEMA_URI = (
        "https://schemas.opengis.net/ogcapi/records/part1/1.0/openapi/schemas/recordGeoJSON.yaml"
    )
    _RESERVED = frozenset(
        {
            "id",
            "type",
            "geometry",
            "properties",
            "links",
            "assets",
            "bbox",
            "collection",
            "stac_extensions",
        }
    )

    # Preserve the positional PySTAC Item constructor API and OGC keyword options.
    def __init__(  # noqa: PLR0917
        self,
        id: str | int,
        geometry: dict[str, Any] | None = None,
        bbox: list[float] | None = None,
        datetime: Datetime | None = None,
        properties: dict[str, Any] | None = None,
        start_datetime: Datetime | None = None,
        end_datetime: Datetime | None = None,
        stac_extensions: list[str] | None = None,
        href: str | None = None,
        collection: str | pystac.Collection | None = None,
        extra_fields: dict[str, Any] | None = None,
        assets: dict[str, pystac.Asset] | None = None,
        *,
        time: dict[str, Any] | None = None,
        conforms_to: list[str] | None = None,
        link_templates: list[dict[str, Any]] | None = None,
    ) -> None:
        """Initialize a record, copying metadata and assigning asset ownership.

        OGC ``time`` is independent of the STAC datetime arguments. Optional
        OGC fields override matching entries in ``extra_fields`` when supplied.

        Raises:
            TypeError: If the identifier is not a string or integer.
            ValueError: If extra fields override managed record fields.
        """
        if isinstance(id, bool) or not isinstance(id, (str, int)):
            raise TypeError("Record id must be a string or integer")
        pystac.STACObject.__init__(self, list(stac_extensions or []))
        self._record_id = id
        self.id = str(id)  # PySTAC path/layout utilities require a string.
        self.geometry = deepcopy(geometry)
        self.bbox = deepcopy(bbox)
        self._null_properties = properties is None
        self.properties = deepcopy(properties) if properties is not None else {}
        self.extra_fields = deepcopy(extra_fields) if extra_fields else {}
        if self._RESERVED.intersection(self.extra_fields):
            raise ValueError("extra_fields must not override managed fields")
        self.assets = {}
        self.collection_id = None
        self._stac_io = None
        self._set_temporal_properties(datetime, start_datetime, end_datetime)
        for field_name, field_value in (
            ("time", time),
            ("conformsTo", conforms_to),
            ("linkTemplates", link_templates),
        ):
            if field_value is not None:
                self.extra_fields[field_name] = deepcopy(field_value)
        self._attach_stac_objects(collection, assets, href)

    def _attach_stac_objects(
        self,
        collection: str | pystac.Collection | None,
        assets: dict[str, pystac.Asset] | None,
        href: str | None,
    ) -> None:
        """Set collection links, asset ownership, and the record's location."""
        if isinstance(collection, pystac.Collection):
            self.set_collection(collection)
        elif collection is not None:
            self.collection_id = collection
        for key, asset in (assets or {}).items():
            self.add_asset(key, asset)
        if href is not None:
            self.set_self_href(href)

    def _set_temporal_properties(
        self,
        instant: Datetime | None,
        start: Datetime | None,
        end: Datetime | None,
    ) -> None:
        """Apply explicit STAC dates, retaining dates already in properties."""
        self.datetime = instant
        if instant is not None:
            self.properties["datetime"] = datetime_to_str(instant)
        elif self.properties.get("datetime") is not None:
            self.datetime = str_to_datetime(self.properties["datetime"])
        for field_name, value in (("start_datetime", start), ("end_datetime", end)):
            if value is not None:
                self.properties[field_name] = datetime_to_str(value)

    @property
    def record_id(self) -> str | int:
        return self._record_id if str(self._record_id) == self.id else self.id

    @property
    def record_metadata(self) -> RecordCommonProperties:
        return RecordCommonProperties(self)

    @property
    def time(self) -> dict[str, Any] | None:
        return self.extra_fields.get("time")

    @time.setter
    def time(self, value: dict[str, Any] | None) -> None:
        self.extra_fields["time"] = value

    @property
    def conforms_to(self) -> list[str]:
        """The live list of conformance URIs.

        Raises:
            TypeError: If the stored value is not a list of strings.
        """
        conformance = self.extra_fields.setdefault("conformsTo", [])
        if not isinstance(conformance, list) or not all(
            isinstance(uri, str) for uri in conformance
        ):
            raise TypeError("conformsTo must be a list of strings")
        return conformance

    @conforms_to.setter
    def conforms_to(self, value: list[str]) -> None:
        self.extra_fields["conformsTo"] = value

    @property
    def link_templates(self) -> list[dict[str, Any]]:
        """The live list of link template dictionaries.

        Raises:
            TypeError: If the stored value is not a list of dictionaries.
        """
        templates = self.extra_fields.setdefault("linkTemplates", [])
        if not isinstance(templates, list) or not all(
            isinstance(template, dict) for template in templates
        ):
            raise TypeError("linkTemplates must be a list of dictionaries")
        return templates

    @link_templates.setter
    def link_templates(self, value: list[dict[str, Any]]) -> None:
        self.extra_fields["linkTemplates"] = value

    def to_dict(
        self, include_self_link: bool = True, transform_hrefs: bool = True
    ) -> dict[str, Any]:
        props = deepcopy(self.properties)
        if self.datetime is not None:
            props["datetime"] = datetime_to_str(self.datetime)
        doc = deepcopy(self.extra_fields)
        doc.update(
            type="Feature",
            id=self.record_id,
            geometry=deepcopy(self.geometry),
            properties=None if self._null_properties and not props else props,
            links=[
                link.to_dict(transform_href=transform_hrefs)
                for link in self.links
                if include_self_link or link.rel != pystac.RelType.SELF
            ],
        )
        if self.bbox is not None:
            doc["bbox"] = list(self.bbox)
        # Preserve extension declarations/assets as GeoJSON foreign members.
        # Never substitute conformsTo for stac_extensions.
        if self.stac_extensions:
            doc["stac_extensions"] = list(self.stac_extensions)
        if self.assets:
            doc["assets"] = {key: deepcopy(asset.to_dict()) for key, asset in self.assets.items()}
        if self.collection_id is not None:
            doc["collection"] = self.collection_id
        return doc

    def to_record_dict(
        self, include_self_link: bool = True, transform_hrefs: bool = True
    ) -> dict[str, Any]:
        return self.to_dict(include_self_link, transform_hrefs)

    @classmethod
    def matches_object_type(cls, d: dict[str, Any]) -> bool:
        # Structural recognition only; GeoJSON/STAC overlap prevents unique dispatch.
        return (
            d.get("type") == "Feature"
            and all(key in d for key in ("id", "geometry", "properties"))
            and isinstance(d["id"], (str, int))
            and not isinstance(d["id"], bool)
            and (d["properties"] is None or isinstance(d["properties"], dict))
        )

    @classmethod
    def from_dict(
        cls,
        d: dict[str, Any],
        href: str | None = None,
        root: pystac.Catalog | None = None,
        migrate: bool = True,
        preserve_dict: bool = True,
    ) -> OGCRecord:
        """Read Records; migrate is accepted for compatibility but never applied.

        Always copies document metadata, even when preserve_dict=False.
        """
        if not cls.matches_object_type(d):
            raise ValueError("Expected a Record GeoJSON Feature with id, geometry, properties")
        obj = cls(
            id=d["id"],
            geometry=d["geometry"],
            properties=d["properties"],
            bbox=d.get("bbox"),
            collection=d.get("collection"),
            stac_extensions=d.get("stac_extensions"),
            extra_fields={k: v for k, v in d.items() if k not in cls._RESERVED},
            assets={k: pystac.Asset.from_dict(deepcopy(v)) for k, v in d.get("assets", {}).items()},
        )
        for link in d.get("links", []):
            if href is None or link.get("rel") != "self":
                obj.add_link(pystac.Link.from_dict(deepcopy(link)))
        if href is not None:
            obj.set_self_href(href)
        if root is not None:
            obj.set_root(root)
        return obj

    def clone(self) -> OGCRecord:
        doc = self.to_dict(transform_hrefs=False)
        doc["links"] = []
        result = type(self).from_dict(doc)
        for link in self.links:
            result.add_link(link.clone())
        result._stac_io = self._stac_io
        return result

    def to_stac_item(self) -> pystac.Item:
        """Explicit export; requires a genuine STAC temporal extent.

        Constructing an Item does not replace full STAC schema validation.
        """
        if self.datetime is None and not all(
            self.properties.get(k) is not None for k in ("start_datetime", "end_datetime")
        ):
            raise ValueError("STAC export requires datetime or start_datetime/end_datetime")
        extra = deepcopy(self.extra_fields)
        extra.pop("stac_version", None)
        item = pystac.Item(
            id=self.id,
            geometry=deepcopy(self.geometry),
            bbox=deepcopy(self.bbox),
            datetime=self.datetime,
            properties=deepcopy(self.properties),
            stac_extensions=list(self.stac_extensions),
            collection=self.collection_id,
            extra_fields=extra,
            assets={k: v.clone() for k, v in self.assets.items()},
        )
        for link in self.links:
            item.add_link(link.clone())
        return item

    def validate(self, validator: Any = None) -> list[Any]:
        """Validate the OGC document with an explicitly supplied validator.

        Supply an object exposing validate(document), configured for the OGC
        OpenAPI 3.0 schema and its references. No implicit STAC validation.
        """
        if validator is None:
            raise ValueError(
                "Supply an OGC schema validator; for STAC use record.to_stac_item().validate()"
            )
        validator.validate(self.to_dict(transform_hrefs=False))
        return [self.SCHEMA_URI]

    def __repr__(self) -> str:
        return f"<OGCRecord id={self.record_id!r}>"

class RecordCommonProperties(RecordMetadataMixin):
    """Backward-compatible metadata view sharing the record properties."""

    def __init__(self, record: OGCRecord) -> None:
        self.record = record

    @property
    def properties(self) -> dict[str, Any]:
        return self.record.properties

    def to_dict(self) -> dict[str, Any]:
        return deepcopy(self.properties)

class RecordMetadataMixin:
    """Direct recordCommonProperties accessors; dates use their JSON string form.

    Nested structures remain open dictionaries, allowing future extensions.
    These accessors do not perform JSON Schema validation.
    """

    created = MetadataField[str]("created")
    updated = MetadataField[str]("updated")
    type = MetadataField[str]("type")
    title = MetadataField[str]("title")
    description = MetadataField[str]("description")
    keywords = MetadataField[list[str]]("keywords")
    themes = MetadataField[list[Theme]]("themes")
    language = MetadataField[Language]("language")
    languages = MetadataField[list[Language]]("languages")
    resource_languages = MetadataField[list[Language]]("resourceLanguages")
    external_ids = MetadataField[list[ExternalId]]("externalIds")
    formats = MetadataField[list[Format]]("formats")
    contacts = MetadataField[list[Contact]]("contacts")
    license = MetadataField[str]("license")
    rights = MetadataField[str]("rights")