Add recording and edge filters to get_coverage

foxglove/client/api.py

@@ -874,36 +874,52 @@ def delete_site_token(self, *, site_token_id: str):
     def get_coverage(
         self,
         *,
-        start: datetime.datetime,
-        end: datetime.datetime,
+        start: Optional[datetime.datetime] = None,
+        end: Optional[datetime.datetime] = None,
         device_id: Optional[str] = None,
         device_name: Optional[str] = None,
         tolerance: Optional[int] = None,
         project_id: Optional[str] = None,
         session_id: Optional[str] = None,
         session_key: Optional[str] = None,
+        recording_id: Optional[str] = None,
+        recording_key: Optional[str] = None,
+        include_edge_recordings: Optional[bool] = None,
     ):
         """
         List coverage ranges for data.
 
-        :param start: The earliest time after which to retrieve data.
-        :param end: The latest time before which to retrieve data.
+        :param start: The earliest time after which to retrieve data. Required with
+            ``end`` unless filtering by recording or session.
+        :param end: The latest time before which to retrieve data. Required with
+            ``start`` unless filtering by recording or session.
         :param device_id: Optional device id to limit data by.
         :param tolerance: Minimum interval (in seconds) that ranges must be separated by
             to be considered discrete.
         :param project_id: Optional Project to filter coverage by.
         :param session_id: Optional Session ID to filter coverage by.
         :param session_key: Optional Session key to filter coverage by.
+        :param recording_id: Optional recording ID to filter coverage by.
+        :param recording_key: Optional recording key to filter coverage by.
+        :param include_edge_recordings: Optionally include recordings from an Edge Site
+            or a device running Foxlet. Ranges then include ``import_status``.
         """
         params = {
             "deviceId": device_id,
             "deviceName": device_name,
             "tolerance": tolerance,
-            "start": start.astimezone().isoformat(),
-            "end": end.astimezone().isoformat(),
+            "start": start.astimezone().isoformat() if start else None,
+            "end": end.astimezone().isoformat() if end else None,
             "projectId": project_id,
             "sessionId": session_id,
             "sessionKey": session_key,
+            "recordingId": recording_id,
+            "recordingKey": recording_key,
+            "includeEdgeRecordings": (
+                bool_query_param(include_edge_recordings)
+                if include_edge_recordings is not None
+                else None
+            ),
         }
         response = self.__session.get(
             self.__url__("/v1/data/coverage"),
@@ -917,6 +933,7 @@ def get_coverage(
                 "device": c.get("device"),
                 "start": arrow.get(c["start"]).datetime,
                 "end": arrow.get(c["end"]).datetime,
+                "import_status": c.get("importStatus"),
             }
             for c in json
         ]

tests/test_coverage.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 
 import responses
+from responses.matchers import query_param_matcher
 from faker import Faker
 from foxglove.client import Client
 
@@ -43,3 +44,29 @@ def test_get_coverage():
     )
     assert len(coverage) == 2
     assert coverage[0]["device_id"] == device_id
+
+
+@responses.activate
+def test_get_coverage_by_recording_with_edge_recordings():
+    responses.add(
+        responses.GET,
+        api_url("/v1/data/coverage"),
+        match=[
+            query_param_matcher(
+                {"recordingKey": "run.mcap", "includeEdgeRecordings": "true"}
+            )
+        ],
+        json=[
+            {
+                "deviceId": "dev_1",
+                "device": {"id": "dev_1", "name": "robot"},
+                "start": datetime(2026, 1, 1).isoformat(),
+                "end": datetime(2026, 1, 2).isoformat(),
+                "importStatus": "pending",
+            }
+        ],
+    )
+    coverage = Client("test").get_coverage(
+        recording_key="run.mcap", include_edge_recordings=True
+    )
+    assert coverage[0]["import_status"] == "pending"