Reolink add TCP push event connection as primary method (#129490)

homeassistant/components/reolink/binary_sensor.py

@@ -42,29 +42,34 @@ class ReolinkBinarySensorEntityDescription(
 BINARY_PUSH_SENSORS = (
     ReolinkBinarySensorEntityDescription(
         key="motion",
+        cmd_id=33,
         device_class=BinarySensorDeviceClass.MOTION,
         value=lambda api, ch: api.motion_detected(ch),
     ),
     ReolinkBinarySensorEntityDescription(
         key=FACE_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="face",
         value=lambda api, ch: api.ai_detected(ch, FACE_DETECTION_TYPE),
         supported=lambda api, ch: api.ai_supported(ch, FACE_DETECTION_TYPE),
     ),
     ReolinkBinarySensorEntityDescription(
         key=PERSON_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="person",
         value=lambda api, ch: api.ai_detected(ch, PERSON_DETECTION_TYPE),
         supported=lambda api, ch: api.ai_supported(ch, PERSON_DETECTION_TYPE),
     ),
     ReolinkBinarySensorEntityDescription(
         key=VEHICLE_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="vehicle",
         value=lambda api, ch: api.ai_detected(ch, VEHICLE_DETECTION_TYPE),
         supported=lambda api, ch: api.ai_supported(ch, VEHICLE_DETECTION_TYPE),
     ),
     ReolinkBinarySensorEntityDescription(
         key=PET_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="pet",
         value=lambda api, ch: api.ai_detected(ch, PET_DETECTION_TYPE),
         supported=lambda api, ch: (
@@ -74,18 +79,21 @@ BINARY_PUSH_SENSORS = (
     ),
     ReolinkBinarySensorEntityDescription(
         key=PET_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="animal",
         value=lambda api, ch: api.ai_detected(ch, PET_DETECTION_TYPE),
         supported=lambda api, ch: api.supported(ch, "ai_animal"),
     ),
     ReolinkBinarySensorEntityDescription(
         key=PACKAGE_DETECTION_TYPE,
+        cmd_id=33,
         translation_key="package",
         value=lambda api, ch: api.ai_detected(ch, PACKAGE_DETECTION_TYPE),
         supported=lambda api, ch: api.ai_supported(ch, PACKAGE_DETECTION_TYPE),
     ),
     ReolinkBinarySensorEntityDescription(
         key="visitor",
+        cmd_id=33,
         translation_key="visitor",
         value=lambda api, ch: api.visitor_detected(ch),
         supported=lambda api, ch: api.is_doorbell(ch),

homeassistant/components/reolink/entity.py

@@ -7,6 +7,7 @@ from dataclasses import dataclass
 
 from reolink_aio.api import DUAL_LENS_MODELS, Chime, Host
 
+from homeassistant.core import callback
 from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
 from homeassistant.helpers.entity import EntityDescription
 from homeassistant.helpers.update_coordinator import (
@@ -23,6 +24,7 @@ class ReolinkEntityDescription(EntityDescription):
     """A class that describes entities for Reolink."""
 
     cmd_key: str | None = None
+    cmd_id: int | None = None
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -90,18 +92,35 @@ class ReolinkHostCoordinatorEntity(CoordinatorEntity[DataUpdateCoordinator[None]
         """Return True if entity is available."""
         return self._host.api.session_active and super().available
 
+    @callback
+    def _push_callback(self) -> None:
+        """Handle incoming TCP push event."""
+        self.async_write_ha_state()
+
+    def register_callback(self, unique_id: str, cmd_id: int) -> None:
+        """Register callback for TCP push events."""
+        self._host.api.baichuan.register_callback(  # pragma: no cover
+            unique_id, self._push_callback, cmd_id
+        )
+
     async def async_added_to_hass(self) -> None:
         """Entity created."""
         await super().async_added_to_hass()
         cmd_key = self.entity_description.cmd_key
+        cmd_id = self.entity_description.cmd_id
         if cmd_key is not None:
             self._host.async_register_update_cmd(cmd_key)
+        if cmd_id is not None and self._attr_unique_id is not None:
+            self.register_callback(self._attr_unique_id, cmd_id)
 
     async def async_will_remove_from_hass(self) -> None:
         """Entity removed."""
         cmd_key = self.entity_description.cmd_key
+        cmd_id = self.entity_description.cmd_id
         if cmd_key is not None:
             self._host.async_unregister_update_cmd(cmd_key)
+        if cmd_id is not None and self._attr_unique_id is not None:
+            self._host.api.baichuan.unregister_callback(self._attr_unique_id)
 
         await super().async_will_remove_from_hass()
 
@@ -160,6 +179,12 @@ class ReolinkChannelCoordinatorEntity(ReolinkHostCoordinatorEntity):
         """Return True if entity is available."""
         return super().available and self._host.api.camera_online(self._channel)
 
+    def register_callback(self, unique_id: str, cmd_id) -> None:
+        """Register callback for TCP push events."""
+        self._host.api.baichuan.register_callback(
+            unique_id, self._push_callback, cmd_id, self._channel
+        )
+
     async def async_added_to_hass(self) -> None:
         """Entity created."""
         await super().async_added_to_hass()

homeassistant/components/reolink/host.py

@@ -41,6 +41,7 @@ from .exceptions import (
 )
 
 DEFAULT_TIMEOUT = 30
+FIRST_TCP_PUSH_TIMEOUT = 10
 FIRST_ONVIF_TIMEOUT = 10
 FIRST_ONVIF_LONG_POLL_TIMEOUT = 90
 SUBSCRIPTION_RENEW_THRESHOLD = 300
@@ -105,6 +106,7 @@ class ReolinkHost:
         self._long_poll_received: bool = False
         self._long_poll_error: bool = False
         self._cancel_poll: CALLBACK_TYPE | None = None
+        self._cancel_tcp_push_check: CALLBACK_TYPE | None = None
         self._cancel_onvif_check: CALLBACK_TYPE | None = None
         self._cancel_long_poll_check: CALLBACK_TYPE | None = None
         self._poll_job = HassJob(self._async_poll_all_motion, cancel_on_shutdown=True)
@@ -220,49 +222,14 @@ class ReolinkHost:
         else:
             self._unique_id = format_mac(self._api.mac_address)
 
-        if self._onvif_push_supported:
+        try:
-            try:
+            await self._api.baichuan.subscribe_events()
-                await self.subscribe()
+        except ReolinkError:
-            except ReolinkError:
+            await self._async_check_tcp_push()
-                self._onvif_push_supported = False
+        else:
-                self.unregister_webhook()
+            self._cancel_tcp_push_check = async_call_later(
-                await self._api.unsubscribe()
+                self._hass, FIRST_TCP_PUSH_TIMEOUT, self._async_check_tcp_push
-            else:
-                if self._api.supported(None, "initial_ONVIF_state"):
-                    _LOGGER.debug(
-                        "Waiting for initial ONVIF state on webhook '%s'",
-                        self._webhook_url,
-                    )
-                else:
-                    _LOGGER.debug(
-                        "Camera model %s most likely does not push its initial state"
-                        " upon ONVIF subscription, do not check",
-                        self._api.model,
-                    )
-                self._cancel_onvif_check = async_call_later(
-                    self._hass, FIRST_ONVIF_TIMEOUT, self._async_check_onvif
-                )
-        if not self._onvif_push_supported:
-            _LOGGER.debug(
-                "Camera model %s does not support ONVIF push, using ONVIF long polling instead",
-                self._api.model,
             )
-            try:
-                await self._async_start_long_polling(initial=True)
-            except NotSupportedError:
-                _LOGGER.debug(
-                    "Camera model %s does not support ONVIF long polling, using fast polling instead",
-                    self._api.model,
-                )
-                self._onvif_long_poll_supported = False
-                await self._api.unsubscribe()
-                await self._async_poll_all_motion()
-            else:
-                self._cancel_long_poll_check = async_call_later(
-                    self._hass,
-                    FIRST_ONVIF_LONG_POLL_TIMEOUT,
-                    self._async_check_onvif_long_poll,
-                )
 
         ch_list: list[int | None] = [None]
         if self._api.is_nvr:
@@ -294,6 +261,67 @@ class ReolinkHost:
             else:
                 ir.async_delete_issue(self._hass, DOMAIN, f"firmware_update_{key}")
 
+    async def _async_check_tcp_push(self, *_) -> None:
+        """Check the TCP push subscription."""
+        if self._api.baichuan.events_active:
+            ir.async_delete_issue(self._hass, DOMAIN, "webhook_url")
+            self._cancel_tcp_push_check = None
+            return
+
+        _LOGGER.debug(
+            "Reolink %s, did not receive initial TCP push event after %i seconds",
+            self._api.nvr_name,
+            FIRST_TCP_PUSH_TIMEOUT,
+        )
+
+        if self._onvif_push_supported:
+            try:
+                await self.subscribe()
+            except ReolinkError:
+                self._onvif_push_supported = False
+                self.unregister_webhook()
+                await self._api.unsubscribe()
+            else:
+                if self._api.supported(None, "initial_ONVIF_state"):
+                    _LOGGER.debug(
+                        "Waiting for initial ONVIF state on webhook '%s'",
+                        self._webhook_url,
+                    )
+                else:
+                    _LOGGER.debug(
+                        "Camera model %s most likely does not push its initial state"
+                        " upon ONVIF subscription, do not check",
+                        self._api.model,
+                    )
+                self._cancel_onvif_check = async_call_later(
+                    self._hass, FIRST_ONVIF_TIMEOUT, self._async_check_onvif
+                )
+
+        # start long polling if ONVIF push failed immediately
+        if not self._onvif_push_supported:
+            _LOGGER.debug(
+                "Camera model %s does not support ONVIF push, using ONVIF long polling instead",
+                self._api.model,
+            )
+            try:
+                await self._async_start_long_polling(initial=True)
+            except NotSupportedError:
+                _LOGGER.debug(
+                    "Camera model %s does not support ONVIF long polling, using fast polling instead",
+                    self._api.model,
+                )
+                self._onvif_long_poll_supported = False
+                await self._api.unsubscribe()
+                await self._async_poll_all_motion()
+            else:
+                self._cancel_long_poll_check = async_call_later(
+                    self._hass,
+                    FIRST_ONVIF_LONG_POLL_TIMEOUT,
+                    self._async_check_onvif_long_poll,
+                )
+
+        self._cancel_tcp_push_check = None
+
     async def _async_check_onvif(self, *_) -> None:
         """Check the ONVIF subscription."""
         if self._webhook_reachable:
@@ -391,6 +419,16 @@ class ReolinkHost:
 
     async def disconnect(self) -> None:
         """Disconnect from the API, so the connection will be released."""
+        try:
+            await self._api.baichuan.unsubscribe_events()
+        except ReolinkError as err:
+            _LOGGER.error(
+                "Reolink error while unsubscribing Baichuan from host %s:%s: %s",
+                self._api.host,
+                self._api.port,
+                err,
+            )
+
         try:
             await self._api.unsubscribe()
         except ReolinkError as err:
@@ -461,6 +499,9 @@ class ReolinkHost:
         if self._cancel_poll is not None:
             self._cancel_poll()
             self._cancel_poll = None
+        if self._cancel_tcp_push_check is not None:
+            self._cancel_tcp_push_check()
+            self._cancel_tcp_push_check = None
         if self._cancel_onvif_check is not None:
             self._cancel_onvif_check()
             self._cancel_onvif_check = None
@@ -494,8 +535,13 @@ class ReolinkHost:
 
     async def renew(self) -> None:
         """Renew the subscription of motion events (lease time is 15 minutes)."""
+        if self._api.baichuan.events_active and self._api.subscribed(SubType.push):
+            # TCP push active, unsubscribe from ONVIF push because not needed
+            self.unregister_webhook()
+            await self._api.unsubscribe()
+
         try:
-            if self._onvif_push_supported:
+            if self._onvif_push_supported and not self._api.baichuan.events_active:
                 await self._renew(SubType.push)
 
             if self._onvif_long_poll_supported and self._long_poll_task is not None:
@@ -608,7 +654,8 @@ class ReolinkHost:
         """Use ONVIF long polling to immediately receive events."""
         # This task will be cancelled once _async_stop_long_polling is called
         while True:
-            if self._webhook_reachable:
+            if self._api.baichuan.events_active or self._webhook_reachable:
+                # TCP push or ONVIF push working, stop long polling
                 self._long_poll_task = None
                 await self._async_stop_long_polling()
                 return
@@ -642,8 +689,12 @@ class ReolinkHost:
 
     async def _async_poll_all_motion(self, *_) -> None:
         """Poll motion and AI states until the first ONVIF push is received."""
-        if self._webhook_reachable or self._long_poll_received:
+        if (
-            # ONVIF push or long polling is working, stop fast polling
+            self._api.baichuan.events_active
+            or self._webhook_reachable
+            or self._long_poll_received
+        ):
+            # TCP push, ONVIF push or long polling is working, stop fast polling
             self._cancel_poll = None
             return
 
@@ -747,6 +798,8 @@ class ReolinkHost:
     @property
     def event_connection(self) -> str:
         """Type of connection to receive events."""
+        if self._api.baichuan.events_active:
+            return "TCP push"
         if self._webhook_reachable:
             return "ONVIF push"
         if self._long_poll_received:

tests/components/reolink/conftest.py

@@ -1,10 +1,12 @@
 """Setup the Reolink tests."""
 
 from collections.abc import Generator
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, create_autospec, patch
 
 import pytest
 from reolink_aio.api import Chime
+from reolink_aio.baichuan import Baichuan
+from reolink_aio.exceptions import ReolinkError
 
 from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
 from homeassistant.components.reolink.const import CONF_USE_HTTPS, DOMAIN
@@ -118,6 +120,12 @@ def reolink_connect_class() -> Generator[MagicMock]:
         host_mock.doorbell_led_list.return_value = ["stayoff", "auto"]
         host_mock.auto_track_method.return_value = 3
         host_mock.daynight_state.return_value = "Black&White"
+
+        # Baichuan
+        host_mock.baichuan = create_autospec(Baichuan)
+        # Disable tcp push by default for tests
+        host_mock.baichuan.events_active = False
+        host_mock.baichuan.subscribe_events.side_effect = ReolinkError("Test error")
         yield host_mock_class
 
 

tests/components/reolink/test_binary_sensor.py

@@ -1,5 +1,6 @@
 """Test the Reolink binary sensor platform."""
 
+from collections.abc import Callable
 from unittest.mock import MagicMock, patch
 
 from freezegun.api import FrozenDateTimeFactory
@@ -8,9 +9,8 @@ from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.const import STATE_OFF, STATE_ON, Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import entity_registry as er
 
-from .conftest import TEST_DUO_MODEL, TEST_NVR_NAME
+from .conftest import TEST_DUO_MODEL, TEST_HOST_MODEL, TEST_NVR_NAME
 
 from tests.common import MockConfigEntry, async_fire_time_changed
 from tests.typing import ClientSessionGenerator
@@ -22,7 +22,6 @@ async def test_motion_sensor(
     freezer: FrozenDateTimeFactory,
     config_entry: MockConfigEntry,
     reolink_connect: MagicMock,
-    entity_registry: er.EntityRegistry,
 ) -> None:
     """Test binary sensor entity with motion sensor."""
     reolink_connect.model = TEST_DUO_MODEL
@@ -42,7 +41,7 @@ async def test_motion_sensor(
 
     assert hass.states.get(entity_id).state == STATE_OFF
 
-    # test webhook callback
+    # test ONVIF webhook callback
     reolink_connect.motion_detected.return_value = True
     reolink_connect.ONVIF_event_callback.return_value = [0]
     webhook_id = config_entry.runtime_data.host.webhook_id
@@ -50,3 +49,43 @@ async def test_motion_sensor(
     await client.post(f"/api/webhook/{webhook_id}", data="test_data")
 
     assert hass.states.get(entity_id).state == STATE_ON
+
+
+async def test_tcp_callback(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    reolink_connect: MagicMock,
+) -> None:
+    """Test tcp callback using motion sensor."""
+
+    class callback_mock_class:
+        callback_func = None
+
+        def register_callback(
+            self, callback_id: str, callback: Callable[[], None], *args, **key_args
+        ) -> None:
+            if callback_id.endswith("_motion"):
+                self.callback_func = callback
+
+    callback_mock = callback_mock_class()
+
+    reolink_connect.model = TEST_HOST_MODEL
+    reolink_connect.baichuan.events_active = True
+    reolink_connect.baichuan.subscribe_events.reset_mock(side_effect=True)
+    reolink_connect.baichuan.register_callback = callback_mock.register_callback
+    reolink_connect.motion_detected.return_value = True
+
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BINARY_SENSOR]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    entity_id = f"{Platform.BINARY_SENSOR}.{TEST_NVR_NAME}_motion"
+    assert hass.states.get(entity_id).state == STATE_ON
+
+    # simulate a TCP push callback
+    reolink_connect.motion_detected.return_value = False
+    assert callback_mock.callback_func is not None
+    callback_mock.callback_func()
+
+    assert hass.states.get(entity_id).state == STATE_OFF

tests/components/reolink/test_host.py

@@ -14,12 +14,14 @@ from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL
 from homeassistant.components.reolink.host import (
     FIRST_ONVIF_LONG_POLL_TIMEOUT,
     FIRST_ONVIF_TIMEOUT,
+    FIRST_TCP_PUSH_TIMEOUT,
     LONG_POLL_COOLDOWN,
     LONG_POLL_ERROR_COOLDOWN,
     POLL_INTERVAL_NO_PUSH,
 )
 from homeassistant.components.webhook import async_handle_webhook
 from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
@@ -31,6 +33,56 @@ from tests.components.diagnostics import get_diagnostics_for_config_entry
 from tests.typing import ClientSessionGenerator
 
 
+async def test_setup_with_tcp_push(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    config_entry: MockConfigEntry,
+    reolink_connect: MagicMock,
+) -> None:
+    """Test successful setup of the integration with TCP push callbacks."""
+    reolink_connect.baichuan.events_active = True
+    reolink_connect.baichuan.subscribe_events.reset_mock(side_effect=True)
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BINARY_SENSOR]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    freezer.tick(timedelta(seconds=FIRST_TCP_PUSH_TIMEOUT))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    # ONVIF push subscription not called
+    assert not reolink_connect.subscribe.called
+
+    reolink_connect.baichuan.events_active = False
+    reolink_connect.baichuan.subscribe_events.side_effect = ReolinkError("Test error")
+
+
+async def test_unloading_with_tcp_push(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    reolink_connect: MagicMock,
+) -> None:
+    """Test successful unloading of the integration with TCP push callbacks."""
+    reolink_connect.baichuan.events_active = True
+    reolink_connect.baichuan.subscribe_events.reset_mock(side_effect=True)
+    with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BINARY_SENSOR]):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    reolink_connect.baichuan.unsubscribe_events.side_effect = ReolinkError("Test error")
+
+    # Unload the config entry
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state is ConfigEntryState.NOT_LOADED
+
+    reolink_connect.baichuan.events_active = False
+    reolink_connect.baichuan.subscribe_events.side_effect = ReolinkError("Test error")
+    reolink_connect.baichuan.unsubscribe_events.reset_mock(side_effect=True)
+
+
 async def test_webhook_callback(
     hass: HomeAssistant,
     hass_client_no_auth: ClientSessionGenerator,
@@ -402,3 +454,8 @@ async def test_diagnostics_event_connection(
 
     diag = await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
     assert diag["event connection"] == "ONVIF push"
+
+    # set TCP push as active
+    reolink_connect.baichuan.events_active = True
+    diag = await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
+    assert diag["event connection"] == "TCP push"