code cleanup

Author: timmarkhuff

pyproject.toml

@@ -22,6 +22,7 @@ python-dateutil = "^2.9.0"
 requests = "^2.28.2"
 typer = "^0.15.4"
 urllib3 = "^2.6.1"
+pyyaml = "^6.0.3"
 
 [tool.poetry.group.dev.dependencies]
 datamodel-code-generator = "^0.35.0"

src/groundlight/edge/__init__.py

@@ -1,4 +1,5 @@
 from .config import (
+    ConfigBase,
     DEFAULT,
     DISABLED,
     EDGE_ANSWERS_WITH_ESCALATION,
@@ -15,6 +16,7 @@
     "DISABLED",
     "EDGE_ANSWERS_WITH_ESCALATION",
     "NO_CLOUD",
+    "ConfigBase",
     "DetectorsConfig",
     "DetectorConfig",
     "EdgeEndpointConfig",

src/groundlight/edge/config.py

@@ -1,11 +1,16 @@
 from typing import Any, Optional, Union
 
 from model import Detector
-from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from typing_extensions import Self
+import yaml
 
 
 class GlobalConfig(BaseModel):
+    """Global runtime settings for edge-endpoint behavior."""
+
+    model_config = ConfigDict(extra="forbid")
+
     refresh_rate: float = Field(
         default=60.0,
         description="The interval (in seconds) at which the inference server checks for a new model binary update.",
@@ -22,7 +27,7 @@ class InferenceConfig(BaseModel):
     """
 
     # Keep shared presets immutable (DEFAULT/NO_CLOUD/etc.) so one mutation cannot globally change behavior.
-    model_config = ConfigDict(frozen=True)
+    model_config = ConfigDict(extra="forbid", frozen=True)
 
     name: str = Field(..., exclude=True, description="A unique name for this inference config preset.")
     enabled: bool = Field(
@@ -71,18 +76,41 @@ class DetectorConfig(BaseModel):
     Configuration for a specific detector.
     """
 
+    model_config = ConfigDict(extra="forbid")
+
     detector_id: str = Field(..., description="Detector ID")
     edge_inference_config: str = Field(..., description="Config for edge inference.")
 
 
-class DetectorsConfig(BaseModel):
-    """
-    Detector and inference-config mappings for edge inference.
-    """
+class ConfigBase(BaseModel):
+    """Shared detector/inference configuration behavior for edge config models."""
+
+    model_config = ConfigDict(extra="forbid")
 
     edge_inference_configs: dict[str, InferenceConfig] = Field(default_factory=dict)
     detectors: list[DetectorConfig] = Field(default_factory=list)
 
+    @field_validator("edge_inference_configs", mode="before")
+    @classmethod
+    def hydrate_inference_config_names(
+        cls, value: dict[str, InferenceConfig | dict[str, Any]] | None
+    ) -> dict[str, InferenceConfig | dict[str, Any]]:
+        """Hydrate InferenceConfig.name from payload mapping keys."""
+        if value is None:
+            return {}
+        if not isinstance(value, dict):
+            return value
+
+        hydrated_configs: dict[str, InferenceConfig | dict[str, Any]] = {}
+        for name, config in value.items():
+            if isinstance(config, InferenceConfig):
+                hydrated_configs[name] = config
+                continue
+            if not isinstance(config, dict):
+                raise TypeError("Each edge inference config must be an object.")
+            hydrated_configs[name] = {"name": name, **config}
+        return hydrated_configs
+
     @model_validator(mode="after")
     def validate_inference_configs(self):
         """
@@ -128,7 +156,7 @@ def add_detector(self, detector: Union[str, Detector], edge_inference_config: In
         self.detectors.append(DetectorConfig(detector_id=detector_id, edge_inference_config=edge_inference_config.name))
 
     def to_payload(self) -> dict[str, Any]:
-        """Return flattened detector payload used by edge-endpoint config HTTP APIs."""
+        """Return detector payload used by edge-endpoint config HTTP APIs."""
         return {
             "edge_inference_configs": {
                 name: config.model_dump() for name, config in self.edge_inference_configs.items()
@@ -137,36 +165,54 @@ def to_payload(self) -> dict[str, Any]:
         }
 
 
-class EdgeEndpointConfig(BaseModel):
+class DetectorsConfig(ConfigBase):
+    """
+    Detector and inference-config mappings for edge inference.
+    """
+
+
+class EdgeEndpointConfig(ConfigBase):
     """
     Top-level edge endpoint configuration.
     """
 
     global_config: GlobalConfig = Field(default_factory=GlobalConfig)
-    detectors_config: DetectorsConfig = Field(default_factory=DetectorsConfig)
-
-    @property
-    def edge_inference_configs(self) -> dict[str, InferenceConfig]:
-        """Convenience accessor for detector inference config map."""
-        return self.detectors_config.edge_inference_configs
 
-    @property
-    def detectors(self) -> list[DetectorConfig]:
-        """Convenience accessor for detector assignments."""
-        return self.detectors_config.detectors
+    @classmethod
+    def from_yaml(
+        cls,
+        filename: Optional[str] = None,
+        yaml_str: Optional[str] = None,
+    ) -> "EdgeEndpointConfig":
+        """Create an EdgeEndpointConfig from a YAML filename or YAML string."""
+        if filename is None and yaml_str is None:
+            raise ValueError("Either filename or yaml_str must be provided.")
+        if filename is not None and yaml_str is not None:
+            raise ValueError("Only one of filename or yaml_str can be provided.")
+        if filename is not None:
+            if not filename.strip():
+                raise ValueError("filename must be a non-empty path when provided.")
+            with open(filename, "r") as f:
+                yaml_str = f.read()
+
+        yaml_text = yaml_str or ""
+        parsed = yaml.safe_load(yaml_text) or {}
+        return cls.model_validate(parsed)
 
-    @model_serializer(mode="plain")
-    def serialize(self):
-        """Serialize to the flattened shape expected by edge-endpoint configs."""
+    def to_payload(self) -> dict[str, Any]:
+        """Return the full edge-endpoint payload shape."""
         return {
             "global_config": self.global_config.model_dump(),
-            **self.detectors_config.to_payload(),
+            "edge_inference_configs": {
+                name: config.model_dump() for name, config in self.edge_inference_configs.items()
+            },
+            "detectors": [detector.model_dump() for detector in self.detectors],
         }
 
-    def add_detector(self, detector: Union[str, Detector], edge_inference_config: InferenceConfig) -> None:
-        """Add a detector with the given inference config. Accepts detector ID or Detector object."""
-        self.detectors_config.add_detector(detector, edge_inference_config)
-
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "EdgeEndpointConfig":
+        """Construct an EdgeEndpointConfig from a payload dictionary."""
+        return cls.model_validate(payload)
 
 # Preset inference configs matching the standard edge-endpoint defaults.
 DEFAULT = InferenceConfig(name="default")

test/unit/test_edge_config.py

@@ -1,5 +1,4 @@
 from datetime import datetime, timezone
-
 import pytest
 from groundlight.edge import (
     DEFAULT,
@@ -108,6 +107,25 @@ def test_constructor_accepts_matching_inference_config_key_and_name():
     assert [detector.detector_id for detector in config.detectors] == ["det_1"]
 
 
+def test_constructor_hydrates_inference_config_name_from_dict_key():
+    """Hydrates inference config names from payload dict keys."""
+    config = DetectorsConfig(
+        edge_inference_configs={"default": {"enabled": True}},
+        detectors=[{"detector_id": "det_1", "edge_inference_config": "default"}],
+    )
+
+    assert config.edge_inference_configs["default"].name == "default"
+
+
+def test_constructor_rejects_detector_map_input():
+    """Rejects detector maps and requires detector list payloads."""
+    with pytest.raises(ValueError):
+        DetectorsConfig(
+            edge_inference_configs={"default": {"enabled": True}},
+            detectors={"det_1": {"detector_id": "det_1", "edge_inference_config": "default"}},
+        )
+
+
 def test_constructor_rejects_undefined_inference_config_reference():
     """Rejects detector entries that reference missing inference configs."""
     with pytest.raises(ValueError, match="not defined"):
@@ -117,7 +135,7 @@ def test_constructor_rejects_undefined_inference_config_reference():
         )
 
 
-def test_edge_endpoint_config_add_detector_delegates_to_detectors_logic():
+def test_edge_endpoint_config_add_detector_uses_shared_config_logic():
     """Adds detectors via EdgeEndpointConfig and preserves inferred config mapping."""
     config = EdgeEndpointConfig()
     config.add_detector("det_1", NO_CLOUD)
@@ -157,23 +175,103 @@ def test_detectors_config_to_payload_shape():
     assert set(payload["edge_inference_configs"].keys()) == {"default", "no_cloud"}
 
 
+def test_edge_endpoint_config_accepts_top_level_payload_shape():
+    """Accepts the top-level edge endpoint payload shape used by APIs."""
+    config = EdgeEndpointConfig.model_validate(
+        {
+            "global_config": {"refresh_rate": CUSTOM_REFRESH_RATE},
+            "edge_inference_configs": {"default": {"enabled": True}},
+            "detectors": [{"detector_id": "det_1", "edge_inference_config": "default"}],
+        }
+    )
+
+    assert config.global_config.refresh_rate == CUSTOM_REFRESH_RATE
+    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+
+
+def test_edge_endpoint_config_from_yaml_accepts_yaml_text():
+    """Parses edge-endpoint YAML text using EdgeEndpointConfig.from_yaml."""
+    config = EdgeEndpointConfig.from_yaml(
+        yaml_str="""
+        global_config:
+          refresh_rate: 15.0
+        edge_inference_configs:
+          default:
+            enabled: true
+        detectors:
+          - detector_id: det_1
+            edge_inference_config: default
+        """
+    )
+
+    assert config.global_config.refresh_rate == 15.0
+    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+
+
+def test_edge_endpoint_config_from_yaml_accepts_filename(tmp_path):
+    """Parses edge-endpoint YAML from a file path."""
+    config_file = tmp_path / "edge-config.yaml"
+    config_file.write_text(
+        "global_config: {}\n"
+        "edge_inference_configs:\n"
+        "  default:\n"
+        "    enabled: true\n"
+        "detectors:\n"
+        "  - detector_id: det_1\n"
+        "    edge_inference_config: default\n"
+    )
+    config = EdgeEndpointConfig.from_yaml(filename=str(config_file))
+
+    assert [detector.detector_id for detector in config.detectors] == ["det_1"]
+
+
+def test_edge_endpoint_config_from_yaml_requires_exactly_one_input():
+    """Rejects missing input and mixed filename/yaml_str input."""
+    with pytest.raises(ValueError, match="Either filename or yaml_str must be provided"):
+        EdgeEndpointConfig.from_yaml()
+
+    with pytest.raises(ValueError, match="Only one of filename or yaml_str can be provided"):
+        EdgeEndpointConfig.from_yaml(filename="a.yaml", yaml_str="global_config: {}")
+
+    with pytest.raises(ValueError, match="filename must be a non-empty path"):
+        EdgeEndpointConfig.from_yaml(filename=" ")
+
+
+def test_edge_endpoint_config_rejects_extra_top_level_fields():
+    """Rejects unknown top-level fields to avoid silent config drift."""
+    with pytest.raises(ValueError, match="Extra inputs are not permitted"):
+        EdgeEndpointConfig.model_validate({"global_config": {}, "unknown_field": True})
+
+
 def test_model_dump_shape_for_edge_endpoint_config():
-    """Serializes full edge endpoint config in flattened wire shape."""
+    """Serializes full edge endpoint config in wire payload shape."""
     config = EdgeEndpointConfig(
         global_config=GlobalConfig(refresh_rate=CUSTOM_REFRESH_RATE, confident_audit_rate=CUSTOM_AUDIT_RATE)
     )
     config.add_detector("det_1", DEFAULT)
     config.add_detector("det_2", EDGE_ANSWERS_WITH_ESCALATION)
     config.add_detector("det_3", NO_CLOUD)
 
-    payload = config.model_dump()
+    payload = config.to_payload()
 
     assert payload["global_config"]["refresh_rate"] == CUSTOM_REFRESH_RATE
     assert payload["global_config"]["confident_audit_rate"] == CUSTOM_AUDIT_RATE
     assert len(payload["detectors"]) == 3  # noqa: PLR2004
     assert set(payload["edge_inference_configs"].keys()) == {"default", "edge_answers_with_escalation", "no_cloud"}
 
 
+def test_edge_endpoint_config_from_payload_round_trip():
+    """Round-trips edge endpoint config through payload helpers."""
+    config = EdgeEndpointConfig()
+    config.add_detector("det_1", DEFAULT)
+    config.add_detector("det_2", NO_CLOUD)
+
+    payload = config.to_payload()
+    reconstructed = EdgeEndpointConfig.from_payload(payload)
+
+    assert reconstructed == config
+
+
 def test_inference_config_validation_errors():
     """Raises on invalid inference config flag combinations and values."""
     with pytest.raises(ValueError, match="disable_cloud_escalation"):