SDK Configures Edge

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 EdgeAPI
 from .config import (
     DEFAULT,
     DISABLED,
@@ -14,6 +15,7 @@
     "DEFAULT",
     "DISABLED",
     "EDGE_ANSWERS_WITH_ESCALATION",
+    "EdgeAPI",
     "NO_CLOUD",
     "DetectorsConfig",
     "DetectorConfig",

src/groundlight/edge/api.py

@@ -0,0 +1,84 @@
+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 EdgeAPI:
+    """Namespace for edge-endpoint operations, accessed via ``gl.edge``."""
+
+    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 d.detector_id}
+        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

@@ -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 EdgeAPI
 from groundlight.images import parse_supported_image_types
 from groundlight.internalapi import _generate_request_id
 from groundlight.optional_imports import Image, np
@@ -103,6 +105,7 @@ def __init__(
         self.detector_reset_api = DetectorResetApi(self.api_client)
 
         self.edge_api = EdgeApi(self.api_client)
+        self.edge = EdgeAPI(self)
 
     ITEMS_PER_PAGE = 100
 
@@ -817,3 +820,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, "", "", "", ""))