Change swe components to use Literal[{type_name}]'s for the type field. added a JSONDatastreamRecordschema for testing serialization/deserialization

Author: tipatterson-dev

src/oshconnect/__init__.py

@@ -33,7 +33,7 @@
     QuantityRangeSchema,
     TimeRangeSchema,
 )
-from .schema_datamodels import SWEDatastreamRecordSchema, JSONCommandSchema
+from .schema_datamodels import SWEDatastreamRecordSchema, JSONDatastreamRecordSchema, JSONCommandSchema
 
 # Event system
 from .events import EventHandler, IEventListener, CallbackListener, DefaultEventTypes, AtomicEventTypes, Event, EventBuilder
@@ -76,6 +76,7 @@
     "QuantityRangeSchema",
     "TimeRangeSchema",
     "SWEDatastreamRecordSchema",
+    "JSONDatastreamRecordSchema",
     "JSONCommandSchema",
     # Event system
     "EventHandler",

src/oshconnect/schema_datamodels.py

@@ -51,7 +51,7 @@ class SWEJSONCommandSchema(CommandSchema):
 
     command_format: str = Field("application/swe+json", alias='commandFormat')
     encoding: SerializeAsAny[Encoding] = Field(...)
-    record_schema: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='recordSchema')
+    record_schema: SerializeAsAny[AnyComponentSchema] = Field(..., alias='recordSchema')
 
 
 class JSONCommandSchema(CommandSchema):
@@ -78,7 +78,7 @@ class DatastreamRecordSchema(BaseModel):
 class SWEDatastreamRecordSchema(DatastreamRecordSchema):
     model_config = ConfigDict(populate_by_name=True)
     encoding: SerializeAsAny[Encoding] = Field(...)
-    record_schema: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='recordSchema')
+    record_schema: SerializeAsAny[AnyComponentSchema] = Field(..., alias='recordSchema')
 
     @field_validator('obs_format')
     @classmethod
@@ -89,6 +89,31 @@ def check_check_obs_format(cls, v):
         return v
 
 
+class JSONDatastreamRecordSchema(DatastreamRecordSchema):
+    """Datastream observation schema for the JSON media types
+    (`application/json`, `application/om+json`).
+
+    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.
+    """
+    model_config = ConfigDict(populate_by_name=True)
+
+    obs_format: str = Field(ObservationFormat.JSON.value, alias='obsFormat')
+    result_schema: SerializeAsAny[AnyComponentSchema] = Field(None, alias='resultSchema')
+    parameters_schema: SerializeAsAny[AnyComponentSchema] = Field(None, alias='parametersSchema')
+    result_link: dict = Field(None, alias='resultLink')
+
+    @field_validator('obs_format')
+    @classmethod
+    def _check_obs_format(cls, v):
+        if v not in (ObservationFormat.JSON.value, "application/json"):
+            raise ValueError(
+                f"obsFormat must be 'application/json' or '{ObservationFormat.JSON.value}'"
+            )
+        return v
+
+
 class ObservationOMJSONInline(BaseModel):
     """
     A class to represent an observation in OM-JSON format

src/oshconnect/swe_components.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 from numbers import Real
-from typing import Union, Any
+from typing import Union, Any, Literal
 
 from pydantic import BaseModel, Field, field_validator, SerializeAsAny
 
@@ -41,14 +41,14 @@ class AnyComponentSchema(BaseModel):
 
 
 class DataRecordSchema(AnyComponentSchema):
-    type: str = "DataRecord"
+    type: Literal["DataRecord"] = "DataRecord"
     fields: SerializeAsAny[list[AnyComponentSchema]] = Field(...)
 
 
 class VectorSchema(AnyComponentSchema):
     label: str = Field(...)
     name: str = Field(...)
-    type: str = "Vector"
+    type: Literal["Vector"] = "Vector"
     definition: str = Field(...)
     reference_frame: str = Field(..., serialization_alias='referenceFrame')
     local_frame: str = Field(None, serialization_alias='localFrame')
@@ -57,7 +57,7 @@ class VectorSchema(AnyComponentSchema):
 
 
 class DataArraySchema(AnyComponentSchema):
-    type: str = "DataArray"
+    type: Literal["DataArray"] = "DataArray"
     name: str = Field(...)
     element_count: dict | str | CountSchema = Field(..., serialization_alias='elementCount')  # Should type of Count
     element_type: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='elementType')
@@ -66,7 +66,7 @@ class DataArraySchema(AnyComponentSchema):
 
 
 class MatrixSchema(AnyComponentSchema):
-    type: str = "Matrix"
+    type: Literal["Matrix"] = "Matrix"
     element_count: dict | str | CountSchema = Field(..., serialization_alias='elementCount')  # Should be type of Count
     element_type: SerializeAsAny[list[AnyComponentSchema]] = Field(..., serialization_alias='elementType')
     encoding: str = Field(...)  # TODO: implement an encodings class
@@ -76,7 +76,7 @@ class MatrixSchema(AnyComponentSchema):
 
 
 class DataChoiceSchema(AnyComponentSchema):
-    type: str = "DataChoice"
+    type: Literal["DataChoice"] = "DataChoice"
     updatable: bool = Field(False)
     optional: bool = Field(False)
     choice_value: CategorySchema = Field(..., serialization_alias='choiceValue')  # TODO: Might be called "choiceValues"
@@ -85,7 +85,7 @@ class DataChoiceSchema(AnyComponentSchema):
 
 class GeometrySchema(AnyComponentSchema):
     label: str = Field(...)
-    type: str = "Geometry"
+    type: Literal["Geometry"] = "Geometry"
     updatable: bool = Field(False)
     optional: bool = Field(False)
     definition: str = Field(...)
@@ -129,17 +129,17 @@ class AnyScalarComponentSchema(AnySimpleComponentSchema):
 
 
 class BooleanSchema(AnyScalarComponentSchema):
-    type: str = "Boolean"
+    type: Literal["Boolean"] = "Boolean"
     value: bool = Field(None)
 
 
 class CountSchema(AnyScalarComponentSchema):
-    type: str = "Count"
+    type: Literal["Count"] = "Count"
     value: int = Field(None)
 
 
 class QuantitySchema(AnyScalarComponentSchema):
-    type: str = "Quantity"
+    type: Literal["Quantity"] = "Quantity"
     value: Union[float, str] = Field(None)
     uom: Union[UCUMCode, URI] = Field(...)
 
@@ -163,45 +163,45 @@ def validate_value(cls, v):
 
 
 class TimeSchema(AnyScalarComponentSchema):
-    type: str = "Time"
+    type: Literal["Time"] = "Time"
     value: str = Field(None)
     reference_time: str = Field(None, serialization_alias='referenceTime')
     local_frame: str = Field(None)
     uom: Union[UCUMCode, URI] = Field(...)
 
 
 class CategorySchema(AnyScalarComponentSchema):
-    type: str = "Category"
+    type: Literal["Category"] = "Category"
     value: str = Field(None)
     code_space: str = Field(None, serialization_alias='codeSpace')
 
 
 class TextSchema(AnyScalarComponentSchema):
-    type: str = "Text"
+    type: Literal["Text"] = "Text"
     value: str = Field(None)
 
 
 class CountRangeSchema(AnySimpleComponentSchema):
-    type: str = "CountRange"
+    type: Literal["CountRange"] = "CountRange"
     value: list[int] = Field(None)
     uom: Union[UCUMCode, URI] = Field(...)
 
 
 class QuantityRangeSchema(AnySimpleComponentSchema):
-    type: str = "QuantityRange"
+    type: Literal["QuantityRange"] = "QuantityRange"
     value: list[Union[float, str]] = Field(None)
     uom: Union[UCUMCode, URI] = Field(...)
 
 
 class TimeRangeSchema(AnySimpleComponentSchema):
-    type: str = "TimeRange"
+    type: Literal["TimeRange"] = "TimeRange"
     value: list[str] = Field(None)
     reference_time: str = Field(None, serialization_alias='referenceTime')
     local_frame: str = Field(None)
     uom: Union[UCUMCode, URI] = Field(...)
 
 
 class CategoryRangeSchema(AnySimpleComponentSchema):
-    type: str = "CategoryRange"
+    type: Literal["CategoryRange"] = "CategoryRange"
     value: list[str] = Field(None)
     code_space: str = Field(None, serialization_alias='codeSpace')