diff --git a/blinkpy/api.py b/blinkpy/api.py index 24f2b9f8..1dbb9c40 100644 --- a/blinkpy/api.py +++ b/blinkpy/api.py @@ -380,6 +380,26 @@ async def request_videos(blink, time=None, page=0): return await http_get(blink, url) +async def request_videos_v4(blink, page_key=None): + """ + Fetch media via the v4 endpoint which includes AI descriptions. + + Uses POST /api/v4/accounts/{id}/media. Returns richer data than the v1 + endpoint, including ai_vd (AI video descriptions) and top-level + cv_detection fields. + + :param blink: Blink instance. + :param page_key: Optional pagination key from previous response. + """ + url = f"{blink.urls.base_url}/api/v4/accounts/{blink.account_id}/media" + if page_key: + query = urlencode({"pagination_key": page_key}) + url = f"{url}?{query}" + + body = dumps({"filters": {}}) + return await http_post(blink, url, data=body) + + async def request_cameras(blink, network): """ Request all camera information. diff --git a/blinkpy/camera.py b/blinkpy/camera.py index d8332e67..fdcc905d 100644 --- a/blinkpy/camera.py +++ b/blinkpy/camera.py @@ -43,6 +43,9 @@ def __init__(self, sync): self.motion_detected = None self.wifi_strength = None self.last_record = None + self.ai_description = None + self.ai_description_short = None + self.cv_detection = None self._cached_image = None self._cached_video = None self.camera_type = "" @@ -74,6 +77,9 @@ def attributes(self): "sync_signal_strength": self.sync_signal_strength, "last_record": self.last_record, "type": self.product_type, + "ai_description": self.ai_description, + "ai_description_short": self.ai_description_short, + "cv_detection": self.cv_detection, } return attributes @@ -341,10 +347,22 @@ def timesort(record): last_records = sorted(self.sync.last_records[self.name], key=timesort) for rec in last_records: clip_addr = rec["clip"] - self.clip = f"{self.sync.urls.base_url}{clip_addr}" + if clip_addr and clip_addr.startswith("http"): + self.clip = clip_addr + else: + self.clip = f"{self.sync.urls.base_url}{clip_addr}" self.last_record = rec["time"] + # Update AI description and CV detection from latest record + self.ai_description = rec.get("ai_description") + self.ai_description_short = rec.get("ai_description_short") + self.cv_detection = rec.get("cv_detection") if self.motion_detected: - recent = {"time": self.last_record, "clip": self.clip} + recent = { + "time": self.last_record, + "clip": self.clip, + "ai_description": self.ai_description, + "cv_detection": self.cv_detection, + } # Prevent duplicates. if recent not in self.recent_clips: self.recent_clips.append(recent) diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 63b6aef6..fc667a57 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -311,7 +311,7 @@ async def check_new_videos(self): _LOGGER.info("No new videos since last refresh.") return False - resp = await api.request_videos(self.blink, time=interval, page=1) + resp = await api.request_videos_v4(self.blink) last_record = {} for camera in self.cameras: @@ -331,14 +331,36 @@ async def check_new_videos(self): _LOGGER.warning("Could not check for motion. Response: %s", resp) return False + # The v4 endpoint returns all recent media (no server-side time + # filter), so we use `interval` as the reference time to filter + # entries client-side. + interval_iso = datetime.datetime.fromtimestamp( + interval, tz=datetime.timezone.utc + ).isoformat() + for entry in info: try: name = entry["device_name"] clip_url = entry["media"] timestamp = entry["created_at"] - if self.check_new_video_time(timestamp): + + if self.check_new_video_time(timestamp, reference=interval_iso): self.motion[name] = True and self.arm record = {"clip": clip_url, "time": timestamp} + + # Extract AI video description (from v4 endpoint) + ai_vd = entry.get("ai_vd") + if ai_vd and isinstance(ai_vd, dict): + if ai_vd.get("full_description"): + record["ai_description"] = ai_vd["full_description"] + if ai_vd.get("short_description"): + record["ai_description_short"] = ai_vd["short_description"] + + # Extract CV detection labels + cv = entry.get("cv_detection") + if cv: + record["cv_detection"] = cv + self.last_records[name].append(record) except KeyError: last_refresh = datetime.datetime.fromtimestamp(self.blink.last_refresh) diff --git a/tests/test_ai_descriptions.py b/tests/test_ai_descriptions.py new file mode 100644 index 00000000..cef268af --- /dev/null +++ b/tests/test_ai_descriptions.py @@ -0,0 +1,262 @@ +"""Tests for AI video descriptions and v4 media endpoint support.""" + +from unittest import mock +from unittest import IsolatedAsyncioTestCase +from blinkpy import api +from blinkpy.blinkpy import Blink +from blinkpy.helpers.util import BlinkURLHandler +from blinkpy.sync_module import BlinkSyncModule +from blinkpy.camera import BlinkCamera + +# Sample v4 media entry with AI description and CV detection +V4_ENTRY_WITH_AI = { + "device_name": "Front Door", + "media": "/api/v4/accounts/1234/media/999/video/contents", + "created_at": "1990-01-01T00:00:00+00:00", + "ai_vd": { + "full_description": "A person is walking on the driveway.", + "short_description": "Person on driveway.", + }, + "cv_detection": ["person"], +} + +# Sample v4 media entry without AI description (SVD not enabled) +V4_ENTRY_NO_AI = { + "device_name": "Front Door", + "media": "/api/v4/accounts/1234/media/888/video/contents", + "created_at": "1990-01-01T00:00:00+00:00", + "ai_vd": None, + "cv_detection": None, +} + +# Sample v4 entry with empty ai_vd object +V4_ENTRY_EMPTY_AI = { + "device_name": "Front Door", + "media": "/api/v4/accounts/1234/media/777/video/contents", + "created_at": "1990-01-01T00:00:00+00:00", + "ai_vd": {}, + "cv_detection": ["vehicle"], +} + +# Full v4 response +V4_RESPONSE = { + "media": [V4_ENTRY_WITH_AI], + "moment_gap_time": 25, + "page_size": 200, + "pagination_key": None, + "smart_video_descriptions": True, +} + +# Camera config used by TestCameraAIAttributes +CAMERA_CFG = { + "name": "foobar", + "id": 1234, + "network_id": 5678, + "serial": "12345678", + "enabled": False, + "battery_state": "ok", + "battery_voltage": 163, + "wifi_strength": -38, + "signals": {"lfr": 5, "wifi": 4, "battery": 3, "temp": 68}, + "thumbnail": "/thumb", +} + + +@mock.patch("blinkpy.auth.Auth.query") +class TestV4MediaEndpoint(IsolatedAsyncioTestCase): + """Test the v4 media endpoint API function.""" + + async def asyncSetUp(self): + """Set up Blink module.""" + self.blink = Blink(session=mock.AsyncMock()) + self.blink.urls = BlinkURLHandler("test") + self.blink.auth.account_id = 1234 + + def tearDown(self): + """Clean up after test.""" + self.blink = None + + async def test_request_videos_v4(self, mock_resp): + """Test v4 media endpoint returns expected data.""" + mock_resp.return_value = V4_RESPONSE + result = await api.request_videos_v4(self.blink) + self.assertIn("media", result) + self.assertEqual(len(result["media"]), 1) + self.assertIn("ai_vd", result["media"][0]) + + async def test_request_videos_v4_empty_response(self, mock_resp): + """Test v4 endpoint with empty response.""" + mock_resp.return_value = None + result = await api.request_videos_v4(self.blink) + self.assertIsNone(result) + + async def test_request_videos_v4_no_media(self, mock_resp): + """Test v4 endpoint with response missing media key.""" + mock_resp.return_value = {"error": "something went wrong"} + result = await api.request_videos_v4(self.blink) + self.assertNotIn("media", result) + + +@mock.patch("blinkpy.auth.Auth.query") +class TestAIDescriptionExtraction(IsolatedAsyncioTestCase): + """Test AI description extraction in check_new_videos.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = Blink(motion_interval=0, session=mock.AsyncMock()) + self.blink.last_refresh = 1000 + self.blink.urls = BlinkURLHandler("test") + self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", []) + self.blink.sync["test"].network_info = {"network": {"armed": True}} + + def tearDown(self): + """Clean up after test.""" + self.blink = None + + async def test_v4_ai_description_extracted(self, mock_resp): + """Test that ai_vd is extracted from v4 media entries.""" + mock_resp.return_value = {"media": [V4_ENTRY_WITH_AI]} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"Front Door": None} + self.assertTrue(await sync_module.check_new_videos()) + records = sync_module.last_records["Front Door"] + self.assertEqual(len(records), 1) + self.assertEqual( + records[0]["ai_description"], + "A person is walking on the driveway.", + ) + self.assertEqual(records[0]["ai_description_short"], "Person on driveway.") + self.assertEqual(records[0]["cv_detection"], ["person"]) + + async def test_v4_no_ai_description(self, mock_resp): + """Test graceful handling when ai_vd is None.""" + mock_resp.return_value = {"media": [V4_ENTRY_NO_AI]} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"Front Door": None} + self.assertTrue(await sync_module.check_new_videos()) + records = sync_module.last_records["Front Door"] + self.assertEqual(len(records), 1) + # No ai_description keys should be present + self.assertNotIn("ai_description", records[0]) + self.assertNotIn("ai_description_short", records[0]) + + async def test_v4_empty_ai_vd_object(self, mock_resp): + """Test graceful handling when ai_vd is an empty dict.""" + mock_resp.return_value = {"media": [V4_ENTRY_EMPTY_AI]} + sync_module = self.blink.sync["test"] + sync_module.cameras = {"Front Door": None} + self.assertTrue(await sync_module.check_new_videos()) + records = sync_module.last_records["Front Door"] + self.assertEqual(len(records), 1) + # Empty dict should not produce ai_description keys + self.assertNotIn("ai_description", records[0]) + # But cv_detection should still be extracted + self.assertEqual(records[0]["cv_detection"], ["vehicle"]) + + +@mock.patch("blinkpy.auth.Auth.query", return_value={}) +class TestCameraAIAttributes(IsolatedAsyncioTestCase): + """Test AI description attributes on camera objects.""" + + def setUp(self): + """Set up Blink module.""" + self.blink = Blink(session=mock.AsyncMock()) + self.blink.urls = BlinkURLHandler("test") + self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", 1234, []) + self.camera = BlinkCamera(self.blink.sync["test"]) + self.camera.name = "foobar" + self.blink.sync["test"].cameras["foobar"] = self.camera + + def tearDown(self): + """Clean up after test.""" + self.blink = None + self.camera = None + + async def test_camera_attributes_include_ai_fields(self, mock_resp): + """Test that camera attributes include AI description fields.""" + attrs = self.camera.attributes + self.assertIn("ai_description", attrs) + self.assertIn("ai_description_short", attrs) + self.assertIn("cv_detection", attrs) + # Initially None + self.assertIsNone(attrs["ai_description"]) + self.assertIsNone(attrs["ai_description_short"]) + self.assertIsNone(attrs["cv_detection"]) + + async def test_camera_ai_description_from_records(self, mock_resp): + """Test that camera picks up AI description from last records.""" + self.camera.sync.last_records["foobar"] = [ + { + "clip": "/clip.mp4", + "time": "2024-01-01T00:00:00+00:00", + "ai_description": "A cat sitting on the porch.", + "ai_description_short": "Cat on porch.", + "cv_detection": ["animal"], + } + ] + self.camera.sync.motion["foobar"] = True + await self.camera.update(CAMERA_CFG, force_cache=False) + self.assertEqual(self.camera.ai_description, "A cat sitting on the porch.") + self.assertEqual(self.camera.ai_description_short, "Cat on porch.") + self.assertEqual(self.camera.cv_detection, ["animal"]) + + async def test_camera_no_ai_description_in_records(self, mock_resp): + """Test camera handles records without AI description gracefully.""" + self.camera.sync.last_records["foobar"] = [ + { + "clip": "/clip.mp4", + "time": "2024-01-01T00:00:00+00:00", + } + ] + self.camera.sync.motion["foobar"] = True + await self.camera.update(CAMERA_CFG, force_cache=False) + # Should be None since record didn't have these keys + self.assertIsNone(self.camera.ai_description) + self.assertIsNone(self.camera.ai_description_short) + self.assertIsNone(self.camera.cv_detection) + + async def test_camera_recent_clips_include_ai_fields(self, mock_resp): + """Test that recent_clips entries include AI description fields.""" + self.camera.sync.last_records["foobar"] = [ + { + "clip": "/clip.mp4", + "time": "2024-01-01T00:00:00+00:00", + "ai_description": "A delivery person at the door.", + "cv_detection": ["person"], + } + ] + self.camera.sync.motion["foobar"] = True + await self.camera.update_images(CAMERA_CFG, expire_clips=False) + self.assertEqual(len(self.camera.recent_clips), 1) + clip = self.camera.recent_clips[0] + self.assertEqual(clip["ai_description"], "A delivery person at the door.") + self.assertEqual(clip["cv_detection"], ["person"]) + + async def test_clip_url_absolute_not_doubled(self, mock_resp): + """Test that absolute clip URLs from v4 are not doubled with base_url.""" + absolute_url = "https://rest-u009.immedia-semi.com/api/v4/media/123/video" + self.camera.sync.last_records["foobar"] = [ + { + "clip": absolute_url, + "time": "2024-01-01T00:00:00+00:00", + } + ] + self.camera.sync.motion["foobar"] = False + await self.camera.update(CAMERA_CFG, force_cache=False) + # URL should NOT be doubled (no base_url prepended) + self.assertEqual(self.camera.clip, absolute_url) + + async def test_clip_url_relative_gets_base_url(self, mock_resp): + """Test that relative clip URLs still get base_url prepended.""" + self.camera.sync.last_records["foobar"] = [ + { + "clip": "/api/v1/media/123.mp4", + "time": "2024-01-01T00:00:00+00:00", + } + ] + self.camera.sync.motion["foobar"] = False + await self.camera.update(CAMERA_CFG, force_cache=False) + self.assertEqual( + self.camera.clip, + f"{self.blink.urls.base_url}/api/v1/media/123.mp4", + ) diff --git a/tests/test_camera_functions.py b/tests/test_camera_functions.py index 0ca75699..22ab55ea 100644 --- a/tests/test_camera_functions.py +++ b/tests/test_camera_functions.py @@ -154,10 +154,20 @@ async def test_recent_video_clips(self, mock_resp): self.camera.sync.last_records["foobar"].append(record1) self.camera.sync.motion["foobar"] = True await self.camera.update_images(CONFIG, expire_clips=False) - record1["clip"] = self.blink.urls.base_url + "/clip1" - record2["clip"] = self.blink.urls.base_url + "/clip2" - self.assertEqual(self.camera.recent_clips[0], record1) - self.assertEqual(self.camera.recent_clips[1], record2) + expected1 = { + "clip": self.blink.urls.base_url + "/clip1", + "time": "2022-12-01 00:00:00+00:00", + "ai_description": None, + "cv_detection": None, + } + expected2 = { + "clip": self.blink.urls.base_url + "/clip2", + "time": "2022-12-01 00:00:10+00:00", + "ai_description": None, + "cv_detection": None, + } + self.assertEqual(self.camera.recent_clips[0], expected1) + self.assertEqual(self.camera.recent_clips[1], expected2) async def test_recent_video_clips_missing_key(self, mock_resp): """Tests that the missing key failst."""