Migrate Edge Configuration to SDK (#413)

timmarkhuff · Auto-format Bot · Tim Huff · web-flow · commit 414ff4283172 · 2026-03-17T21:24:42.000-07:00
Moves the edge endpoint config schema from `edge-endpoint` into the
Python SDK so users can programmatically build, validate, serialize, and
deserialize edge configurations.

## What's new

- New `groundlight.edge` config models ported from edge-endpoint:
  - `EdgeEndpointConfig` (top-level config with `global_config`)
  - `DetectorsConfig` (detector-scoped workflows)
  - `InferenceConfig` (immutable/frozen per-detector inference behavior)
  - `GlobalConfig`, `DetectorConfig` (supporting models)
- Preset constants aligned with edge-endpoint defaults:
  - `DEFAULT`
  - `EDGE_ANSWERS_WITH_ESCALATION`
  - `NO_CLOUD`
  - `DISABLED`
- Strict validation added with `extra="forbid"` across config models.

## Model design

- Shared detector/inference behavior is implemented once in internal
`ConfigBase`.
- `DetectorsConfig` and `EdgeEndpointConfig` both inherit from
`ConfigBase`.
- `EdgeEndpointConfig` is top-level and flat (`global_config`,
`edge_inference_configs`, `detectors`).
- `DetectorsConfig` remains available as a detector-only model.

## Validation behavior

- `disable_cloud_escalation=True` requires
`always_return_edge_prediction=True`.
- Duplicate detector IDs are rejected.
- Every detector's `edge_inference_config` must exist in
`edge_inference_configs`.
- Inference config dict-key/name mismatch is rejected.
- For payload dict input, `InferenceConfig.name` is hydrated from map
keys.

## Serialization / parsing API

- `EdgeEndpointConfig.from_yaml(filename=..., yaml_str=...)`
  - Accepts exactly one input source.
  - Supports file-path and YAML-string workflows explicitly.
- `EdgeEndpointConfig.to_payload()`
  - Returns payload dict via `model_dump()`.
- `EdgeEndpointConfig.from_payload(payload)`
  - Reconstructs from payload dict via `model_validate()`.
- `DetectorsConfig.to_payload()`
  - Returns detector-scoped payload dict via `model_dump()`.

## Usage

```python
from groundlight.edge import (
    DEFAULT,
    EDGE_ANSWERS_WITH_ESCALATION,
    GlobalConfig,
    NO_CLOUD,
    DetectorsConfig,
    EdgeEndpointConfig,
)

# Full edge endpoint config
full_config = EdgeEndpointConfig(
    global_config=GlobalConfig(refresh_rate=30.0, confident_audit_rate=0.0),
)
full_config.add_detector("det_abc123", NO_CLOUD)
full_config.add_detector("det_def456", DEFAULT)

payload = full_config.to_payload()
round_trip = EdgeEndpointConfig.from_payload(payload)

# Detector-only config
detectors_config = DetectorsConfig()
detectors_config.add_detector("det_abc123", NO_CLOUD)
detectors_config.add_detector("det_xyz789", EDGE_ANSWERS_WITH_ESCALATION)
detectors_payload = detectors_config.to_payload()
```

## Tests

Updated unit coverage in `test/unit/test_edge_config.py` includes:

- add-detector behavior and conflict handling
- constructor-time duplicate/cross-reference validation
- hydration of inference-config names from payload keys
- top-level payload acceptance and unknown-field rejection
- `from_yaml` (yaml string + filename + argument-validation cases)
- `to_payload` / `from_payload` round-trip
- direct literal-payload `from_payload` constructor coverage
- `InferenceConfig` validation constraints

## Dependencies

- `pyyaml` added as a runtime dependency in `python-sdk`
(`pyproject.toml`).

---------

Co-authored-by: Auto-format Bot &lt;autoformatbot@groundlight.ai&gt;
Co-authored-by: Tim Huff &lt;thuff@axon.com&gt;
diff --git a/pyproject.toml b/pyproject.toml
@@ -9,7 +9,7 @@ packages = [
     {include = "**/*.py", from = "src"},
 ]
 readme = "README.md"
-version = "0.24.0"
+version = "0.25.0"
 
 [tool.poetry.dependencies]
 # For certifi, use ">=" instead of "^" since it upgrades its "major version" every year, not really following semver
@@ -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"
diff --git a/src/groundlight/edge/__init__.py b/src/groundlight/edge/__init__.py
@@ -0,0 +1,23 @@
+from .config import (
+    DEFAULT,
+    DISABLED,
+    EDGE_ANSWERS_WITH_ESCALATION,
+    NO_CLOUD,
+    DetectorConfig,
+    DetectorsConfig,
+    EdgeEndpointConfig,
+    GlobalConfig,
+    InferenceConfig,
+)
+
+__all__ = [
+    "DEFAULT",
+    "DISABLED",
+    "EDGE_ANSWERS_WITH_ESCALATION",
+    "NO_CLOUD",
+    "DetectorsConfig",
+    "DetectorConfig",
+    "EdgeEndpointConfig",
+    "GlobalConfig",
+    "InferenceConfig",
+]
diff --git a/src/groundlight/edge/config.py b/src/groundlight/edge/config.py
@@ -0,0 +1,215 @@
+from typing import Any, Optional, Union
+
+import yaml
+from model import Detector
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from typing_extensions import Self
+
+
+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.",
+    )
+    confident_audit_rate: float = Field(
+        default=1e-5,  # A detector running at 1 FPS = ~100,000 IQ/day, so 1e-5 is ~1 confident IQ/day audited
+        description="The probability that any given confident prediction will be sent to the cloud for auditing.",
+    )
+
+
+class InferenceConfig(BaseModel):
+    """
+    Configuration for edge inference on a specific detector.
+    """
+
+    # Keep shared presets immutable (DEFAULT/NO_CLOUD/etc.) so one mutation cannot globally change behavior.
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    name: str = Field(..., exclude=True, description="A unique name for this inference config preset.")
+    enabled: bool = Field(
+        default=True, description="Whether the edge endpoint should accept image queries for this detector."
+    )
+    api_token: Optional[str] = Field(
+        default=None, description="API token used to fetch the inference model for this detector."
+    )
+    always_return_edge_prediction: bool = Field(
+        default=False,
+        description=(
+            "Indicates if the edge-endpoint should always provide edge ML predictions, regardless of confidence. "
+            "When this setting is true, whether or not the edge-endpoint should escalate low-confidence predictions "
+            "to the cloud is determined by `disable_cloud_escalation`."
+        ),
+    )
+    disable_cloud_escalation: bool = Field(
+        default=False,
+        description=(
+            "Never escalate ImageQueries from the edge-endpoint to the cloud. "
+            "Requires `always_return_edge_prediction=True`."
+        ),
+    )
+    min_time_between_escalations: float = Field(
+        default=2.0,
+        description=(
+            "The minimum time (in seconds) to wait between cloud escalations for a given detector. "
+            "Cannot be less than 0.0. "
+            "Only applies when `always_return_edge_prediction=True` and `disable_cloud_escalation=False`."
+        ),
+    )
+
+    @model_validator(mode="after")
+    def validate_configuration(self) -> Self:
+        if self.disable_cloud_escalation and not self.always_return_edge_prediction:
+            raise ValueError(
+                "The `disable_cloud_escalation` flag is only valid when `always_return_edge_prediction` is set to True."
+            )
+        if self.min_time_between_escalations < 0.0:
+            raise ValueError("`min_time_between_escalations` cannot be less than 0.0.")
+        return self
+
+
+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 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):
+        """
+        Validates detector config state.
+        Raises ValueError if dict keys mismatch InferenceConfig.name, detector IDs are duplicated,
+        or any detector references an undefined inference config.
+        """
+        for name, config in self.edge_inference_configs.items():
+            if name != config.name:
+                raise ValueError(f"Edge inference config key '{name}' must match InferenceConfig.name '{config.name}'.")
+
+        seen_detector_ids = set()
+        duplicate_detector_ids = set()
+        for detector_config in self.detectors:
+            detector_id = detector_config.detector_id
+            if detector_id in seen_detector_ids:
+                duplicate_detector_ids.add(detector_id)
+            else:
+                seen_detector_ids.add(detector_id)
+        if duplicate_detector_ids:
+            duplicates = ", ".join(sorted(duplicate_detector_ids))
+            raise ValueError(f"Duplicate detector IDs are not allowed: {duplicates}.")
+
+        for detector_config in self.detectors:
+            if detector_config.edge_inference_config not in self.edge_inference_configs:
+                raise ValueError(f"Edge inference config '{detector_config.edge_inference_config}' not defined.")
+        return self
+
+    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."""
+        detector_id = detector.id if isinstance(detector, Detector) else detector
+        if any(existing.detector_id == detector_id for existing in self.detectors):
+            raise ValueError(f"A detector with ID '{detector_id}' already exists.")
+
+        existing = self.edge_inference_configs.get(edge_inference_config.name)
+        if existing is None:
+            self.edge_inference_configs[edge_inference_config.name] = edge_inference_config
+        elif existing != edge_inference_config:
+            raise ValueError(
+                f"A different inference config named '{edge_inference_config.name}' is already registered."
+            )
+
+        self.detectors.append(DetectorConfig(detector_id=detector_id, edge_inference_config=edge_inference_config.name))
+
+    def to_payload(self) -> dict[str, Any]:
+        """Return this config as a payload dictionary."""
+        return self.model_dump()
+
+
+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)
+
+    @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)
+
+    @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")
+EDGE_ANSWERS_WITH_ESCALATION = InferenceConfig(
+    name="edge_answers_with_escalation",
+    always_return_edge_prediction=True,
+    min_time_between_escalations=2.0,
+)
+NO_CLOUD = InferenceConfig(
+    name="no_cloud",
+    always_return_edge_prediction=True,
+    disable_cloud_escalation=True,
+)
+DISABLED = InferenceConfig(name="disabled", enabled=False)
diff --git a/test/unit/test_edge_config.py b/test/unit/test_edge_config.py
