implement fixes to controlstream discovery and insertion as well as add networked tests to verify that commands can be sent based off source schemas and maintain coherence across the wire.

Author: tipatterson-dev

docs/source/architecture/construction.md

@@ -173,30 +173,46 @@ 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, `Datastream` has three dedicated fetch
-methods — one per `obsFormat` the server supports. Each returns a
-typed schema model:
+OM+JSON or logical variant, hit the schema endpoint directly through
+the parent `Node`'s `APIHelper` and parse with the matching schema
+model:
 
 ```python
-ds = Datastream(parent_node=node, datastream_resource=DatastreamResource.from_csapi_dict(server_response))
+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
 
-# Wire-format schemas (CS API spec)
-sw = ds.fetch_swejson_schema()    # -> SWEDatastreamRecordSchema (application/swe+json)
-om = ds.fetch_omjson_schema()     # -> OMJSONDatastreamRecordSchema (application/om+json)
+# 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 = ds.fetch_logical_schema()    # -> LogicalDatastreamRecordSchema (obsFormat=logical)
+lg_resp = api.get_resource(APIResourceTypes.DATASTREAM, ds_id,
+                           APIResourceTypes.SCHEMA,
+                           params={'obsFormat': 'logical'})
+lg = LogicalDatastreamRecordSchema.from_logical_dict(lg_resp.json())
 ```
 
-Each method:
-
-1. Hits ``GET /datastreams/{id}/schema?obsFormat={format}`` using the
-   parent `Node`'s `APIHelper` for base URL + auth.
-2. Parses the response into the corresponding pydantic model.
-3. Returns the parsed model — does *not* mutate the datastream's
-   `_underlying_resource.record_schema`. (Discovery is the one place
-   that opts into caching the SWE+JSON variant; if you want to cache
-   an OM+JSON or logical fetch, assign it yourself.)
+`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

docs/source/architecture/serialization.md

@@ -124,9 +124,10 @@ A third schema model, `LogicalDatastreamRecordSchema`, covers OSH's
 extension keywords (`x-ogc-definition`, `x-ogc-refFrame`, `x-ogc-unit`,
 `x-ogc-axis`) carrying SWE Common metadata. Distinct from the SWE+JSON
 and OM+JSON envelopes (no `obsFormat` field, no `recordSchema`
-wrapper). See [Construction → "I want the schema for an existing
-datastream from the server"](construction.md) for the
-`Datastream.fetch_logical_schema()` method that retrieves it.
+wrapper). To retrieve it, use the per-`Node` `APIHelper`:
+`api.get_resource(APIResourceTypes.DATASTREAM, ds_id, APIResourceTypes.SCHEMA, params={'obsFormat': 'logical'})`,
+then parse the response with
+`LogicalDatastreamRecordSchema.from_logical_dict(...)`.
 
 ## Deprecated factories
 

docs/source/tutorial.rst

@@ -139,6 +139,27 @@ Discover all datastreams across all discovered systems:
 
    app.discover_datastreams()
 
+Each discovered ``Datastream`` arrives with its SWE+JSON record schema
+already cached on ``ds._underlying_resource.record_schema`` — discovery
+makes a follow-up ``GET /datastreams/{id}/schema`` per stream so callers
+that build observations don't need a second round trip.
+
+Discover control streams the same way, per system:
+
+.. code-block:: python
+
+   for system in node.get_systems():
+       control_streams = system.discover_controlstreams()
+       for cs in control_streams:
+           print(cs.get_id(), cs._underlying_resource.input_name)
+
+Discovered control streams arrive with their command schema cached on
+``cs._underlying_resource.command_schema`` (a ``JSONCommandSchema`` —
+OSH normalizes responses to the JSON envelope). Reach the inner SWE
+Common component via ``cs._underlying_resource.command_schema.params_schema``;
+its ``items`` (for ``DataChoice``) or ``fields`` (for ``DataRecord``)
+list the parameters the stream accepts.
+
 
 Streaming Observations (MQTT)
 ------------------------------
@@ -239,6 +260,146 @@ Build a schema using SWE Common component classes, then attach it to a system:
    A ``TimeSchema`` must be the first field in the ``DataRecordSchema`` when targeting OpenSensorHub.
 
 
+Inserting a New Control Stream
+------------------------------
+A control stream is the input counterpart to a datastream — it accepts
+commands and emits status reports. Build a ``DataRecordSchema``
+describing the command structure, then attach it to a system via
+``System.add_and_insert_control_stream(...)``:
+
+.. code-block:: python
+
+   from oshconnect import DataRecordSchema, BooleanSchema, CountSchema
+
+   command_record = DataRecordSchema(
+       name='counterControl',
+       label='Counter Control',
+       description='Commands to control the counter behavior',
+       fields=[
+           BooleanSchema(name='setCountDown', label='Set Count Down',
+                         definition='http://sensorml.com/ont/swe/property/SetCountDown'),
+           CountSchema(name='setStep', label='Set Step',
+                       definition='http://sensorml.com/ont/swe/property/SetStep'),
+       ],
+   )
+
+   control_stream = new_system.add_and_insert_control_stream(command_record)
+
+By default the wire form is ``application/swe+json`` (spec-compliant CS API
+Part 2 — ``commandFormat: "application/swe+json"`` plus ``recordSchema`` plus
+a ``JSONEncoding`` block). To target the JSON envelope instead (which is
+what OSH echoes back from ``/controlstreams/{id}/schema``), pass
+``command_format='application/json'``:
+
+.. code-block:: python
+
+   control_stream = new_system.add_and_insert_control_stream(
+       command_record,
+       command_format='application/json',
+   )
+
+The JSON form emits ``commandFormat: "application/json"`` with a
+``parametersSchema`` block (no ``encoding``).
+
+For full control over the resource body — for example, when copying a
+control stream from one node to another and you already have a
+``ControlStreamResource`` in hand — use ``add_insert_controlstream(...)``
+instead. It takes a fully-built resource and POSTs it as-is:
+
+.. code-block:: python
+
+   from oshconnect.resource_datamodels import ControlStreamResource
+   from oshconnect.schema_datamodels import JSONCommandSchema
+
+   resource = ControlStreamResource(
+       name='Counter Control',
+       input_name='counterControl',
+       command_schema=JSONCommandSchema(
+           command_format='application/json',
+           params_schema=command_record,
+       ),
+   )
+   control_stream = new_system.add_insert_controlstream(resource)
+
+After insert, the returned ``ControlStream`` carries the server-assigned
+ID (``control_stream.get_id()``) and is appended to ``new_system.control_channels``.
+
+
+Sending Commands
+----------------
+A control stream is the input side of a system. Once you have one — either
+freshly inserted or reconstructed from ``System.discover_controlstreams()`` —
+there are two ways to deliver a command:
+
+**Over MQTT (preferred for real-time control).** Initialize the stream's
+MQTT client, then publish to the command topic:
+
+.. code-block:: python
+
+   from oshconnect import StreamableModes
+
+   control_stream.set_connection_mode(StreamableModes.BIDIRECTIONAL)
+   control_stream.initialize()
+   control_stream.start()
+
+   control_stream.publish_command({
+       'params': {'setStep': 5},
+   })
+
+``publish_command(payload)`` is sugar for ``publish(payload, topic='command')``;
+it routes to the CS API Part 3 ``:commands`` topic for this stream
+(``…/controlstreams/{id}/commands``). The payload shape is whatever the
+control stream's command schema accepts — a dict matching the field names
+under ``params``, or a SWE+JSON envelope if the stream uses the SWE form.
+
+**Over HTTP (stateless, one-shot).** POST a command directly to the
+``/controlstreams/{id}/commands`` endpoint via the node's
+``APIHelper``:
+
+.. code-block:: python
+
+   from oshconnect.csapi4py.constants import APIResourceTypes
+   from oshconnect.schema_datamodels import CommandJSON
+
+   command = CommandJSON(params={'setStep': 5})
+   api = node.get_api_helper()
+   resp = api.create_resource(
+       APIResourceTypes.COMMAND,
+       command.to_csapi_dict(),
+       parent_res_id=control_stream.get_id(),
+       req_headers={'Content-Type': 'application/json'},
+   )
+   resp.raise_for_status()
+   command_id = resp.headers['Location'].rsplit('/', 1)[-1]
+
+The server responds with ``201 Created`` and a ``Location`` header pointing
+at the newly-created command resource (``/commands/{id}``); poll its
+``/status`` sub-resource (or subscribe to the MQTT status topic — next
+section) to see whether the system accepted and executed it.
+
+Subscribing to Command Status
+-----------------------------
+Control streams emit two MQTT topics: ``:commands`` (input) and ``:status``
+(output, where the system reports execution results). Subscribe to status
+updates:
+
+.. code-block:: python
+
+   def on_status(client, userdata, msg):
+       print(f"Status on {msg.topic}: {msg.payload}")
+
+   control_stream.subscribe(topic='status', callback=on_status)
+
+Inbound status reports are also pushed onto an internal deque — drain it
+exactly like a datastream's inbound queue:
+
+.. code-block:: python
+
+   while control_stream.get_status_deque_inbound():
+       status = control_stream.get_status_deque_inbound().popleft()
+       print(status)
+
+
 Inserting an Observation
 ------------------------
 Once a datastream is registered, send observation data using ``insert_observation_dict()``:

src/oshconnect/schema_datamodels.py

@@ -16,6 +16,15 @@
 from .encoding import Encoding
 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:
@@ -35,9 +44,12 @@ class CommandJSON(BaseModel):
     """
     model_config = ConfigDict(populate_by_name=True)
     control_id: str = Field(None, serialization_alias="control@id")
-    issue_time: Union[str, float] = Field(datetime.now().isoformat(), serialization_alias="issueTime")
+    issue_time: Union[str, float] = Field(default_factory=_now_iso8601_z,
+                                          serialization_alias="issueTime")
     sender: str = Field(None)
-    params: Union[dict, list, int, float, 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")
 
     def to_csapi_dict(self) -> dict:
         """Render as the CS API `application/json` command body."""