improve control stream discovery to have it get the command schema in the same process.

Author: tipatterson-dev

docs/source/tutorial.rst

@@ -285,26 +285,31 @@ describing the command structure, then attach it to a system via
 
    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'``:
+The default wire form is ``application/json`` —
+``commandFormat: "application/json"`` with a ``parametersSchema`` block
+(no ``encoding``). It matches what OSH echoes back from
+``GET /controlstreams/{id}/schema?f=json``, which is the form
+``discover_controlstreams`` parses, so cross-node sync round-trips
+without any format conversion. It also sidesteps the SWE+JSON
+``encoding``-omission deviation documented in
+``docs/osh_spec_deviations.md`` §1.
+
+For the spec-canonical SWE+JSON form (``recordSchema`` plus a
+``JSONEncoding`` block), pass ``command_format='application/swe+json'``:
 
 .. code-block:: python
 
    control_stream = new_system.add_and_insert_control_stream(
        command_record,
-       command_format='application/json',
+       command_format='application/swe+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:
+instead. It takes a fully-built resource and POSTs it as-is. Build the
+embedded ``command_schema`` as a ``JSONCommandSchema`` for the
+recommended JSON form:
 
 .. code-block:: python
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oshconnect"
-version = "0.5.1a2"
+version = "0.5.1a3"
 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."
 readme = "README.md"
 authors = [

src/oshconnect/schema_datamodels.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Union, List
+from typing import Union, List, Literal
 
 from pydantic import BaseModel, Field, SerializeAsAny, field_validator, model_validator, HttpUrl, ConfigDict
 
@@ -101,7 +101,7 @@ class JSONCommandSchema(CommandSchema):
     """
     model_config = ConfigDict(populate_by_name=True)
 
-    command_format: str = Field("application/json", alias='commandFormat')
+    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')

src/oshconnect/streamableresource.py

@@ -993,15 +993,18 @@ def discover_controlstreams(self) -> list[ControlStream]:
         ``self.control_channels`` and also returned.
 
         For each discovered control stream we additionally fetch the
-        command schema (``GET /controlstreams/{id}/schema``, which OSH
-        returns as ``application/json`` with a ``parametersSchema``
-        SWE Common component) and cache it on
-        ``_underlying_resource.command_schema``. 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.
+        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,
@@ -1016,6 +1019,7 @@ def discover_controlstreams(self) -> list[ControlStream]:
                 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 = (
@@ -1165,12 +1169,21 @@ def add_insert_controlstream(self, controlstream_resource: ControlStreamResource
         the system's parent node via HTTP POST.
 
         Mirrors `add_insert_datastream`: caller assembles the full
-        `ControlStreamResource` (including the embedded `command_schema`
-        — a `JSONCommandSchema` for ``application/json`` or a
-        `SWEJSONCommandSchema` for ``application/swe+json``) and this
-        method posts it to ``/systems/{id}/controlstreams``, captures
-        the new resource ID from the ``Location`` header, and returns a
-        wrapped `ControlStream`.
+        `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``,
@@ -1201,18 +1214,24 @@ def add_insert_controlstream(self, controlstream_resource: ControlStreamResource
 
     def add_and_insert_control_stream(self, control_stream_record_schema: DataRecordSchema, input_name: str = None,
                                       valid_time: TimePeriod = None,
-                                      command_format: str = "application/swe+json") -> ControlStream:
+                                      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/swe+json`` → `SWEJSONCommandSchema` carrying
-          `recordSchema` (the SWE Common component) and `encoding`
-          (`JSONEncoding`). This is the spec-compliant default.
         - ``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
@@ -1222,8 +1241,9 @@ def add_and_insert_control_stream(self, control_stream_record_schema: DataRecord
             is lowercased and whitespace-stripped.
         :param valid_time: Optional `TimePeriod`; defaults to
             ``[now, now + 1 year]``.
-        :param command_format: ``"application/swe+json"`` (default) or
-            ``"application/json"``. Anything else raises ``ValueError``.
+        :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(

tests/test_controlstream_insert_schema.py

@@ -84,10 +84,10 @@ def _captured_body_json(captured: dict) -> dict:
     return json.loads(body)
 
 
-def test_swejson_default_emits_recordschema_and_encoding(system, monkeypatch):
-    """Default `command_format='application/swe+json'` must produce the
-    spec-compliant wire form: ``commandFormat: application/swe+json`` plus
-    ``recordSchema`` plus ``encoding`` (JSONEncoding). NOT ``parametersSchema``."""
+def test_json_default_emits_parametersschema_no_encoding(system, monkeypatch):
+    """Default ``command_format='application/json'`` must produce the JSON
+    wire form: ``commandFormat: application/json`` plus ``parametersSchema``.
+    NOT ``recordSchema`` and NOT ``encoding``."""
     captured: dict = {}
     monkeypatch.setattr(
         "oshconnect.csapi4py.request_wrappers.requests.post", _capture_post(captured),
@@ -97,37 +97,37 @@ def test_swejson_default_emits_recordschema_and_encoding(system, monkeypatch):
 
     body = _captured_body_json(captured)
     schema = body["schema"]
-    assert schema["commandFormat"] == "application/swe+json"
-    assert "recordSchema" in schema, "SWE+JSON form must carry recordSchema"
-    assert "parametersSchema" not in schema, (
-        "SWE+JSON form must NOT carry parametersSchema (that's the JSON form)"
+    assert schema["commandFormat"] == "application/json"
+    assert "parametersSchema" in schema, "JSON form must carry parametersSchema"
+    assert "recordSchema" not in schema, (
+        "JSON form must NOT carry recordSchema (that's the SWE+JSON form)"
+    )
+    assert "encoding" not in schema, (
+        "JSON form has no encoding block — that's SWE+JSON only"
     )
-    assert schema["encoding"]["type"] == "JSONEncoding"
 
 
-def test_json_emits_parametersschema_no_encoding(system, monkeypatch):
-    """`command_format='application/json'` must produce the JSON wire form:
-    ``commandFormat: application/json`` plus ``parametersSchema``. NOT
-    ``recordSchema`` and NOT ``encoding``."""
+def test_swejson_emits_recordschema_and_encoding(system, monkeypatch):
+    """`command_format='application/swe+json'` must produce the
+    spec-canonical wire form: ``commandFormat: application/swe+json`` plus
+    ``recordSchema`` plus ``encoding`` (JSONEncoding). NOT ``parametersSchema``."""
     captured: dict = {}
     monkeypatch.setattr(
         "oshconnect.csapi4py.request_wrappers.requests.post", _capture_post(captured),
     )
 
     system.add_and_insert_control_stream(
-        _record_schema(), command_format="application/json",
+        _record_schema(), command_format="application/swe+json",
     )
 
     body = _captured_body_json(captured)
     schema = body["schema"]
-    assert schema["commandFormat"] == "application/json"
-    assert "parametersSchema" in schema, "JSON form must carry parametersSchema"
-    assert "recordSchema" not in schema, (
-        "JSON form must NOT carry recordSchema (that's the SWE+JSON form)"
-    )
-    assert "encoding" not in schema, (
-        "JSON form has no encoding block — that's SWE+JSON only"
+    assert schema["commandFormat"] == "application/swe+json"
+    assert "recordSchema" in schema, "SWE+JSON form must carry recordSchema"
+    assert "parametersSchema" not in schema, (
+        "SWE+JSON form must NOT carry parametersSchema (that's the JSON form)"
     )
+    assert schema["encoding"]["type"] == "JSONEncoding"
 
 
 def test_unsupported_command_format_raises(system):

tests/test_node_to_node_sync.py

@@ -22,8 +22,13 @@
 
 from oshconnect import Node, System
 from oshconnect.csapi4py.constants import APIResourceTypes
+from oshconnect.encoding import JSONEncoding
 from oshconnect.resource_datamodels import ControlStreamResource, DatastreamResource
-from oshconnect.schema_datamodels import CommandJSON, JSONCommandSchema, SWEDatastreamRecordSchema
+from oshconnect.schema_datamodels import (
+    CommandJSON,
+    SWEDatastreamRecordSchema,
+    SWEJSONCommandSchema,
+)
 from oshconnect.timemanagement import TimeInstant, TimePeriod, TimeUtils
 
 SRC_PORT = int(os.environ.get("OSHC_SRC_PORT", "8282"))