SDK Configures Edge

docs/docs/guide/8-edge.md

@@ -38,6 +38,46 @@ python your_app.py
 In the above example, the `edge-endpoint` is running on the same machine as the application, so the endpoint URL is `http://localhost:30101`. If the `edge-endpoint` is running on a different machine, you should replace `localhost` with the IP address or hostname of the machine running the `edge-endpoint`.
 :::
 
+## Configuring the Edge Endpoint at Runtime
+
+:::note
+Runtime edge configuration is currently in beta and available through the `ExperimentalApi`.
+:::
+
+You can programmatically configure which detectors run on the edge and how they behave, without redeploying. 
+
+This allows applications to define the desired state of the Edge Endpoint, thereby eliminating the need to manually configure the Edge Endpoint separately.
+
+```python notest
+from groundlight import ExperimentalApi
+from groundlight.edge import EdgeEndpointConfig
+from groundlight.edge.config import NO_CLOUD, EDGE_ANSWERS_WITH_ESCALATION
+
+# Connect to an Edge Endpoint
+gl = ExperimentalApi(endpoint="http://localhost:30101")
+
+# Build a configuration with detectors and inference presets
+config = EdgeEndpointConfig()
+config.add_detector("det_YOUR_DETECTOR_ID_HERE_01", NO_CLOUD)
+config.add_detector("det_YOUR_DETECTOR_ID_HERE_02", EDGE_ANSWERS_WITH_ESCALATION)
+
+# Apply the configuration and wait for detectors to be ready
+print("Applying configuration...")
+config = gl.edge.set_config(config)
+print(f"Applied config with {len(config.detectors)} detector(s)")
+```
+
+`set_config` replaces the current configuration and blocks until all detectors have inference pods ready to serve requests (or until the timeout expires).
+
+You can also inspect the current configuration:
+
+```python notest
+# Retrieve the active configuration
+config = gl.edge.get_config()
+for det in config.detectors:
+    print(f"  {det.detector_id} -> {det.edge_inference_config}")
+```
+
 ## Edge Endpoint performance
 
 We have benchmarked the `edge-endpoint` handling 500 requests/sec at a latency of less than 50ms on an off-the-shelf [Katana 15 B13VGK-1007US](https://us.msi.com/Laptop/Katana-15-B13VX/Specification) laptop (Intel® Core™ i9-13900H CPU, NVIDIA® GeForce RTX™ 4070 Laptop GPU, 32GB DDR5 5200MHz RAM) running Ubuntu 20.04.

pyproject.toml

@@ -9,7 +9,7 @@ packages = [
     {include = "**/*.py", from = "src"},
 ]
 readme = "README.md"
-version = "0.25.1"
+version = "0.26.0"
 
 [tool.poetry.dependencies]
 # For certifi, use ">=" instead of "^" since it upgrades its "major version" every year, not really following semver

src/groundlight/__init__.py

@@ -7,7 +7,7 @@
 
 # Imports from our code
 from .client import Groundlight
-from .client import GroundlightClientError, ApiTokenError, NotFoundError
+from .client import GroundlightClientError, ApiTokenError, EdgeNotAvailableError, NotFoundError
 from .experimental_api import ExperimentalApi
 from .binary_labels import Label
 from .version import get_version

src/groundlight/client.py

@@ -69,6 +69,10 @@ class ApiTokenError(GroundlightClientError):
     pass
 
 
+class EdgeNotAvailableError(GroundlightClientError):
+    """Raised when an edge-only method is called against a non-edge endpoint."""
+
+
 class Groundlight:  # pylint: disable=too-many-instance-attributes,too-many-public-methods
     """
     Client for accessing the Groundlight cloud service. Provides methods to create visual detectors,

src/groundlight/edge/__init__.py

@@ -1,3 +1,4 @@
+from .api import EdgeEndpointApi
 from .config import (
     DEFAULT,
     DISABLED,
@@ -14,6 +15,7 @@
     "DEFAULT",
     "DISABLED",
     "EDGE_ANSWERS_WITH_ESCALATION",
+    "EdgeEndpointApi",
     "NO_CLOUD",
     "DetectorsConfig",
     "DetectorConfig",

src/groundlight/edge/api.py

@@ -0,0 +1,87 @@
+import time
+from http import HTTPStatus
+
+import requests
+
+from groundlight.client import EdgeNotAvailableError
+from groundlight.edge.config import EdgeEndpointConfig
+
+_EDGE_METHOD_UNAVAILABLE_HINT = (
+    "Make sure the client is pointed at a running Edge Endpoint "
+    "(via GROUNDLIGHT_ENDPOINT env var or the endpoint= constructor arg)."
+)
+
+
+class EdgeEndpointApi:
+    """
+    Namespace for operations that are specific to the Edge Endpoint,
+    such as setting and getting the EdgeEndpoint configuration.
+    """
+
+    def __init__(self, client) -> None:
+        self._client = client
+
+    def _base_url(self) -> str:
+        return self._client.edge_base_url()
+
+    def _request(self, method: str, path: str, **kwargs) -> requests.Response:
+        url = f"{self._base_url()}{path}"
+        headers = self._client.get_raw_headers()
+        try:
+            response = requests.request(
+                method, url, headers=headers, verify=self._client.configuration.verify_ssl, timeout=10, **kwargs
+            )
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            if e.response is not None and e.response.status_code == HTTPStatus.NOT_FOUND:
+                raise EdgeNotAvailableError(
+                    f"Edge method not available at {url}. {_EDGE_METHOD_UNAVAILABLE_HINT}"
+                ) from e
+            raise
+        except requests.exceptions.ConnectionError as e:
+            raise EdgeNotAvailableError(
+                f"Could not connect to {self._base_url()}. {_EDGE_METHOD_UNAVAILABLE_HINT}"
+            ) from e
+        return response
+
+    def get_config(self) -> EdgeEndpointConfig:
+        """Retrieve the active edge endpoint configuration."""
+        response = self._request("GET", "/edge-config")
+        return EdgeEndpointConfig.from_payload(response.json())
+
+    def get_detector_readiness(self) -> dict[str, bool]:
+        """Check which configured detectors have inference pods ready to serve.
+
+        :return: Dict mapping detector_id to readiness (True/False).
+        """
+        response = self._request("GET", "/edge-detector-readiness")
+        return {det_id: info["ready"] for det_id, info in response.json().items()}
+
+    def set_config(
+        self,
+        config: EdgeEndpointConfig,
+        timeout_sec: float = 600,
+    ) -> EdgeEndpointConfig:
+        """Replace the edge endpoint configuration and wait until all detectors are ready.
+
+        :param config: The new configuration to apply.
+        :param timeout_sec: Max seconds to wait for all detectors to become ready.
+        :return: The applied configuration as reported by the edge endpoint.
+        """
+        self._request("PUT", "/edge-config", json=config.to_payload())
+
+        desired_ids = {d.detector_id for d in config.detectors}
+        if not desired_ids:
+            return self.get_config()
+
+        deadline = time.time() + timeout_sec
+        while time.time() < deadline:
+            readiness = self.get_detector_readiness()
+            if all(readiness.get(did, False) for did in desired_ids):
+                return self.get_config()
+            time.sleep(1)
+
+        raise TimeoutError(
+            f"Edge detectors were not all ready within {timeout_sec}s. "
+            "The edge endpoint may still be converging, or may have encountered an error."
+        )

src/groundlight/edge/config.py

@@ -13,10 +13,12 @@ class GlobalConfig(BaseModel):  # pylint: disable=too-few-public-methods
 
     refresh_rate: float = Field(
         default=60.0,
+        gt=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
+        ge=0,
         description="The probability that any given confident prediction will be sent to the cloud for auditing.",
     )
 
@@ -29,7 +31,7 @@ class InferenceConfig(BaseModel):  # pylint: disable=too-few-public-methods
     # Keep shared presets immutable (DEFAULT/NO_CLOUD/etc.) so one mutation cannot globally change behavior.
     model_config = ConfigDict(extra="ignore", frozen=True)
 
-    name: str = Field(..., exclude=True, description="A unique name for this inference config preset.")
+    name: str = Field(..., min_length=1, 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."
     )
@@ -53,9 +55,9 @@ class InferenceConfig(BaseModel):  # pylint: disable=too-few-public-methods
     )
     min_time_between_escalations: float = Field(
         default=2.0,
+        gt=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`."
         ),
     )
@@ -66,8 +68,6 @@ def validate_configuration(self) -> Self:
             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
 
 
@@ -78,7 +78,7 @@ class DetectorConfig(BaseModel):  # pylint: disable=too-few-public-methods
 
     model_config = ConfigDict(extra="ignore")
 
-    detector_id: str = Field(..., description="Detector ID")
+    detector_id: str = Field(..., pattern=r"^det_[A-Za-z0-9]{27}$", description="Detector ID")
     edge_inference_config: str = Field(..., description="Config for edge inference.")
 
 

src/groundlight/experimental_api.py

@@ -11,6 +11,7 @@
 from io import BufferedReader, BytesIO
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
+from urllib.parse import urlparse, urlunparse
 
 import requests
 from groundlight_openapi_client.api.actions_api import ActionsApi
@@ -40,6 +41,7 @@
 )
 from urllib3.response import HTTPResponse
 
+from groundlight.edge.api import EdgeEndpointApi
 from groundlight.images import parse_supported_image_types
 from groundlight.internalapi import _generate_request_id
 from groundlight.optional_imports import Image, np
@@ -102,7 +104,11 @@ def __init__(
         self.detector_group_api = DetectorGroupsApi(self.api_client)
         self.detector_reset_api = DetectorResetApi(self.api_client)
 
-        self.edge_api = EdgeApi(self.api_client)
+        # API client for fetching Edge models
+        self._edge_model_download_api = EdgeApi(self.api_client)
+
+        # API client for interacting with the EdgeEndpoint (getting/setting configuration, etc.)
+        self.edge = EdgeEndpointApi(self)
 
     ITEMS_PER_PAGE = 100
 
@@ -704,7 +710,7 @@ def _download_mlbinary_url(self, detector: Union[str, Detector]) -> EdgeModelInf
         """
         if isinstance(detector, Detector):
             detector = detector.id
-        obj = self.edge_api.get_model_urls(detector)
+        obj = self._edge_model_download_api.get_model_urls(detector)
         return EdgeModelInfo.parse_obj(obj.to_dict())
 
     def download_mlbinary(self, detector: Union[str, Detector], output_dir: str) -> None:
@@ -817,3 +823,8 @@ 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 edge_base_url(self) -> str:
+        """Return the scheme+host+port of the configured endpoint, without the /device-api path."""
+        parsed = urlparse(self.configuration.host)
+        return urlunparse((parsed.scheme, parsed.netloc, "", "", "", ""))