limit requests in config flow (#101)

custom_components/hikvision_next/__init__.py

@@ -47,7 +47,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     try:
         await isapi.get_hardware_info()
         await isapi.get_cameras()
-        device_info = isapi.get_device_info()
+        device_info = isapi.hass_device_info()
         device_registry = dr.async_get(hass)
         device_registry.async_get_or_create(config_entry_id=entry.entry_id, **device_info)
     except (asyncio.TimeoutError, TimeoutException) as ex:

custom_components/hikvision_next/binary_sensor.py

@@ -43,4 +43,4 @@ def __init__(self, isapi, device_id: int, event: EventInfo) -> None:
         self._attr_unique_id = self.entity_id
         self._attr_name = f"{EVENTS[event.id]['label']}{' ' + str(event.io_port_id) if event.io_port_id != 0 else ''}"
         self._attr_device_class = EVENTS[event.id]["device_class"]
-        self._attr_device_info = isapi.get_device_info(device_id)
+        self._attr_device_info = isapi.hass_device_info(device_id)

custom_components/hikvision_next/camera.py

@@ -40,7 +40,7 @@ def __init__(
         """Initialize Hikvision camera stream."""
         Camera.__init__(self)
 
-        self._attr_device_info = isapi.get_device_info(camera.id)
+        self._attr_device_info = isapi.hass_device_info(camera.id)
         self._attr_name = f"{camera.name} {stream_info.type}"
         self._attr_unique_id = slugify(f"{isapi.device_info.serial_no.lower()}_{stream_info.id}")
         self.entity_id = f"camera.{self.unique_id}"

custom_components/hikvision_next/config_flow.py

@@ -58,7 +58,7 @@ async def async_step_user(self, user_input: dict[str, Any] | None = None) -> Flo
                 password = user_input[CONF_PASSWORD]
 
                 isapi = ISAPI(host, username, password)
-                await isapi.get_hardware_info()
+                await isapi.get_device_info()
 
                 if self._reauth_entry:
                     self.hass.config_entries.async_update_entry(self._reauth_entry, data=user_input)

custom_components/hikvision_next/isapi.py

@@ -181,11 +181,25 @@ def __init__(self, host: str, username: str, password: str) -> None:
         self.cameras: list[IPCamera | AnalogCamera] = []
         self.supported_events: list[SupportedEventsInfo] = []
 
+    async def get_device_info(self):
+        """Get device info"""
+        hw_info = (await self.isapi.System.deviceInfo(method=GET)).get("DeviceInfo", {})
+        _LOGGER.debug("%s/ISAPI/System/deviceInfo %s", self.isapi.host, hw_info)
+        self.device_info = HikDeviceInfo(
+            name=hw_info.get("deviceName"),
+            manufacturer=str(hw_info.get("manufacturer", "Hikvision")).title(),
+            model=hw_info.get("model"),
+            serial_no=hw_info.get("serialNumber"),
+            firmware=hw_info.get("firmwareVersion"),
+            mac_address=hw_info.get("macAddress"),
+            ip_address=urlparse(self.host).hostname,  # type: ignore
+            device_type=hw_info.get("deviceType"),
+        )
+
     async def get_hardware_info(self):
         """Get base device data."""
         # Get base hw info
-        hw_info = (await self.isapi.System.deviceInfo(method=GET)).get("DeviceInfo", {})
-        _LOGGER.debug("%s/ISAPI/System/deviceInfo %s", self.isapi.host, hw_info)
+        await self.get_device_info()
 
         # Get device capabilities
         capabilities = (await self.isapi.System.capabilities(method=GET)).get("DeviceCap", {})
@@ -195,28 +209,19 @@ async def get_hardware_info(self):
         self.supported_events = await self.get_supported_events_info()
 
         # Set DeviceInfo
-        self.device_info = HikDeviceInfo(
-            name=hw_info.get("deviceName"),
-            manufacturer=str(hw_info.get("manufacturer", "Hikvision")).title(),
-            model=hw_info.get("model"),
-            serial_no=hw_info.get("serialNumber"),
-            firmware=hw_info.get("firmwareVersion"),
-            mac_address=hw_info.get("macAddress"),
-            ip_address=urlparse(self.host).hostname,  # type: ignore
-            device_type=hw_info.get("deviceType"),
-            support_analog_cameras=int(deep_get(capabilities, "SysCap.VideoCap.videoInputPortNums", 0)),
-            support_digital_cameras=int(deep_get(capabilities, "RacmCap.inputProxyNums", 0)),
-            support_holiday_mode=deep_get(capabilities, "SysCap.isSupportHolidy", False),
-            support_alarm_server=bool(await self.get_alarm_server()),
-            support_channel_zero=deep_get(capabilities, "RacmCap.isSupportZeroChan", False),
-            support_event_mutex_checking=capabilities.get("isSupportGetmutexFuncErrMsg", False),
-            input_ports=int(deep_get(capabilities, "SysCap.IOCap.IOInputPortNums", 0)),
-            output_ports=int(deep_get(capabilities, "SysCap.IOCap.IOOutputPortNums", 0)),
-            storage=await self.get_storage_devices(),
-            supported_events=await self.get_device_event_capabilities(
-                self.supported_events, hw_info.get("serialNumber"), 0
-            )
+        self.device_info.support_analog_cameras = int(deep_get(capabilities, "SysCap.VideoCap.videoInputPortNums", 0))
+        self.device_info.support_digital_cameras = int(deep_get(capabilities, "RacmCap.inputProxyNums", 0))
+        self.device_info.support_holiday_mode = deep_get(capabilities, "SysCap.isSupportHolidy", False)
+        self.device_info.support_channel_zero = deep_get(capabilities, "RacmCap.isSupportZeroChan", False)
+        self.device_info.support_event_mutex_checking = capabilities.get("isSupportGetmutexFuncErrMsg", False)
+        self.device_info.input_ports = int(deep_get(capabilities, "SysCap.IOCap.IOInputPortNums", 0))
+        self.device_info.output_ports = int(deep_get(capabilities, "SysCap.IOCap.IOOutputPortNums", 0))
+
+        self.device_info.storage = await self.get_storage_devices()
+        self.device_info.supported_events = await self.get_device_event_capabilities(
+            self.supported_events, self.device_info.serial_no, 0
         )
+        self.device_info.support_alarm_server = bool(await self.get_alarm_server())
 
         await self.get_protocols()
 
@@ -329,10 +334,7 @@ async def get_cameras(self):
                             connection_type=CONNECTION_TYPE_DIRECT,
                             streams=await self.get_camera_streams(camera_id),
                             supported_events=await self.get_device_event_capabilities(
-                                self.supported_events,
-                                self.device_info.serial_no,
-                                camera_id,
-                                CONNECTION_TYPE_DIRECT,
+                                self.supported_events, self.device_info.serial_no, camera_id, CONNECTION_TYPE_DIRECT
                             ),
                         )
                     )
@@ -372,22 +374,19 @@ async def get_device_event_capabilities(
         events = []
 
         if device_id == 0:  # NVR
-            device_supported_events = [s for s in supported_events if (
-                s.event_id in EVENTS and EVENTS[s.event_id].get("type") == EVENT_IO
-            )]
+            device_supported_events = [
+                s for s in supported_events if (s.event_id in EVENTS and EVENTS[s.event_id].get("type") == EVENT_IO)
+            ]
         else:  # Camera
-            device_supported_events = [s for s in supported_events if (
-                s.channel_id == int(device_id)
-                and s.event_id in EVENTS
-            )]
+            device_supported_events = [
+                s for s in supported_events if (s.channel_id == int(device_id) and s.event_id in EVENTS)
+            ]
 
         for event in device_supported_events:
             # Build unique_id
             device_id_param = f"_{device_id}" if device_id != 0 else ""
             io_port_id_param = f"_{event.io_port_id}" if event.io_port_id != 0 else ""
-            unique_id = (
-                f"{slugify(serial_no.lower())}{device_id_param}{io_port_id_param}_{event.event_id}"
-            )
+            unique_id = f"{slugify(serial_no.lower())}{device_id_param}{io_port_id_param}_{event.event_id}"
 
             if EVENTS.get(event.event_id):
                 event_info = EventInfo(
@@ -546,8 +545,8 @@ def get_storage_device_by_id(self, device_id: int) -> HDDInfo | None:
             # Storage id does not exist
             return None
 
-    def get_device_info(self, device_id: int = 0) -> DeviceInfo:
-        """Return device registry information."""
+    def hass_device_info(self, device_id: int = 0) -> DeviceInfo:
+        """Return Home Assistant entity device information."""
         if device_id == 0:
             return DeviceInfo(
                 manufacturer=self.device_info.manufacturer,

custom_components/hikvision_next/sensor.py

@@ -61,7 +61,7 @@ def __init__(self, coordinator, key: str) -> None:
         isapi = coordinator.isapi
         self._attr_unique_id = f"{isapi.device_info.serial_no}_{DATA_ALARM_SERVER_HOST}_{key}"
         self.entity_id = ENTITY_ID_FORMAT.format(self.unique_id)
-        self._attr_device_info = isapi.get_device_info()
+        self._attr_device_info = isapi.hass_device_info()
         self._attr_name = ALARM_SERVER_SENSOR_LABEL_FORMAT.format(ALARM_SERVER_SETTINGS[key])
         self.key = key
 
@@ -83,7 +83,7 @@ def __init__(self, coordinator, hdd: HDDInfo) -> None:
         isapi = coordinator.isapi
         self._attr_unique_id = f"{isapi.device_info.serial_no}_{hdd.id}_{hdd.name}"
         self.entity_id = ENTITY_ID_FORMAT.format(self.unique_id)
-        self._attr_device_info = isapi.get_device_info()
+        self._attr_device_info = isapi.hass_device_info()
         self._attr_name = f"HDD {hdd.id}"
         self.hdd = hdd
 

custom_components/hikvision_next/switch.py

@@ -63,7 +63,7 @@ def __init__(self, device_id: int, event: EventInfo, coordinator) -> None:
         super().__init__(coordinator)
         self.entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
         self._attr_unique_id = self.entity_id
-        self._attr_device_info = coordinator.isapi.get_device_info(device_id)
+        self._attr_device_info = coordinator.isapi.hass_device_info(device_id)
         self._attr_name = EVENT_SWITCH_LABEL_FORMAT.format(
             f"{EVENTS[event.id]['label']}{' ' + str(event.io_port_id) if event.io_port_id != 0 else ''}"
         )
@@ -109,7 +109,7 @@ def __init__(self, coordinator, port_no: int) -> None:
             f"{slugify(coordinator.isapi.device_info.serial_no.lower())}_{port_no}_alarm_output"
         )
         self._attr_unique_id = self.entity_id
-        self._attr_device_info = coordinator.isapi.get_device_info(0)
+        self._attr_device_info = coordinator.isapi.hass_device_info(0)
         self._attr_name = f"Alarm Output {port_no}"
         self._port_no = port_no
 
@@ -144,7 +144,7 @@ def __init__(self, coordinator) -> None:
         super().__init__(coordinator)
         self._attr_unique_id = f"{slugify(coordinator.isapi.device_info.serial_no.lower())}_{HOLIDAY_MODE}"
         self.entity_id = ENTITY_ID_FORMAT.format(self.unique_id)
-        self._attr_device_info = coordinator.isapi.get_device_info()
+        self._attr_device_info = coordinator.isapi.hass_device_info()
         self._attr_name = HOLIDAY_MODE_SWITCH_LABEL
 
     @property