Cache for Image Query ID's Originating from the Edge Endpoint (#17)

* iqe_cache

* add `iqe_` cache

* Automatically reformatting code with black and isort

* edge detector manager class

* Automatically reformatting code with black and isort

* formatting

---------

Co-authored-by: Blaise Munyampirwa <blaise@paleo.positronix.ai>
Co-authored-by: Auto-format Bot <autoformatbot@groundlight.ai>

app/api/routes/image_queries.py

@@ -2,11 +2,17 @@
 from io import BytesIO
 
 import numpy as np
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from model import ImageQuery
 from PIL import Image, ImageFile
 
-from app.core.utils import get_groundlight_sdk_instance, get_motion_detector_instance, prefixed_ksuid, safe_call_api
+from app.core.utils import (
+    get_edge_detector_manager,
+    get_groundlight_sdk_instance,
+    get_motion_detector_instance,
+    prefixed_ksuid,
+    safe_call_api,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +29,7 @@ async def post_image_query(
     request: Request = None,
     gl: Depends = Depends(get_groundlight_sdk_instance),
     motion_detector: Depends = Depends(get_motion_detector_instance),
+    edge_detector_manager: Depends = Depends(get_edge_detector_manager),
 ):
     image = await request.body()
     img = Image.open(BytesIO(image))
@@ -44,9 +51,20 @@ async def post_image_query(
     logger.debug("No motion detected")
     new_image_query = ImageQuery(**motion_detector.image_query_response.dict())
     new_image_query.id = prefixed_ksuid(prefix="iqe_")
+    edge_detector_manager.iqe_cache[new_image_query.id] = new_image_query
+
     return new_image_query
 
 
 @router.get("/{id}", response_model=ImageQuery)
-async def get_image_query(id: str, gl: Depends = Depends(get_groundlight_sdk_instance)):
+async def get_image_query(
+    id: str,
+    gl: Depends = Depends(get_groundlight_sdk_instance),
+    edge_detector_manager: Depends = Depends(get_edge_detector_manager),
+):
+    if id.startswith("iqe_"):
+        image_query = edge_detector_manager.iqe_cache.get(id, None)
+        if not image_query:
+            raise HTTPException(status_code=404, detail=f"Image query with ID {id} not found")
+        return image_query
     return safe_call_api(gl.get_image_query, id=id)

app/core/edge_detector_manager.py

@@ -0,0 +1,11 @@
+class EdgeDetectorManager:
+    """
+    Fow now this class is just a container for the IQE cache.
+    The cache allows us to control the SDK's polling behavior for image queries
+    without having to change the SDK itself.
+    """
+
+    def __init__(self) -> None:
+        # Cache for image query responses whose IDs start with "iqe_". This is needed
+        # because the cloud API does not currently recognize these IDs.
+        self.iqe_cache = {}

app/core/motion_detection.py

@@ -68,7 +68,7 @@ async def motion_detected(self, new_img: np.ndarray) -> bool:
             current_time = time.monotonic()
             if current_time - self._previous_motion_detection_time > self._max_time_between_images:
                 self._previous_motion_detection_time = current_time
-                logger.debug("Maximum time between images exceeded")
+                logger.debug("Maximum time between cloud-submitted images exceeded")
                 return True
 
         motion_is_detected = await asyncio.to_thread(self._motion_detector.motion_detected, new_img)

app/core/utils.py

@@ -14,6 +14,10 @@ def get_motion_detector_instance(request: Request):
     return request.app.state.motion_detector
 
 
+def get_edge_detector_manager(request: Request):
+    return request.app.state.edge_detector_manager
+
+
 def safe_call_api(api_method: Callable, **kwargs):
     """
     This ensures that we correctly handle HTTP error status codes. In some cases,

app/main.py

@@ -7,6 +7,7 @@
 from app.api.api import api_router, ping_router
 from app.api.naming import API_BASE_PATH
 
+from .core.edge_detector_manager import EdgeDetectorManager
 from .core.motion_detection import AsyncMotionDetector, MotdetParameterSettings
 
 LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
@@ -21,7 +22,8 @@
 # Create global shared Groundlight SDK client object in the app's state
 app.state.groundlight = Groundlight()
 
-parameters = MotdetParameterSettings()
-
 # Create global shared motion detector object in the app's state
 app.state.motion_detector = AsyncMotionDetector(parameters=MotdetParameterSettings())
+
+# Create global shared edge detector manager object in the app's state
+app.state.edge_detector_manager = EdgeDetectorManager()