SDK Configures Edge

src/groundlight/experimental_api.py

@@ -40,6 +40,7 @@
 )
 from urllib3.response import HTTPResponse
 
+from groundlight.edge.config import EdgeEndpointConfig
 from groundlight.images import parse_supported_image_types
 from groundlight.internalapi import _generate_request_id
 from groundlight.optional_imports import Image, np
@@ -817,3 +818,19 @@ def make_generic_api_request(  # noqa: PLR0913 # pylint: disable=too-many-argume
             auth_settings=["ApiToken"],
             _preload_content=False,  # This returns the urllib3 response rather than trying any type of processing
         )
+
+    def get_edge_config(self) -> EdgeEndpointConfig:
+        """Retrieve the active edge endpoint configuration.
+
+        Only works when the client is pointed at an edge endpoint
+        (via GROUNDLIGHT_ENDPOINT or the endpoint constructor arg).
+        """
+        from urllib.parse import urlparse, urlunparse
+
+        parsed = urlparse(self.configuration.host)
+        base_url = urlunparse((parsed.scheme, parsed.netloc, "", "", "", ""))
+        url = f"{base_url}/edge-config"
+        headers = self.get_raw_headers()
+        response = requests.get(url, headers=headers, verify=self.configuration.verify_ssl)
+        response.raise_for_status()
+        return EdgeEndpointConfig.from_payload(response.json())

test/unit/test_edge_config.py

@@ -307,3 +307,30 @@ def test_inference_config_validation_errors():
             always_return_edge_prediction=True,
             min_time_between_escalations=-1.0,
         )
+
+
+def test_get_edge_config_parses_response():
+    """ExperimentalApi.get_edge_config() parses the HTTP response into an EdgeEndpointConfig."""
+    from unittest.mock import Mock, patch
+
+    from groundlight import ExperimentalApi
+
+    payload = {
+        "global_config": {"refresh_rate": REFRESH_RATE_SECONDS},
+        "edge_inference_configs": {"default": {"enabled": True}},
+        "detectors": [{"detector_id": "det_1", "edge_inference_config": "default"}],
+    }
+
+    mock_response = Mock()
+    mock_response.json.return_value = payload
+    mock_response.raise_for_status = Mock()
+
+    gl = ExperimentalApi()
+    with patch("requests.get", return_value=mock_response) as mock_get:
+        config = gl.get_edge_config()
+
+    mock_get.assert_called_once()
+    assert isinstance(config, EdgeEndpointConfig)
+    assert config.global_config.refresh_rate == REFRESH_RATE_SECONDS
+    assert config.edge_inference_configs["default"].name == "default"
+    assert [d.detector_id for d in config.detectors] == ["det_1"]