Add Privacy Mode Toggle

custom_components/dahua/__init__.py

@@ -24,6 +24,7 @@
 from custom_components.dahua.thread import DahuaEventThread, DahuaVtoEventThread
 from . import dahua_utils
 from .client import DahuaClient
+from .rpc2 import DahuaRpc2Client
 
 from .const import (
     CONF_EVENTS,
@@ -102,6 +103,10 @@ def __init__(self, hass: HomeAssistant, events: list, address: str, port: int, r
 
         # The client used to communicate with Dahua devices
         self.client: DahuaClient = DahuaClient(username, password, address, port, rtsp_port, self._session)
+
+        # RPC2 client for advanced features like privacy mode
+        self.rpc2_client: DahuaRpc2Client = DahuaRpc2Client(username, password, address, port, rtsp_port, self._session)
+        self._rpc2_available = False
 
         self.platforms = []
         self.initialized = False
@@ -292,6 +297,14 @@ async def _async_update_data(self):
                     pass
                 _LOGGER.info("Device supports Lighting_V2=%s", self._supports_lighting_v2)
 
+                # Test RPC2 client availability
+                try:
+                    await self.rpc2_client.login()
+                    self._rpc2_available = True
+                    _LOGGER.info("RPC2 client initialized successfully")
+                except Exception as exception:
+                    self._rpc2_available = False
+                    _LOGGER.debug("RPC2 client not available: %s", exception)
 
                 if not is_doorbell:
                     # Start the event listeners for IP cameras
@@ -366,6 +379,8 @@ async def _async_update_data(self):
                 coros.append(asyncio.ensure_future(self.client.async_get_light_global_enabled()))
             if self._supports_lighting_v2:   #add lighing_v2 API if it is supported
                 coros.append(asyncio.ensure_future(self.client.async_get_lighting_v2()))
+            if self._rpc2_available:
+                coros.append(asyncio.ensure_future(self._fetch_privacy_mode_status()))
 
 
             # Gather results and update the data map
@@ -755,6 +770,29 @@ def supports_smart_motion_detection_amcrest(self) -> bool:
         """ True if smart motion detection is supported for an amcrest device"""
         return self.model == "AD410" or self.model == "DB61i"
 
+    def supports_privacy_mode(self) -> bool:
+        """ True if privacy mode is supported (RPC2 client available)"""
+        return self._rpc2_available
+
+    def is_privacy_mode_enabled(self) -> bool:
+        """ True if privacy mode is enabled"""
+        return self.data.get("privacy_mode_enabled", False)
+
+    async def _fetch_privacy_mode_status(self) -> dict:
+        """ Fetch the current privacy mode status from the camera"""
+        try:
+            config = await self.rpc2_client.get_privacy_mode_config()
+            # Extract the Enable flag from the table array
+            table = config.get("table", [])
+            if table and len(table) > 0:
+                enabled = table[0].get("Enable", False)
+            else:
+                enabled = False
+            return {"privacy_mode_enabled": enabled}
+        except Exception as exception:
+            _LOGGER.debug("Failed to fetch privacy mode status: %s", exception)
+            return {"privacy_mode_enabled": False}
+
     def get_vto_client(self) -> DahuaVTOClient:
         """
         Returns an instance of the connected VTO client if this is a VTO device. We need this because there's different

custom_components/dahua/camera.py

@@ -23,6 +23,7 @@
 SERVICE_SET_VIDEO_PROFILE_MODE = "set_video_profile_mode"
 SERVICE_SET_FOCUS_ZOOM = "set_focus_zoom"
 SERVICE_SET_PRIVACY_MASKING = "set_privacy_masking"
+SERVICE_SET_PRIVACY_MODE = "set_privacy_mode"
 SERVICE_SET_CHANNEL_TITLE = "set_channel_title"
 SERVICE_SET_TEXT_OVERLAY = "set_text_overlay"
 SERVICE_SET_CUSTOM_OVERLAY = "set_custom_overlay"
@@ -94,6 +95,14 @@ async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entitie
         "async_set_privacy_masking"
     )
 
+    platform.async_register_entity_service(
+        SERVICE_SET_PRIVACY_MODE,
+        {
+            vol.Required("enabled", default=False): bool,
+        },
+        "async_set_privacy_mode"
+    )
+
     platform.async_register_entity_service(
         SERVICE_ENABLE_CHANNEL_TITLE,
         {
@@ -344,6 +353,10 @@ async def async_set_privacy_masking(self, index: int, enabled: bool):
         """ Handles the service call from SERVICE_SET_PRIVACY_MASKING to control the privacy masking """
         await self._coordinator.client.async_setprivacymask(index, enabled)
 
+    async def async_set_privacy_mode(self, enabled: bool):
+        """ Handles the service call from SERVICE_SET_PRIVACY_MODE to control the privacy mode """
+        await self._coordinator.rpc2_client.set_privacy_mode(enabled)
+
     async def async_set_enable_channel_title(self, enabled: bool):
         """ Handles the service call from SERVICE_ENABLE_CHANNEL_TITLE """
         channel = self._coordinator.get_channel()

custom_components/dahua/const.py

@@ -15,6 +15,7 @@
 DISARMING_ICON = "mdi:alarm-check"
 VOLUME_HIGH_ICON = "mdi:volume-high"
 BELL_ICON = "mdi:bell-ring"
+PRIVACY_MODE_ICON = "mdi:shield-lock"
 
 # Device classes - https://www.home-assistant.io/integrations/binary_sensor/#device-class
 MOTION_SENSOR_DEVICE_CLASS = "motion"

custom_components/dahua/rpc2.py

@@ -53,10 +53,47 @@ async def request(self, method, params=None, object_id=None, extra=None, url=Non
             url = "{0}/RPC2".format(self._base)
 
         resp = await self._session.post(url, data=json.dumps(data))
-        resp_json = json.loads(await resp.text())
-
-        if verify_result and resp_json['result'] is False:
-            raise ConnectionError(str(resp))
+        resp_text = await resp.text()
+
+        try:
+            resp_json = json.loads(resp_text)
+        except json.JSONDecodeError as e:
+            _LOGGER.error("Failed to parse JSON response: %s", resp_text)
+            raise ConnectionError(f"Invalid JSON response: {resp_text}")
+
+        if verify_result and resp_json.get('result') is False:
+            error_msg = resp_json.get('error', {})
+            error_code = error_msg.get('code', 0)
+
+            # Check if it's a session error (code 287637505 means "Invalid session")
+            if error_code == 287637505 or 'session' in str(error_msg).lower():
+                _LOGGER.info("Session expired, attempting to re-login")
+                # Clear the session and try to login again
+                self._session_id = None
+                await self.login()
+
+                # Retry the request with the new session
+                if self._session_id:
+                    data['session'] = self._session_id
+                    resp = await self._session.post(url, data=json.dumps(data))
+                    resp_text = await resp.text()
+                    try:
+                        resp_json = json.loads(resp_text)
+                    except json.JSONDecodeError as e:
+                        _LOGGER.error("Failed to parse JSON response after re-login: %s", resp_text)
+                        raise ConnectionError(f"Invalid JSON response: {resp_text}")
+
+                    # Check the result again
+                    if resp_json.get('result') is False:
+                        error_msg = resp_json.get('error', {})
+                        _LOGGER.error("RPC2 request failed after re-login: %s", error_msg)
+                        raise ConnectionError(f"RPC2 error: {error_msg}")
+                else:
+                    _LOGGER.error("Failed to re-login after session expiry")
+                    raise ConnectionError(f"RPC2 error: {error_msg}")
+            else:
+                _LOGGER.error("RPC2 request failed: %s", error_msg)
+                raise ConnectionError(f"RPC2 error: {error_msg}")
 
         return resp_json
 
@@ -135,3 +172,37 @@ async def get_coaxial_control_io_status(self, channel: int) -> CoaxialControlIOS
         """ async_get_coaxial_control_io_status returns the the current state of the speaker and white light. """
         response = await self.request(method="CoaxialControlIO.getStatus", params={"channel": channel})
         return CoaxialControlIOStatus(response)
+
+    async def get_privacy_mode_config(self) -> dict:
+        """Gets the current privacy mode (LeLensMask) configuration."""
+        await self._ensure_logged_in()
+        response = await self.get_config({"name": "LeLensMask"})
+        return response
+
+    async def _ensure_logged_in(self):
+        """Ensure we have a valid session, login if needed."""
+        if not self._session_id:
+            await self.login()
+
+    async def set_privacy_mode(self, enabled: bool) -> bool:
+        """Sets privacy mode on or off."""
+        await self._ensure_logged_in()
+
+        # Default time sections for all days, all hours
+        default_time_sections = [
+            ["1 00:00:00-23:59:59", "0 00:00:00-23:59:59", "0 00:00:00-23:59:59", 
+             "0 00:00:00-23:59:59", "0 00:00:00-23:59:59", "0 00:00:00-23:59:59"]
+        ] * 7
+
+        params = {
+            "name": "LeLensMask",
+            "table": [{
+                "Enable": enabled,
+                "TimeSection": default_time_sections
+            }],
+            "options": []
+        }
+
+        response = await self.request(method="configManager.setConfig", params=params)
+        # For configManager.setConfig, success is indicated by result being True or the method completing without error
+        return response.get('result', True) is not False

custom_components/dahua/services.yaml

@@ -445,6 +445,23 @@ set_privacy_masking:
       selector:
         boolean:
 
+set_privacy_mode:
+  name: Set Privacy Mode
+  description: Enables or disables the camera's privacy mode (lens masking)
+  target:
+    entity:
+      integration: dahua
+      domain: camera
+  fields:
+    enabled:
+      name: Enabled
+      description: "If true enables privacy mode, otherwise disables it"
+      example: true
+      required: true
+      default: true
+      selector:
+        boolean:
+
 goto_preset_position:
   name: Go to a preset position
   description: Go to a position already preset

custom_components/dahua/switch.py

@@ -1,13 +1,16 @@
 """Switch platform for dahua."""
+import logging
 from aiohttp import ClientError
 from homeassistant.core import HomeAssistant
 from homeassistant.components.switch import SwitchEntity
 from custom_components.dahua import DahuaDataUpdateCoordinator
 
-from .const import DOMAIN, DISARMING_ICON, MOTION_DETECTION_ICON, SIREN_ICON, BELL_ICON
+from .const import DOMAIN, DISARMING_ICON, MOTION_DETECTION_ICON, SIREN_ICON, BELL_ICON, PRIVACY_MODE_ICON
 from .entity import DahuaBaseEntity
 from .client import SIREN_TYPE
 
+_LOGGER: logging.Logger = logging.getLogger(__package__)
+
 
 async def async_setup_entry(hass: HomeAssistant, entry, async_add_devices):
     """Setup sensor platform."""
@@ -31,6 +34,10 @@ async def async_setup_entry(hass: HomeAssistant, entry, async_add_devices):
     except ClientError as exception:
         pass
 
+    # Add privacy mode switch if RPC2 client is available
+    if coordinator.supports_privacy_mode():
+        devices.append(DahuaPrivacyModeBinarySwitch(coordinator, entry))
+
     async_add_devices(devices)
 
 
@@ -243,3 +250,47 @@ def is_on(self):
         Value is fetched from api.get_motion_detection_config
         """
         return self._coordinator.is_siren_on()
+
+
+class DahuaPrivacyModeBinarySwitch(DahuaBaseEntity, SwitchEntity):
+    """Dahua privacy mode switch class. Used to enable or disable privacy mode."""
+
+    async def async_turn_on(self, **kwargs):
+        """Turn on privacy mode."""
+        try:
+            result = await self._coordinator.rpc2_client.set_privacy_mode(True)
+            if result:
+                await self._coordinator.async_refresh()
+        except Exception as e:
+            _LOGGER.error("Failed to turn on privacy mode: %s", e)
+            raise
+
+    async def async_turn_off(self, **kwargs):
+        """Turn off privacy mode."""
+        try:
+            result = await self._coordinator.rpc2_client.set_privacy_mode(False)
+            if result:
+                await self._coordinator.async_refresh()
+        except Exception as e:
+            _LOGGER.error("Failed to turn off privacy mode: %s", e)
+            raise
+
+    @property
+    def name(self):
+        """Return the name of the switch."""
+        return self._coordinator.get_device_name() + " Privacy Mode"
+
+    @property
+    def unique_id(self):
+        """Return unique ID."""
+        return self._coordinator.get_serial_number() + "_privacy_mode"
+
+    @property
+    def icon(self):
+        """Return the icon of this switch."""
+        return PRIVACY_MODE_ICON
+
+    @property
+    def is_on(self):
+        """Return true if privacy mode is on."""
+        return self._coordinator.is_privacy_mode_enabled()