Source code for oshconnect.resources.system

#  =============================================================================
#  Copyright (c) 2026 Georobotix Innovative Research
#  Date: 2026/5/18
#  Author: Ian Patterson
#  Contact Email: ian.patterson@georobotix.us
#  =============================================================================

"""`System` — a sensor system on an OSH server.

Concrete `StreamableResource` subclass. Logical grouping of one or more
`Datastream` outputs and `ControlStream` inputs sharing a single URN.
Exposes discovery and creation flows for both child resource types.
"""
from __future__ import annotations

import datetime
import logging
import uuid
import warnings
from typing import TYPE_CHECKING

from ..csapi4py.constants import APIResourceTypes, ContentTypes
from ..encoding import JSONEncoding
from ..resource_datamodels import ControlStreamResource, DatastreamResource, SystemResource
from ..schema_datamodels import (
    JSONCommandSchema, SWEBinaryDatastreamRecordSchema,
    SWEDatastreamRecordSchema, SWEFlatBuffersDatastreamRecordSchema,
    SWEJSONCommandSchema, SWEProtobufDatastreamRecordSchema,
)
from ..swe_components import DataRecordSchema
from ..timemanagement import TimeInstant, TimePeriod, TimeUtils
from .base import SchemaFetchWarning, StreamableResource
from .controlstream import ControlStream
from .datastream import Datastream

if TYPE_CHECKING:
    from ..node import Node


[docs] class System(StreamableResource[SystemResource]): """A sensor system on an OSH server: a logical grouping of one or more `Datastream` outputs and `ControlStream` inputs sharing a single URN. Construct directly to define a new system, or build one from a parsed `SystemResource` via `from_system_resource`. Use `discover_datastreams` / `discover_controlstreams` to populate child resources from the server, or `add_insert_datastream` / `add_and_insert_control_stream` to create new ones server-side. """ label: str datastreams: list[Datastream] control_channels: list[ControlStream] description: str urn: str _parent_node: Node
[docs] def __init__(self, label: str = None, urn: str = None, parent_node: Node = None, **kwargs): """ :param label: The display string for the system. Maps to SML's ``label`` and GeoJSON's ``properties.name`` on the wire — the OGC CS API only carries one display string per system. :param urn: The URN of the system, typically formed as such: ``'urn:general_identifier:specific_identifier:…'``. :param parent_node: The `Node` this system attaches to. :param kwargs: - 'description': A description of the system - 'resource_id': The server-assigned ID once known - 'name': Deprecated alias for ``label``. Emits ``DeprecationWarning``; if ``label`` is also supplied, ``name`` is ignored. Will be removed in a future release. """ super().__init__(node=parent_node) # Back-compat: `name` was a separate constructor parameter that # always carried the same value as `label` because the wire only # has one display string. Route deprecated callers to `label`. if 'name' in kwargs: import warnings warnings.warn( "`System(name=...)` is deprecated; use `label=` instead. " "The wire-format only carries one display string per " "system and `name` was always populated from the same " "source as `label`.", DeprecationWarning, stacklevel=2, ) legacy_name = kwargs.pop('name') if label is None: label = legacy_name self.label = label self.datastreams = [] self.control_channels = [] self.urn = urn if kwargs.get('resource_id'): self._resource_id = kwargs['resource_id'] if kwargs.get('description'): self.description = kwargs['description'] self._underlying_resource = self.to_system_resource()
@property def name(self) -> str: """Deprecated alias for `label`. Will be removed in a future release. SWE Common 3 / OGC CS API only carry one display string per system (SML's ``label``, GeoJSON's ``properties.name``). The wrapper's prior `name` field was always set to the same value as `label`. Use `self.label` directly going forward. """ import warnings warnings.warn( "`System.name` is deprecated; use `.label` instead.", DeprecationWarning, stacklevel=2, ) return self.label @name.setter def name(self, value: str) -> None: import warnings warnings.warn( "Setting `System.name` is deprecated; set `.label` instead.", DeprecationWarning, stacklevel=2, ) self.label = value @staticmethod def _pick_datastream_schema_format(formats: list[str]): """Choose an ``obsFormat`` for the schema fetch, plus the parser that knows how to validate the response. Preference order: SWE+JSON (textual, easiest to inspect) → SWE+binary (the only choice for video/blob datastreams that don't advertise SWE+JSON, e.g. Axis cameras' ``video1``). Returns ``(None, None)`` if neither is advertised, so the caller can skip the fetch with a warning instead of crashing. """ if formats is None: return None, None if "application/swe+json" in formats: return ("application/swe+json", SWEDatastreamRecordSchema.from_swejson_dict) if "application/swe+proto" in formats: return ("application/swe+proto", SWEProtobufDatastreamRecordSchema.from_sweproto_dict) if "application/swe+flatbuffers" in formats: return ("application/swe+flatbuffers", SWEFlatBuffersDatastreamRecordSchema.from_sweflatbuffers_dict) if "application/swe+binary" in formats: return ("application/swe+binary", SWEBinaryDatastreamRecordSchema.from_swebinary_dict) return None, None
[docs] def discover_datastreams(self) -> list[Datastream]: """GET ``/systems/{id}/datastreams`` and instantiate `Datastream` objects for every entry. New datastreams are appended to ``self.datastreams`` and also returned. For each discovered datastream we additionally fetch its record schema (``GET /datastreams/{id}/schema?obsFormat=…``) and cache it on ``_underlying_resource.record_schema``. The schema variant is chosen from the datastream's advertised ``formats`` list: ``application/swe+json`` is preferred when available (parsed as `SWEDatastreamRecordSchema`); otherwise ``application/swe+binary`` is used (parsed as `SWEBinaryDatastreamRecordSchema`). Datastreams like Axis camera ``video1`` outputs advertise *only* the binary variant — without this fallback every video datastream would land without a schema. The CS API listing endpoint omits the inner schema, so without this step every discovered datastream would be missing the schema callers need for observation construction or cross-node sync. A failure on a single datastream's schema fetch is downgraded to a warning so it doesn't poison the whole call. """ api = self._parent_node.get_api_helper() res = api.get_resource(APIResourceTypes.SYSTEM, self._resource_id, APIResourceTypes.DATASTREAM) datastream_json = res.json()['items'] datastreams = [] for ds in datastream_json: datastream_objs = DatastreamResource.model_validate(ds, by_alias=True) new_ds = Datastream(self._parent_node, datastream_objs) obs_format, parser = self._pick_datastream_schema_format( datastream_objs.formats) if obs_format is None: msg = ( f"Datastream {datastream_objs.ds_id} advertises no " f"supported schema format (have: {datastream_objs.formats}); " "skipping schema fetch." ) logging.warning(msg) warnings.warn(msg, SchemaFetchWarning, stacklevel=2) else: try: schema_resp = api.get_resource( APIResourceTypes.DATASTREAM, datastream_objs.ds_id, APIResourceTypes.SCHEMA, params={'obsFormat': obs_format}, ) schema_resp.raise_for_status() new_ds._underlying_resource.record_schema = parser(schema_resp.json()) except Exception as e: msg = ( f"Failed to fetch {obs_format} schema for datastream " f"{datastream_objs.ds_id}: {type(e).__name__}: {e}" ) logging.error(msg, exc_info=True) warnings.warn(msg, SchemaFetchWarning, stacklevel=2) datastreams.append(new_ds) # Dedup by server-side id so a re-discovery refreshes nothing # but also duplicates nothing. (A previous expression here # evaluated truthiness of a list comprehension, which only # ever admitted the first datastream.) if all(ds.get_id() != datastream_objs.ds_id for ds in self.datastreams): self.datastreams.append(new_ds) return datastreams
[docs] def discover_controlstreams(self) -> list[ControlStream]: """GET ``/systems/{id}/controlstreams`` and instantiate `ControlStream` objects for every entry. New control streams are appended to ``self.control_channels`` and also returned. For each discovered control stream we additionally fetch the command schema (``GET /controlstreams/{id}/schema?f=json``, which OSH returns as ``application/json`` with a ``parametersSchema`` SWE Common component) and cache it on ``_underlying_resource.command_schema`` as a `JSONCommandSchema`. ``f=json`` is the OGC API standard format-selector and pins the response shape to the JSON variant — without it the server default could change. The CS API listing endpoint omits the inner schema, so without this step every discovered control stream would be missing the schema callers need for command construction or cross-node sync. A failure on a single control stream's schema fetch is downgraded to a warning so it doesn't poison the whole call. """ api = self._parent_node.get_api_helper() res = api.get_resource(APIResourceTypes.SYSTEM, self._resource_id, APIResourceTypes.CONTROL_CHANNEL) controlstream_json = res.json()['items'] controlstreams = [] for cs_json in controlstream_json: controlstream_objs = ControlStreamResource.model_validate(cs_json) new_cs = ControlStream(self._parent_node, controlstream_objs) try: schema_resp = api.get_resource( APIResourceTypes.CONTROL_CHANNEL, controlstream_objs.cs_id, APIResourceTypes.SCHEMA, params={'f': 'json'}, ) schema_resp.raise_for_status() new_cs._underlying_resource.command_schema = ( JSONCommandSchema.from_json_dict(schema_resp.json()) ) except Exception as e: msg = ( f"Failed to fetch command schema for control stream " f"{controlstream_objs.cs_id}: {type(e).__name__}: {e}" ) logging.error(msg, exc_info=True) warnings.warn(msg, SchemaFetchWarning, stacklevel=2) controlstreams.append(new_cs) # Same id-based dedup as discover_datastreams (the previous # list-truthiness expression only ever admitted the first one). if all(cs.get_id() != controlstream_objs.cs_id for cs in self.control_channels): self.control_channels.append(new_cs) return controlstreams
@classmethod def _construct_from_resource(cls, system_resource: SystemResource, parent_node: Node) -> "System": """Build a `System` from a parsed `SystemResource`. Internal helper shared by `from_csapi_dict` / `from_smljson_dict` / `from_geojson_dict` and the deprecated `from_system_resource`. """ # exclude_none avoids triggering TimePeriod.ser_model on None-valued # optional time fields (it does `str(self.start)` unconditionally). other_props = system_resource.model_dump(exclude_none=True) # GeoJSON form carries `properties.name`/`properties.uid`; SML form # has `label`/`uid` directly on the resource. Both wire shapes # carry exactly one display string, mapped to `System.label`. if other_props.get('properties'): props = other_props['properties'] new_system = cls(label=props.get('name'), urn=props.get('uid'), resource_id=system_resource.system_id, parent_node=parent_node) else: new_system = cls(label=system_resource.label, urn=system_resource.uid, resource_id=system_resource.system_id, parent_node=parent_node) new_system.set_system_resource(system_resource) return new_system
[docs] @classmethod def from_resource(cls, system_resource: SystemResource, parent_node: Node) -> "System": """Build a `System` from an already-parsed `SystemResource`. Mirror of `Datastream.__init__(parent_node=, datastream_resource=)` and `ControlStream.__init__(node=, controlstream_resource=)` — provides the same "I have a parsed pydantic resource model in memory and want a wrapper attached to a node" entry point for Systems, whose constructor takes individual fields rather than a full resource model. Handles both wire shapes that round-trip through `SystemResource`: the GeoJSON form (with a ``properties`` block carrying ``name``/``uid``) and the SML form (``label``/``uid`` directly on the resource). Source of the resource doesn't matter — built locally, validated from `from_smljson_dict` / `from_geojson_dict` / `from_csapi_dict`, returned by some other library, etc. :param system_resource: A populated `SystemResource` instance. :param parent_node: The `Node` the new `System` will attach to. :return: A `System` wrapper bound to ``parent_node`` with ``_underlying_resource`` set to ``system_resource``. """ return cls._construct_from_resource(system_resource, parent_node)
[docs] @staticmethod def from_system_resource(system_resource: SystemResource, parent_node: Node) -> System: """Build a `System` from an already-parsed `SystemResource`. .. deprecated:: 0.5.1 Use :meth:`System.from_resource` instead — same behavior, more consistent name with other wrappers' resource-taking factories. Handles both shapes the OSH server emits: the GeoJSON form (with a ``properties`` block carrying ``name``/``uid``) and the SML form (``label``/``uid`` directly on the resource). """ warnings.warn("System.from_system_resource is deprecated; use System.from_resource instead " "(then dump it to a dict if you need wire JSON).", DeprecationWarning, stacklevel=2, ) return System._construct_from_resource(system_resource, parent_node)
[docs] def to_system_resource(self) -> SystemResource: """Render this `System` as a `SystemResource` pydantic model suitable for POSTing to the server. When this wrapper already carries an ``_underlying_resource`` (e.g. populated by ``from_csapi_dict``, ``set_system_resource``, or a prior ``retrieve_resource`` call), all of its fields are preserved into a deep copy — so cross-node sync, partial updates, and re-POSTs round-trip everything the source carried, not just ``uniqueId`` / ``label`` / a hardcoded ``PhysicalSystem`` type. Currently-attached datastreams are always reflected into ``outputs`` so newly-added children come along. When no underlying resource is present (i.e. during this wrapper's own ``__init__``), a thin shell is built from wrapper attrs and the SML type defaults to ``PhysicalSystem``. """ underlying = getattr(self, '_underlying_resource', None) if underlying is not None: resource = underlying.model_copy(deep=True) # Pick up any wrapper-side updates the user made directly # on the System (the wrapper doesn't proxy these into the # resource on assignment). if self.urn and not resource.uid: resource.uid = self.urn if self.label and not resource.label: resource.label = self.label else: resource = SystemResource(uid=self.urn, label=self.label, feature_type='PhysicalSystem') if self.datastreams: resource.outputs = [ds.get_underlying_resource() for ds in self.datastreams] return resource
[docs] def set_system_resource(self, sys_resource: SystemResource): """Replace the underlying `SystemResource` model.""" self._underlying_resource = sys_resource
[docs] def get_system_resource(self) -> SystemResource: """Return the underlying `SystemResource` model.""" return self._underlying_resource
[docs] def add_insert_datastream(self, datastream_schema: DatastreamResource): """Adds a datastream to the system while also inserting it into the system's parent node via HTTP POST. :param datastream_schema: DataRecordSchema to be used to define the datastream. Must carry a ``name`` matching NameToken (``^[A-Za-z][A-Za-z0-9_\\-]*$``); SWE Common 3 wraps DataStream.elementType in SoftNamedProperty, so the root component requires a name. :return: """ api = self._parent_node.get_api_helper() res = api.create_resource(APIResourceTypes.DATASTREAM, datastream_schema.model_dump_json(by_alias=True, exclude_none=True), req_headers={'Content-Type': ContentTypes.JSON.value}, parent_res_id=self._resource_id) if res.ok: datastream_id = res.headers['Location'].split('/')[-1] datastream_schema.ds_id = datastream_id else: raise Exception( f'Failed to create datastream {datastream_schema.name!r}: ' f'HTTP {res.status_code}{res.text}' ) new_ds = Datastream(self._parent_node, datastream_schema) new_ds.set_parent_resource_id(self._underlying_resource.system_id) self.datastreams.append(new_ds) return new_ds
[docs] def add_insert_controlstream(self, controlstream_resource: ControlStreamResource) -> ControlStream: """Adds a control stream to the system while also inserting it into the system's parent node via HTTP POST. Mirrors `add_insert_datastream`: caller assembles the full `ControlStreamResource` (including the embedded `command_schema`) and this method posts it to ``/systems/{id}/controlstreams``, captures the new resource ID from the ``Location`` header, and returns a wrapped `ControlStream`. For the embedded `command_schema`, prefer `JSONCommandSchema` (`commandFormat: application/json` with a ``parametersSchema``). It matches what OSH returns from ``GET /controlstreams/{id}/schema?f=json`` (the form ``discover_controlstreams`` parses), keeps round-trip sync symmetric, and avoids the SWE+JSON ``encoding``-omission deviation documented in ``docs/osh_spec_deviations.md`` §1. `SWEJSONCommandSchema` (``application/swe+json`` with ``recordSchema`` plus ``encoding``) is also accepted for spec-strict scenarios. :param controlstream_resource: A fully-built `ControlStreamResource` carrying ``name``, ``input_name``, and ``command_schema``. :return: ControlStream object added to the system. """ api = self._parent_node.get_api_helper() res = api.create_resource( APIResourceTypes.CONTROL_CHANNEL, controlstream_resource.model_dump_json(by_alias=True, exclude_none=True), req_headers={'Content-Type': ContentTypes.JSON.value}, parent_res_id=self._resource_id, ) if res.ok: cs_id = res.headers['Location'].split('/')[-1] controlstream_resource.cs_id = cs_id else: raise Exception( f'Failed to create control stream {controlstream_resource.name!r}: ' f'HTTP {res.status_code}{res.text}' ) new_cs = ControlStream(node=self._parent_node, controlstream_resource=controlstream_resource) new_cs.set_parent_resource_id(self._underlying_resource.system_id) self.control_channels.append(new_cs) return new_cs
[docs] def add_and_insert_control_stream(self, control_stream_record_schema: DataRecordSchema, input_name: str = None, valid_time: TimePeriod = None, command_format: str = "application/json") -> ControlStream: """Accepts a DataRecordSchema and creates a ControlStreamResource with the matching command-schema variant, then POSTs it to the parent node. Per CS API Part 2 §16.x, command schemas come in two wire forms: - ``application/json`` → `JSONCommandSchema` carrying `parametersSchema` (the SWE Common component); no `encoding`. **This is the default.** It matches what OSH returns from ``GET /controlstreams/{id}/schema?f=json`` (the form ``discover_controlstreams`` parses), keeps round-trip sync symmetric, and avoids the SWE+JSON ``encoding``-omission deviation documented in ``docs/osh_spec_deviations.md`` §1. - ``application/swe+json`` → `SWEJSONCommandSchema` carrying `recordSchema` (the SWE Common component) and `encoding` (`JSONEncoding`). Spec-canonical; pass ``command_format='application/swe+json'`` to opt in. :param control_stream_record_schema: DataRecordSchema to wrap. Must carry a ``name`` matching NameToken (``^[A-Za-z][A-Za-z0-9_\\-]*$``); the schema is the root named component required by both command-schema variants. :param input_name: Name of the input. If None, the schema label is lowercased and whitespace-stripped. :param valid_time: Optional `TimePeriod`; defaults to ``[now, now + 1 year]``. :param command_format: ``"application/json"`` (default) or ``"application/swe+json"``. Anything else raises ``ValueError``. :return: ControlStream object added to the system. """ input_name_checked = input_name if input_name is not None else control_stream_record_schema.label.lower().replace( ' ', '') now = datetime.datetime.now() future_time = now.replace(year=now.year + 1) future_str = future_time.strftime("%Y-%m-%dT%H:%M:%SZ") valid_time_checked = valid_time if valid_time else TimePeriod(start=TimeInstant.now_as_time_instant(), end=TimeInstant( utc_time=TimeUtils.to_utc_time(future_str))) if command_format == "application/swe+json": command_schema = SWEJSONCommandSchema( command_format="application/swe+json", record_schema=control_stream_record_schema, encoding=JSONEncoding(), ) elif command_format == "application/json": command_schema = JSONCommandSchema( command_format="application/json", params_schema=control_stream_record_schema, ) else: raise ValueError( f"Unsupported command_format: {command_format!r}. " f"Expected 'application/swe+json' or 'application/json'." ) control_stream_resource = ControlStreamResource(name=control_stream_record_schema.label, input_name=input_name_checked, command_schema=command_schema, validTime=valid_time_checked) api = self._parent_node.get_api_helper() res = api.create_resource(APIResourceTypes.CONTROL_CHANNEL, control_stream_resource.model_dump_json(by_alias=True, exclude_none=True), req_headers={'Content-Type': 'application/json'}, parent_res_id=self._resource_id) if res.ok: control_channel_id = res.headers['Location'].split('/')[-1] control_stream_resource.cs_id = control_channel_id else: raise Exception( f'Failed to create control stream {control_stream_resource.name!r}: ' f'HTTP {res.status_code}{res.text}' ) new_cs = ControlStream(node=self._parent_node, controlstream_resource=control_stream_resource) new_cs.set_parent_resource_id(self._underlying_resource.system_id) self.control_channels.append(new_cs) return new_cs
[docs] def insert_self(self): """POST this system to the server (Content-Type ``application/sml+json``) and capture the new resource ID from the ``Location`` response header. Server-assigned fields (``id``, ``links``) are stripped from the body before POST so a re-POSTed (e.g. cross-node-synced) system doesn't leak the source server's identifier or links to the destination — the destination assigns its own. """ body_resource = self.to_system_resource().model_copy(deep=True) body_resource.system_id = None body_resource.links = None res = self._parent_node.get_api_helper().create_resource( APIResourceTypes.SYSTEM, body_resource.model_dump_json(by_alias=True, exclude_none=True), req_headers={'Content-Type': 'application/sml+json'}) if res.ok: location = res.headers['Location'] sys_id = location.split('/')[-1] self._resource_id = sys_id if self._underlying_resource is not None: self._underlying_resource.system_id = sys_id
[docs] def retrieve_resource(self): """GET ``/systems/{id}`` and refresh the underlying `SystemResource`. Returns ``None`` either way (kept for API symmetry). """ if self._resource_id is None: return None res = self._parent_node.get_api_helper().retrieve_resource(res_type=APIResourceTypes.SYSTEM, res_id=self._resource_id) if res.ok: system_json = res.json() system_resource = SystemResource.model_validate(system_json) self._underlying_resource = system_resource return None
[docs] def to_storage_dict(self) -> dict: """Return a JSON-safe snapshot of this system, its child datastreams / control streams, and the dumped underlying `SystemResource`, for OSHConnect's persistence layer. Not a CS API server-shaped payload — the ``underlying_resource`` block is the only piece that matches the CS API system shape. """ data = super().to_storage_dict() data["label"] = getattr(self, "label", None) data["urn"] = getattr(self, "urn", None) data["description"] = getattr(self, "description", None) datastreams = getattr(self, "datastreams", None) if datastreams is not None: data["datastreams"] = [ds.to_storage_dict() for ds in datastreams] else: data["datastreams"] = None control_channels = getattr(self, "control_channels", None) if control_channels is not None: data["control_channels"] = [cc.to_storage_dict() for cc in control_channels] else: data["control_channels"] = None underlying = getattr(self, "_underlying_resource", None) if underlying is not None: dump = getattr(underlying, 'model_dump', None) if callable(dump): data["underlying_resource"] = underlying.model_dump(by_alias=True, exclude_none=True, mode='json') elif hasattr(underlying, 'to_dict'): data["underlying_resource"] = underlying.to_dict() else: data["underlying_resource"] = str(underlying) else: data["underlying_resource"] = None # Remove any 'resource' key if present data.pop("resource", None) return data
[docs] @classmethod def from_storage_dict(cls, data: dict, node: 'Node') -> 'System': """Build a `System` from a dict produced by `to_storage_dict`. Expects ``label``, ``urn``, optional ``description`` / ``resource_id``, and optional ``datastreams`` / ``control_channels`` / ``underlying_resource`` blocks. The embedded ``underlying_resource`` is parsed via `SystemResource.model_validate`, so that nested block can also be a CS API server response body. For backwards compatibility, ``data["name"]`` is accepted as a legacy alias for ``label`` if ``label`` is missing — older snapshots written before the `name`/`label` consolidation still load. :param data: Source dict. :param node: Parent `Node` the rebuilt system attaches to. """ label = data.get("label") or data.get("name") obj = cls( label=label, urn=data["urn"], parent_node=node, description=data.get("description"), resource_id=data.get("resource_id")) obj._id = uuid.UUID(data["id"]) obj.datastreams = [Datastream.from_storage_dict(ds, node) for ds in data.get("datastreams", [])] obj.control_channels = [ControlStream.from_storage_dict(cc, node) for cc in data.get("control_channels", [])] underlying = data.get("underlying_resource") obj._underlying_resource = SystemResource.model_validate(underlying) if underlying else None return obj