API Reference ¶

find_system(system_id)[source]¶

Find a system in the OSHConnect instance. :param system_id: :return: the found system or None if not found

Parameters:: system_id (str)
Return type:: System | None

add_system_to_node(system, target_node, insert_resource=False)[source]¶

Add a system to the target node. :param system: System object :param target_node: Node object, must be within the OSHConnect instance :param insert_resource: Whether to insert the system into the target node’s server, default is False :return:

Parameters:

system (System)
target_node (Node)
insert_resource (bool)

create_and_insert_system(system_opts, target_node)[source]¶

Create a system on the target node. :param system_opts: System object parameters :param target_node: Node object, must be within the OSHConnect instance :return: the created system

Parameters:

system_opts (dict)
target_node (Node)

remove_system(system_id)[source]¶

Parameters:: system_id (str)

get_datastreams()[source]¶

Return type:: list[Datastream]

get_datastream_ids()[source]¶

Return type:: list[UUID]

connect_session_streams(session_id)[source]¶

Connects all datastreams that are associated with the given session ID. :param session_id: :return:

Parameters:: session_id (str)

get_resource_group(resource_ids)[source]¶

Get a group of resources by their IDs. Can be any mix of systems, datastreams, and controlstreams. :param resource_ids: list of resource IDs (internal UUID)

Parameters:: resource_ids (list[UUID])
Return type:: tuple[list[System], list[Datastream]]

initialize_resource_groups(resource_ids=None)[source]¶

Initializes the datastreams that are specified.

Parameters:: resource_ids (list)

start_datastreams(dsid_list=None)[source]¶

Starts the datastreams that are specified.

Parameters:: dsid_list (list)

start_systems(sysid_list=None)[source]¶

Starts the systems that are specified.

Parameters:: sysid_list (list)

on_observation(callback, datastream_id=None)[source]¶

Subscribe to incoming observation events.

Parameters:

callback (Callable) – fn(event: Event) called for each matching event.
datastream_id (str) – When provided, only events from that datastream are delivered (matched via the datastream’s MQTT data topic). When omitted, all observation events are delivered.

Returns:

CallbackListener — pass to event_bus.unregister_listener() to cancel.

Return type:

CallbackListener

on_system_added(callback)[source]¶

Subscribe to system-discovered / system-added events.

Parameters:: callback (Callable) – fn(event: Event) where event.data is the System.
Returns:: CallbackListener for later removal.
Return type:: CallbackListener

on_command(callback, controlstream_id=None)[source]¶

Subscribe to incoming command events.

Parameters:

callback (Callable) – fn(event: Event) called for each matching event.
controlstream_id (str) – When provided, only events from that control stream are delivered. When omitted, all command events are delivered.

Returns:

CallbackListener for later removal.

Return type:

CallbackListener

property event_bus: EventHandler¶: Direct access to the EventHandler for advanced subscriptions.

—

Streamable Resources¶

These are the primary objects for interacting with systems, datastreams, and control streams on an OSH node. Includes Node, System, Datastream, ControlStream, and supporting enums.

Backward-compatible re-export shim.

The classes that used to live in this module have moved into focused sibling modules:

Node, SessionManager, OSHClientSession, Endpoints, Utilities → oshconnect.node
StreamableResource, Status, StreamableModes, SchemaFetchWarning → oshconnect.resources.base
System → oshconnect.resources.system
Datastream → oshconnect.resources.datastream
ControlStream → oshconnect.resources.controlstream

Existing from oshconnect.streamableresource import X paths continue to resolve through this shim. Prefer importing from oshconnect directly or from the new sibling modules in new code.

class oshconnect.streamableresource.ControlStream(node=None, controlstream_resource=None)[source]¶

Bases: StreamableResource[ControlStreamResource]

An input channel of a System: accepts commands and emits status.

Unlike Datastream, a control stream has TWO MQTT topics — one for commands (self._topic) and one for status updates (self._status_topic) — and two pairs of inbound/outbound deques to match. Construct from a parsed ControlStreamResource (typically from System.discover_controlstreams) or build locally and insert via System.add_and_insert_control_stream.

Parameters:

node (Node) – The Node this control stream lives under.
controlstream_resource (ControlStreamResource) – The pydantic ControlStreamResource model that backs this stream.

add_underlying_resource(resource)[source]¶

Replace the underlying ControlStreamResource model.

Parameters:: resource (ControlStreamResource)

get_id()[source]¶

Return the server-side control-stream ID.

Return type:: str

init_mqtt()[source]¶: Set self._topic to the control stream’s command data topic. When this control stream has a command_schema the topic is suffixed with the matching format subtopic (e.g. …/commands:data/swe-json); otherwise a bare :data topic is used and the server’s default format applies.

get_mqtt_status_topic()[source]¶

Return the topic/subject for command status updates. Status payloads are always application/json, so it is suffixed with the json format subtopic (…/status:data/json for MQTT, or the nested …commands... status:data.json equivalent for NATS).

Return type:: str

start()[source]¶: Start the control stream. PULL/BIDIRECTIONAL subscribes to the command topic; PUSH spawns the async MQTT write loop. Requires an active asyncio event loop for PUSH mode.

get_inbound_deque()[source]¶

Return the deque receiving inbound command payloads.

Return type:: deque

get_outbound_deque()[source]¶

Return the deque feeding outbound command publishes.

Return type:: deque

get_status_deque_inbound()[source]¶

Return the deque receiving inbound status updates.

Return type:: deque

get_status_deque_outbound()[source]¶

Return the deque feeding outbound status publishes.

Return type:: deque

publish_command(payload)[source]¶: Publish payload to the command MQTT topic. Convenience wrapper for publish(payload, APIResourceTypes.COMMAND.value).

publish_status(payload)[source]¶: Publish payload to the status MQTT topic. Convenience wrapper for publish(payload, APIResourceTypes.STATUS.value).

publish(payload, topic='Command')[source]¶

Publishes data to the MQTT topic associated with this control stream resource.

Parameters:

payload – Data to be published; subclass determines specifically allowed types.
topic (str) – One of APIResourceTypes.COMMAND.value ("Command", the default) or APIResourceTypes.STATUS.value ("Status"). Pass the enum value rather than a lowercase shorthand — the comparison is case-sensitive against the canonical CS API resource-type strings.

subscribe(topic=None, callback=None, qos=0)[source]¶

Subscribes to the MQTT topic associated with this control stream resource.

Parameters:

topic – None (defaults to the command topic), APIResourceTypes.COMMAND.value ("Command"), or APIResourceTypes.STATUS.value ("Status"). Comparison is case-sensitive against the canonical CS API resource-type strings.
callback – Optional callback function to handle incoming messages, if None the default handler is used.
qos – Quality of Service level for the subscription, default is 0.

to_storage_dict()[source]¶

Return a JSON-safe snapshot of this control stream — local identity, connection state, status topic, and the dumped underlying ControlStreamResource — 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 control-stream shape.

Return type:: dict

classmethod from_storage_dict(data, node)[source]¶

Build a ControlStream from a dict produced by to_storage_dict. The embedded underlying_resource is parsed via ControlStreamResource.model_validate, so that nested block can also be a CS API server response body for the control stream.

Parameters:

data (dict)
node (Node)

Return type:

ControlStream

class oshconnect.streamableresource.Datastream(parent_node=None, datastream_resource=None)[source]¶

Bases: StreamableResource[DatastreamResource]

An output channel of a System: produces observations.

Created from a parsed DatastreamResource (typically returned by System.discover_datastreams) or built locally and inserted via System.add_insert_datastream. Subscribes to its observation MQTT topic when started.

Parameters:

parent_node (Node) – The Node this datastream lives under.
datastream_resource (DatastreamResource) – The pydantic DatastreamResource model.

should_poll: bool¶

get_id()[source]¶

Return the server-side datastream ID.

Return type:: str

static from_resource(ds_resource, parent_node)[source]¶

Build a Datastream from an already-parsed DatastreamResource.

Deprecated since version 0.5.1: Use the constructor directly instead: Datastream(parent_node=node, datastream_resource=ds_resource). For raw JSON, parse first via DatastreamResource.from_csapi_dict(data).

Parameters:

ds_resource (DatastreamResource)
parent_node (Node)

Return type:

Datastream

set_resource(resource)[source]¶

Replace the underlying DatastreamResource model.

Parameters:: resource (DatastreamResource)

get_resource()[source]¶

Return the underlying DatastreamResource model.

Return type:: DatastreamResource

create_observation(obs_data)[source]¶

Build an ObservationResource from a result dict, validating against this datastream’s record schema if one is set.

Does NOT insert the observation server-side — pair with insert_observation_dict if you want to POST it.

Parameters:: obs_data (dict)
Return type:: ObservationResource

insert_observation_dict(obs_data)[source]¶

POST an observation dict to /datastreams/{id}/observations.

Raises:: Exception – if the server returns a non-OK response.
Parameters:: obs_data (dict)

start()[source]¶: Start the datastream. PULL/BIDIRECTIONAL subscribes to the observation topic; PUSH spawns the async MQTT write loop. Requires an active asyncio event loop for PUSH mode.

init_mqtt()[source]¶: Set self._topic to the datastream’s observation data topic (CS API Part 3 :data suffix). When this datastream has a record_schema the topic is suffixed with the matching format subtopic (e.g. …/observations:data/swe-binary); otherwise a bare :data topic is used and the server’s default format applies.

insert(data)[source]¶

Encode data and publish it to this datastream’s observation MQTT topic. Bypasses the outbound deque.

Encoding is chosen from the datastream’s record schema:

application/swe+binary → uses SWEBinaryCodec to pack a dict (or Sequence in declared member order) into the binary wire form. Raw bytes/bytearray/memoryview payloads are passed through verbatim — useful when the caller has already framed a record (e.g. a pre-encoded H.264 NAL unit with the standard [ts][size][bytes] blob framing from oshconnect.swe_binary.encode_swe_binary_blob).
everything else (incl. application/swe+json, application/om+json) → json.dumps of a dict.

decode_observation(raw, obs_format=None)[source]¶

Decode one observation off the wire using this datastream’s schema.

For application/swe+binary datastreams: walks the record encoding’s members and returns a dict keyed by field name. Block members come back as bytes (opaque — the codec does not demux H.264 / JPEG / etc.).

For JSON-family datastreams: returns json.loads(raw).

Parameters:

obs_format (str | None) – Optional MIME content-type overriding the datastream’s own schema format. Use this to decode data received via a NATS format-wildcard subscription, where the concrete format is only known per-message from the delivered subject (pass the result of nats_content_type_from_subject()). application/swe+flatbuffers is schemaless and decodes regardless of the datastream’s own schema type.
raw (bytes)

Raises:

ValueError – if no schema has been fetched.

Return type:

dict

to_storage_dict()[source]¶

Return a JSON-safe snapshot of this datastream — local identity, connection state, polling flag, and the dumped underlying DatastreamResource — 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 datastream shape.

Return type:: dict

classmethod from_storage_dict(data, node)[source]¶

Build a Datastream from a dict produced by to_storage_dict. The embedded underlying_resource is parsed via DatastreamResource.model_validate, so that nested block can also be a CS API server response body for the datastream.

Parameters:

data (dict)
node (Node)

Return type:

Datastream

subscribe(topic=None, callback=None, qos=0)[source]¶

Subscribe to this datastream’s observation MQTT topic.

Parameters:

topic – None or "observation" — both resolve to the datastream’s data topic. Any other string raises.
callback – Override the default callback (which appends payloads to _inbound_deque).
qos – MQTT QoS level. Default 0.

Raises:

ValueError – if topic is anything other than None / "observation".

class oshconnect.streamableresource.Endpoints(*, root='sensorhub', sos='sensorhub/sos', connected_systems='sensorhub/api')[source]¶

Bases: object

Default URL path segments for an OSH server’s REST APIs.

Parameters:

root (str)
sos (str)
connected_systems (str)

root: str = 'sensorhub'¶

sos: str = 'sensorhub/sos'¶

connected_systems: str = 'sensorhub/api'¶

class oshconnect.streamableresource.Node(protocol, address, port, username=None, password=None, server_root='sensorhub', api_root='api', mqtt_topic_root=None, session_manager=None, enable_mqtt=False, mqtt_port=1883, enable_nats=False, nats_port=4222, nats_token=None, mqtt_legacy_topics=False)[source]¶

Bases: object

One connection to a single OSH server.

A Node is the unit of “where to talk to”. It owns the APIHelper that builds and executes HTTP requests, an optional MQTTCommClient for Pub/Sub, and the list of System objects discovered from or inserted into that server. Most user code creates a Node and then either calls discover_systems() or attaches user-built systems via add_system().

Parameters:

protocol (str) – "http" or "https".
address (str) – Hostname or IP (no scheme).
port (int) – HTTP port the server is listening on.
username (str) – Optional Basic-Auth username.
password (str) – Optional Basic-Auth password.
server_root (str) – First path segment of the server URL (default "sensorhub").
api_root (str) – Second path segment under server_root (default "api").
mqtt_topic_root (str) – Override for the MQTT topic root if it diverges from the HTTP api root (CS API Part 3 § A.1).
session_manager (SessionManager) – Optional SessionManager; if given the node registers itself and gets a fresh OSHClientSession.
enable_mqtt (bool) – If True, connects an MQTT client to address.
mqtt_port (int) – MQTT broker port. Default 1883.
enable_nats (bool)
nats_port (int)
nats_token (str)
mqtt_legacy_topics (bool)

protocol: str¶

address: str¶

server_root: str = 'sensorhub'¶

port: int¶

is_secure: bool¶

endpoints: Endpoints¶

get_id()[source]¶

Return the locally-generated node ID (node-<uuid4>).

Return type:: str

get_address()[source]¶

Return the configured server hostname/IP.

Return type:: str

get_port()[source]¶

Return the configured server port.

Return type:: int

get_api_endpoint()[source]¶

Return the fully-qualified CS API root URL for this node.

Return type:: str

add_basicauth(username, password)[source]¶

Attach Basic-Auth credentials and mark the node as secure.

Parameters:

username (str)
password (str)

get_decoded_auth()[source]¶

Return the Base64 Basic-Auth header value as a UTF-8 string.

Return type:: str

get_mqtt_client()[source]¶

Return the connected MQTTCommClient or None if MQTT was not enabled at construction (enable_mqtt=True).

Return type:: MQTTCommClient

get_nats_client()[source]¶

Return the connected NatsCommClient or None if NATS was not enabled at construction (enable_nats=True).

Return type:: NatsCommClient

get_comm_client()[source]¶

Return the active pub/sub transport client for this node.

Prefers NATS when it was enabled, otherwise falls back to MQTT (and None if neither was enabled). StreamableResource reads this so it can drive whichever transport the node was configured with.

discover_systems()[source]¶

GET /systems?f=application/sml+json and create a System for each entry.

We pin SML+JSON because the GeoJSON listing variant (OSH’s default when no format is specified) is a summary that drops SensorML detail — identifiers, classifiers, keywords, characteristics, definition, typeOf, configuration, contacts, documentation, inputs/outputs/parameters, modes, method, featuresOfInterest. SML+JSON delivers all of those, which cross-node sync and any caller round-tripping _underlying_resource need.

Accept: application/sml+json is ignored by the OSH listing endpoint (still returns GeoJSON), so the format is selected via the ?f= query parameter — the OGC API standard format selector. SystemResource.model_validate parses both shapes, so the wrapper still copes if a server returns GeoJSON anyway.

The new systems are appended to this node’s internal list and also returned for convenience.

Returns:: List of newly-created System objects, or None if the HTTP request failed.
Return type:: list[System] | None

get_api_helper()[source]¶

Return the APIHelper this node uses for HTTP calls.

Return type:: APIHelper

add_system(system, insert_resource=False)[source]¶

Attach a system to this node.

When insert_resource=True, the system is first POSTed to the server via system.insert_self() (which populates its server-assigned resource id), then attached locally — so the system enters this node’s collection already carrying its real id. With insert_resource=False the system is attached in-memory only; useful when reconstructing state from a datastore or staging a system before a deferred POST.

Parameters:

system (System) – System object to attach.
insert_resource (bool) – Whether to POST the system to the server before attaching it locally.

Returns:

The same System (now parented to this node and tracked in self.systems()).

Return type:

systems()[source]¶

Return the list of System objects currently attached to this node.

Return type:: list[System]

register_with_session_manager(session_manager)[source]¶

Registers this node with the provided session manager, creating a new client session. :param session_manager: SessionManager instance

Parameters:: session_manager (SessionManager)

register_streamable(streamable)[source]¶

Register a streamable with this node’s session so its lifecycle is driven by OSHClientSession.connect_streamables / close_streamables.

Soft no-op when no SessionManager was attached at construction; the caller can still drive the streamable manually via initialize() / start() / stop().

Parameters:: streamable (StreamableResource)

get_session()[source]¶

Return the OSHClientSession bound to this node.

Return type:: OSHClientSession

to_storage_dict()[source]¶

Return a JSON-safe dict snapshot of this node — connection params, attached systems / streamables, and any locally-tracked state — for OSHConnect’s persistence layer (see OSHConnect.save_config, oshconnect.datastores.sqlite_store).

Not a CS API server-shaped payload; the dict format is OSHConnect’s own. For a CS API-shaped representation, use the underlying pydantic resource model’s model_dump(by_alias=True).

Return type:: dict

classmethod from_storage_dict(data, session_manager=None)[source]¶

Build a Node from a dict produced by to_storage_dict (i.e., from OSHConnect’s persistence layer, not from a CS API server response).

Expects connection params (protocol, address, port, optional username/password/server_root/api_root/ mqtt_topic_root), an _id, and a _systems list.

Parameters:

data (dict) – Source dict.
session_manager (SessionManager) – Optional SessionManager to register the rebuilt node with — required if any child StreamableResource in _systems was originally registered.

Return type:

Node

class oshconnect.streamableresource.OSHClientSession(base_url, *args, verify_ssl=True, **kwargs)[source]¶

Bases: object

One client session against a Node, owning its registered streamables.

Created by SessionManager.register_session and used by Node to manage the lifecycle (start/stop) of every StreamableResource attached to that node. Holds the streamables in a dict keyed by streamable ID.

Parameters:

base_url – Base URL of the OSH server (passed by Node, not used directly by this class today).
verify_ssl – Whether to verify TLS certificates. Default True.

verify_ssl = True¶

connect_streamables()[source]¶: Call start() on every registered streamable.

close_streamables()[source]¶: Call stop() on every registered streamable.

register_streamable(streamable)[source]¶

Track a streamable so its lifecycle is driven by this session.

Parameters:: streamable (StreamableResource)

exception oshconnect.streamableresource.SchemaFetchWarning[source]¶

Bases: UserWarning

A datastream/control-stream schema fetch or parse failed during Node.discover_systems / System.discover_datastreams / System.discover_controlstreams.

Discovery deliberately does not raise on per-resource schema failures — one broken schema would otherwise poison the entire listing. The matching wrapper is still appended (with record_schema / command_schema left as None), but the original exception is surfaced both here (via warnings.warn) and in the root logger at ERROR level (with a full traceback via exc_info=True). Filter or capture this category if you want to react programmatically.

class oshconnect.streamableresource.SessionManager(session_tokens=None)[source]¶

Bases: object

Top-level registry for OSHClientSession instances, one per Node.

The application owns one SessionManager; passing it to Node(...) causes the node to call register_session and bind itself to a fresh OSHClientSession. start_session_streams / start_all_streams are convenience entry points for booting streams on a single node or all nodes at once.

Parameters:: session_tokens (dict[str, str]) – Optional dict of session tokens keyed by ID (reserved for future auth schemes; currently unused).

sessions: dict[str, OSHClientSession] = None¶

register_session(session_id, session)[source]¶

Store session under session_id and return it.

Parameters:: session (OSHClientSession)
Return type:: OSHClientSession

unregister_session(session_id)[source]¶: Remove the session and call close() on it.

get_session(session_id)[source]¶

Return the session for session_id or None if unknown.

Return type:: OSHClientSession | None

start_session_streams(session_id)[source]¶

Start every streamable on the session identified by session_id.

Raises:: ValueError – if no session is registered for that ID.

start_all_streams()[source]¶: Start every streamable across every registered session.

class oshconnect.streamableresource.Status(*values)[source]¶

Bases: Enum

Lifecycle states a StreamableResource transitions through: STOPPED → INITIALIZING → INITIALIZED → STARTING → STARTED → STOPPING → STOPPED.

INITIALIZING = 'initializing'¶

INITIALIZED = 'initialized'¶

STARTING = 'starting'¶

STARTED = 'started'¶

STOPPING = 'stopping'¶

STOPPED = 'stopped'¶

class oshconnect.streamableresource.StreamableModes(*values)[source]¶

Bases: Enum

Direction(s) in which a streamable resource exchanges messages.

PUSH: this client publishes outbound messages only.
PULL: this client subscribes to inbound messages only.
BIDIRECTIONAL: both publish and subscribe.

PUSH = 'push'¶

PULL = 'pull'¶

BIDIRECTIONAL = 'bidirectional'¶

class oshconnect.streamableresource.StreamableResource(node, connection_mode='push')[source]¶

Bases: Generic[T], ABC

Abstract base for System, Datastream, and ControlStream.

Encapsulates the streaming machinery shared by all three: MQTT subscribe/ publish, optional WebSocket I/O, inbound and outbound message deques, and lifecycle (initialize → start → stop). Subclasses set _underlying_resource (a SystemResource / DatastreamResource / ControlStreamResource pydantic model) and override init_mqtt to derive the appropriate topic.

Parameters:

node (Node) – The parent Node this resource lives under.
connection_mode (StreamableModes) – One of StreamableModes. Default PUSH.

ws_url: str¶

get_streamable_id()[source]¶

Return the local UUID assigned at construction (not the server-side ID).

Return type:: UUID

get_streamable_id_str()[source]¶

Return the local UUID as a hex string.

Return type:: str

initialize()[source]¶

Build the WebSocket URL, allocate I/O queues, and configure MQTT.

Must be called before start. Inspects _underlying_resource to determine the right resource type and constructs the WS URL via the parent node’s APIHelper.

Raises:: ValueError – if _underlying_resource is not set or is not one of System / Datastream / ControlStream.

start()[source]¶: Subclasses override to also kick off MQTT subscribe / async write tasks. Logs and returns silently if initialize hasn’t been called.

async stream()[source]¶

Open a WebSocket to ws_url and run read/write loops in parallel.

Used as an alternative to MQTT for resources that prefer WS streaming. Reads incoming frames into the message handler and drains _msg_writer_queue to the socket.

init_mqtt()[source]¶

Wire the MQTT subscribe-acknowledged callback if a client exists.

Subclasses override to additionally derive their resource-specific topic into self._topic (see Datastream.init_mqtt / ControlStream.init_mqtt).

get_mqtt_topic(subresource=None, data_topic=True, format=None)[source]¶

Retrieves the MQTT topic for this streamable resource based on its underlying resource type. By default, returns a Resource Data Topic (:data suffix per CS API Part 3). :param subresource: Optional subresource type to get the topic for, defaults to None :param data_topic: If True (default), produces a Resource Data Topic with ‘:data’ suffix. Set False for Resource Event Topics. :param format: Optional MIME content-type for the :data/<token> format subtopic. None (default) emits bare :data so the server’s default format applies. Ignored when data_topic=False.

Parameters:

subresource (APIResourceTypes | None)
data_topic (bool)
format (str | None)

uses_nats()[source]¶

True when this resource’s active transport is NATS.

Drives the flat-vs-nested topic choice: NATS data subjects nest under systems (and need the parent system id) where the MQTT topics are flat. See get_nats_subject.

Return type:: bool

get_stream_topic(subresource=None, data_topic=True, format=None)[source]¶

Return the data topic/subject for the active transport.

Dispatches to get_nats_subject (nested-under-systems, dot-delimited) when NATS is active, else get_mqtt_topic (flat, slash-delimited). Subclasses call this from init_mqtt so they stay transport-agnostic.

Parameters:

subresource (APIResourceTypes | None)
data_topic (bool)
format (str | None)

Return type:

get_subscribe_topic(subresource=None, format=None)[source]¶

Return the topic/subject to subscribe to for inbound data.

For MQTT this is the exact per-format topic (the broker negotiates the format per subscription). For NATS it is a format-wildcard subject (…:data.*): in PROACTIVE mode the server publishes each stream on a single server-chosen format regardless of the subscriber, so we accept whatever it emits and read the concrete format back from each delivered subject via nats_content_type_from_subject().

Parameters:

subresource (APIResourceTypes | None)
format (str | None)

Return type:

get_nats_subject(subresource=None, data_topic=True, format=None, format_wildcard=False)[source]¶

Build the CS API Part 3 NATS data subject for this resource.

Unlike the flat MQTT topics, NATS data subjects are the canonical nested-under-systems resource path, dot-delimited, with a :data[.<token>] suffix — e.g. api.systems.{sysId}.datastreams.{dsId}.observations:data.swe-proto. This mirrors ConSysApiNatsConnector.getResourceUri on the server, which reverses a subject back into /systems/.../observations by stripping the :data suffix and replacing . with /.

When format_wildcard is set, the format subtopic is a NATS single-token wildcard (…:data.*) instead of a concrete token. This is used to subscribe in PROACTIVE mode, where the server publishes on one server-chosen format subtopic regardless of the datastream’s own obs format — the actual format is then read back from each delivered subject’s trailing token (see nats_content_type_from_subject()).

The parent system id required for the nesting is read from the datastream’s system_id (system@id) or — for control streams and locally-created datastreams — from _parent_resource_id.

Raises:

ValueError – if the underlying resource type is unsupported or the parent system id cannot be resolved.

Parameters:

subresource (APIResourceTypes | None)
data_topic (bool)
format (str | None)
format_wildcard (bool)

Return type:

get_event_topic()[source]¶

Returns the Resource Event Topic for this streamable resource per CS API Part 3. Event topics point to the resource itself (no ‘:data’ suffix) and are used to receive CloudEvents lifecycle notifications (create/update/delete) published by the server.

For Datastream/ControlStream, includes the parent system path when a parent resource ID is available.

Return type:: str

subscribe_events(callback=None, qos=0)[source]¶

Subscribes to the Resource Event Topic for this streamable resource. Event messages are CloudEvents v1.0 JSON payloads published by the server when the resource is created, updated, or deleted.

Parameters:

callback – Optional message callback. If None, uses the default handler (appends to inbound deque).
qos (int) – MQTT Quality of Service level, default 0.

Returns:

The event topic string that was subscribed to.

Return type:

stop()[source]¶

Tear down the streaming process and mark the resource STOPPED.

Note: currently calls Process.terminate(); cleaner shutdown (graceful drain, auth state preservation) is a known follow-up.

set_parent_node(node)[source]¶

Attach this resource to the given Node.

Parameters:: node (Node)

get_parent_node()[source]¶

Return the Node this resource is attached to.

Return type:: Node

set_parent_resource_id(res_id)[source]¶

Set the server-side ID of the parent resource (e.g. the parent System for a Datastream / ControlStream).

Parameters:: res_id (str)

get_parent_resource_id()[source]¶

Return the server-side ID of the parent resource, if set.

Return type:: str

set_connection_mode(connection_mode)[source]¶

Switch direction (PUSH / PULL / BIDIRECTIONAL).

Parameters:: connection_mode (StreamableModes)

poll()[source]¶: Poll for new data. Hook for subclass implementations; no-op here.

fetch(time_period)[source]¶

Fetch data over a TimePeriod. Hook for subclass implementations; no-op here.

Parameters:: time_period (TimePeriod)

get_msg_reader_queue()[source]¶

Returns the message queue for this streamable resource. In cases where a custom message handler is used this is not guaranteed to return anything or provided a queue with data. :return: Queue object

Return type:: Queue

get_msg_writer_queue()[source]¶

Returns the message queue for writing messages to this streamable resource. :return: Queue object

Return type:: Queue

get_underlying_resource()[source]¶

Return the pydantic resource model (System/Datastream/ControlStream) that backs this streamable.

Return type:: T

get_internal_id()[source]¶

Return the local UUID. Alias for get_streamable_id.

Return type:: UUID

insert_data(data)[source]¶: Inserts data into the message writer queue to be sent over the WebSocket / MQTT connection. Encoding is delegated to _encode_for_wire, which subclasses can override to honour their datastream’s wire format (e.g. Datastream routes through SWEBinaryCodec when its schema is application/swe+binary). No semantic validation is performed. :param data: Data to be sent (dict, sequence, or bytes-like).

subscribe_mqtt(topic, qos=0)[source]¶

Subscribe to an arbitrary MQTT topic using the default callback (appends incoming payloads to _inbound_deque).

Parameters:

topic (str) – MQTT topic string. The caller is responsible for any topic-prefix conventions (CS API Part 3 :data etc.).
qos (int) – MQTT QoS level. Default 0.

publish(payload, topic=None)[source]¶

Publishes data to the MQTT topic associated with this streamable resource. :param payload: Data to be published, subclass should determine specifically allowed types :param topic: Specific implementation determines the topic from the provided string, if None the default topic is used

Parameters:: topic (str)

subscribe(topic=None, callback=None, qos=0)[source]¶: Subscribes to the MQTT topic associated with this streamable resource. :param topic: Specific implementation determines the topic from the provided string, if None the default topic is used :param callback: Optional callback function to handle incoming messages, if None the default handler is used :param qos: Quality of Service level for the subscription, default is 0

get_inbound_deque()[source]¶

Return the deque that receives inbound MQTT message payloads.

Return type:: deque

get_outbound_deque()[source]¶

Return the deque feeding outbound MQTT publishes.

Return type:: deque

to_storage_dict()[source]¶

Return a JSON-safe snapshot of the streamable’s identity and connection state, for OSHConnect’s persistence layer. Subclasses extend this with their own fields and the dumped underlying resource. Safely handles missing / None attributes.

Not a CS API server-shaped payload.

Return type:: dict

classmethod from_storage_dict(data, node)[source]¶

Rebuild common attributes from a to_storage_dict payload. Subclasses override and call super() to wire in their own fields and the underlying resource.

Parameters:

data (dict)
node (Node)

Return type:

StreamableResource

class oshconnect.streamableresource.System(label=None, urn=None, parent_node=None, **kwargs)[source]¶

Bases: 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.

Parameters:

label (str)
urn (str)
parent_node (Node)

__init__(label=None, urn=None, parent_node=None, **kwargs)[source]¶

Parameters:

label (str) – 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.
urn (str) – The URN of the system, typically formed as such: 'urn:general_identifier:specific_identifier:…'.
parent_node (Node) – The Node this system attaches to.
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.

label: str¶

datastreams: list[Datastream]¶

control_channels: list[ControlStream]¶

urn: str¶

description: str¶

property name: 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.

discover_datastreams()[source]¶

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.

Return type:: list[Datastream]

discover_controlstreams()[source]¶

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.

Return type:: list[ControlStream]

classmethod from_resource(system_resource, parent_node)[source]¶

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.

Parameters:

system_resource (SystemResource) – A populated SystemResource instance.
parent_node (Node) – The Node the new System will attach to.

Returns:

A System wrapper bound to parent_node with _underlying_resource set to system_resource.

Return type:

static from_system_resource(system_resource, parent_node)[source]¶

Build a System from an already-parsed SystemResource.

Deprecated since version 0.5.1: Use 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).

Parameters:

system_resource (SystemResource)
parent_node (Node)

Return type:

to_system_resource()[source]¶

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.

Return type:: SystemResource

set_system_resource(sys_resource)[source]¶

Replace the underlying SystemResource model.

Parameters:: sys_resource (SystemResource)

get_system_resource()[source]¶

Return the underlying SystemResource model.

Return type:: SystemResource

add_insert_datastream(datastream_schema)[source]¶

Adds a datastream to the system while also inserting it into the system’s parent node via HTTP POST.

Parameters:: datastream_schema (DatastreamResource) – 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.
Returns:

add_insert_controlstream(controlstream_resource)[source]¶

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.

Parameters:: controlstream_resource (ControlStreamResource) – A fully-built ControlStreamResource carrying name, input_name, and command_schema.
Returns:: ControlStream object added to the system.
Return type:: ControlStream

add_and_insert_control_stream(control_stream_record_schema, input_name=None, valid_time=None, command_format='application/json')[source]¶

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.

Parameters:

control_stream_record_schema (DataRecordSchema) – 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.
input_name (str) – Name of the input. If None, the schema label is lowercased and whitespace-stripped.
valid_time (TimePeriod) – Optional TimePeriod; defaults to [now, now + 1 year].
command_format (str) – "application/json" (default) or "application/swe+json". Anything else raises ValueError.

Returns:

ControlStream object added to the system.

Return type:

ControlStream

insert_self()[source]¶

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.

retrieve_resource()[source]¶: GET /systems/{id} and refresh the underlying SystemResource. Returns None either way (kept for API symmetry).

to_storage_dict()[source]¶

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.

Return type:: dict

classmethod from_storage_dict(data, node)[source]¶

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.

Parameters:

data (dict) – Source dict.
node (Node) – Parent Node the rebuilt system attaches to.

Return type:

class oshconnect.streamableresource.Utilities[source]¶

Bases: object

Module-level helper namespace; intentionally just static methods.

static convert_auth_to_base64(username, password)[source]¶

Return username:password Base64-encoded for HTTP Basic Auth.

Parameters:

username (str)
password (str)

Return type:

—

Resource Data Models¶

Pydantic models that represent CS API resources returned from or sent to an OSH server.

class oshconnect.resource_datamodels.BoundingBox(*, lower_left_corner, upper_right_corner, min_value=None, max_value=None)[source]¶

Bases: BaseModel

Parameters:

lower_left_corner (Point)
upper_right_corner (Point)
min_value (float)
max_value (float)

lower_left_corner: Point¶

upper_right_corner: Point¶

min_value: float¶

max_value: float¶

class oshconnect.resource_datamodels.BaseResource(*, id, name, description=None, type=None, links=None)[source]¶

Bases: BaseModel

Parameters:

id (str)
name (str)
description (str)
type (str)
links (List[Link])

id: str¶

name: str¶

description: str¶

type: str¶

links: List[Link]¶

class oshconnect.resource_datamodels.SystemResource(*, type=None, id=None, properties=None, geometry=None, bbox=None, links=None, description=None, uniqueId=None, label=None, lang=None, keywords=None, identifiers=None, classifiers=None, validTime=None, securityConstraints=None, legalConstraints=None, characteristics=None, capabilities=None, contacts=None, documentation=None, history=None, definition=None, typeOf=None, configuration=None, featuresOfInterest=None, inputs=None, outputs=None, parameters=None, modes=None, method=None, **extra_data)[source]¶

Bases: BaseModel

Parameters:

type (str)
id (str)
properties (dict)
geometry (Geometry | None)
bbox (BoundingBox)
links (List[Link])
description (str)
uniqueId (str)
label (str)
lang (str)
keywords (List[str])
identifiers (list[Term])
classifiers (list[Term])
validTime (TimePeriod)
securityConstraints (list[dict])
legalConstraints (list[dict])
characteristics (list[Characteristics])
capabilities (list[Capabilities])
contacts (list[dict])
documentation (list[dict])
history (list[dict])
definition (str)
typeOf (str)
configuration (dict)
featuresOfInterest (list[dict])
inputs (list[dict])
outputs (list[dict])
parameters (list[dict])
modes (list[dict])
method (dict)
extra_data (Any)

feature_type: str¶

system_id: str¶

properties: dict¶

geometry: Geometry | None¶

bbox: BoundingBox¶

links: List[Link]¶

description: str¶

uid: str¶

label: str¶

lang: str¶

keywords: List[str]¶

identifiers: list[Term]¶

classifiers: list[Term]¶

valid_time: TimePeriod¶

security_constraints: list[dict]¶

legal_constraints: list[dict]¶

characteristics: list[Characteristics]¶

capabilities: list[Capabilities]¶

contacts: list[dict]¶

documentation: list[dict]¶

history: list[dict]¶

definition: str¶

type_of: str¶

configuration: dict¶

features_of_interest: list[dict]¶

inputs: list[dict]¶

outputs: list[dict]¶

parameters: list[dict]¶

modes: list[dict]¶

method: dict¶

to_smljson_dict()[source]¶

Render this system as an application/sml+json dict (SensorML JSON encoding).

The type discriminator (PhysicalSystem, PhysicalComponent, SimpleProcess, AggregateProcess, etc.) is preserved from self.feature_type when set — important for cross-node sync, where the source’s SML kind determines how OSH surfaces featureType (e.g. Sensor vs. System). Defaults to "PhysicalSystem" only when feature_type is unset, so callers building a bare SystemResource still get a valid SML body. Does not mutate self.

Return type:: dict

to_smljson()[source]¶

JSON-string variant of to_smljson_dict.

Return type:: str

to_geojson_dict()[source]¶

Render this system as an application/geo+json dict.

The type field is always set to "Feature" per the GeoJSON spec, regardless of self.feature_type — that’s the whole point of this rendering variant. Does not mutate self.

Return type:: dict

to_geojson()[source]¶

JSON-string variant of to_geojson_dict.

Return type:: str

classmethod from_smljson_dict(data)[source]¶

Build a SystemResource from an application/sml+json dict (e.g., a CS API server response body for a system in SML form).

Parameters:: data (dict)
Return type:: SystemResource

classmethod from_geojson_dict(data)[source]¶

Build a SystemResource from an application/geo+json dict (e.g., a CS API server response body for a system in GeoJSON form).

Parameters:: data (dict)
Return type:: SystemResource

classmethod from_csapi_dict(data)[source]¶

Build a SystemResource from a CS API system dict, auto-dispatching on the type field: "PhysicalSystem" → SML+JSON path, "Feature" → GeoJSON path. Anything else falls through to a permissive validate.

Parameters:: data (dict)
Return type:: SystemResource

class oshconnect.resource_datamodels.DatastreamResource(*, id, name, description=None, validTime, outputName=None, procedure_link=None, deployment_link=None, feature_of_interest_link=None, sampling_feature_link=None, parameters=None, phenomenonTime=None, resultTimeInterval=None, type=None, resultType=None, formats=<factory>, observedProperties=<factory>, system_id=None, links=None, schema=None)[source]¶

Bases: BaseModel

The DatastreamResource class is a Pydantic model that represents a datastream resource in the OGC SensorThings API. It contains all the necessary and optional properties listed in the OGC Connected Systems API documentation. Note that, depending on the format of the request, the fields needed may differ. There may be derived models in a later release that will have different sets of required fields to ease the validation process for users.

Parameters:

id (str)
name (str)
description (str)
validTime (TimePeriod)
outputName (str)
procedure_link (Link)
deployment_link (Link)
feature_of_interest_link (Link)
sampling_feature_link (Link)
parameters (dict)
phenomenonTime (TimePeriod)
resultTimeInterval (TimePeriod)
type (str)
resultType (str)
formats (List[str])
observedProperties (List[dict])
system_id (str)
links (List[Link])
schema (SWEDatastreamRecordSchema | SWEBinaryDatastreamRecordSchema | SWEProtobufDatastreamRecordSchema | SWEFlatBuffersDatastreamRecordSchema | OMJSONDatastreamRecordSchema)

ds_id: str¶

name: str¶

description: str¶

valid_time: TimePeriod¶

output_name: str¶

procedure_link: Link¶

deployment_link: Link¶

feature_of_interest_link: Link¶

sampling_feature_link: Link¶

parameters: dict¶

phenomenon_time: TimePeriod¶

result_time: TimePeriod¶

ds_type: str¶

result_type: str¶

formats: List[str]¶

observed_properties: List[dict]¶

system_id: str¶

links: List[Link]¶

record_schema: AnyDatastreamRecordSchema¶

to_csapi_dict()[source]¶

Render this datastream as the CS API application/json resource body. The embedded schema field is dumped polymorphically per whichever variant (SWEDatastreamRecordSchema / OMJSONDatastreamRecordSchema) it holds.

Return type:: dict

to_csapi_json()[source]¶

JSON-string variant of to_csapi_dict.

Return type:: str

classmethod from_csapi_dict(data)[source]¶

Build a DatastreamResource from a CS API datastream dict (e.g., a server response body or an entry from a /datastreams listing).

Parameters:: data (dict)
Return type:: DatastreamResource

class oshconnect.resource_datamodels.ObservationResource(*, sampling_feature_id=None, procedure_link=None, phenomenonTime=None, resultTime, parameters=None, result, result_link=None)[source]¶

Bases: BaseModel

Parameters:

sampling_feature_id (str)
procedure_link (Link)
phenomenonTime (TimeInstant)
resultTime (TimeInstant)
parameters (dict)
result (dict)
result_link (Link)

sampling_feature_id: str¶

procedure_link: Link¶

phenomenon_time: TimeInstant¶

result_time: TimeInstant¶

parameters: dict¶

result: dict¶

result_link: Link¶

to_omjson_dict(datastream_id=None)[source]¶

Render this observation as an application/om+json dict (the ObservationOMJSONInline shape).

Parameters:: datastream_id (str | None) – Optional ID to include as datastream@id on the output. The CS API typically supplies this from URL context, so it’s not required on the model itself.
Return type:: dict

to_swejson_dict(schema=None)[source]¶

Render this observation as an application/swe+json payload (the SWE Common JSON encoding of one record).

SWE+JSON encodes a single observation as a flat JSON object whose keys are the schema field names; self.result is already that dict, so this is essentially a passthrough. The optional schema argument is accepted for forward compatibility (when we add field-order / encoding-aware emission).

Parameters:: schema (AnyComponent)
Return type:: dict

classmethod from_omjson_dict(data)[source]¶

Build an ObservationResource from an application/om+json dict.

Parses through ObservationOMJSONInline to validate the OM+JSON envelope, then strips the datastream@id / foi@id envelope fields (those live on the surrounding context, not the resource) and returns the inner observation.

Parameters:: data (dict)
Return type:: ObservationResource

classmethod from_swejson_dict(data, schema=None, result_time=None)[source]¶

Build an ObservationResource from an application/swe+json observation payload.

SWE+JSON observations don’t carry an envelope (no resultTime / phenomenonTime fields); pass result_time explicitly when you have it, otherwise the current UTC time is used.

Parameters:

data (dict) – The flat SWE+JSON record dict.
schema (AnyComponent) – Optional schema, reserved for future per-field type coercion. Currently ignored.
result_time (str | None) – ISO 8601 timestamp for resultTime; defaults to TimeInstant.now_as_time_instant().isoformat() if omitted.

Return type:

ObservationResource

class oshconnect.resource_datamodels.ControlStreamResource(*, id=None, name, description=None, validTime=None, inputName=None, procedure_link=None, deployment_link=None, feature_of_interest_link=None, sampling_feature_link=None, issueTime=None, executionTime=None, live=None, asynchronous=True, schema=None, links=None)[source]¶

Bases: BaseModel

Parameters:

id (str)
name (str)
description (str)
validTime (TimePeriod)
inputName (str)
procedure_link (Link)
deployment_link (Link)
feature_of_interest_link (Link)
sampling_feature_link (Link)
issueTime (TimePeriod)
executionTime (TimePeriod)
live (bool)
asynchronous (bool)
schema (SWEJSONCommandSchema | JSONCommandSchema)
links (List[Link])

cs_id: str¶

name: str¶

description: str¶

valid_time: TimePeriod¶

input_name: str¶

procedure_link: Link¶

deployment_link: Link¶

feature_of_interest_link: Link¶

sampling_feature_link: Link¶

issue_time: TimePeriod¶

execution_time: TimePeriod¶

live: bool¶

asynchronous: bool¶

command_schema: AnyCommandSchema¶

links: List[Link]¶

to_csapi_dict()[source]¶

Render this control stream as the CS API application/json resource body. The embedded schema field is dumped polymorphically per whichever variant (SWEJSONCommandSchema / JSONCommandSchema) it holds.

Return type:: dict

to_csapi_json()[source]¶

JSON-string variant of to_csapi_dict.

Return type:: str

classmethod from_csapi_dict(data)[source]¶

Build a ControlStreamResource from a CS API control-stream dict (e.g., a server response body or an entry from a /controlstreams listing).

Parameters:: data (dict)
Return type:: ControlStreamResource

—

SWE Schema Components¶

Builder classes for constructing datastream and command schemas using SWE Common data types.

oshconnect.swe_components.check_named(component, location)[source]¶

Validate that a component bound via SoftNamedProperty carries a NameToken name.

Parameters:: location (str)
Return type:: None

class oshconnect.swe_components.AnyComponentSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Bases: BaseModel

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: str¶

id: str¶

name: str¶

label: str¶

description: str¶

updatable: bool¶

optional: bool¶

definition: str¶

class oshconnect.swe_components.DataRecordSchema(*, type='DataRecord', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None, fields)[source]¶

Parameters:

type (Literal['DataRecord'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
fields (Annotated[list[Annotated[DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], MinLen(min_length=1)])

type: Literal['DataRecord']¶

fields: list['AnyComponent']¶

class oshconnect.swe_components.VectorSchema(*, type='Vector', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition, referenceFrame, localFrame=None, coordinates)[source]¶

Parameters:

type (Literal['Vector'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
referenceFrame (str)
localFrame (str)
coordinates (list[CountSchema] | list[QuantitySchema] | list[TimeSchema])

type: Literal['Vector']¶

definition: str¶

reference_frame: str¶

local_frame: str¶

coordinates: list[CountSchema] | list[QuantitySchema] | list[TimeSchema]¶

class oshconnect.swe_components.DataArraySchema(*, type='DataArray', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None, elementCount, elementType, encoding=None, values=None)[source]¶

Parameters:

type (Literal['DataArray'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
elementCount (dict | str | CountSchema)
elementType (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
encoding (str)
values (list)

type: Literal['DataArray']¶

element_count: dict | str | CountSchema¶

element_type: AnyComponent¶

encoding: str¶

values: list¶

class oshconnect.swe_components.MatrixSchema(*, type='Matrix', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None, elementCount, elementType, encoding=None, values=None, reference_frame=None, local_frame=None)[source]¶

Parameters:

type (Literal['Matrix'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
elementCount (dict | str | CountSchema)
elementType (list[Annotated[DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])
encoding (str)
values (list)
reference_frame (str)
local_frame (str)

type: Literal['Matrix']¶

element_count: dict | str | CountSchema¶

element_type: list['AnyComponent']¶

encoding: str¶

values: list¶

reference_frame: str¶

local_frame: str¶

class oshconnect.swe_components.DataChoiceSchema(*, type='DataChoice', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None, choiceValue=None, items)[source]¶

Parameters:

type (Literal['DataChoice'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
choiceValue (CategorySchema)
items (list[Annotated[DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])

type: Literal['DataChoice']¶

updatable: bool¶

optional: bool¶

choice_value: CategorySchema¶

items: list['AnyComponent']¶

class oshconnect.swe_components.GeometrySchema(*, type='Geometry', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition, constraint=<factory>, nilValues=None, srs, value=None)[source]¶

Parameters:

type (Literal['Geometry'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
constraint (dict)
nilValues (list)
srs (str)
value (Geometry)

type: Literal['Geometry']¶

updatable: bool¶

optional: bool¶

definition: str¶

constraint: dict¶

nil_values: list¶

srs: str¶

value: Geometry¶

class oshconnect.swe_components.AnySimpleComponentSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

description: str¶

type: str¶

updatable: bool¶

optional: bool¶

definition: str¶

reference_frame: str¶

axis_id: str¶

quality: list[Annotated[QuantitySchema | QuantityRangeSchema | CategorySchema | TextSchema, Field(discriminator='type')]]¶

nil_values: list¶

constraint: Any¶

value: Any¶

class oshconnect.swe_components.AnyScalarComponentSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

A base class for all scalar components. The structure is essentially that of AnySimpleComponent

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

class oshconnect.swe_components.BooleanSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Boolean']¶

value: bool¶

class oshconnect.swe_components.CountSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Count']¶

value: int¶

class oshconnect.swe_components.QuantitySchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Quantity']¶

value: float | str¶

uom: UCUMCode | URI¶

classmethod validate_value(v)[source]¶

class oshconnect.swe_components.TimeSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Time']¶

value: str¶

reference_time: str¶

local_frame: str¶

uom: UCUMCode | URI¶

class oshconnect.swe_components.CategorySchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Category']¶

value: str¶

code_space: str¶

class oshconnect.swe_components.TextSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['Text']¶

value: str¶

class oshconnect.swe_components.CountRangeSchema(*, type, id=None, name=None, label=None, description=None, updatable=False, optional=False, definition=None)[source]¶

Parameters:

type (str)
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)

type: Literal['CountRange']¶

value: list[int]¶

uom: UCUMCode | URI¶

class oshconnect.swe_components.QuantityRangeSchema(*, type='QuantityRange', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition, referenceFrame=None, axisID=None, quality=None, nilValues=None, constraint=None, value=None, uom)[source]¶

Parameters:

type (Literal['QuantityRange'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
referenceFrame (str)
axisID (str)
quality (list[Annotated[QuantitySchema | QuantityRangeSchema | CategorySchema | TextSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])
nilValues (list)
constraint (Any)
value (list[float | str])
uom (UCUMCode | URI)

type: Literal['QuantityRange']¶

value: list[float | str]¶

uom: UCUMCode | URI¶

class oshconnect.swe_components.TimeRangeSchema(*, type='TimeRange', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition, referenceFrame=None, axisID=None, quality=None, nilValues=None, constraint=None, value=None, referenceTime=None, local_frame=None, uom)[source]¶

Parameters:

type (Literal['TimeRange'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
referenceFrame (str)
axisID (str)
quality (list[Annotated[QuantitySchema | QuantityRangeSchema | CategorySchema | TextSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])
nilValues (list)
constraint (Any)
value (list[str])
referenceTime (str)
local_frame (str)
uom (UCUMCode | URI)

type: Literal['TimeRange']¶

value: list[str]¶

reference_time: str¶

local_frame: str¶

uom: UCUMCode | URI¶

class oshconnect.swe_components.CategoryRangeSchema(*, type='CategoryRange', id=None, name=None, label=None, description=None, updatable=False, optional=False, definition, referenceFrame=None, axisID=None, quality=None, nilValues=None, constraint=None, value=None, codeSpace=None)[source]¶

Parameters:

type (Literal['CategoryRange'])
id (str)
name (str)
label (str)
description (str)
updatable (bool)
optional (bool)
definition (str)
referenceFrame (str)
axisID (str)
quality (list[Annotated[QuantitySchema | QuantityRangeSchema | CategorySchema | TextSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])
nilValues (list)
constraint (Any)
value (list[str])
codeSpace (str)

type: Literal['CategoryRange']¶

value: list[str]¶

code_space: str¶

class oshconnect.schema_datamodels.CommandJSON(*, control_id=None, issue_time=<factory>, sender=None, parameters=None)[source]¶

Bases: BaseModel

A class to represent a command in JSON format

Parameters:

control_id (str)
issue_time (str | float)
sender (str)
parameters (dict | list | int | float | str)

control_id: str¶

issue_time: str | float¶

sender: str¶

params: dict | list | int | float | str¶

to_csapi_dict()[source]¶

Render as the CS API application/json command body.

Return type:: dict

classmethod from_csapi_dict(data)[source]¶

Build from a CS API command JSON dict.

Parameters:: data (dict)
Return type:: CommandJSON

class oshconnect.schema_datamodels.CommandSchema(*, commandFormat)[source]¶

Bases: BaseModel

Base class representation for control streams’ command schemas

Parameters:: commandFormat (str)

command_format: str¶

class oshconnect.schema_datamodels.SWEJSONCommandSchema(*, commandFormat='application/swe+json', encoding, recordSchema)[source]¶

Bases: CommandSchema

SWE+JSON command schema

Parameters:

commandFormat (Literal['application/swe+json'])
encoding (JSONEncoding)
recordSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)

command_format: Literal['application/swe+json']¶

encoding: JSONEncoding¶

record_schema: AnyComponent¶

to_swejson_dict()[source]¶

Render as an application/swe+json command-schema document.

Return type:: dict

classmethod from_swejson_dict(data)[source]¶

Build from an application/swe+json command-schema dict.

Parameters:: data (dict)
Return type:: SWEJSONCommandSchema

class oshconnect.schema_datamodels.JSONCommandSchema(*, commandFormat='application/json', parametersSchema, resultSchema=None, feasibilityResultSchema=None)[source]¶

Bases: CommandSchema

JSON command schema

Parameters:

commandFormat (Literal['application/json'])
parametersSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
resultSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
feasibilityResultSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)

command_format: Literal['application/json']¶

params_schema: AnyComponent¶

result_schema: AnyComponent¶

feasibility_schema: AnyComponent¶

to_json_dict()[source]¶

Render as an application/json command-schema document.

Return type:: dict

classmethod from_json_dict(data)[source]¶

Build from an application/json command-schema dict.

Parameters:: data (dict)
Return type:: JSONCommandSchema

class oshconnect.schema_datamodels.DatastreamRecordSchema(*, obsFormat)[source]¶

Bases: BaseModel

A class to represent the schema of a datastream

Parameters:: obsFormat (str)

obs_format: str¶

class oshconnect.schema_datamodels.SWEDatastreamRecordSchema(*, obsFormat, encoding=None, recordSchema)[source]¶

Parameters:

obsFormat (Literal['application/swe+json', 'application/swe+csv', 'application/swe+text'])
encoding (JSONEncoding)
recordSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)

obs_format: Literal['application/swe+json', 'application/swe+csv', 'application/swe+text']¶

encoding: JSONEncoding¶

record_schema: AnyComponent¶

to_swejson_dict()[source]¶

Render as an application/swe+json datastream-schema document.

Return type:: dict

classmethod from_swejson_dict(data)[source]¶

Build from an application/swe+json datastream-schema dict (e.g., a CS API /datastreams/{id}/schema response in SWE form).

Parameters:: data (dict)
Return type:: SWEDatastreamRecordSchema

class oshconnect.schema_datamodels.SWEBinaryDatastreamRecordSchema(*, obsFormat='application/swe+binary', recordSchema, recordEncoding)[source]¶

Datastream observation schema for application/swe+binary.

Split from SWEDatastreamRecordSchema because the encoding block is a BinaryEncoding (with a members list mapping component refs to dataType / compression), not a JSONEncoding. The recordSchema side mirrors the SWE+JSON form — it describes the semantic shape of the record. The recordEncoding side describes the wire shape, overriding the semantic shape where needed (e.g. a DataArray in the recordSchema may be replaced by a single Block member with compression="H264" on the wire, as Axis cameras do for video).

Use oshconnect.swe_binary.SWEBinaryCodec(schema) to encode dicts to bytes and decode bytes back to dicts.

Parameters:

obsFormat (Literal['application/swe+binary'])
recordSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
recordEncoding (BinaryEncoding)

obs_format: Literal['application/swe+binary']¶

record_schema: AnyComponent¶

record_encoding: BinaryEncoding¶

to_swebinary_dict()[source]¶

Render as an application/swe+binary datastream-schema document.

Return type:: dict

classmethod from_swebinary_dict(data)[source]¶

Build from an application/swe+binary datastream-schema dict (a CS API /datastreams/{id}/schema?obsFormat=application/swe+binary response body).

Parameters:: data (dict)
Return type:: SWEBinaryDatastreamRecordSchema

class oshconnect.schema_datamodels.SWEProtobufDatastreamRecordSchema(*, obsFormat='application/swe+proto', fileDescriptorSet, messageType=None, recordEncoding=<factory>)[source]¶

SWEProtobufDatastreamRecordSchema

Datastream observation schema for application/swe+proto.

application/swe+proto is the per-datastream descriptor Protobuf encoding: each DataStream ships a pre-compiled Protobuf schema (a serialized google.protobuf.FileDescriptorSet) describing one per-datastream observation message — envelope fields 1–5 (id, datastream_id, foi_id, phenomenon/result time) plus the SWE Common record at fields 6+. An observation on the wire is a serialized instance of that message; receivers register the descriptor in a DescriptorPool and decode dynamically (no protoc).

This carries the descriptor itself rather than a SWE recordSchema tree — the descriptor is the source of truth for the wire layout, and SWE semantics (uom, definition, …) travel as field options inside it. The codec lives in oshconnect.swe_protobuf.SWEProtobufCodec.

Note

The JSON envelope that delivers the descriptor over the /datastreams/{id}/schema endpoint is a “meet in the middle” contract with the OSH node side — currently assumed to be {"obsFormat", "messageType", "fileDescriptorSet": <base64>}. See docs/osh_spec_deviations.md (swe-proto-descriptor-format).

Parameters:

obsFormat (Literal['application/swe+proto'])
fileDescriptorSet (bytes)
messageType (str)
recordEncoding (ProtobufEncoding)

obs_format: Literal['application/swe+proto']¶

file_descriptor_set: bytes¶

message_type: str¶

record_encoding: ProtobufEncoding¶

to_sweproto_dict()[source]¶

Render as an application/swe+proto datastream-schema document (fileDescriptorSet base64-encoded).

Return type:: dict

classmethod from_sweproto_dict(data)[source]¶

Build from an application/swe+proto datastream-schema dict.

Parameters:: data (dict)
Return type:: SWEProtobufDatastreamRecordSchema

classmethod from_record_schema(record, *, message_name='Observation', package='oshconnect.sweproto', datatype_by_path=None)[source]¶

Generate a swe+proto schema from a SWE Common DataRecord.

Builds the per-datastream observation descriptor (envelope fields 1–5 + the record’s components as result fields 6+, nested records and vectors recursed) — the inverse of OSH’s ProtoSchemaWriter — and wraps it in this schema model. Use this to produce a swe+proto datastream schema from the record_schema you already hold for a SWE+JSON or SWE+Binary datastream.

Parameters:

record (Annotated[DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema, FieldInfo(annotation=NoneType, required=True, discriminator='type')]) – a DataRecordSchema describing the observation.
message_name (str) – generated message name (e.g. f"Observation_{ds_id}").
package (str) – proto package for the generated message.
datatype_by_path (dict) – optional {ref: dataType_uri} map (same refs a BinaryEncoding uses) so float32 / int-width leaves map to the matching proto wire type instead of the defaults.

Return type:

classmethod from_other_schema(other, *, message_name='Observation', package='oshconnect.sweproto')[source]¶

Translate another datastream schema into swe+proto.

Accepts any record-bearing datastream schema (SWEDatastreamRecordSchema, SWEBinaryDatastreamRecordSchema, …) — pulling its record_schema — or a bare DataRecordSchema. When the source is SWE+Binary, the recordEncoding members are mined for each leaf’s OGC dataType so the generated proto fields use the matching wire type (e.g. float32 → float); SWE+JSON / SWE+CSV / SWE+text sources carry no dataType, so the node defaults apply.

Parameters:

message_name (str)
package (str)

Return type:

SWEProtobufDatastreamRecordSchema

to_proto_source()[source]¶

Render this schema’s descriptor as editable .proto source text.

Works for any swe+proto schema — one generated here or one delivered by the node — since it renders the carried FileDescriptorSet. Edit the text and recompile it with from_proto_source() to apply changes.

Return type:: str

classmethod from_proto_source(proto_text, *, message_type=None, protoc='protoc')[source]¶

Build a schema from .proto source text (compiles via protoc).

The round-trip companion to to_proto_source(): generate the text, hand-edit it (add/rename fields, tweak types, add annotations), then compile it back into a schema. Requires protoc on PATH (or pass protoc=<path>); the binary-descriptor paths (from_record_schema / from_other_schema) need no protoc. message_type defaults to the first message in the compiled file.

Parameters:

proto_text (str)
message_type (str)
protoc (str)

Return type:

SWEProtobufDatastreamRecordSchema

class oshconnect.schema_datamodels.SWEFlatBuffersDatastreamRecordSchema(*, obsFormat='application/swe+flatbuffers', recordSchema, recordEncoding=<factory>)[source]¶

Datastream observation schema for application/swe+flatbuffers.

Mirrors SWEProtobufDatastreamRecordSchema. The wire format is a FlatBuffers-serialized SWE Common 3 message; the codec lives in oshconnect.swe_flatbuffers.SWEFlatBuffersCodec.

Warning

The FlatBuffers codec is not currently functional — flatc –python does not yet support vectors-of-unions, which the SWE Common 3 schema uses for BinaryEncoding.members. The schema class is provided so the SDK can already parse and round-trip schemas that name this format; calling SWEFlatBuffersCodec.encode/decode raises NotImplementedError. See docs/osh_spec_deviations.md (flatc-python-vector-of-union).

Parameters:

obsFormat (Literal['application/swe+flatbuffers'])
recordSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
recordEncoding (FlatBuffersEncoding)

obs_format: Literal['application/swe+flatbuffers']¶

record_schema: AnyComponent¶

record_encoding: FlatBuffersEncoding¶

to_sweflatbuffers_dict()[source]¶

Render as an application/swe+flatbuffers datastream-schema document.

Return type:: dict

classmethod from_sweflatbuffers_dict(data)[source]¶

Build from an application/swe+flatbuffers datastream-schema dict.

Parameters:: data (dict)
Return type:: SWEFlatBuffersDatastreamRecordSchema

class oshconnect.schema_datamodels.OMJSONDatastreamRecordSchema(*, obsFormat='application/om+json', resultSchema=None, parametersSchema=None, resultLink=None)[source]¶

Datastream observation schema for the OM+JSON media type (application/om+json, also accepts application/json as a synonym on parse since OSH treats them equivalently for datastream schemas).

Per CS API Part 2 §16.1.4, this form does not carry a SWE encoding block; structure is fully described by resultSchema (inline result) or resultLink (out-of-band). parametersSchema is optional.

Parameters:

obsFormat (Literal['application/om+json', 'application/json'])
resultSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
parametersSchema (DataRecordSchema | VectorSchema | DataArraySchema | MatrixSchema | DataChoiceSchema | GeometrySchema | BooleanSchema | CountSchema | QuantitySchema | TimeSchema | CategorySchema | TextSchema | CountRangeSchema | QuantityRangeSchema | TimeRangeSchema | CategoryRangeSchema)
resultLink (dict)

obs_format: Literal['application/om+json', 'application/json']¶

result_schema: AnyComponent¶

parameters_schema: AnyComponent¶

result_link: dict¶

to_omjson_dict()[source]¶

Render as an application/om+json datastream-schema document.

Return type:: dict

classmethod from_omjson_dict(data)[source]¶

Build from an application/om+json (or application/json) datastream-schema dict (e.g., a CS API /datastreams/{id}/schema response in OM+JSON form).

Parameters:: data (dict)
Return type:: OMJSONDatastreamRecordSchema

class oshconnect.schema_datamodels.LogicalProperty(*, title=None, type, format=None, enum=None, items=None, properties=None, ogc_definition=None, ogc_ref_frame=None, ogc_unit=None, ogc_axis=None, **extra_data)[source]¶

Bases: BaseModel

One entry in LogicalDatastreamRecordSchema.properties.

The logical schema is OSH’s JSON-Schema-flavored representation of a SWE Common DataRecord. Each property is a JSON Schema field with OGC extension keywords (x-ogc-definition, x-ogc-refFrame, x-ogc-unit, x-ogc-axis) that carry the SWE Common metadata.

Permissive: extra='allow' accepts JSON Schema fields we haven’t modeled (e.g. description, default, minimum, maximum, nested items for arrays).

Parameters:

title (str)
type (str)
format (str)
enum (list)
items (dict)
properties (dict)
ogc_definition (str)
ogc_ref_frame (str)
ogc_unit (str)
ogc_axis (str)
extra_data (Any)

title: str¶

type: str¶

format: str¶

enum: list¶

items: dict¶

properties: dict¶

ogc_definition: str¶

ogc_ref_frame: str¶

ogc_unit: str¶

ogc_axis: str¶

class oshconnect.schema_datamodels.LogicalDatastreamRecordSchema(*, type, title=None, properties, required=None, **extra_data)[source]¶

Bases: BaseModel

Logical schema document — OSH’s obsFormat=logical representation.

Returned by GET /datastreams/{id}/schema?obsFormat=logical. Distinct from SWEDatastreamRecordSchema and OMJSONDatastreamRecordSchema:

No obsFormat envelope field
No recordSchema wrapper — the schema is the document
JSON Schema flavor (type: "object" + properties) instead of a SWE Common AnyComponent tree
Each property carries SWE Common metadata via x-ogc-* extension keywords

OSH-specific (not in the OGC CS API spec) but useful for tooling that speaks JSON Schema natively. Permissive (extra='allow') so future JSON Schema fields don’t break parsing.

Parameters:

type (str)
title (str)
properties (dict[str, LogicalProperty])
required (list[str])
extra_data (Any)

type: str¶

title: str¶

properties: dict[str, LogicalProperty]¶

required: list[str]¶

to_logical_dict()[source]¶

Render as an OSH obsFormat=logical JSON Schema dict.

Return type:: dict

classmethod from_logical_dict(data)[source]¶

Build from a logical schema dict (e.g., a CS API /datastreams/{id}/schema?obsFormat=logical response body).

Parameters:: data (dict)
Return type:: LogicalDatastreamRecordSchema

class oshconnect.schema_datamodels.ObservationOMJSONInline(*, datastream_id=None, foi_id=None, phenomenonTime=None, resultTime='2026-07-09T22:19:18.127423', parameters=None, result, result_links=None)[source]¶

Bases: BaseModel

A class to represent an observation in OM-JSON format

Parameters:

datastream_id (str)
foi_id (str)
phenomenonTime (str)
resultTime (str)
parameters (dict)
result (int | float | str | dict | list)
result_links (List[Link])

datastream_id: str¶

foi_id: str¶

phenomenon_time: str¶

result_time: str¶

parameters: dict¶

result: int | float | str | dict | list¶

result_links: List[Link]¶

to_csapi_dict()[source]¶

Render as an application/om+json observation body.

Return type:: dict

classmethod from_csapi_dict(data)[source]¶

Build from an application/om+json observation dict.

Parameters:: data (dict)
Return type:: ObservationOMJSONInline

class oshconnect.schema_datamodels.SystemEventOMJSON(*, label, description=None, definition, identifiers=None, classifiers=None, contacts=None, documentation=None, time, properties=None, configuration=None, links=None)[source]¶

Bases: BaseModel

A class to represent the schema of a system event

Parameters:

label (str)
description (str)
definition (HttpUrl)
identifiers (list)
classifiers (list)
contacts (list)
documentation (list)
time (str)
properties (list)
configuration (dict)
links (list[Link])

label: str¶

description: str¶

definition: HttpUrl¶

identifiers: list¶

classifiers: list¶

contacts: list¶

documentation: list¶

time: str¶

properties: list¶

configuration: dict¶

links: list[Link]¶

class oshconnect.schema_datamodels.SystemHistoryGeoJSON(**data)[source]¶

Bases: BaseModel

A class to represent the schema of a system history

Parameters:: data (Any)

type: str¶

id: str¶

properties: SystemHistoryProperties¶

geometry: Geometry¶

bbox: list¶

links: list[Link]¶

class oshconnect.schema_datamodels.SystemHistoryProperties(*, feature_type, uid, name, description=None, asset_type=None, valid_time=None, parent_system_link=None, procedure_link=None)[source]¶

Bases: BaseModel

Parameters:

feature_type (str)
uid (URI)
name (str)
description (str)
asset_type (str)
valid_time (list)
parent_system_link (str)
procedure_link (str)

feature_type: str¶

uid: URI¶

name: str¶

description: str¶

asset_type: str¶

valid_time: list¶

parent_system_link: str¶

procedure_link: str¶

oshconnect.schema_datamodels.AnyDatastreamRecordSchema¶

obs_format.

Type:: Public alias for DatastreamResource.record_schema. Discriminator

alias of Annotated[SWEDatastreamRecordSchema | SWEBinaryDatastreamRecordSchema | SWEProtobufDatastreamRecordSchema | SWEFlatBuffersDatastreamRecordSchema | OMJSONDatastreamRecordSchema, FieldInfo(annotation=NoneType, required=True, discriminator=’obs_format’)]

oshconnect.schema_datamodels.AnyCommandSchema¶

command_format.

Type:: Public alias for ControlStreamResource.command_schema. Discriminator

alias of Annotated[SWEJSONCommandSchema | JSONCommandSchema, FieldInfo(annotation=NoneType, required=True, discriminator=’command_format’)]

—

Event System¶

Pub/sub event bus for in-process notifications. Implement IEventListener to receive events.

The names below are re-exported from oshconnect.events.core, oshconnect.events.handler, etc.; :no-index: keeps Sphinx from reporting them as duplicate object descriptions.

class oshconnect.eventbus.Event(*, timestamp, type, topic, data, producer)[source]

Bases: BaseModel

Parameters:

timestamp (datetime)
type (DefaultEventTypes)
topic (str)
data (Any)
producer (Any)

timestamp: datetime.datetime

type: DefaultEventTypes

topic: str

data: Any

producer: Any

classmethod blank_event()[source]

Return type:: Event

class oshconnect.eventbus.DefaultEventTypes(*values)[source]

Bases: Enum

ADD_NODE: str = 'add_node'

REMOVE_NODE: str = 'remove_node'

ADD_SYSTEM: str = 'add_system'

REMOVE_SYSTEM: str = 'remove_system'

ADD_DATASTREAM: str = 'add_datastream'

REMOVE_DATASTREAM: str = 'remove_datastream'

ADD_CONTROLSTREAM: str = 'add_controlstream'

REMOVE_CONTROLSTREAM: str = 'remove_controlstream'

NEW_OBSERVATION: str = 'new_observation'

NEW_COMMAND: str = 'new_command'

NEW_COMMAND_STATUS: str = 'new_command_status'

class oshconnect.eventbus.AtomicEventTypes(*values)[source]

Bases: Enum

Defines atomic event types for local resource operations.

CREATE

Creating a resource within OSHConnect (local, in-app).

Type:: str

POST

Posting a resource to an external server.

Type:: str

GET

Retrieving a resource from an external server.

Type:: str

MODIFY

Modifying a resource within OSHConnect (local, in-app).

Type:: str

UPDATE

Updating a resource on an external server.

Type:: str

REMOVE

Removing a resource within OSHConnect (local, in-app).

Type:: str

DELETE

Deleting a resource from an external server.

Type:: str

CREATE: str = 'create'

POST: str = 'post'

GET: str = 'get'

MODIFY: str = 'modify'

UPDATE: str = 'update'

REMOVE: str = 'remove'

DELETE: str = 'delete'

class oshconnect.eventbus.EventHandler[source]

Bases: object

Singleton event bus. Manages listener registration and event dispatch.

Listeners are filtered by type and topic before dispatch — a listener only receives events whose type is in listener.types (empty = all types) AND whose topic is in listener.topics (empty = all topics).

Usage — functional style (no subclassing):

handler = EventHandler()

def on_obs(event: Event):
    print(event.data)

listener = handler.subscribe(on_obs, types=[DefaultEventTypes.NEW_OBSERVATION])
# later: handler.unregister_listener(listener)

Usage — subclass style:

class MyListener(IEventListener):
    def handle_events(self, event: Event):
        ...

handler.register_listener(MyListener(types=[DefaultEventTypes.ADD_SYSTEM]))

listeners: list[IEventListener] = []

to_add: list[IEventListener] = []

to_remove: list[IEventListener] = []

event_queue: deque[Event] = deque([])

publish_lock: bool = False

register_listener(listener)[source]

Parameters:: listener (IEventListener)

unregister_listener(listener)[source]

Parameters:: listener (IEventListener)

subscribe(callback, types=None, topics=None)[source]

Parameters:

callback (Callable[[Event], None]) – Function to call when a matching event is published.
types (list[DefaultEventTypes]) – Event types to filter on. None / empty = all types.
topics (list[str]) – MQTT/event topics to filter on. None / empty = all topics.

Returns:

The CallbackListener — keep a reference to unregister later.

Return type:

CallbackListener

publish(evt)[source]

Parameters:: evt (Event)

commit_changes()[source]

commit_adds()[source]

commit_removes()[source]

clear_listeners()[source]

get_num_listeners()[source]

Return type:: int

class oshconnect.eventbus.IEventListener(topics=<factory>, types=<factory>)[source]

Bases: ABC

Interface for event listeners. Subscribe to specific event types and/or topics. Empty lists mean “subscribe to all” — the handler filters before dispatching.

Parameters:

topics (list[str])
types (list[DefaultEventTypes])

topics: list[str]

types: list[DefaultEventTypes]

abstractmethod handle_events(event)[source]

Parameters:: event (Event)

class oshconnect.eventbus.CallbackListener(topics=<factory>, types=<factory>, callback=None)[source]

Bases: IEventListener

Concrete IEventListener that wraps a Python callable. The primary user-facing subscription mechanism — no subclassing required.

Example:

def my_handler(event: Event):
    print(event.data)

listener = CallbackListener(
    types=[DefaultEventTypes.NEW_OBSERVATION],
    callback=my_handler,
)
EventHandler().register_listener(listener)

Parameters:

topics (list[str])
types (list[DefaultEventTypes])
callback (Callable[[Event], None])

callback: Callable[[Event], None] = None

handle_events(event)[source]

Parameters:: event (Event)

class oshconnect.eventbus.EventBuilder[source]

Bases: ABC

with_type(event_type)[source]

Parameters:: event_type (DefaultEventTypes)
Return type:: EventBuilder

with_topic(topic)[source]

Parameters:: topic (str)
Return type:: EventBuilder

with_data(data)[source]

Parameters:: data (Any)
Return type:: EventBuilder

with_producer(producer)[source]

Parameters:: producer (Any)
Return type:: EventBuilder

with_timestamp(timestamp)[source]

Parameters:: timestamp (datetime)
Return type:: EventBuilder

build()[source]

Return type:: Event

reset()[source]

Return type:: None

static create_topic(base_topic, resource_id=None)[source]

Parameters:

base_topic (DefaultEventTypes)
resource_id (str | None)

Return type: