From c3ae471feba5e846aca9dfcf7216572edd2446d9 Mon Sep 17 00:00:00 2001 From: Realcrash <131798491+Realcrash@users.noreply.github.com> Date: Wed, 5 Aug 2026 20:20:13 +0200 Subject: [PATCH 1/2] Add SDT4E425 dual-sensor and PTZ preset support --- custom_components/dahua/__init__.py | 17 +-- custom_components/dahua/camera.py | 121 ++++++++++++++------- custom_components/dahua/client.py | 73 +++++++++++++ custom_components/dahua/model_profiles.py | 14 +++ custom_components/dahua/rpc2.py | 125 +++++++++++++++++++--- custom_components/dahua/select.py | 75 ++++++++----- 6 files changed, 343 insertions(+), 82 deletions(-) create mode 100644 custom_components/dahua/model_profiles.py diff --git a/custom_components/dahua/__init__.py b/custom_components/dahua/__init__.py index 984e55e..ab5a163 100755 --- a/custom_components/dahua/__init__.py +++ b/custom_components/dahua/__init__.py @@ -23,6 +23,7 @@ from . import dahua_utils from .client import DahuaClient +from .model_profiles import is_sdt4e425 from .const import ( CONF_EVENTS, @@ -304,13 +305,17 @@ async def _async_update_data(self): self._supports_event_notifications = False _LOGGER.debug("Device supports event notifications=%s", self._supports_event_notifications) - # PTZ - # The following lines are for Dahua devices - try: - await self.client.async_get_ptz_position() - self._supports_ptz_position = True - except ClientError: + # PTZ position readback. The SDT4E425 PTZ sensor is controllable, + # but firmware V3.200.0000027.6.R returns HTTP 400 for CGI getStatus. + # Do not conflate PTZ/preset control with CGI position readback. + if is_sdt4e425(self.model): self._supports_ptz_position = False + else: + try: + await self.client.async_get_ptz_position() + self._supports_ptz_position = True + except ClientError: + self._supports_ptz_position = False _LOGGER.debug("Device supports PTZ position=%s", self._supports_ptz_position) # Smart motion detection is enabled/disabled/fetched differently on Dahua devices compared to Amcrest diff --git a/custom_components/dahua/camera.py b/custom_components/dahua/camera.py index ced874a..54b74db 100755 --- a/custom_components/dahua/camera.py +++ b/custom_components/dahua/camera.py @@ -10,6 +10,7 @@ from custom_components.dahua import DahuaDataUpdateCoordinator from custom_components.dahua.entity import DahuaBaseEntity +from custom_components.dahua.model_profiles import is_sdt4e425 from .const import ( DOMAIN, @@ -44,19 +45,48 @@ async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entitie """Add a Dahua IP camera from a config entry.""" coordinator: DahuaDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] - max_streams = coordinator.get_max_streams() - - # Note the stream_index is 0 based. The main stream is index 0 - for stream_index in range(max_streams): - async_add_entities( - [ - DahuaCamera( - coordinator, - stream_index, - config_entry, - ) - ] + if is_sdt4e425(coordinator.get_model()): + # This physical camera exposes two sensors. Preserve RRoller's native + # Main/Sub/Sub_2 creation for each media channel from one config entry. + sensors = ( + # logical channel, media channel, display name, unique-id prefix + (0, 1, "Panorama", ""), + (1, 2, "PTZ", "1_"), ) + entities = [] + for logical_channel, media_channel, sensor_name, unique_prefix in sensors: + for stream_index in range(coordinator.get_max_streams()): + stream_name = coordinator.client.to_stream_name(stream_index) + display_name = ( + sensor_name + if stream_index == 0 + else f"{sensor_name} {stream_name}" + ) + entities.append( + DahuaCamera( + coordinator, + stream_index, + config_entry, + logical_channel=logical_channel, + media_channel=media_channel, + display_name=display_name, + unique_suffix=f"{unique_prefix}{stream_name}", + ) + ) + async_add_entities(entities) + else: + max_streams = coordinator.get_max_streams() + # Note the stream_index is 0 based. The main stream is index 0 + for stream_index in range(max_streams): + async_add_entities( + [ + DahuaCamera( + coordinator, + stream_index, + config_entry, + ) + ] + ) platform = entity_platform.async_get_current_platform() @@ -236,19 +266,33 @@ async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entitie class DahuaCamera(DahuaBaseEntity, Camera): """An implementation of a Dahua IP camera.""" - def __init__(self, coordinator: DahuaDataUpdateCoordinator, stream_index: int, config_entry): + def __init__( + self, coordinator: DahuaDataUpdateCoordinator, stream_index: int, config_entry, + *, logical_channel: int | None = None, media_channel: int | None = None, + display_name: str | None = None, unique_suffix: str | None = None, + ): """Initialize the Dahua camera.""" DahuaBaseEntity.__init__(self, coordinator, config_entry) Camera.__init__(self) - - name = coordinator.client.to_stream_name(stream_index) - self._channel_number = coordinator.get_channel_number() + stream_name = coordinator.client.to_stream_name(stream_index) + self._logical_channel = ( + coordinator.get_channel() if logical_channel is None else logical_channel + ) + self._channel_number = ( + coordinator.get_channel_number() if media_channel is None else media_channel + ) self._coordinator = coordinator - self._name = "{0} {1}".format(config_entry.title, name) - self._unique_id = coordinator.get_serial_number() + "_" + name + self._name = ( + f"{config_entry.title} {display_name}" + if display_name else f"{config_entry.title} {stream_name}" + ) + suffix = unique_suffix or stream_name + self._unique_id = coordinator.get_serial_number() + "_" + suffix self._stream_index = stream_index self._motion_status = False - self._stream_source = coordinator.client.get_rtsp_stream_url(self._channel_number, stream_index) + self._stream_source = coordinator.client.get_rtsp_stream_url( + self._channel_number, stream_index + ) self._attr_frontend_stream_type = StreamType.WEB_RTC @property @@ -278,7 +322,7 @@ def motion_detection_enabled(self): async def async_enable_motion_detection(self): """Enable motion detection in camera.""" try: - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.enable_motion_detection(channel, True) await self._coordinator.async_refresh() except TypeError: @@ -287,7 +331,7 @@ async def async_enable_motion_detection(self): async def async_disable_motion_detection(self): """Disable motion detection.""" try: - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.enable_motion_detection(channel, False) await self._coordinator.async_refresh() except TypeError: @@ -300,19 +344,22 @@ def name(self): async def async_set_infrared_mode(self, mode: str, brightness: int): """ Handles the service call from SERVICE_SET_INFRARED_MODE to set infrared mode and brightness """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_lighting_v1_mode(channel, mode, brightness) await self._coordinator.async_refresh() async def async_goto_preset_position(self, position: int): - """ Handles the service call from SERVICE_GOTO_PRESET_POSITION to go to a specific preset position """ - channel = self._coordinator.get_channel() - await self._coordinator.client.async_goto_preset_position(channel, position) + """Go to a preset, using RPC2 only for the SDT4E425.""" + channel = self._logical_channel + if is_sdt4e425(self._coordinator.get_model()): + await self._coordinator.client.async_goto_preset_rpc2(1, position) + else: + await self._coordinator.client.async_goto_preset_position(channel, position) await self._coordinator.async_refresh() async def async_set_video_in_day_night_mode(self, config_type: str, mode: str): """ Handles the service call from SERVICE_SET_DAY_NIGHT_MODE to set the day/night color mode """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_video_in_day_night_mode(channel, config_type, mode) await self._coordinator.async_refresh() @@ -322,13 +369,13 @@ async def async_reboot(self): async def async_set_record_mode(self, mode: str): """ Handles the service call from SERVICE_SET_RECORD_MODE to set the record mode """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_record_mode(channel, mode) await self._coordinator.async_refresh() async def async_set_video_profile_mode(self, mode: str): """ Handles the service call from SERVICE_SET_VIDEO_PROFILE_MODE to set profile mode to day/night """ - channel = self._coordinator.get_channel() + channel = self._logical_channel model = self._coordinator.get_model() # Some NVRs like the Lorex DHI-NVR4108HS-8P-4KS2 change the day/night mode through a switch if any(substring in model for substring in ['NVR4108HS', 'IPC-Color4K']): @@ -347,32 +394,32 @@ async def async_set_privacy_masking(self, index: int, enabled: bool): async def async_set_enable_channel_title(self, enabled: bool): """ Handles the service call from SERVICE_ENABLE_CHANNEL_TITLE """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_enable_channel_title(channel, enabled) async def async_set_enable_time_overlay(self, enabled: bool): """ Handles the service call from SERVICE_ENABLE_TIME_OVERLAY """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_enable_time_overlay(channel, enabled) async def async_set_enable_text_overlay(self, group: int, enabled: bool): """ Handles the service call from SERVICE_ENABLE_TEXT_OVERLAY """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_enable_text_overlay(channel, group, enabled) async def async_set_enable_custom_overlay(self, group: int, enabled: bool): """ Handles the service call from SERVICE_ENABLE_CUSTOM_OVERLAY """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_enable_custom_overlay(channel, group, enabled) async def async_set_enable_all_ivs_rules(self, enabled: bool): """ Handles the service call from SERVICE_ENABLE_ALL_IVS_RULES """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_all_ivs_rules(channel, enabled) async def async_enable_ivs_rule(self, index: int, enabled: bool): """ Handles the service call from SERVICE_ENABLE_IVS_RULE """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_ivs_rule(channel, index, enabled) async def async_vto_open_door(self, door_id: int): @@ -385,16 +432,16 @@ async def async_vto_cancel_call(self): async def async_set_service_set_channel_title(self, text1: str, text2: str): """ Handles the service call from SERVICE_SET_CHANNEL_TITLE to set profile mode to day/night """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_service_set_channel_title(channel, text1, text2) async def async_set_service_set_text_overlay(self, group: int, text1: str, text2: str, text3: str, text4: str): """ Handles the service call from SERVICE_SET_TEXT_OVERLAY to set profile mode to day/night """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_service_set_text_overlay(channel, group, text1, text2, text3, text4) async def async_set_service_set_custom_overlay(self, group: int, text1: str, text2: str): """ Handles the service call from SERVICE_SET_CUSTOM_OVERLAY to set profile mode to day/night """ - channel = self._coordinator.get_channel() + channel = self._logical_channel await self._coordinator.client.async_set_service_set_custom_overlay(channel, group, text1, text2) diff --git a/custom_components/dahua/client.py b/custom_components/dahua/client.py index 7d573c8..36c98bf 100644 --- a/custom_components/dahua/client.py +++ b/custom_components/dahua/client.py @@ -6,6 +6,7 @@ import async_timeout from .digest import DigestAuth +from .rpc2 import DahuaRpc2Client from hashlib import md5 from urllib.parse import quote @@ -316,6 +317,78 @@ async def async_get_ptz_position(self) -> dict: url = "/cgi-bin/ptz.cgi?action=getStatus" return await self.get(url) + @staticmethod + def parse_ptz_preset_ids(data: list) -> list[int]: + """Return sorted positive preset IDs from ptz.getPresets.""" + preset_ids: set[int] = set() + if not isinstance(data, list): + return [] + for preset in data: + if not isinstance(preset, dict): + continue + value = preset.get("Index") + if isinstance(value, bool): + continue + try: + preset_id = int(value) + except (TypeError, ValueError): + continue + if preset_id > 0: + preset_ids.add(preset_id) + return sorted(preset_ids) + + @staticmethod + def _new_rpc2_session() -> aiohttp.ClientSession: + """Use an isolated RPC2 session whose cookie jar accepts IP hosts.""" + return aiohttp.ClientSession( + connector=aiohttp.TCPConnector(enable_cleanup_closed=True, ssl=False), + cookie_jar=aiohttp.CookieJar(unsafe=True), + ) + + async def async_get_ptz_preset_ids(self, channel_index: int) -> list[int]: + """Read the real preset IDs exposed by Web5.0 RPC2.""" + async with self._new_rpc2_session() as session: + rpc2 = DahuaRpc2Client( + self._username, self._password, self._address, self._port, + self._rtsp_port, session + ) + try: + async with async_timeout.timeout(5): + presets = await rpc2.async_get_ptz_presets(channel_index) + ids = self.parse_ptz_preset_ids(presets) + if presets and not ids: + raise ValueError("Dahua RPC2 preset response contains no valid IDs") + return ids + finally: + try: + async with async_timeout.timeout(3): + logout_ok = await rpc2.logout() + if not logout_ok: + _LOGGER.debug( + "RPC2 logout reported failure after preset discovery" + ) + except Exception: + _LOGGER.debug("RPC2 logout failed after preset discovery", exc_info=True) + + async def async_goto_preset_rpc2(self, channel: int, position: int) -> dict: + """Go to a real preset through the hardware-validated RPC2 contract.""" + async with self._new_rpc2_session() as session: + rpc2 = DahuaRpc2Client( + self._username, self._password, self._address, self._port, + self._rtsp_port, session + ) + try: + async with async_timeout.timeout(5): + return await rpc2.async_goto_preset_position(channel, position) + finally: + try: + async with async_timeout.timeout(3): + logout_ok = await rpc2.logout() + if not logout_ok: + _LOGGER.debug("RPC2 logout reported failure after GotoPreset") + except Exception: + _LOGGER.debug("RPC2 logout failed after GotoPreset", exc_info=True) + async def async_get_light_global_enabled(self) -> dict: """ Returns the state of the Amcrest blue ring light (if it's on or off) diff --git a/custom_components/dahua/model_profiles.py b/custom_components/dahua/model_profiles.py new file mode 100644 index 0000000..e234ccc --- /dev/null +++ b/custom_components/dahua/model_profiles.py @@ -0,0 +1,14 @@ +"""Small model-specific helpers for upstream Dahua compatibility.""" +from __future__ import annotations + +import re + +_SDT4E425_RE = re.compile( + r"^(?:DH-)?SDT4E425-4F-GB-A-PV1(?:-.+)?$", + re.IGNORECASE, +) + + +def is_sdt4e425(model: str | None) -> bool: + """Return True for the hardware-validated Dahua SDT4E425 model family.""" + return bool(_SDT4E425_RE.match((model or "").strip())) diff --git a/custom_components/dahua/rpc2.py b/custom_components/dahua/rpc2.py index f5a7f85..cbaacde 100644 --- a/custom_components/dahua/rpc2.py +++ b/custom_components/dahua/rpc2.py @@ -9,10 +9,10 @@ import sys import aiohttp - from custom_components.dahua.models import CoaxialControlIOStatus _LOGGER: logging.Logger = logging.getLogger(__package__) +_PARAMS_UNSET = object() if sys.version_info > (3, 0): unicode = str @@ -33,15 +33,16 @@ def __init__( self._session = session self._rtsp_port = rtsp_port self._session_id = None + self._ptz_objects: dict[int, int] = {} self._id = 0 protocol = "https" if int(port) == 443 else "http" self._base = "{0}://{1}:{2}".format(protocol, address, port) - async def request(self, method, params=None, object_id=None, extra=None, url=None, verify_result=True): + async def request(self, method, params=_PARAMS_UNSET, object_id=None, extra=None, url=None, verify_result=True): """Make an RPC request.""" self._id += 1 data = {'method': method, 'id': self._id} - if params is not None: + if params is not _PARAMS_UNSET: data['params'] = params if object_id: data['object'] = object_id @@ -52,11 +53,24 @@ async def request(self, method, params=None, object_id=None, extra=None, url=Non if not url: url = "{0}/RPC2".format(self._base) - resp = await self._session.post(url, data=json.dumps(data)) + resp = await self._session.post(url, json=data) resp_json = json.loads(await resp.text()) if verify_result and resp_json['result'] is False: - raise ConnectionError(str(resp)) + error = resp_json.get("error") + details = [] + if isinstance(error, dict): + if error.get("code") is not None: + details.append("code={0}".format(error["code"])) + if isinstance(error.get("message"), str): + message = error["message"].replace("\r", " ").replace("\n", " ") + details.append("message={0}".format(message[:200])) + suffix = " ({0})".format(", ".join(details)) if details else "" + raise ConnectionError( + "Dahua RPC2 method {0} returned result=false{1}".format( + method, suffix + ) + ) return resp_json @@ -69,17 +83,21 @@ async def login(self): # login1: get session, realm & random for real login self._session_id = None + self._ptz_objects.clear() self._id = 0 url = '{0}/RPC2_Login'.format(self._base) method = "global.login" params = {'userName': self._username, 'password': "", - 'clientType': "Dahua3.0-Web3.0"} - r = await self.request(method=method, params=params, url=url, verify_result=False) + 'clientType': "Web5.0"} + r = await self.request( + method=method, params=params, url=url, verify_result=False + ) self._session_id = r['session'] realm = r['params']['realm'] random = r['params']['random'] + authority_type = r['params'].get('encryption') or "Default" # Password encryption algorithm. Reversed from rpcCore.getAuthByType pwd_phrase = self._username + ":" + realm + ":" + self._password @@ -94,22 +112,99 @@ async def login(self): # login2: the real login params = {'userName': self._username, 'password': pass_hash, - 'clientType': "Dahua3.0-Web3.0", - 'authorityType': "Default", - 'passwordType': "Default"} - return await self.request(method=method, params=params, url=url) + 'clientType': "Web5.0", + 'realm': realm, + 'random': random, + 'passwordType': "Default", + 'authorityType': authority_type} + response = await self.request(method=method, params=params, url=url) + authenticated_session = response.get('session') + if not isinstance(authenticated_session, str) or not authenticated_session: + raise ConnectionError( + "Dahua RPC2 authenticated login response is missing session" + ) + self._session_id = authenticated_session + _LOGGER.debug("RPC2 login succeeded") + return response async def logout(self) -> bool: """Logs out of the current session. Returns true if the logout was successful""" + if not self._session_id: + self._ptz_objects.clear() + return True try: response = await self.request(method="global.logout") if response['result'] is True: + _LOGGER.debug("RPC2 logout succeeded") return True - else: - _LOGGER.debug("Failed to log out of Dahua device %s", self._base) - return False - except Exception as exception: + _LOGGER.debug("RPC2 logout reported result=false") + return False + except Exception: + _LOGGER.debug("RPC2 logout failed", exc_info=True) return False + finally: + self._session_id = None + self._ptz_objects.clear() + + async def async_get_ptz_object(self, channel: int) -> int: + """Return the session-scoped PTZ object for one logical channel.""" + if channel in self._ptz_objects: + return self._ptz_objects[channel] + if not self._session_id: + await self.login() + response = await self.request( + method="ptz.factory.instance", + params={"channel": channel}, + ) + object_id = response.get("result") + if ( + isinstance(object_id, bool) + or not isinstance(object_id, int) + or object_id <= 0 + ): + raise ConnectionError("Dahua RPC2 returned an invalid PTZ object") + self._ptz_objects[channel] = object_id + _LOGGER.debug("RPC2 ptz.factory.instance succeeded channel=%d", channel) + return object_id + + async def async_goto_preset_position(self, channel: int, position: int) -> dict: + """Move to a preset using the exact Dahua RPC2 GotoPreset contract.""" + object_id = await self.async_get_ptz_object(channel) + response = await self.request( + method="ptz.start", + object_id=object_id, + params={ + "code": "GotoPreset", + "arg1": position, + "arg2": 0, + "arg3": 0, + }, + ) + _LOGGER.debug( + "RPC2 GotoPreset succeeded channel=%d preset_id=%d", + channel, + position, + ) + return response + + async def async_get_ptz_presets(self, channel: int) -> list[dict]: + """Return the firmware's real presets for one dynamic PTZ object.""" + object_id = await self.async_get_ptz_object(channel) + response = await self.request( + method="ptz.getPresets", + object_id=object_id, + params=None, + ) + params = response.get("params") + if not isinstance(params, dict) or not isinstance(params.get("presets"), list): + raise ValueError("Dahua RPC2 response is missing params.presets") + presets = params["presets"] + _LOGGER.debug( + "RPC2 ptz.getPresets succeeded channel=%d preset_count=%d", + channel, + len(presets), + ) + return presets async def current_time(self): """Get the current time on the device.""" diff --git a/custom_components/dahua/select.py b/custom_components/dahua/select.py index ee53b13..f02d5de 100755 --- a/custom_components/dahua/select.py +++ b/custom_components/dahua/select.py @@ -1,33 +1,46 @@ -""" -Select entity platform for dahua. -https://developers.home-assistant.io/docs/core/entity/select -Requires HomeAssistant 2021.7.0 or greater -""" +"""Select entity platform for Dahua.""" +import logging + from homeassistant.core import HomeAssistant from homeassistant.components.select import SelectEntity from custom_components.dahua import DahuaDataUpdateCoordinator from .const import DOMAIN from .entity import DahuaBaseEntity +from .model_profiles import is_sdt4e425 + +_LOGGER = logging.getLogger(__package__) async def async_setup_entry(hass: HomeAssistant, entry, async_add_devices): """Setup select platform.""" coordinator: DahuaDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] - devices = [] if coordinator.is_amcrest_doorbell() and coordinator.supports_security_light(): devices.append(DahuaDoorbellLightSelect(coordinator, entry)) - #if coordinator._supports_ptz_position: - devices.append(DahuaCameraPresetPositionSelect(coordinator, entry)) - + if is_sdt4e425(coordinator.get_model()): + try: + preset_ids = await coordinator.client.async_get_ptz_preset_ids(1) + except Exception: + _LOGGER.warning( + "Unable to enumerate SDT4E425 presets through RPC2", exc_info=True + ) + preset_ids = [] + devices.append( + DahuaCameraPresetPositionSelect( + coordinator, entry, preset_ids=preset_ids, rpc2_channel=1 + ) + ) + else: + devices.append(DahuaCameraPresetPositionSelect(coordinator, entry)) + async_add_devices(devices) class DahuaDoorbellLightSelect(DahuaBaseEntity, SelectEntity): - """allows one to turn the doorbell light on/off/strobe""" + """Allow one to turn the doorbell light on/off/strobe.""" def __init__(self, coordinator: DahuaDataUpdateCoordinator, config_entry): DahuaBaseEntity.__init__(self, coordinator, config_entry) @@ -41,13 +54,10 @@ def __init__(self, coordinator: DahuaDataUpdateCoordinator, config_entry): def current_option(self) -> str: mode = self._coordinator.data.get("table.Lighting_V2[0][0][1].Mode", "") state = self._coordinator.data.get("table.Lighting_V2[0][0][1].State", "") - if mode == "ForceOn" and state == "On": return "On" - if mode == "ForceOn" and state == "Flicker": return "Strobe" - return "Off" async def async_select_option(self, option: str) -> None: @@ -60,31 +70,49 @@ def name(self): @property def unique_id(self): - """ https://developers.home-assistant.io/docs/entity_registry_index/#unique-id-requirements """ return self._attr_unique_id class DahuaCameraPresetPositionSelect(DahuaBaseEntity, SelectEntity): - """allows """ + """Select a camera preset position.""" - def __init__(self, coordinator: DahuaDataUpdateCoordinator, config_entry): + def __init__( + self, coordinator: DahuaDataUpdateCoordinator, config_entry, + *, preset_ids: list[int] | None = None, rpc2_channel: int | None = None, + ): DahuaBaseEntity.__init__(self, coordinator, config_entry) SelectEntity.__init__(self) self._coordinator = coordinator + self._rpc2_channel = rpc2_channel self._attr_name = f"{coordinator.get_device_name()} Preset Position" - self._attr_unique_id = f"{coordinator.get_serial_number()}_preset_position" - self._attr_options = ["Manual","1","2","3","4","5","6","7","8","9","10"] + suffix = "1_preset_position" if rpc2_channel == 1 else "preset_position" + self._attr_unique_id = f"{coordinator.get_serial_number()}_{suffix}" + if preset_ids is None: + self._attr_options = ["Manual", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10"] + else: + self._attr_options = ["Manual", *[str(value) for value in preset_ids]] @property def current_option(self) -> str: - presetID = self._coordinator.data.get("status.PresetID", "0") - if presetID == "0": + if self._rpc2_channel is not None: + # This firmware has no supported CGI position readback. Do not claim + # a position we cannot verify. + return "Manual" + preset_id = self._coordinator.data.get("status.PresetID", "0") + if preset_id == "0": return "Manual" - return presetID + return preset_id async def async_select_option(self, option: str) -> None: - channel = self._coordinator.get_channel() - await self._coordinator.client.async_goto_preset_position(channel, int(option)) + if option == "Manual": + return + if self._rpc2_channel is not None: + await self._coordinator.client.async_goto_preset_rpc2( + self._rpc2_channel, int(option) + ) + else: + channel = self._coordinator.get_channel() + await self._coordinator.client.async_goto_preset_position(channel, int(option)) await self._coordinator.async_refresh() @property @@ -93,5 +121,4 @@ def name(self): @property def unique_id(self): - """ https://developers.home-assistant.io/docs/entity_registry_index/#unique-id-requirements """ return self._attr_unique_id From afccaf49599d71e6c5e0eb2c3d5e8b94768c937b Mon Sep 17 00:00:00 2001 From: Realcrash <131798491+Realcrash@users.noreply.github.com> Date: Wed, 5 Aug 2026 20:33:38 +0200 Subject: [PATCH 2/2] Add tests for SDT4E425 support --- tests/dahua/test_issue589.py | 246 +++++++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 tests/dahua/test_issue589.py diff --git a/tests/dahua/test_issue589.py b/tests/dahua/test_issue589.py new file mode 100644 index 0000000..cfed445 --- /dev/null +++ b/tests/dahua/test_issue589.py @@ -0,0 +1,246 @@ +"""Tests for SDT4E425 dual-sensor and RPC2 preset support.""" + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from custom_components.dahua import camera as camera_platform +from custom_components.dahua import select as select_platform +from custom_components.dahua.client import DahuaClient +from custom_components.dahua.const import DOMAIN +from custom_components.dahua.model_profiles import is_sdt4e425 +from custom_components.dahua.rpc2 import DahuaRpc2Client + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + async def text(self): + return json.dumps(self._payload) + + +class _FakeSession: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + async def post(self, url, json): + self.calls.append((url, json)) + return _FakeResponse(self.responses.pop(0)) + + +class _FakePlatform: + def async_register_entity_service(self, *args, **kwargs): + return None + + +class _FakeClient: + @staticmethod + def to_stream_name(subtype): + return ("Main", "Sub", "Sub_2")[subtype] + + +class _FakeCoordinator: + def __init__(self, model="DH-SDT4E425-4F-GB-A-PV1"): + self.client = _FakeClient() + self._model = model + + def get_model(self): + return self._model + + def get_max_streams(self): + return 3 + + +class _CapturedCamera: + def __init__( + self, + coordinator, + stream_index, + config_entry, + *, + logical_channel=None, + media_channel=None, + display_name=None, + unique_suffix=None, + ): + self.stream_index = stream_index + self.logical_channel = logical_channel + self.media_channel = media_channel + self.display_name = display_name + self.unique_suffix = unique_suffix + + +def test_model_profile_is_narrow(): + assert is_sdt4e425("DH-SDT4E425-4F-GB-A-PV1") + assert is_sdt4e425("SDT4E425-4F-GB-A-PV1") + assert is_sdt4e425("DH-SDT4E425-4F-GB-A-PV1-S2") + assert not is_sdt4e425("SDT4E425") + assert not is_sdt4e425(None) + + +def test_sdt4e425_creates_two_sensors_with_three_streams(monkeypatch): + monkeypatch.setattr(camera_platform, "DahuaCamera", _CapturedCamera) + monkeypatch.setattr( + camera_platform.entity_platform, + "async_get_current_platform", + lambda: _FakePlatform(), + ) + + coordinator = _FakeCoordinator() + hass = SimpleNamespace(data={DOMAIN: {"entry": coordinator}}) + entry = SimpleNamespace(entry_id="entry", title="Camera") + entities = [] + + asyncio.run(camera_platform.async_setup_entry(hass, entry, entities.extend)) + + assert [ + ( + entity.logical_channel, + entity.media_channel, + entity.stream_index, + entity.display_name, + entity.unique_suffix, + ) + for entity in entities + ] == [ + (0, 1, 0, "Panorama", "Main"), + (0, 1, 1, "Panorama Sub", "Sub"), + (0, 1, 2, "Panorama Sub_2", "Sub_2"), + (1, 2, 0, "PTZ", "1_Main"), + (1, 2, 1, "PTZ Sub", "1_Sub"), + (1, 2, 2, "PTZ Sub_2", "1_Sub_2"), + ] + + +def test_other_models_keep_native_stream_setup(monkeypatch): + monkeypatch.setattr(camera_platform, "DahuaCamera", _CapturedCamera) + monkeypatch.setattr( + camera_platform.entity_platform, + "async_get_current_platform", + lambda: _FakePlatform(), + ) + + coordinator = _FakeCoordinator(model="OTHER") + hass = SimpleNamespace(data={DOMAIN: {"entry": coordinator}}) + entry = SimpleNamespace(entry_id="entry", title="Camera") + entities = [] + + asyncio.run(camera_platform.async_setup_entry(hass, entry, entities.extend)) + + assert len(entities) == 3 + assert [entity.stream_index for entity in entities] == [0, 1, 2] + assert all(entity.logical_channel is None for entity in entities) + assert all(entity.media_channel is None for entity in entities) + + +def test_rpc2_web5_login_promotes_authenticated_session(): + session = _FakeSession( + [ + { + "result": False, + "session": "S1", + "params": { + "realm": "realm", + "random": "random", + "encryption": "Default", + }, + }, + {"result": True, "session": "S2", "params": {}}, + ] + ) + client = DahuaRpc2Client("user", "password", "192.0.2.1", 80, 554, session) + + asyncio.run(client.login()) + + assert client._session_id == "S2" + assert session.calls[0][1]["params"]["clientType"] == "Web5.0" + assert session.calls[1][1]["session"] == "S1" + + +def test_rpc2_get_presets_sends_explicit_null_params(): + session = _FakeSession( + [ + { + "result": True, + "params": {"presets": [{"Index": 1}, {"Index": 5}]}, + } + ] + ) + client = DahuaRpc2Client("user", "password", "192.0.2.1", 80, 554, session) + client._session_id = "S2" + client._ptz_objects[1] = 42 + + presets = asyncio.run(client.async_get_ptz_presets(1)) + + assert [preset["Index"] for preset in presets] == [1, 5] + payload = session.calls[0][1] + assert payload["method"] == "ptz.getPresets" + assert payload["object"] == 42 + assert "params" in payload + assert payload["params"] is None + + +def test_rpc2_goto_preset_uses_observed_payload(): + session = _FakeSession([{"result": True, "params": {}}]) + client = DahuaRpc2Client("user", "password", "192.0.2.1", 80, 554, session) + client._session_id = "S2" + client._ptz_objects[1] = 42 + + asyncio.run(client.async_goto_preset_position(1, 3)) + + payload = session.calls[0][1] + assert payload["method"] == "ptz.start" + assert payload["object"] == 42 + assert payload["params"] == { + "code": "GotoPreset", + "arg1": 3, + "arg2": 0, + "arg3": 0, + } + + +def test_parse_ptz_preset_ids_filters_and_sorts(): + assert DahuaClient.parse_ptz_preset_ids( + [ + {"Index": "5"}, + {"Index": 1}, + {"Index": 5}, + {"Index": 0}, + {"Index": True}, + {"Index": "invalid"}, + {}, + ] + ) == [1, 5] + + +def test_sdt4e425_select_uses_real_preset_ids(monkeypatch): + monkeypatch.setattr( + select_platform.DahuaBaseEntity, + "__init__", + lambda self, coordinator, config_entry: None, + ) + coordinator = SimpleNamespace( + client=SimpleNamespace(async_goto_preset_rpc2=AsyncMock()), + async_refresh=AsyncMock(), + get_device_name=lambda: "Camera", + get_serial_number=lambda: "SERIAL", + ) + + entity = select_platform.DahuaCameraPresetPositionSelect( + coordinator, + SimpleNamespace(), + preset_ids=[1, 3, 5], + rpc2_channel=1, + ) + + assert entity.unique_id == "SERIAL_1_preset_position" + assert entity.options == ["Manual", "1", "3", "5"] + assert entity.current_option == "Manual" + + asyncio.run(entity.async_select_option("3")) + + coordinator.client.async_goto_preset_rpc2.assert_awaited_once_with(1, 3) + coordinator.async_refresh.assert_awaited_once()