Feature add blink doorbell without sync module

blink_credentials.json

@@ -0,0 +1,17 @@
+{
+  "username": "nirgal@live.com",
+  "password": "yHuJ91)3",
+  "uid": "BlinkCamera_0286240e-db42-cd59-b7a5-2f64fc13bde8",
+  "device_id": "Blinkpy",
+  "token": "eyJhbGciOiJSUzI1NiIsImprdSI6Ii9vYXV0aC9pbnRlcm5hbC9qd2tzIiwia2lkIjoiNjQ1NWZmNDEiLCJ0eXAiOiJKV1QifQ.eyJhcHBfaWQiOiJpb3MiLCJjaWQiOiJpb3MiLCJleHAiOjE3NzQ2Mzk2MzgsImhhcmR3YXJlX2lkIjoiNkVBQzU1OEYtMEY4MS00NDM0LTgyMTYtRTU2MzM3RjU3NzBCIiwiaWF0IjoxNzc0NjI1MjM4LCJpc3MiOiJCbGlua09hdXRoU2VydmljZS1wcm9kOnVzLWVhc3QtMTo1YmM2OTg4ZCIsIm9pYXQiOjE3NzQ2MjUyMzgsInJuZCI6Ikl0VV9OdG1kWGIiLCJzY29wZXMiOlsiY2xpZW50Il0sInNlc3Npb25faWQiOiJibGluay1zZXNzaW9uLTA1MmQ0MDA5LTc2MjgtNDVlZC1hOGJkLWMzMmRiYzY3NWI4NiIsInVzZXJfaWQiOjEyNjk3NDExMH0.VOGUbc44jryHbFC_Trmgq46d0qo0PBBevsoRcgFa8lTtwpgH3zbjQFZatN8Xbs5MvpKNTr-Z7ED2BBL2aj10C6rI61ioBdef0Jae2GNPxgAzMnJwASKAQB2H1ZLPid6TaTZzujSvYnigckMNekYjc23b_rweJ_rXZwHOrsLT7ws1_Rc1rjxY5k8N9irPJy-bWBh_S6mv50_bFMSSu1tLFxGvWItsp5sDni7Bz64RhEJFO4AXFzqYcZ4RDW3qEknt6Zeh0z0cxcB7ZDuWxuMw3WF5CBcsjbvPLRHhniYMRPImT2fEcZnpWZMKS7d9nuB1a6A-QcirJ91rOA5GNK6XmA",
+  "expires_in": 14400,
+  "expiration_date": 1774639638.2754378,
+  "refresh_token": "eyJhbGciOiJSUzI1NiIsImprdSI6Ii9vYXV0aC9pbnRlcm5hbC9qd2tzIiwia2lkIjoiNjQ1NWZmNDEiLCJ0eXAiOiJKV1QifQ.eyJpYXQiOjE3NzQ2MjUyMzgsImlzcyI6IkJsaW5rT2F1dGhTZXJ2aWNlLXByb2Q6dXMtZWFzdC0xOjViYzY5ODhkIiwib2lhdCI6MTc3NDYyNTIzOCwicmVmcmVzaF9jaWQiOiJpb3MiLCJyZWZyZXNoX3Njb3BlcyI6WyJjbGllbnQiXSwicmVmcmVzaF91c2VyX2lkIjoxMjY5NzQxMTAsInJuZCI6IlJOZDA4dDRJazYiLCJzZXNzaW9uX2lkIjoiYmxpbmstc2Vzc2lvbi0wNTJkNDAwOS03NjI4LTQ1ZWQtYThiZC1jMzJkYmM2NzViODYiLCJ0eXBlIjoicmVmcmVzaC10b2tlbiJ9.kSRBILMQHxrZhWhbUm78xJDo011plQhotvOiyxoqVgbXZ0cueVvYw8hnaMfdocCpiPUrcibtiuBQYOP9zgzO7e-n98zMaCF_t1l-mVPl5s2j4qcWALoZFCs2fL_Szx-UuYjUxLm6iZrRX7CfDFz2BVaHwcYHgWaNR4V6IBXf_UeaSYrcXWjfdFrx24RnrWFpZQX9j6z2U64T22Z5Tc1ePrChnG7G_4-ELBZZuiw-FNNyZGr_oVsaxabmCI5_zBWZESEiuWBInKPGoBLcF9sENVJd-Bl6hS9eiaGDNCn4vdbN3hpZqlr4_2vCRY6oxJL5cj92jOOyoUk3m1eH_vqc7g",
+  "host": "prde.immedia-semi.com",
+  "region_id": "prde",
+  "client_id": "2340966",
+  "account_id": 230552,
+  "user_id": "230551",
+  "hardware_id": "6EAC558F-0F81-4434-8216-E56337F5770B",
+  "2fa_code": "190161"
+}

blink_list_devices_json.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+import asyncio
+import builtins
+import json
+from pathlib import Path
+
+from aiohttp import ClientSession
+from blinkpy.auth import Auth
+from blinkpy.auth import BlinkTwoFARequiredError
+from blinkpy.blinkpy import Blink
+
+
+CRED_FILE = Path("blink_credentials.json")
+
+
+def load_credentials(path: Path) -> dict:
+    if not path.exists():
+        return {}
+    raw = path.read_text(encoding="utf-8").strip()
+    if not raw:
+        return {}
+    try:
+        data = json.loads(raw)
+        if not isinstance(data, dict):
+            return {}
+        return data
+    except json.JSONDecodeError:
+        return {}
+
+
+def save_credentials(path: Path, data: dict) -> None:
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+
+
+def prompt_if_missing(creds: dict, key: str, prompt_text: str) -> str:
+    value = (creds.get(key) or "").strip()
+    if not value:
+        value = input(prompt_text).strip()
+        creds[key] = value
+    return value
+
+
+async def close_session_if_needed(session_obj) -> None:
+    if session_obj is None:
+        return
+    if getattr(session_obj, "closed", False):
+        return
+    close_fn = getattr(session_obj, "close", None)
+    if close_fn is None:
+        return
+    result = close_fn()
+    if asyncio.iscoroutine(result):
+        await result
+
+
+async def prompt_2fa_with_optional_stored_code(blink: Blink, stored_code: str) -> str:
+    code = (stored_code or "").strip()
+    if not code:
+        code = input("Enter Blink 2FA code: ").strip()
+
+    # blinkpy 0.25.x exposes prompt_2fa(), but not send_auth_key().
+    # Feed the stored code into the prompt automatically.
+    original_input = builtins.input
+    try:
+        builtins.input = lambda _prompt="": code
+        await blink.prompt_2fa()
+        return code
+    finally:
+        builtins.input = original_input
+
+
+async def main():
+    creds = load_credentials(CRED_FILE)
+
+    username = prompt_if_missing(creds, "username", "Blink username/email: ")
+    password = prompt_if_missing(creds, "password", "Blink password: ")
+    creds.setdefault("2fa_code", "")
+
+    session = ClientSession()
+    blink = None
+    try:
+        blink = Blink(session=session)
+        auth_payload = dict(creds)
+        auth_payload["username"] = username
+        auth_payload["password"] = password
+        auth_payload.pop("2fa_code", None)
+        auth = Auth(auth_payload, no_prompt=True)
+        blink.auth = auth
+
+        try:
+            await blink.start()
+        except BlinkTwoFARequiredError:
+            try:
+                used_code = await prompt_2fa_with_optional_stored_code(
+                    blink, creds.get("2fa_code", "")
+                )
+                if used_code:
+                    creds["2fa_code"] = used_code
+            except Exception:
+                code = input("Stored 2FA code failed. Enter new 2FA code: ").strip()
+                creds["2fa_code"] = code
+                await prompt_2fa_with_optional_stored_code(blink, code)
+
+        # Persist Blink auth cache/tokens so future runs avoid repeated 2FA.
+        await blink.save(str(CRED_FILE))
+        saved = load_credentials(CRED_FILE)
+        saved["username"] = username
+        saved["password"] = password
+        saved["2fa_code"] = creds.get("2fa_code", "")
+        save_credentials(CRED_FILE, saved)
+
+        print("\n=== Cameras (including doorbells) ===")
+        if blink.cameras:
+            for name, camera in blink.cameras.items():
+                cam_type = getattr(camera, "camera_type", "unknown")
+                print(f"- {name} [{cam_type}]")
+        else:
+            print("No cameras found.")
+
+        print("\n=== Doorbells ===")
+        doorbells = getattr(blink, "doorbells", None)
+        if doorbells:
+            for name, bell in doorbells.items():
+                bell_type = getattr(bell, "camera_type", "doorbell")
+                print(f"- {name} [{bell_type}]")
+        else:
+            derived = [
+                (name, cam)
+                for name, cam in blink.cameras.items()
+                if "doorbell" in str(getattr(cam, "camera_type", "")).lower()
+            ]
+            if derived:
+                for name, bell in derived:
+                    print(f"- {name} [{getattr(bell, 'camera_type', 'doorbell')}]")
+            else:
+                print("No doorbells found.")
+    finally:
+        # Some blinkpy flows can leave extra aiohttp sessions open.
+        # Close all known session handles explicitly.
+        seen = set()
+        session_candidates = [
+            getattr(blink, "session", None) if blink else None,
+            getattr(getattr(blink, "auth", None), "session", None) if blink else None,
+            session,
+        ]
+        for item in session_candidates:
+            if item is None or id(item) in seen:
+                continue
+            seen.add(id(item))
+            await close_session_if_needed(item)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

blinkpy-changes.patch

@@ -0,0 +1,98 @@
+diff --git a/blinkpy/blinkpy.py b/share/blinkpy-upstream-0.25.5/blinkpy/blinkpy.py
+index c970420..375f9a6 100644
+--- a/blinkpy/blinkpy.py
++++ b/share/blinkpy-upstream-0.25.5/blinkpy/blinkpy.py
+@@ -75,6 +75,51 @@ class Blink:
+         self.homescreen = {}
+         self.no_owls = no_owls
+
++    def _iter_device_dicts(self, payload):
++        """Yield device-like dictionaries from potentially nested payloads."""
++        if isinstance(payload, list):
++            for item in payload:
++                yield from self._iter_device_dicts(item)
++            return
++        if isinstance(payload, dict):
++            # Treat this as a device entry when it has the minimum fields.
++            if "name" in payload and "network_id" in payload:
++                yield payload
++            for value in payload.values():
++                yield from self._iter_device_dicts(value)
++
++    def get_homescreen_devices(self, kind):
++        """Return homescreen devices for a given kind."""
++        key_candidates = {
++            "mini": ["owls", "mini_cameras", "minis"],
++            "doorbell": ["doorbells", "lotus", "doorbell_cameras"],
++        }
++        devices = []
++        seen = set()
++        for key in key_candidates.get(kind, []):
++            payload = self.homescreen.get(key)
++            for device in self._iter_device_dicts(payload):
++                signature = (
++                    str(device.get("id")),
++                    str(device.get("network_id")),
++                    str(device.get("name")),
++                )
++                if signature in seen:
++                    continue
++                seen.add(signature)
++                devices.append(device)
++        return devices
++
++    def has_sync_module_for_network(self, network_id):
++        """Check whether homescreen reports a real sync module for network."""
++        try:
++            for sync in self.homescreen.get("sync_modules", []):
++                if str(sync.get("network_id")) == str(network_id):
++                    return True
++        except AttributeError:
++            return False
++        return False
++
+     @property
+     def client_id(self):
+         """Return the client id."""
+@@ -220,10 +265,12 @@ class Blink:
+         network_list = []
+         camera_list = []
+         try:
+-            for owl in self.homescreen["owls"]:
++            for owl in self.get_homescreen_devices("mini"):
+                 name = owl["name"]
+                 network_id = str(owl["network_id"])
+-                if network_id in self.network_ids:
++                if network_id in self.network_ids and self.has_sync_module_for_network(
++                    network_id
++                ):
+                     camera_list.append(
+                         {network_id: {"name": name, "id": network_id, "type": "mini"}}
+                     )
+@@ -244,10 +291,12 @@ class Blink:
+         network_list = []
+         camera_list = []
+         try:
+-            for lotus in self.homescreen["doorbells"]:
++            for lotus in self.get_homescreen_devices("doorbell"):
+                 name = lotus["name"]
+                 network_id = str(lotus["network_id"])
+-                if network_id in self.network_ids:
++                if network_id in self.network_ids and self.has_sync_module_for_network(
++                    network_id
++                ):
+                     camera_list.append(
+                         {
+                             network_id: {
+@@ -287,9 +336,11 @@ class Blink:
+             lotus_cameras = await self.setup_lotus()
+             for camera in mini_cameras:
+                 for network, camera_info in camera.items():
++                    all_cameras.setdefault(network, [])
+                     all_cameras[network].append(camera_info)
+             for camera in lotus_cameras:
+                 for network, camera_info in camera.items():
++                    all_cameras.setdefault(network, [])
+                     all_cameras[network].append(camera_info)
+             return all_cameras
+         except (KeyError, TypeError) as ex:

blinkpy/blinkpy.py

@@ -75,6 +75,51 @@ def __init__(
         self.homescreen = {}
         self.no_owls = no_owls
 
+    def _iter_device_dicts(self, payload):
+        """Yield device-like dictionaries from potentially nested payloads."""
+        if isinstance(payload, list):
+            for item in payload:
+                yield from self._iter_device_dicts(item)
+            return
+        if isinstance(payload, dict):
+            # Treat this as a device entry when it has the minimum fields.
+            if "name" in payload and "network_id" in payload:
+                yield payload
+            for value in payload.values():
+                yield from self._iter_device_dicts(value)
+
+    def get_homescreen_devices(self, kind):
+        """Return homescreen devices for a given kind."""
+        key_candidates = {
+            "mini": ["owls", "mini_cameras", "minis"],
+            "doorbell": ["doorbells", "lotus", "doorbell_cameras"],
+        }
+        devices = []
+        seen = set()
+        for key in key_candidates.get(kind, []):
+            payload = self.homescreen.get(key)
+            for device in self._iter_device_dicts(payload):
+                signature = (
+                    str(device.get("id")),
+                    str(device.get("network_id")),
+                    str(device.get("name")),
+                )
+                if signature in seen:
+                    continue
+                seen.add(signature)
+                devices.append(device)
+        return devices
+
+    def has_sync_module_for_network(self, network_id):
+        """Check whether homescreen reports a real sync module for network."""
+        try:
+            for sync in self.homescreen.get("sync_modules", []):
+                if str(sync.get("network_id")) == str(network_id):
+                    return True
+        except AttributeError:
+            return False
+        return False
+
     @property
     def client_id(self):
         """Return the client id."""
@@ -220,10 +265,12 @@ async def setup_owls(self):
         network_list = []
         camera_list = []
         try:
-            for owl in self.homescreen["owls"]:
+            for owl in self.get_homescreen_devices("mini"):
                 name = owl["name"]
                 network_id = str(owl["network_id"])
-                if network_id in self.network_ids:
+                if network_id in self.network_ids and self.has_sync_module_for_network(
+                    network_id
+                ):
                     camera_list.append(
                         {network_id: {"name": name, "id": network_id, "type": "mini"}}
                     )
@@ -244,10 +291,12 @@ async def setup_lotus(self):
         network_list = []
         camera_list = []
         try:
-            for lotus in self.homescreen["doorbells"]:
+            for lotus in self.get_homescreen_devices("doorbell"):
                 name = lotus["name"]
                 network_id = str(lotus["network_id"])
-                if network_id in self.network_ids:
+                if network_id in self.network_ids and self.has_sync_module_for_network(
+                    network_id
+                ):
                     camera_list.append(
                         {
                             network_id: {
@@ -287,9 +336,11 @@ async def setup_camera_list(self):
             lotus_cameras = await self.setup_lotus()
             for camera in mini_cameras:
                 for network, camera_info in camera.items():
+                    all_cameras.setdefault(network, [])
                     all_cameras[network].append(camera_info)
             for camera in lotus_cameras:
                 for network, camera_info in camera.items():
+                    all_cameras.setdefault(network, [])
                     all_cameras[network].append(camera_info)
             return all_cameras
         except (KeyError, TypeError) as ex:

blinkpy/sync_module.py

@@ -225,8 +225,8 @@ async def update_cameras(self, camera_type=BlinkCamera):
     def get_unique_info(self, name):
         """Extract unique information for Minis and Doorbells."""
         try:
-            for type_key in self.type_key_map.values():
-                for device in self.blink.homescreen[type_key]:
+            for type_name in self.type_key_map:
+                for device in self.blink.get_homescreen_devices(type_name):
                     _LOGGER.debug("checking device %s", device)
                     if device["name"] == name:
                         _LOGGER.debug("Found unique_info %s", device)
@@ -551,7 +551,7 @@ async def update_cameras(self, camera_type=BlinkCameraMini):
     async def get_camera_info(self, camera_id, **kwargs):
         """Retrieve camera information."""
         try:
-            for owl in self.blink.homescreen["owls"]:
+            for owl in self.blink.get_homescreen_devices("mini"):
                 if owl["name"] == self.name:
                     self.status = owl["enabled"]
                     return owl
@@ -614,7 +614,7 @@ async def update_cameras(self, camera_type=BlinkDoorbell):
     async def get_camera_info(self, camera_id, **kwargs):
         """Retrieve camera information."""
         try:
-            for doorbell in self.blink.homescreen["doorbells"]:
+            for doorbell in self.blink.get_homescreen_devices("doorbell"):
                 if doorbell["name"] == self.name:
                     self.status = doorbell["enabled"]
                     return doorbell