Skip to content

Commit 8758df3

Browse files
Added format-explicit to_/from_smljson_dict, to_/from_omjson_dict, to_/from_swejson_dict, to_/from_geojson_dict, and to_/from_csapi_dict methods across System/Datastream/ControlStream and their underlying pydantic resource/schema models for round-tripping CS API server JSON, deprecated the older
System.from_system_resource and Datastream.from_resource factories, and fixed three latent bugs (Node._client_session initialization, TimeUtils.time_to_iso UTC handling, ObservationOMJSONInline alias direction) exposed by the new tests.
1 parent dd6b22f commit 8758df3

7 files changed

Lines changed: 535 additions & 30 deletions

File tree

README.md

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,41 @@ uv run interrogate -vv src/oshconnect # per-symbol (shows which symbols
6363
Once we agree on a baseline, raise `[tool.interrogate].fail-under` from `0` so
6464
new code without docstrings starts failing locally and in CI.
6565

66+
## OGC Format Serialization
67+
68+
Format-explicit conversion methods on the wrapper classes (`System`,
69+
`Datastream`, `ControlStream`) and the underlying pydantic resource models.
70+
Use these to round-trip CS API server JSON in **SML+JSON**, **OM+JSON**, and
71+
**SWE+JSON** without having to remember the `model_dump(by_alias=True, …)`
72+
incantation, and to construct OSHConnect wrappers from raw server payloads.
73+
74+
```python
75+
from oshconnect import Node, System, Datastream
76+
77+
node = Node(protocol="http", address="localhost", port=8282)
78+
79+
# Build a System from an SML+JSON server response
80+
sys_dict = {"type": "PhysicalSystem", "uniqueId": "urn:test:1", "label": "Sensor"}
81+
sys = System.from_csapi_dict(sys_dict, node) # auto-detects SML vs GeoJSON
82+
sys.to_smljson_dict() # -> dict ready to POST
83+
84+
# Build a Datastream from a CS API listing entry
85+
ds = Datastream.from_csapi_dict(ds_json, node)
86+
ds.to_csapi_dict() # the resource body
87+
ds.schema_to_swejson_dict() # the SWE+JSON schema doc
88+
ds.observation_to_omjson_dict({"temperature": 22.5}) # one OM+JSON observation
89+
90+
# Single observations / commands
91+
from oshconnect.resource_datamodels import ObservationResource
92+
obs = ObservationResource.from_omjson_dict(om_json_payload)
93+
obs.to_swejson_dict() # flat SWE+JSON record
94+
```
95+
96+
The two older static factories `System.from_system_resource` and
97+
`Datastream.from_resource` are deprecated in favor of `from_csapi_dict` and
98+
emit `DeprecationWarning` on use. They'll be removed in a future major
99+
version.
100+
66101
## Generating the Docs
67102

68103
The documentation is built with [MkDocs](https://www.mkdocs.org/) using the

pyproject.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[project]
22
name = "oshconnect"
3-
version = "0.5.0a1"
3+
version = "0.5.1a0"
44
description = "Library for interfacing with OSH, helping guide visualization efforts, and providing a place to store configurations. Implements OGC CS API Part 3 (Pub/Sub) MQTT topic conventions including :data topics and resource event topics."
55
readme = "README.md"
66
authors = [

src/oshconnect/resource_datamodels.py

Lines changed: 174 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@
66
# ==============================================================================
77
from __future__ import annotations
88

9-
from typing import List
9+
import json
10+
from typing import List, TYPE_CHECKING
1011

1112
from pydantic import BaseModel, ConfigDict, Field, SerializeAsAny, model_validator
1213
from shapely import Point
@@ -16,6 +17,9 @@
1617
from .schema_datamodels import DatastreamRecordSchema, CommandSchema
1718
from .timemanagement import TimeInstant, TimePeriod
1819

20+
if TYPE_CHECKING:
21+
from .swe_components import AnyComponent
22+
1923

2024
class BoundingBox(BaseModel):
2125
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
@@ -132,6 +136,59 @@ class SystemResource(BaseModel):
132136
modes: List[Mode] = Field(None)
133137
method: ProcessMethod = Field(None)
134138

139+
def to_smljson_dict(self) -> dict:
140+
"""Render this system as an `application/sml+json` dict (SensorML JSON encoding).
141+
142+
Sets ``feature_type = "PhysicalSystem"`` to match the SML discriminator
143+
before dumping. Output keys are camelCase per the CS API wire format.
144+
"""
145+
self.feature_type = "PhysicalSystem"
146+
return self.model_dump(by_alias=True, exclude_none=True, mode='json')
147+
148+
def to_smljson(self) -> str:
149+
"""JSON-string variant of `to_smljson_dict`."""
150+
return json.dumps(self.to_smljson_dict())
151+
152+
def to_geojson_dict(self) -> dict:
153+
"""Render this system as an `application/geo+json` dict.
154+
155+
Sets ``feature_type = "Feature"`` to match the GeoJSON discriminator
156+
before dumping. Useful when posting to endpoints that expect the
157+
GeoJSON Feature shape.
158+
"""
159+
self.feature_type = "Feature"
160+
return self.model_dump(by_alias=True, exclude_none=True, mode='json')
161+
162+
def to_geojson(self) -> str:
163+
"""JSON-string variant of `to_geojson_dict`."""
164+
return json.dumps(self.to_geojson_dict())
165+
166+
@classmethod
167+
def from_smljson_dict(cls, data: dict) -> "SystemResource":
168+
"""Build a `SystemResource` from an `application/sml+json` dict
169+
(e.g., a CS API server response body for a system in SML form)."""
170+
return cls.model_validate(data, by_alias=True)
171+
172+
@classmethod
173+
def from_geojson_dict(cls, data: dict) -> "SystemResource":
174+
"""Build a `SystemResource` from an `application/geo+json` dict
175+
(e.g., a CS API server response body for a system in GeoJSON form)."""
176+
return cls.model_validate(data, by_alias=True)
177+
178+
@classmethod
179+
def from_csapi_dict(cls, data: dict) -> "SystemResource":
180+
"""Build a `SystemResource` from a CS API system dict, auto-dispatching
181+
on the ``type`` field: ``"PhysicalSystem"`` → SML+JSON path,
182+
``"Feature"`` → GeoJSON path. Anything else falls through to a
183+
permissive validate.
184+
"""
185+
feature_type = data.get("type")
186+
if feature_type == "PhysicalSystem":
187+
return cls.from_smljson_dict(data)
188+
if feature_type == "Feature":
189+
return cls.from_geojson_dict(data)
190+
return cls.model_validate(data, by_alias=True)
191+
135192

136193
class DatastreamResource(BaseModel):
137194
"""
@@ -175,6 +232,25 @@ def handle_aliases(cls, values):
175232
break
176233
return values
177234

235+
def to_csapi_dict(self) -> dict:
236+
"""Render this datastream as the CS API `application/json` resource
237+
body. The embedded ``schema`` field is dumped polymorphically per
238+
whichever variant (`SWEDatastreamRecordSchema` /
239+
`JSONDatastreamRecordSchema`) it holds.
240+
"""
241+
return self.model_dump(by_alias=True, exclude_none=True, mode='json')
242+
243+
def to_csapi_json(self) -> str:
244+
"""JSON-string variant of `to_csapi_dict`."""
245+
return json.dumps(self.to_csapi_dict())
246+
247+
@classmethod
248+
def from_csapi_dict(cls, data: dict) -> "DatastreamResource":
249+
"""Build a `DatastreamResource` from a CS API datastream dict
250+
(e.g., a server response body or an entry from a /datastreams
251+
listing)."""
252+
return cls.model_validate(data, by_alias=True)
253+
178254

179255
class ObservationResource(BaseModel):
180256
model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
@@ -187,6 +263,84 @@ class ObservationResource(BaseModel):
187263
result: dict = Field(...)
188264
result_link: Link = Field(None, alias="result@link")
189265

266+
def to_omjson_dict(self, datastream_id: str | None = None) -> dict:
267+
"""Render this observation as an `application/om+json` dict
268+
(the ``ObservationOMJSONInline`` shape).
269+
270+
:param datastream_id: Optional ID to include as ``datastream@id``
271+
on the output. The CS API typically supplies this from URL
272+
context, so it's not required on the model itself.
273+
"""
274+
from .schema_datamodels import ObservationOMJSONInline
275+
kwargs = {"result": self.result}
276+
if datastream_id is not None:
277+
kwargs["datastream_id"] = datastream_id
278+
if self.phenomenon_time:
279+
kwargs["phenomenon_time"] = self.phenomenon_time.get_iso_time()
280+
if self.result_time:
281+
kwargs["result_time"] = self.result_time.get_iso_time()
282+
if self.parameters is not None:
283+
kwargs["parameters"] = self.parameters
284+
wrapper = ObservationOMJSONInline(**kwargs)
285+
return wrapper.model_dump(by_alias=True, exclude_none=True, mode='json')
286+
287+
def to_swejson_dict(self, schema: "AnyComponent" = None) -> dict:
288+
"""Render this observation as an `application/swe+json` payload
289+
(the SWE Common JSON encoding of one record).
290+
291+
SWE+JSON encodes a single observation as a flat JSON object whose
292+
keys are the schema field names; ``self.result`` is already that
293+
dict, so this is essentially a passthrough. The optional
294+
``schema`` argument is accepted for forward compatibility (when
295+
we add field-order / encoding-aware emission).
296+
"""
297+
# ``schema`` reserved for future encoding rules (vector-as-arrays,
298+
# JSONEncoding handling, etc.); current behavior is passthrough.
299+
del schema
300+
return dict(self.result) if self.result is not None else {}
301+
302+
@classmethod
303+
def from_omjson_dict(cls, data: dict) -> "ObservationResource":
304+
"""Build an `ObservationResource` from an `application/om+json` dict.
305+
306+
Parses through `ObservationOMJSONInline` to validate the OM+JSON
307+
envelope, then strips the ``datastream@id`` / ``foi@id`` envelope
308+
fields (those live on the surrounding context, not the resource)
309+
and returns the inner observation.
310+
"""
311+
from .schema_datamodels import ObservationOMJSONInline
312+
wrapper = ObservationOMJSONInline.model_validate(data)
313+
kwargs = {
314+
"result_time": TimeInstant.from_string(wrapper.result_time),
315+
"result": wrapper.result,
316+
}
317+
if wrapper.phenomenon_time:
318+
kwargs["phenomenon_time"] = TimeInstant.from_string(wrapper.phenomenon_time)
319+
if wrapper.parameters is not None:
320+
kwargs["parameters"] = wrapper.parameters
321+
return cls(**kwargs)
322+
323+
@classmethod
324+
def from_swejson_dict(cls, data: dict, schema: "AnyComponent" = None,
325+
result_time: str | None = None) -> "ObservationResource":
326+
"""Build an `ObservationResource` from an `application/swe+json`
327+
observation payload.
328+
329+
SWE+JSON observations don't carry an envelope (no ``resultTime`` /
330+
``phenomenonTime`` fields); pass ``result_time`` explicitly when
331+
you have it, otherwise the current UTC time is used.
332+
333+
:param data: The flat SWE+JSON record dict.
334+
:param schema: Optional schema, reserved for future per-field
335+
type coercion. Currently ignored.
336+
:param result_time: ISO 8601 timestamp for ``resultTime``;
337+
defaults to ``TimeInstant.now_as_time_instant().isoformat()``
338+
if omitted.
339+
"""
340+
del schema # future use
341+
rt = TimeInstant.from_string(result_time) if result_time is not None else TimeInstant.now_as_time_instant()
342+
return cls(result_time=rt, result=dict(data))
343+
190344

191345
class ControlStreamResource(BaseModel):
192346
model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
@@ -206,3 +360,22 @@ class ControlStreamResource(BaseModel):
206360
asynchronous: bool = Field(True, alias="async")
207361
command_schema: SerializeAsAny[CommandSchema] = Field(None, alias="schema")
208362
links: List[Link] = Field(None)
363+
364+
def to_csapi_dict(self) -> dict:
365+
"""Render this control stream as the CS API `application/json`
366+
resource body. The embedded ``schema`` field is dumped
367+
polymorphically per whichever variant
368+
(`SWEJSONCommandSchema` / `JSONCommandSchema`) it holds.
369+
"""
370+
return self.model_dump(by_alias=True, exclude_none=True, mode='json')
371+
372+
def to_csapi_json(self) -> str:
373+
"""JSON-string variant of `to_csapi_dict`."""
374+
return json.dumps(self.to_csapi_dict())
375+
376+
@classmethod
377+
def from_csapi_dict(cls, data: dict) -> "ControlStreamResource":
378+
"""Build a `ControlStreamResource` from a CS API control-stream dict
379+
(e.g., a server response body or an entry from a /controlstreams
380+
listing)."""
381+
return cls.model_validate(data, by_alias=True)

src/oshconnect/schema_datamodels.py

Lines changed: 68 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,12 @@
1717
from .geometry import Geometry
1818
from .swe_components import AnyComponent, check_named
1919

20+
21+
def _dump_csapi(model: BaseModel) -> dict:
22+
"""Internal: canonical CS API serialization (alias keys, exclude None, JSON-mode)."""
23+
return model.model_dump(by_alias=True, exclude_none=True, mode='json')
24+
25+
2026
"""
2127
In many of the top level resource models there is a "schema" field of some description. These models are meant to ease
2228
the burden on the end user to create those.
@@ -33,6 +39,15 @@ class CommandJSON(BaseModel):
3339
sender: str = Field(None)
3440
params: Union[dict, list, int, float, str] = Field(None)
3541

42+
def to_csapi_dict(self) -> dict:
43+
"""Render as the CS API `application/json` command body."""
44+
return _dump_csapi(self)
45+
46+
@classmethod
47+
def from_csapi_dict(cls, data: dict) -> "CommandJSON":
48+
"""Build from a CS API command JSON dict."""
49+
return cls.model_validate(data)
50+
3651

3752
class CommandSchema(BaseModel):
3853
"""
@@ -58,6 +73,15 @@ def _root_record_schema_requires_name(self):
5873
check_named(self.record_schema, "SWEJSONCommandSchema.recordSchema")
5974
return self
6075

76+
def to_swejson_dict(self) -> dict:
77+
"""Render as an `application/swe+json` command-schema document."""
78+
return _dump_csapi(self)
79+
80+
@classmethod
81+
def from_swejson_dict(cls, data: dict) -> "SWEJSONCommandSchema":
82+
"""Build from an `application/swe+json` command-schema dict."""
83+
return cls.model_validate(data, by_alias=True)
84+
6185

6286
class JSONCommandSchema(CommandSchema):
6387
"""
@@ -79,6 +103,15 @@ def _root_schemas_require_name(self):
79103
check_named(self.feasibility_schema, "JSONCommandSchema.feasibilityResultSchema")
80104
return self
81105

106+
def to_json_dict(self) -> dict:
107+
"""Render as an `application/json` command-schema document."""
108+
return _dump_csapi(self)
109+
110+
@classmethod
111+
def from_json_dict(cls, data: dict) -> "JSONCommandSchema":
112+
"""Build from an `application/json` command-schema dict."""
113+
return cls.model_validate(data, by_alias=True)
114+
82115

83116
class DatastreamRecordSchema(BaseModel):
84117
"""
@@ -111,6 +144,16 @@ def _root_record_schema_requires_name(self):
111144
check_named(self.record_schema, "SWEDatastreamRecordSchema.recordSchema")
112145
return self
113146

147+
def to_swejson_dict(self) -> dict:
148+
"""Render as an `application/swe+json` datastream-schema document."""
149+
return _dump_csapi(self)
150+
151+
@classmethod
152+
def from_swejson_dict(cls, data: dict) -> "SWEDatastreamRecordSchema":
153+
"""Build from an `application/swe+json` datastream-schema dict
154+
(e.g., a CS API ``/datastreams/{id}/schema`` response in SWE form)."""
155+
return cls.model_validate(data, by_alias=True)
156+
114157

115158
class JSONDatastreamRecordSchema(DatastreamRecordSchema):
116159
"""Datastream observation schema for the JSON media types
@@ -144,19 +187,39 @@ def _root_schemas_require_name(self):
144187
check_named(self.parameters_schema, "JSONDatastreamRecordSchema.parametersSchema")
145188
return self
146189

190+
def to_omjson_dict(self) -> dict:
191+
"""Render as an `application/om+json` datastream-schema document."""
192+
return _dump_csapi(self)
193+
194+
@classmethod
195+
def from_omjson_dict(cls, data: dict) -> "JSONDatastreamRecordSchema":
196+
"""Build from an `application/om+json` (or `application/json`)
197+
datastream-schema dict (e.g., a CS API ``/datastreams/{id}/schema``
198+
response in OM+JSON form)."""
199+
return cls.model_validate(data, by_alias=True)
200+
147201

148202
class ObservationOMJSONInline(BaseModel):
149203
"""
150204
A class to represent an observation in OM-JSON format
151205
"""
152206
model_config = ConfigDict(populate_by_name=True)
153-
datastream_id: str = Field(None, serialization_alias="datastream@id")
154-
foi_id: str = Field(None, serialization_alias="foi@id")
155-
phenomenon_time: str = Field(None, serialization_alias="phenomenonTime")
156-
result_time: str = Field(datetime.now().isoformat(), serialization_alias="resultTime")
207+
datastream_id: str = Field(None, alias="datastream@id")
208+
foi_id: str = Field(None, alias="foi@id")
209+
phenomenon_time: str = Field(None, alias="phenomenonTime")
210+
result_time: str = Field(datetime.now().isoformat(), alias="resultTime")
157211
parameters: dict = Field(None)
158212
result: Union[int, float, str, dict, list] = Field(...)
159-
result_links: List[Link] = Field(None, serialization_alias="result@links")
213+
result_links: List[Link] = Field(None, alias="result@links")
214+
215+
def to_csapi_dict(self) -> dict:
216+
"""Render as an `application/om+json` observation body."""
217+
return _dump_csapi(self)
218+
219+
@classmethod
220+
def from_csapi_dict(cls, data: dict) -> "ObservationOMJSONInline":
221+
"""Build from an `application/om+json` observation dict."""
222+
return cls.model_validate(data)
160223

161224

162225
class SystemEventOMJSON(BaseModel):

0 commit comments

Comments
 (0)