Source code for oshconnect.schema_datamodels

#  =============================================================================
#  Copyright (c) 2025 Botts Innovative Research Inc.
#  Date: 2025/9/30
#  Author: Ian Patterson
#  Contact Email: ian@botts-inc.com
#  =============================================================================
from __future__ import annotations

import base64
from datetime import datetime
from typing import Annotated, Union, List, Literal

from pydantic import (
    BaseModel, ConfigDict, Field, HttpUrl, field_serializer, field_validator,
    model_validator,
)

from .api_utils import Link, URI
from .encoding import BinaryEncoding, FlatBuffersEncoding, JSONEncoding, ProtobufEncoding
from .geometry import Geometry
from .swe_components import AnyComponent, check_named
from .timemanagement import TimeInstant


def _now_iso8601_z() -> str:
    """Per-call default for ``CommandJSON.issue_time``: a UTC timestamp with
    trailing ``Z`` (CS API Part 2 / SWE Common 3 expect a valid ISO8601
    with zone info — OSH 400s on the bare ``datetime.now().isoformat()``
    form because it has no zone designator)."""
    return TimeInstant.now_as_time_instant().get_iso_time()


def _dump_csapi(model: BaseModel) -> dict:
    """Internal: canonical CS API serialization (alias keys, exclude None, JSON-mode)."""
    return model.model_dump(by_alias=True, exclude_none=True, mode='json')


"""
In many of the top level resource models there is a "schema" field of some description. These models are meant to ease
the burden on the end user to create those.
"""


[docs] class CommandJSON(BaseModel): """ A class to represent a command in JSON format """ model_config = ConfigDict(populate_by_name=True) control_id: str = Field(None, serialization_alias="control@id") issue_time: Union[str, float] = Field(default_factory=_now_iso8601_z, serialization_alias="issueTime") sender: str = Field(None) # CS API Part 2 — and OSH — call this field ``parameters`` on the wire. # ``populate_by_name=True`` keeps the Python attribute readable as ``params``. params: Union[dict, list, int, float, str] = Field(None, alias="parameters")
[docs] def to_csapi_dict(self) -> dict: """Render as the CS API `application/json` command body.""" return _dump_csapi(self)
[docs] @classmethod def from_csapi_dict(cls, data: dict) -> "CommandJSON": """Build from a CS API command JSON dict.""" return cls.model_validate(data)
[docs] class CommandSchema(BaseModel): """ Base class representation for control streams' command schemas """ model_config = ConfigDict(populate_by_name=True) command_format: str = Field(..., alias='commandFormat')
[docs] class SWEJSONCommandSchema(CommandSchema): """ SWE+JSON command schema """ model_config = ConfigDict(populate_by_name=True) # Literal pin powers the discriminated `AnyCommandSchema` union below # and removes the need for a runtime field_validator. command_format: Literal["application/swe+json"] = Field( "application/swe+json", alias='commandFormat') # Concrete subclass instead of `SerializeAsAny[Encoding]` — `JSONEncoding` # is the only Encoding type used in practice, and a concrete type # serializes deterministically without `SerializeAsAny`. If/when more # encoding types arrive, migrate this to a discriminated Union on # `Encoding.type`. encoding: JSONEncoding = Field(...) record_schema: AnyComponent = Field(..., alias='recordSchema') @model_validator(mode="after") def _root_record_schema_requires_name(self): check_named(self.record_schema, "SWEJSONCommandSchema.recordSchema") return self
[docs] def to_swejson_dict(self) -> dict: """Render as an `application/swe+json` command-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_swejson_dict(cls, data: dict) -> "SWEJSONCommandSchema": """Build from an `application/swe+json` command-schema dict.""" return cls.model_validate(data, by_alias=True)
[docs] class JSONCommandSchema(CommandSchema): """ JSON command schema """ model_config = ConfigDict(populate_by_name=True) command_format: Literal["application/json"] = Field("application/json", alias='commandFormat') params_schema: AnyComponent = Field(..., alias='parametersSchema') result_schema: AnyComponent = Field(None, alias='resultSchema') feasibility_schema: AnyComponent = Field(None, alias='feasibilityResultSchema') @model_validator(mode="after") def _root_schemas_require_name(self): check_named(self.params_schema, "JSONCommandSchema.parametersSchema") if self.result_schema is not None: check_named(self.result_schema, "JSONCommandSchema.resultSchema") if self.feasibility_schema is not None: check_named(self.feasibility_schema, "JSONCommandSchema.feasibilityResultSchema") return self
[docs] def to_json_dict(self) -> dict: """Render as an `application/json` command-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_json_dict(cls, data: dict) -> "JSONCommandSchema": """Build from an `application/json` command-schema dict.""" return cls.model_validate(data, by_alias=True)
[docs] class DatastreamRecordSchema(BaseModel): """ A class to represent the schema of a datastream """ model_config = ConfigDict(populate_by_name=True) obs_format: str = Field(..., alias='obsFormat')
# `encoding` is required per CS API Part 2 §16.2.3 Requirement 109.B, but the # OSH server omits it from /datastreams/{id}/schema responses. We accept it as # optional to be able to parse what the server returns. See # docs/osh_spec_deviations.md (swe-json-missing-encoding).
[docs] class SWEDatastreamRecordSchema(DatastreamRecordSchema): model_config = ConfigDict(populate_by_name=True) # Multi-Literal acts as the discriminator value(s) for AnyDatastreamRecordSchema # below. Replaces the previous runtime field_validator. # # Note: `application/swe+binary` is NOT included here — it has a distinct # `encoding` shape (`BinaryEncoding`, not `JSONEncoding`) and gets its own # class (`SWEBinaryDatastreamRecordSchema`) so the discriminated union can # dispatch on `obsFormat` without runtime branching on the encoding type. obs_format: Literal[ "application/swe+json", "application/swe+csv", "application/swe+text", ] = Field(..., alias='obsFormat') encoding: JSONEncoding = Field(None) record_schema: AnyComponent = Field(..., alias='recordSchema') @model_validator(mode="after") def _root_record_schema_requires_name(self): check_named(self.record_schema, "SWEDatastreamRecordSchema.recordSchema") return self
[docs] def to_swejson_dict(self) -> dict: """Render as an `application/swe+json` datastream-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_swejson_dict(cls, data: dict) -> "SWEDatastreamRecordSchema": """Build from an `application/swe+json` datastream-schema dict (e.g., a CS API ``/datastreams/{id}/schema`` response in SWE form).""" return cls.model_validate(data, by_alias=True)
[docs] class SWEBinaryDatastreamRecordSchema(DatastreamRecordSchema): """Datastream observation schema for `application/swe+binary`. Split from `SWEDatastreamRecordSchema` because the encoding block is a `BinaryEncoding` (with a `members` list mapping component refs to `dataType` / `compression`), not a `JSONEncoding`. The `recordSchema` side mirrors the SWE+JSON form — it describes the *semantic* shape of the record. The `recordEncoding` side describes the *wire* shape, overriding the semantic shape where needed (e.g. a `DataArray` in the recordSchema may be replaced by a single `Block` member with ``compression="H264"`` on the wire, as Axis cameras do for video). Use ``oshconnect.swe_binary.SWEBinaryCodec(schema)`` to encode dicts to bytes and decode bytes back to dicts. """ model_config = ConfigDict(populate_by_name=True) obs_format: Literal["application/swe+binary"] = Field( "application/swe+binary", alias='obsFormat') record_schema: AnyComponent = Field(..., alias='recordSchema') # OSH emits ``recordEncoding`` for the binary variant; the JSON-family # variant calls the same slot ``encoding``. Accept either via alias. record_encoding: BinaryEncoding = Field(..., alias='recordEncoding') @model_validator(mode="after") def _root_record_schema_requires_name(self): check_named(self.record_schema, "SWEBinaryDatastreamRecordSchema.recordSchema") return self
[docs] def to_swebinary_dict(self) -> dict: """Render as an `application/swe+binary` datastream-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_swebinary_dict(cls, data: dict) -> "SWEBinaryDatastreamRecordSchema": """Build from an `application/swe+binary` datastream-schema dict (a CS API ``/datastreams/{id}/schema?obsFormat=application/swe+binary`` response body).""" return cls.model_validate(data, by_alias=True)
[docs] class SWEProtobufDatastreamRecordSchema(DatastreamRecordSchema): """Datastream observation schema for ``application/swe+proto``. ``application/swe+proto`` is the **per-datastream descriptor** Protobuf encoding: each DataStream ships a pre-compiled Protobuf schema (a serialized ``google.protobuf.FileDescriptorSet``) describing one per-datastream observation message — envelope fields 1–5 (id, datastream_id, foi_id, phenomenon/result time) plus the SWE Common record at fields 6+. An observation on the wire is a serialized instance of that message; receivers register the descriptor in a ``DescriptorPool`` and decode dynamically (no ``protoc``). This carries the descriptor itself rather than a SWE ``recordSchema`` tree — the descriptor is the source of truth for the wire layout, and SWE semantics (uom, definition, …) travel as field options inside it. The codec lives in ``oshconnect.swe_protobuf.SWEProtobufCodec``. .. note:: The JSON envelope that delivers the descriptor over the ``/datastreams/{id}/schema`` endpoint is a "meet in the middle" contract with the OSH node side — currently assumed to be ``{"obsFormat", "messageType", "fileDescriptorSet": <base64>}``. See ``docs/osh_spec_deviations.md`` (swe-proto-descriptor-format). """ model_config = ConfigDict(populate_by_name=True) obs_format: Literal["application/swe+proto"] = Field( "application/swe+proto", alias='obsFormat') # Serialized google.protobuf.FileDescriptorSet (carries the # per-datastream message file plus its transitive imports). Stored as # raw bytes in Python; (de)serialized as base64 in JSON so it survives # the CS API schema document and the discriminated-union round-trip. file_descriptor_set: bytes = Field(..., alias='fileDescriptorSet') # Fully-qualified per-datastream message name (e.g. # "georobotix.csapi....WeatherObservation"). Optional when the # descriptor set carries exactly one message type. message_type: str = Field(None, alias='messageType') # Marker only — the wire layout is fully defined by the descriptor. record_encoding: ProtobufEncoding = Field( default_factory=ProtobufEncoding, alias='recordEncoding') @field_validator('file_descriptor_set', mode='before') @classmethod def _decode_base64_fds(cls, value): """Accept the descriptor as base64 text (JSON wire form) or raw bytes.""" if isinstance(value, str): return base64.b64decode(value) return value @field_serializer('file_descriptor_set', when_used='json') def _encode_base64_fds(self, value: bytes) -> str: return base64.b64encode(value).decode('ascii')
[docs] def to_sweproto_dict(self) -> dict: """Render as an `application/swe+proto` datastream-schema document (``fileDescriptorSet`` base64-encoded).""" return _dump_csapi(self)
[docs] @classmethod def from_sweproto_dict(cls, data: dict) -> "SWEProtobufDatastreamRecordSchema": """Build from an `application/swe+proto` datastream-schema dict.""" return cls.model_validate(data, by_alias=True)
[docs] @classmethod def from_record_schema( cls, record: AnyComponent, *, message_name: str = "Observation", package: str = "oshconnect.sweproto", datatype_by_path: dict = None, ) -> "SWEProtobufDatastreamRecordSchema": """Generate a swe+proto schema from a SWE Common ``DataRecord``. Builds the per-datastream observation descriptor (envelope fields 1–5 + the record's components as result fields 6+, nested records and vectors recursed) — the inverse of OSH's ``ProtoSchemaWriter`` — and wraps it in this schema model. Use this to produce a swe+proto datastream schema from the ``record_schema`` you already hold for a SWE+JSON or SWE+Binary datastream. :param record: a ``DataRecordSchema`` describing the observation. :param message_name: generated message name (e.g. ``f"Observation_{ds_id}"``). :param package: proto package for the generated message. :param datatype_by_path: optional ``{ref: dataType_uri}`` map (same refs a ``BinaryEncoding`` uses) so float32 / int-width leaves map to the matching proto wire type instead of the defaults. """ from .swe_protobuf import build_observation_descriptor_set fds, message_type = build_observation_descriptor_set( record, message_name=message_name, package=package, datatype_by_path=datatype_by_path) return cls(file_descriptor_set=fds, message_type=message_type)
[docs] @classmethod def from_other_schema( cls, other, *, message_name: str = "Observation", package: str = "oshconnect.sweproto", ) -> "SWEProtobufDatastreamRecordSchema": """Translate another datastream schema into swe+proto. Accepts any record-bearing datastream schema (``SWEDatastreamRecordSchema``, ``SWEBinaryDatastreamRecordSchema``, …) — pulling its ``record_schema`` — or a bare ``DataRecordSchema``. When the source is SWE+Binary, the ``recordEncoding`` members are mined for each leaf's OGC ``dataType`` so the generated proto fields use the matching wire type (e.g. float32 → ``float``); SWE+JSON / SWE+CSV / SWE+text sources carry no dataType, so the node defaults apply. """ record = getattr(other, "record_schema", other) datatype_by_path = None record_encoding = getattr(other, "record_encoding", None) members = getattr(record_encoding, "members", None) if members: datatype_by_path = { m.ref: m.data_type for m in members if getattr(m, "type", None) == "Component" and getattr(m, "ref", None) and getattr(m, "data_type", None) } return cls.from_record_schema( record, message_name=message_name, package=package, datatype_by_path=datatype_by_path)
[docs] def to_proto_source(self) -> str: """Render this schema's descriptor as editable ``.proto`` source text. Works for any swe+proto schema — one generated here *or* one delivered by the node — since it renders the carried ``FileDescriptorSet``. Edit the text and recompile it with :meth:`from_proto_source` to apply changes. """ from .swe_protobuf import render_proto_source return render_proto_source(self.file_descriptor_set, self.message_type)
[docs] @classmethod def from_proto_source( cls, proto_text: str, *, message_type: str = None, protoc: str = "protoc", ) -> "SWEProtobufDatastreamRecordSchema": """Build a schema from ``.proto`` source text (compiles via ``protoc``). The round-trip companion to :meth:`to_proto_source`: generate the text, hand-edit it (add/rename fields, tweak types, add annotations), then compile it back into a schema. Requires ``protoc`` on PATH (or pass ``protoc=<path>``); the binary-descriptor paths (`from_record_schema` / `from_other_schema`) need no protoc. ``message_type`` defaults to the first message in the compiled file. """ from .swe_protobuf import compile_proto_source, primary_message_type fds = compile_proto_source(proto_text, protoc=protoc) if message_type is None: message_type = primary_message_type(fds) return cls(file_descriptor_set=fds, message_type=message_type)
[docs] class SWEFlatBuffersDatastreamRecordSchema(DatastreamRecordSchema): """Datastream observation schema for ``application/swe+flatbuffers``. Mirrors `SWEProtobufDatastreamRecordSchema`. The wire format is a FlatBuffers-serialized SWE Common 3 message; the codec lives in ``oshconnect.swe_flatbuffers.SWEFlatBuffersCodec``. .. warning:: The FlatBuffers codec is not currently functional — `flatc --python` does not yet support vectors-of-unions, which the SWE Common 3 schema uses for `BinaryEncoding.members`. The schema class is provided so the SDK can already parse and round-trip schemas that name this format; calling ``SWEFlatBuffersCodec.encode``/``decode`` raises `NotImplementedError`. See ``docs/osh_spec_deviations.md`` (flatc-python-vector-of-union). """ model_config = ConfigDict(populate_by_name=True) obs_format: Literal["application/swe+flatbuffers"] = Field( "application/swe+flatbuffers", alias='obsFormat') record_schema: AnyComponent = Field(..., alias='recordSchema') record_encoding: FlatBuffersEncoding = Field( default_factory=FlatBuffersEncoding, alias='recordEncoding') @model_validator(mode="after") def _root_record_schema_requires_name(self): check_named(self.record_schema, "SWEFlatBuffersDatastreamRecordSchema.recordSchema") return self
[docs] def to_sweflatbuffers_dict(self) -> dict: """Render as an `application/swe+flatbuffers` datastream-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_sweflatbuffers_dict(cls, data: dict) -> "SWEFlatBuffersDatastreamRecordSchema": """Build from an `application/swe+flatbuffers` datastream-schema dict.""" return cls.model_validate(data, by_alias=True)
[docs] class OMJSONDatastreamRecordSchema(DatastreamRecordSchema): """Datastream observation schema for the OM+JSON media type (`application/om+json`, also accepts `application/json` as a synonym on parse since OSH treats them equivalently for datastream schemas). Per CS API Part 2 §16.1.4, this form does not carry a SWE `encoding` block; structure is fully described by `resultSchema` (inline result) or `resultLink` (out-of-band). `parametersSchema` is optional. """ model_config = ConfigDict(populate_by_name=True) # Multi-Literal — both wire forms are spec-equivalent for OM+JSON. obs_format: Literal[ "application/om+json", "application/json", ] = Field("application/om+json", alias='obsFormat') result_schema: AnyComponent = Field(None, alias='resultSchema') parameters_schema: AnyComponent = Field(None, alias='parametersSchema') result_link: dict = Field(None, alias='resultLink') @model_validator(mode="after") def _root_schemas_require_name(self): if self.result_schema is not None: check_named(self.result_schema, "OMJSONDatastreamRecordSchema.resultSchema") if self.parameters_schema is not None: check_named(self.parameters_schema, "OMJSONDatastreamRecordSchema.parametersSchema") return self
[docs] def to_omjson_dict(self) -> dict: """Render as an `application/om+json` datastream-schema document.""" return _dump_csapi(self)
[docs] @classmethod def from_omjson_dict(cls, data: dict) -> "OMJSONDatastreamRecordSchema": """Build from an `application/om+json` (or `application/json`) datastream-schema dict (e.g., a CS API ``/datastreams/{id}/schema`` response in OM+JSON form).""" return cls.model_validate(data, by_alias=True)
[docs] class LogicalProperty(BaseModel): """One entry in `LogicalDatastreamRecordSchema.properties`. The logical schema is OSH's JSON-Schema-flavored representation of a SWE Common DataRecord. Each property is a JSON Schema field with OGC extension keywords (`x-ogc-definition`, `x-ogc-refFrame`, `x-ogc-unit`, `x-ogc-axis`) that carry the SWE Common metadata. Permissive: ``extra='allow'`` accepts JSON Schema fields we haven't modeled (e.g. ``description``, ``default``, ``minimum``, ``maximum``, nested ``items`` for arrays). """ model_config = ConfigDict(populate_by_name=True, extra='allow') title: str = Field(None) type: str = Field(...) # "string" | "number" | "integer" | "boolean" | "object" | "array" format: str = Field(None) # e.g. "date-time" enum: list = Field(None) items: dict = Field(None) # for type="array" properties: dict = Field(None) # for type="object" (nested) # OGC SWE Common extensions (hyphenated keys → aliased) ogc_definition: str = Field(None, alias='x-ogc-definition') ogc_ref_frame: str = Field(None, alias='x-ogc-refFrame') ogc_unit: str = Field(None, alias='x-ogc-unit') ogc_axis: str = Field(None, alias='x-ogc-axis')
[docs] class LogicalDatastreamRecordSchema(BaseModel): """Logical schema document — OSH's `obsFormat=logical` representation. Returned by ``GET /datastreams/{id}/schema?obsFormat=logical``. Distinct from `SWEDatastreamRecordSchema` and `OMJSONDatastreamRecordSchema`: - No ``obsFormat`` envelope field - No ``recordSchema`` wrapper — the schema is the document - JSON Schema flavor (``type: "object"`` + ``properties``) instead of a SWE Common AnyComponent tree - Each property carries SWE Common metadata via ``x-ogc-*`` extension keywords OSH-specific (not in the OGC CS API spec) but useful for tooling that speaks JSON Schema natively. Permissive (``extra='allow'``) so future JSON Schema fields don't break parsing. """ model_config = ConfigDict(populate_by_name=True, extra='allow') type: str = Field(...) # always "object" for OSH datastream schemas title: str = Field(None) properties: dict[str, LogicalProperty] = Field(...) required: list[str] = Field(None)
[docs] def to_logical_dict(self) -> dict: """Render as an OSH `obsFormat=logical` JSON Schema dict.""" return _dump_csapi(self)
[docs] @classmethod def from_logical_dict(cls, data: dict) -> "LogicalDatastreamRecordSchema": """Build from a logical schema dict (e.g., a CS API ``/datastreams/{id}/schema?obsFormat=logical`` response body).""" return cls.model_validate(data, by_alias=True)
[docs] class ObservationOMJSONInline(BaseModel): """ A class to represent an observation in OM-JSON format """ model_config = ConfigDict(populate_by_name=True) datastream_id: str = Field(None, alias="datastream@id") foi_id: str = Field(None, alias="foi@id") phenomenon_time: str = Field(None, alias="phenomenonTime") result_time: str = Field(datetime.now().isoformat(), alias="resultTime") parameters: dict = Field(None) result: Union[int, float, str, dict, list] = Field(...) result_links: List[Link] = Field(None, alias="result@links")
[docs] def to_csapi_dict(self) -> dict: """Render as an `application/om+json` observation body.""" return _dump_csapi(self)
[docs] @classmethod def from_csapi_dict(cls, data: dict) -> "ObservationOMJSONInline": """Build from an `application/om+json` observation dict.""" return cls.model_validate(data)
[docs] class SystemEventOMJSON(BaseModel): """ A class to represent the schema of a system event """ model_config = ConfigDict(populate_by_name=True) label: str = Field(...) description: str = Field(None) definition: HttpUrl = Field(...) identifiers: list = Field(None) classifiers: list = Field(None) contacts: list = Field(None) documentation: list = Field(None) time: str = Field(...) properties: list = Field(None) configuration: dict = Field(None) links: list[Link] = Field(None)
[docs] class SystemHistoryGeoJSON(BaseModel): """ A class to represent the schema of a system history """ model_config = ConfigDict(populate_by_name=True) type: str = Field(...) id: str = Field(None) properties: SystemHistoryProperties = Field(...) geometry: Geometry = Field(None) bbox: list = Field(None) links: list[Link] = Field(None)
[docs] class SystemHistoryProperties(BaseModel): model_config = ConfigDict(populate_by_name=True) feature_type: str = Field(...) uid: URI = Field(...) name: str = Field(...) description: str = Field(None) asset_type: str = Field(None) valid_time: list = Field(None) parent_system_link: str = Field(None, serialization_alias='parentSystem@link') procedure_link: str = Field(None, serialization_alias='procedure@link')
# Discriminated unions replace the earlier `SerializeAsAny[<base>]` pattern # on resource models. Pydantic dispatches by the literal value of the # discriminator field — `obsFormat` / `commandFormat` — so validate and # dump round-trip without polymorphism quirks. AnyDatastreamRecordSchema = Annotated[ Union[ SWEDatastreamRecordSchema, SWEBinaryDatastreamRecordSchema, SWEProtobufDatastreamRecordSchema, SWEFlatBuffersDatastreamRecordSchema, OMJSONDatastreamRecordSchema, ], Field(discriminator='obs_format'), ] """Public alias for `DatastreamResource.record_schema`. Discriminator: `obs_format`.""" AnyCommandSchema = Annotated[ Union[SWEJSONCommandSchema, JSONCommandSchema], Field(discriminator='command_format'), ] """Public alias for `ControlStreamResource.command_schema`. Discriminator: `command_format`.""" # Defense-in-depth: rebuild every container model that forward-references # `AnyComponent`. See the matching block in swe_components.py for the # `MockValSer` rationale — same fault recurs here because each schema # class threads `AnyComponent` through its body. SWEJSONCommandSchema.model_rebuild(force=True) JSONCommandSchema.model_rebuild(force=True) SWEDatastreamRecordSchema.model_rebuild(force=True) SWEBinaryDatastreamRecordSchema.model_rebuild(force=True) # SWEProtobufDatastreamRecordSchema no longer threads `AnyComponent` (it # carries the protobuf descriptor instead), so it needs no forced rebuild. SWEFlatBuffersDatastreamRecordSchema.model_rebuild(force=True) OMJSONDatastreamRecordSchema.model_rebuild(force=True)