Fix/refactoring isapi (#236)

* tests fix for python 3.12

* make use of config entry runtime_data

* isapi models refactoring

* improved getting of single element arrays from xml

* moved http client into isapi

* remove HA imports from isapi

* events and constants refactoring

* _get_event_notification_host simplified

* separated the ISAPI client from the HA integration

* use python 3.12 in CI

* HA core version sync for test env

.github/workflows/testing.yml

@@ -9,7 +9,7 @@ jobs:
     strategy:
       max-parallel: 4
       matrix:
-        python-version: ["3.11"]
+        python-version: ["3.12"]
 
     steps:
       - uses: actions/checkout@v4

.pre-commit-config.yaml

@@ -1,13 +1,13 @@
 repos:
 
--   repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: 'v0.0.261'
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.6.9
     hooks:
     -   id: ruff
         args: [--fix, --show-fixes, --exit-non-zero-on-fix]
         exclude: ^.*\b(assets)\b.*$
 
--   repo: https://github.com/pre-commit/pre-commit-hooks
+  - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.3.0
     hooks:
     -   id: end-of-file-fixer

custom_components/hikvision_next/__init__.py

@@ -14,24 +14,14 @@
 )
 from homeassistant.components.switch import ENTITY_ID_FORMAT as SWITCH_ENTITY_ID_FORMAT
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
+from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr, entity_registry as er
-from homeassistant.helpers.httpx_client import get_async_client
 from homeassistant.helpers.typing import ConfigType
 
-from .const import (
-    ALARM_SERVER_PATH,
-    DATA_ALARM_SERVER_HOST,
-    DATA_ISAPI,
-    DATA_SET_ALARM_SERVER,
-    DOMAIN,
-    EVENTS_COORDINATOR,
-    SECONDARY_COORDINATOR,
-)
-from .coordinator import EventsCoordinator, SecondaryCoordinator
-from .isapi import ISAPI
+from .const import DOMAIN
+from .hikvision_device import HikvisionDevice
 from .notifications import EventNotificationsView
 from .services import setup_services
 
@@ -45,6 +35,8 @@
     Platform.IMAGE,
 ]
 
+type HikvisionConfigEntry = ConfigEntry[HikvisionDevice]
+
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the Hikvision component."""
@@ -54,58 +46,35 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     return True
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool:
     """Set up integration from a config entry."""
-    hass.data.setdefault(DOMAIN, {})
-
-    host = entry.data[CONF_HOST]
-    username = entry.data[CONF_USERNAME]
-    password = entry.data[CONF_PASSWORD]
-    session = get_async_client(hass)
-    isapi = ISAPI(host, username, password, session)
-    isapi.pending_initialization = True
+    device = HikvisionDevice(hass, entry)
+    device.pending_initialization = True
     try:
-        await isapi.get_hardware_info()
-        await isapi.get_cameras()
-        device_info = isapi.hass_device_info()
+        await device.get_hardware_info()
+        device_info = device.hass_device_info()
         device_registry = dr.async_get(hass)
         device_registry.async_get_or_create(config_entry_id=entry.entry_id, **device_info)
     except (TimeoutError, TimeoutException) as ex:
-        raise ConfigEntryNotReady(f"Timeout while connecting to {host}. Cannot initialize {DOMAIN}") from ex
+        raise ConfigEntryNotReady(f"Timeout while connecting to {device.host}. Cannot initialize {DOMAIN}") from ex
     except Exception as ex:  # pylint: disable=broad-except
-        msg = f"Cannot initialize {DOMAIN} {host}. Error: {ex}\n"
+        msg = f"Cannot initialize {DOMAIN} {device.host}. Error: {ex}\n"
         _LOGGER.error(msg + traceback.format_exc())
         raise ConfigEntryNotReady(msg) from ex
 
-    coordinators = {}
-
-    coordinators[EVENTS_COORDINATOR] = EventsCoordinator(hass, isapi)
-
-    if isapi.device_info.support_holiday_mode or isapi.device_info.support_alarm_server:
-        coordinators[SECONDARY_COORDINATOR] = SecondaryCoordinator(hass, isapi)
+    entry.runtime_data = device
 
-    hass.data[DOMAIN][entry.entry_id] = {
-        DATA_SET_ALARM_SERVER: entry.data[DATA_SET_ALARM_SERVER],
-        DATA_ALARM_SERVER_HOST: entry.data[DATA_ALARM_SERVER_HOST],
-        DATA_ISAPI: isapi,
-        **coordinators,
-    }
-
-    if entry.data[DATA_SET_ALARM_SERVER] and isapi.device_info.support_alarm_server:
-        await isapi.set_alarm_server(entry.data[DATA_ALARM_SERVER_HOST], ALARM_SERVER_PATH)
-
-    for coordinator in coordinators.values():
-        await coordinator.async_config_entry_first_refresh()
+    await device.init_coordinators()
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
-    isapi.pending_initialization = False
+    device.pending_initialization = False
 
     # Only initialise view once if multiple instances of integration
     if get_first_instance_unique_id(hass) == entry.unique_id:
         hass.http.register_view(EventNotificationsView(hass))
 
-    refresh_disabled_entities_in_registry(hass, isapi)
+    refresh_disabled_entities_in_registry(hass, device)
 
     return True
 
@@ -120,11 +89,9 @@ async def async_remove_config_entry_device(hass: HomeAssistant, config_entry, de
     return True
 
 
-async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+async def async_unload_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool:
     """Unload a config entry."""
 
-    config = hass.data[DOMAIN][entry.entry_id]
-
     # Unload a config entry
     unload_ok = all(
         await asyncio.gather(
@@ -133,13 +100,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     )
 
     # Reset alarm server after it has been set
-    if config[DATA_SET_ALARM_SERVER]:
-        isapi = config[DATA_ISAPI]
+    device = entry.runtime_data
+    if device.control_alarm_server_host:
         with suppress(Exception):
-            await isapi.set_alarm_server("http://0.0.0.0:80", "/")
-
-    if unload_ok:
-        hass.data[DOMAIN].pop(entry.entry_id)
+            await device.set_alarm_server("http://0.0.0.0:80", "/")
 
     return unload_ok
 
@@ -176,7 +140,7 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):
     return True
 
 
-def refresh_disabled_entities_in_registry(hass: HomeAssistant, isapi: ISAPI):
+def refresh_disabled_entities_in_registry(hass: HomeAssistant, device: HikvisionDevice):
     """Set disable state according to Notify Surveillance Center flag."""
 
     def update_entity(event, ENTITY_ID_FORMAT):
@@ -189,11 +153,11 @@ def update_entity(event, ENTITY_ID_FORMAT):
             entity_registry.async_update_entity(entity_id, disabled_by=disabled_by)
 
     entity_registry = er.async_get(hass)
-    for camera in isapi.cameras:
+    for camera in device.cameras:
         for event in camera.events_info:
             update_entity(event, SWITCH_ENTITY_ID_FORMAT)
             update_entity(event, BINARY_SENSOR_ENTITY_ID_FORMAT)
 
-    for event in isapi.device_info.events_info:
+    for event in device.events_info:
         update_entity(event, SWITCH_ENTITY_ID_FORMAT)
         update_entity(event, BINARY_SENSOR_ENTITY_ID_FORMAT)

custom_components/hikvision_next/binary_sensor.py

@@ -3,31 +3,36 @@
 from __future__ import annotations
 
 from homeassistant.components.binary_sensor import ENTITY_ID_FORMAT, BinarySensorEntity
-from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from .const import DATA_ISAPI, DOMAIN, EVENTS, EVENT_IO
+from . import HikvisionConfigEntry
+from .isapi.const import EVENT_IO
+from .const import EVENTS
+from .hikvision_device import HikvisionDevice
 from .isapi import EventInfo
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: HikvisionConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
     """Add binary sensors for hikvision events states."""
 
-    config = hass.data[DOMAIN][entry.entry_id]
-    isapi = config[DATA_ISAPI]
+    device = entry.runtime_data
 
     entities = []
 
     # Camera Events
-    for camera in isapi.cameras:
+    for camera in device.cameras:
         for event in camera.events_info:
-            entities.append(EventBinarySensor(isapi, camera.id, event))
+            entities.append(EventBinarySensor(device, camera.id, event))
 
     # NVR Events
-    if isapi.device_info.is_nvr:
-        for event in isapi.device_info.events_info:
-            entities.append(EventBinarySensor(isapi, 0, event))
+    if device.device_info.is_nvr:
+        for event in device.events_info:
+            entities.append(EventBinarySensor(device, 0, event))
 
     async_add_entities(entities)
 
@@ -38,13 +43,13 @@ class EventBinarySensor(BinarySensorEntity):
     _attr_has_entity_name = True
     _attr_is_on = False
 
-    def __init__(self, isapi, device_id: int, event: EventInfo) -> None:
+    def __init__(self, device: HikvisionDevice, device_id: int, event: EventInfo) -> None:
         """Initialize."""
         self.entity_id = ENTITY_ID_FORMAT.format(event.unique_id)
         self._attr_unique_id = self.entity_id
         self._attr_translation_key = event.id
         if event.id == EVENT_IO:
             self._attr_translation_placeholders = {"io_port_id": event.io_port_id}
         self._attr_device_class = EVENTS[event.id]["device_class"]
-        self._attr_device_info = isapi.hass_device_info(device_id)
+        self._attr_device_info = device.hass_device_info(device_id)
         self._attr_entity_registry_enabled_default = not event.disabled

custom_components/hikvision_next/camera.py

@@ -3,25 +3,26 @@
 from __future__ import annotations
 
 from homeassistant.components.camera import Camera, CameraEntityFeature
-from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.util import slugify
 
-from .const import DATA_ISAPI, DOMAIN
-from .isapi import ISAPI, AnalogCamera, CameraStreamInfo, IPCamera
+from . import HikvisionConfigEntry
+from .hikvision_device import HikvisionDevice
+from .isapi import AnalogCamera, CameraStreamInfo, IPCamera
 
 
-async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
+async def async_setup_entry(
+    hass: HomeAssistant, entry: HikvisionConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
     """Set up a Hikvision IP Camera."""
 
-    config = hass.data[DOMAIN][entry.entry_id]
-    isapi: ISAPI = config[DATA_ISAPI]
+    device = entry.runtime_data
 
     entities = []
-    for camera in isapi.cameras:
+    for camera in device.cameras:
         for stream in camera.streams:
-            entities.append(HikvisionCamera(isapi, camera, stream))
+            entities.append(HikvisionCamera(device, camera, stream))
 
     async_add_entities(entities)
 
@@ -33,17 +34,15 @@ class HikvisionCamera(Camera):
 
     def __init__(
         self,
-        isapi: ISAPI,
+        device: HikvisionDevice,
         camera: AnalogCamera | IPCamera,
         stream_info: CameraStreamInfo,
     ) -> None:
         """Initialize Hikvision camera stream."""
         Camera.__init__(self)
 
-        self._attr_device_info = isapi.hass_device_info(camera.id)
-        self._attr_unique_id = slugify(
-            f"{isapi.device_info.serial_no.lower()}_{stream_info.id}"
-        )
+        self._attr_device_info = device.hass_device_info(camera.id)
+        self._attr_unique_id = slugify(f"{device.device_info.serial_no.lower()}_{stream_info.id}")
         if stream_info.type_id > 1:
             self._attr_has_entity_name = True
             self._attr_translation_key = f"stream{stream_info.type_id}"
@@ -52,13 +51,13 @@ def __init__(
             # for the main stream use just its name
             self._attr_name = camera.name
         self.entity_id = f"camera.{self.unique_id}"
-        self.isapi = isapi
+        self.device = device
         self.stream_info = stream_info
 
     async def stream_source(self) -> str | None:
         """Return the source of the stream."""
-        return self.isapi.get_stream_source(self.stream_info)
+        return self.device.get_stream_source(self.stream_info)
 
     async def async_camera_image(self, width: int | None = None, height: int | None = None) -> bytes | None:
         """Return a still image response from the camera."""
-        return await self.isapi.get_camera_image(self.stream_info, width, height)
+        return await self.device.get_camera_image(self.stream_info, width, height)

custom_components/hikvision_next/config_flow.py

@@ -16,8 +16,8 @@
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers.httpx_client import get_async_client
 
-from .const import DATA_ALARM_SERVER_HOST, DATA_SET_ALARM_SERVER, DOMAIN
-from .isapi import ISAPI
+from .const import CONF_ALARM_SERVER_HOST, CONF_SET_ALARM_SERVER, DOMAIN
+from .isapi import ISAPIClient
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -38,12 +38,12 @@ async def get_schema(self, user_input: dict[str, Any]):
                 vol.Required(CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")): str,
                 vol.Required(CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")): str,
                 vol.Required(
-                    DATA_SET_ALARM_SERVER,
-                    default=user_input.get(DATA_SET_ALARM_SERVER, True),
+                    CONF_SET_ALARM_SERVER,
+                    default=user_input.get(CONF_SET_ALARM_SERVER, True),
                 ): bool,
                 vol.Required(
-                    DATA_ALARM_SERVER_HOST,
-                    default=user_input.get(DATA_ALARM_SERVER_HOST, f"http://{local_ip}:8123"),
+                    CONF_ALARM_SERVER_HOST,
+                    default=user_input.get(CONF_ALARM_SERVER_HOST, f"http://{local_ip}:8123"),
                 ): str,
             }
         )
@@ -64,7 +64,7 @@ async def async_step_user(self, user_input: dict[str, Any] | None = None) -> Flo
                 }
 
                 session = get_async_client(self.hass)
-                isapi = ISAPI(host, username, password, session)
+                isapi = ISAPIClient(host, username, password, session)
                 await isapi.get_device_info()
 
                 if self._reauth_entry: