Switch to edge config models defined in python-sdk

app/core/edge_config_loader.py

@@ -2,75 +2,51 @@
 import os
 from typing import Dict
 
-import yaml
+from groundlight.edge import EdgeEndpointConfig, InferenceConfig
 
-from .configs import EdgeInferenceConfig, RootEdgeConfig
 from .file_paths import DEFAULT_EDGE_CONFIG_PATH
 
 logger = logging.getLogger(__name__)
 
 
-def load_edge_config() -> RootEdgeConfig:
+def load_edge_config() -> EdgeEndpointConfig:
     """
     Reads the edge config from the EDGE_CONFIG environment variable if it exists.
     If EDGE_CONFIG is not set, reads the default edge config file.
     """
     yaml_config = os.environ.get("EDGE_CONFIG", "").strip()
     if yaml_config:
-        return _load_config_from_yaml(yaml_config)
+        return EdgeEndpointConfig.from_yaml(yaml_str=yaml_config)
 
-    logger.warning("EDGE_CONFIG environment variable not set. Checking default locations.")
-
-    if os.path.exists(DEFAULT_EDGE_CONFIG_PATH):
-        logger.info(f"Loading edge config from {DEFAULT_EDGE_CONFIG_PATH}")
-        with open(DEFAULT_EDGE_CONFIG_PATH, "r") as f:
-            return _load_config_from_yaml(f)
-
-    raise FileNotFoundError(f"Could not find edge config file in default location: {DEFAULT_EDGE_CONFIG_PATH}")
-
-
-def _load_config_from_yaml(yaml_config) -> RootEdgeConfig:
-    """
-    Creates a `RootEdgeConfig` from the config yaml. Raises an error if there are duplicate detector ids.
-    """
-    config = yaml.safe_load(yaml_config)
-
-    detectors = config.get("detectors", [])
-    detector_ids = [det["detector_id"] for det in detectors]
-
-    # Check for duplicate detector IDs
-    if len(detector_ids) != len(set(detector_ids)):
-        raise ValueError("Duplicate detector IDs found in the configuration. Each detector should only have one entry.")
-
-    config["detectors"] = {det["detector_id"]: det for det in detectors}
-
-    return RootEdgeConfig(**config)
+    logger.info(
+        f"EDGE_CONFIG environment variable not set. Checking default Edge Config path: {DEFAULT_EDGE_CONFIG_PATH}."
+    )
+    return EdgeEndpointConfig.from_yaml(filename=DEFAULT_EDGE_CONFIG_PATH)
 
 
 def get_detector_inference_configs(
-    root_edge_config: RootEdgeConfig,
-) -> dict[str, EdgeInferenceConfig] | None:
+    root_edge_config: EdgeEndpointConfig,
+) -> dict[str, InferenceConfig] | None:
     """
-    Produces a dict mapping detector IDs to their associated `EdgeInferenceConfig`.
+    Produces a dict mapping detector IDs to their associated `InferenceConfig`.
     Returns None if there are no detectors in the config file.
     """
-    # Mapping of config names to EdgeInferenceConfig objects
-    edge_inference_configs: dict[str, EdgeInferenceConfig] = root_edge_config.edge_inference_configs
+    # Mapping of config names to InferenceConfig objects
+    edge_inference_configs: dict[str, InferenceConfig] = root_edge_config.edge_inference_configs
 
-    # Filter out detectors whose ID's are empty strings
-    detectors = {det_id: detector for det_id, detector in root_edge_config.detectors.items() if det_id != ""}
+    # Filter out detectors whose IDs are empty strings.
+    detectors = [detector for detector in root_edge_config.detectors if detector.detector_id != ""]
 
-    detector_to_inference_config: dict[str, EdgeInferenceConfig] | None = None
+    detector_to_inference_config: dict[str, InferenceConfig] | None = None
     if detectors:
         detector_to_inference_config = {
-            detector_id: edge_inference_configs[detector_config.edge_inference_config]
-            for detector_id, detector_config in detectors.items()
+            detector.detector_id: edge_inference_configs[detector.edge_inference_config] for detector in detectors
         }
 
     return detector_to_inference_config
 
 
-def get_detector_edge_configs_by_id() -> Dict[str, EdgeInferenceConfig]:
+def get_detector_edge_configs_by_id() -> Dict[str, InferenceConfig]:
     """
     Convenience helper that loads the edge config and returns detector-level inference configs,
     defaulting to an empty dict when none are defined.

app/core/edge_inference.py

@@ -10,10 +10,10 @@
 import yaml
 from cachetools import TTLCache, cached
 from fastapi import HTTPException, status
+from groundlight.edge import InferenceConfig
 from jinja2 import Template
 from model import ModeEnum
 
-from app.core.configs import EdgeInferenceConfig
 from app.core.file_paths import MODEL_REPOSITORY_PATH
 from app.core.speedmon import SpeedMonitor
 from app.core.utils import ModelInfoBase, ModelInfoWithBinary, parse_model_info
@@ -229,15 +229,14 @@ class EdgeInferenceManager:
 
     def __init__(
         self,
-        detector_inference_configs: dict[str, EdgeInferenceConfig] | None,
+        detector_inference_configs: dict[str, InferenceConfig] | None,
         verbose: bool = False,
         separate_oodd_inference: bool = True,
     ) -> None:
         """
         Initializes the edge inference manager.
         Args:
-            detector_inference_configs: Dictionary of detector IDs to EdgeInferenceConfig objects
-            edge_config: RootEdgeConfig object
+            detector_inference_configs: Dictionary of detector IDs to InferenceConfig objects
             verbose: Whether to print verbose logs from the inference server client
             separate_oodd_inference: Whether to run inference separately for the OODD model
         """
@@ -277,7 +276,9 @@ def update_inference_config(self, detector_id: str, api_token: str) -> None:
 
         """
         if detector_id not in self.detector_inference_configs.keys():
-            self.detector_inference_configs[detector_id] = EdgeInferenceConfig(enabled=True, api_token=api_token)
+            self.detector_inference_configs[detector_id] = InferenceConfig(
+                name="runtime_detector_config", enabled=True, api_token=api_token
+            )
             self.inference_client_urls[detector_id] = get_edge_inference_service_name(detector_id) + ":8000"
             if self.separate_oodd_inference:
                 logger.info(f"Performing separate OODD inference, updating OODD inference URL for {detector_id}")

app/metrics/system_metrics.py

@@ -6,18 +6,18 @@
 import psutil
 import tzlocal
 import yaml
+from groundlight.edge import InferenceConfig
 from kubernetes import client, config
 
-from app.core.configs import EdgeInferenceConfig
 from app.core.edge_config_loader import get_detector_edge_configs_by_id
 from app.core.edge_inference import get_current_pipeline_config, get_predictor_metadata, get_primary_edge_model_dir
 from app.core.file_paths import MODEL_REPOSITORY_PATH
 
 logger = logging.getLogger(__name__)
 
 
-def _edge_config_to_dict(config: EdgeInferenceConfig | None) -> dict | None:
-    """Convert an EdgeInferenceConfig to a plain dict for JSON serialization."""
+def _edge_config_to_dict(config: InferenceConfig | None) -> dict | None:
+    """Convert an InferenceConfig to a plain dict for JSON serialization."""
     if config is None:
         return None
     return {

app/model_updater/update_models.py

@@ -2,7 +2,8 @@
 import os
 import time
 
-from app.core.configs import RootEdgeConfig
+from groundlight.edge import EdgeEndpointConfig
+
 from app.core.database import DatabaseManager
 from app.core.edge_config_loader import get_detector_inference_configs, load_edge_config
 from app.core.edge_inference import (
@@ -234,7 +235,7 @@ def manage_update_models(
 if __name__ == "__main__":
     logger.info("Starting model updater.")
 
-    edge_config: RootEdgeConfig = load_edge_config()
+    edge_config: EdgeEndpointConfig = load_edge_config()
     logger.info(f"{edge_config=}")
 
     refresh_rate = edge_config.global_config.refresh_rate

pyproject.toml

@@ -9,7 +9,7 @@ license = "MIT"
 APScheduler = "3.10.4"
 cachetools = "^5.3.1"
 fastapi = "^0.115.0"
-groundlight = ">=0.23.3, <0.24.0"
+groundlight = ">=0.25.0, <0.26.0"
 httpx = "^0.27.2"
 jinja2 = "^3.1.6"
 kubernetes = "^27.2.0"