Constructing wrappers¶

A System, Datastream, or ControlStream wrapper is a thin shell around a pydantic resource model (SystemResource, DatastreamResource, ControlStreamResource) plus a Node for HTTP / MQTT / streaming context. Wrappers handle node-attached operations (insertion, MQTT pub/sub, schema fetches over HTTP, storage-layer round-trip); format conversion lives entirely on the resource models.

That separation drives the construction story: build the resource via the resource model’s parsers, then bind it to a parent node via the wrapper’s constructor / factory.

At-a-glance matrix¶

Input	System	Datastream	ControlStream
Individual fields (new local resource)	`System(name=, label=, urn=, parent_node=, …)`	via `System.add_insert_datastream(DataRecordSchema)` — also POSTs server-side	via `System.add_and_insert_control_stream(DataRecordSchema, input_name=…)` — also POSTs server-side
Parsed `*Resource` model	`System.from_resource(sys_res, node)` (`from_system_resource` is deprecated)	`Datastream(parent_node=node, datastream_resource=ds_res)` (`from_resource` is deprecated)	`ControlStream(node=node, controlstream_resource=cs_res)`
Storage dict (round-trip from `to_storage_dict`)	`System.from_storage_dict(data, node)`	`Datastream.from_storage_dict(data, node)`	`ControlStream.from_storage_dict(data, node)`

For raw CS API JSON, parse it through the resource model first:

Raw input	Resource-model parser
SML+JSON dict	`SystemResource.from_smljson_dict(data)`
GeoJSON dict	`SystemResource.from_geojson_dict(data)`
Any system shape (auto-detect)	`SystemResource.from_csapi_dict(data)`
CS API datastream dict	`DatastreamResource.from_csapi_dict(data)`
CS API control-stream dict	`ControlStreamResource.from_csapi_dict(data)`
Single OM+JSON observation	`ObservationResource.from_omjson_dict(data)`
Single SWE+JSON observation	`ObservationResource.from_swejson_dict(data, schema=…, result_time=…)`
SWE+JSON schema document	`SWEDatastreamRecordSchema.from_swejson_dict(data)`
OM+JSON schema document	`OMJSONDatastreamRecordSchema.from_omjson_dict(data)`
OSH logical schema (`obsFormat=logical`)	`LogicalDatastreamRecordSchema.from_logical_dict(data)`

When to use which¶

“I’m building a brand-new system from scratch”¶

Use the System constructor directly, then insert_self() (or let OSHConnect.create_and_insert_system(...) do both). The wrapper generates a SystemResource internally via to_system_resource().

from oshconnect import Node, System

node = Node(protocol='http', address='localhost', port=8282,
            username='admin', password='admin')
sys = System(
    name='WeatherStation',
    label='Weather Station #1',
    urn='urn:osh:sensor:weather:001',
    parent_node=node,
)
sys.insert_self()                  # POST /systems
print(sys.get_streamable_id())     # local UUID
print(sys._resource_id)            # server-assigned ID from Location header

“I just got a JSON response back from a CS API server”¶

Two steps: parse the JSON via the matching resource-model factory, then hand the resource to the wrapper.

import requests
from oshconnect import Node, System, Datastream
from oshconnect.resource_datamodels import SystemResource, DatastreamResource

node = Node(protocol='http', address='localhost', port=8282,
            username='admin', password='admin')

# System: SML+JSON or GeoJSON, auto-detected by the resource model
resp = requests.get('http://localhost:8282/sensorhub/api/systems/abc')
sys = System.from_resource(SystemResource.from_csapi_dict(resp.json()), node)

# Datastream: single shape (application/json)
resp = requests.get('http://localhost:8282/sensorhub/api/datastreams/def')
ds = Datastream(
    parent_node=node,
    datastream_resource=DatastreamResource.from_csapi_dict(resp.json()),
)

If you already know the format and want to skip the auto-detect, swap in from_smljson_dict(...) / from_geojson_dict(...) on SystemResource. The wrapper layer doesn’t care — it just receives a pydantic model.

“I have a `*Resource` already in memory”¶

from oshconnect import Datastream, ControlStream, System

# System — `from_resource` binds a parsed SystemResource to a node
sys = System.from_resource(sys_resource, node)

# Datastream — constructor takes the parsed resource directly
ds = Datastream(parent_node=node, datastream_resource=ds_resource)

# ControlStream — same pattern
cs = ControlStream(node=node, controlstream_resource=cs_resource)

System.from_resource handles both wire shapes that round-trip through SystemResource — the GeoJSON form (with name/uid under properties) and the SML form (label/uid directly on the resource). The deprecated System.from_system_resource emits a DeprecationWarning and is a shim for from_resource.

“I want to dump the wrapper back to JSON”¶

Reach down to the resource model. Format conversion isn’t on the wrapper:

sys.to_system_resource().to_smljson_dict()       # SML+JSON
sys.to_system_resource().to_geojson_dict()       # GeoJSON
ds._underlying_resource.to_csapi_dict()           # datastream resource body
cs._underlying_resource.to_csapi_dict()           # control-stream resource body

# Schema documents: through the schema model
ds._underlying_resource.record_schema.to_swejson_dict()
ds._underlying_resource.record_schema.to_omjson_dict()
cs._underlying_resource.command_schema.to_json_dict()

“I want the schema for an existing datastream from the server”¶

For datastreams that came back from System.discover_datastreams(), the SWE+JSON schema is already cached on _underlying_resource.record_schema. The CS API listing endpoint omits the inner schema, so discovery makes a second HTTP call per datastream (GET /datastreams/{id}/schema?obsFormat=application/swe+json) and assigns the result onto the underlying resource. Reading ds._underlying_resource.record_schema post-discovery returns the populated SWEDatastreamRecordSchema without another network call. A schema fetch that fails for a single datastream is downgraded to a warning so it doesn’t poison the rest of the discovery; that datastream’s record_schema stays None.

For datastreams built locally (no discovery), or when you need the OM+JSON or logical variant, hit the schema endpoint directly through the parent Node’s APIHelper and parse with the matching schema model:

from oshconnect.csapi4py.constants import APIResourceTypes
from oshconnect.schema_datamodels import (
    SWEDatastreamRecordSchema,
    OMJSONDatastreamRecordSchema,
    LogicalDatastreamRecordSchema,
)

api = node.get_api_helper()
ds_id = ds._underlying_resource.ds_id

# SWE+JSON (CS API spec)
sw_resp = api.get_resource(APIResourceTypes.DATASTREAM, ds_id,
                           APIResourceTypes.SCHEMA,
                           params={'obsFormat': 'application/swe+json'})
sw = SWEDatastreamRecordSchema.from_swejson_dict(sw_resp.json())

# OM+JSON (CS API spec)
om_resp = api.get_resource(APIResourceTypes.DATASTREAM, ds_id,
                           APIResourceTypes.SCHEMA,
                           params={'obsFormat': 'application/om+json'})
om = OMJSONDatastreamRecordSchema.from_omjson_dict(om_resp.json())

# OSH-specific JSON Schema flavor
lg_resp = api.get_resource(APIResourceTypes.DATASTREAM, ds_id,
                           APIResourceTypes.SCHEMA,
                           params={'obsFormat': 'logical'})
lg = LogicalDatastreamRecordSchema.from_logical_dict(lg_resp.json())

api.get_resource(...) returns a requests.Response; the from_*_dict classmethods on each schema model parse it into the typed pydantic class. None of these calls mutate the datastream’s _underlying_resource.record_schema — only discover_datastreams populates that, and only with the SWE+JSON variant. If you want to cache an OM+JSON or logical fetch, assign it yourself.

The logical schema is OSH-specific (not in the OGC CS API spec): a JSON Schema document with OGC extension keywords (x-ogc-definition, x-ogc-refFrame, x-ogc-unit, x-ogc-axis) carrying the SWE Common metadata.

“I’m restoring state from local storage”¶

from_storage_dict() rebuilds wrappers from the dicts produced by to_storage_dict(). Used by OSHConnect.load_config() and the SQLite datastore (oshconnect.datastores.sqlite_store); not what you want for parsing CS API server responses (those have a different shape — use the resource models for those).

import json
from oshconnect import Node, System

with open('my_app_config.json') as f:
    cfg = json.load(f)

node = Node.from_storage_dict(cfg['nodes'][0])
for sys_dict in cfg['systems']:
    sys = System.from_storage_dict(sys_dict, node)
    node.add_system(sys)

What about new datastreams/controlstreams without going through System?¶

The Datastream(...) and ControlStream(...) constructors require an already-built resource object — there’s no “build from individual fields” path because building one of these correctly requires defining the schema (SWEDatastreamRecordSchema or JSONCommandSchema) and threading it through a DatastreamResource / ControlStreamResource. The high-level entry points handle that for you:

System.add_insert_datastream(DataRecordSchema) — wraps a schema as SWEDatastreamRecordSchema (with JSONEncoding), builds the DatastreamResource, POSTs to the server, and returns the Datastream.
System.add_and_insert_control_stream(DataRecordSchema, input_name=…) — symmetric for ControlStreams via JSONCommandSchema.

If you really want to build from scratch without inserting, copy what those two methods do (see streamableresource.py for the recipe).

Why no `to__dict` / `from__dict` on the wrappers?¶

Because format conversion is the resource model’s job. Keeping it there gives one canonical entry point per format, so there’s no question of “is System.to_smljson_dict() the same as system_resource.to_smljson_dict()?” — there’s only the latter. The wrapper’s job is to bind a resource to a parent node and run the operations that need that node (HTTP, MQTT, storage). Two layers, two responsibilities.

The deprecated System.from_system_resource and Datastream.from_resource shims remain for one release as compatibility — both delegate to the new canonical paths.