diff --git a/custom_components/hikvision_next/__init__.py b/custom_components/hikvision_next/__init__.py index 91681bf..d907a34 100644 --- a/custom_components/hikvision_next/__init__.py +++ b/custom_components/hikvision_next/__init__.py @@ -1,22 +1,16 @@ """hikvision component""" from __future__ import annotations + import asyncio import logging from httpx import TimeoutException from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ( - CONF_HOST, - CONF_PASSWORD, - CONF_USERNAME, - Platform, -) +from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady - - from homeassistant.helpers import device_registry as dr from .const import ( @@ -32,7 +26,6 @@ from .isapi import ISAPI from .notifications import EventNotificationsView - _LOGGER = logging.getLogger(__name__) PLATFORMS = [ @@ -56,13 +49,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: await isapi.get_cameras() device_info = isapi.get_device_info() device_registry = dr.async_get(hass) - device_registry.async_get_or_create( - config_entry_id=entry.entry_id, **device_info - ) + device_registry.async_get_or_create(config_entry_id=entry.entry_id, **device_info) except (asyncio.TimeoutError, TimeoutException) as ex: - raise ConfigEntryNotReady( - f"Timeout while connecting to {host}. Cannot initialize {DOMAIN}" - ) from ex + raise ConfigEntryNotReady(f"Timeout while connecting to {host}. Cannot initialize {DOMAIN}") from ex except Exception as ex: # pylint: disable=broad-except raise ConfigEntryNotReady( f"Unknown error connecting to {host}. Cannot initialize {DOMAIN}. Error is {ex}" @@ -72,10 +61,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: coordinators[EVENTS_COORDINATOR] = EventsCoordinator(hass, isapi) - if ( - isapi.device_info.support_holiday_mode - or isapi.device_info.support_alarm_server - ): + if isapi.device_info.support_holiday_mode or isapi.device_info.support_alarm_server: coordinators[SECONDARY_COORDINATOR] = SecondaryCoordinator(hass, isapi) hass.data[DOMAIN][entry.entry_id] = { @@ -85,13 +71,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: **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 - ) + 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() @@ -105,9 +86,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return True -async def async_remove_config_entry_device( - hass: HomeAssistant, config_entry, device_entry -) -> bool: +async def async_remove_config_entry_device(hass: HomeAssistant, config_entry, device_entry) -> bool: """Delete device if not entities""" if not device_entry.via_device_id: _LOGGER.error( @@ -125,10 +104,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # Unload a config entry unload_ok = all( await asyncio.gather( - *[ - hass.config_entries.async_forward_entry_unload(entry, platform) - for platform in PLATFORMS - ] + *[hass.config_entries.async_forward_entry_unload(entry, platform) for platform in PLATFORMS] ) ) @@ -148,9 +124,5 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: def get_first_instance_unique_id(hass: HomeAssistant) -> int: """Get entry unique_id for first instance of integration""" - entry = [ - entry - for entry in hass.config_entries.async_entries(DOMAIN) - if not entry.disabled_by - ][0] + entry = [entry for entry in hass.config_entries.async_entries(DOMAIN) if not entry.disabled_by][0] return entry.unique_id diff --git a/custom_components/hikvision_next/binary_sensor.py b/custom_components/hikvision_next/binary_sensor.py index e1dc04d..f4f9ac1 100644 --- a/custom_components/hikvision_next/binary_sensor.py +++ b/custom_components/hikvision_next/binary_sensor.py @@ -11,9 +11,7 @@ from .isapi import AnalogCamera, EventInfo, IPCamera -async def async_setup_entry( - hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback -) -> None: +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None: """Add binary sensors for hikvision events states.""" config = hass.data[DOMAIN][entry.entry_id] @@ -33,9 +31,7 @@ class EventBinarySensor(BinarySensorEntity): _attr_has_entity_name = True _attr_is_on = False - def __init__( - self, isapi, camera: AnalogCamera | IPCamera, event: EventInfo - ) -> None: + def __init__(self, isapi, camera: AnalogCamera | IPCamera, event: EventInfo) -> None: self.entity_id = ENTITY_ID_FORMAT.format(event.unique_id) self._attr_unique_id = self.entity_id self._attr_name = EVENTS[event.id]["label"] diff --git a/custom_components/hikvision_next/camera.py b/custom_components/hikvision_next/camera.py index e244729..9c214db 100644 --- a/custom_components/hikvision_next/camera.py +++ b/custom_components/hikvision_next/camera.py @@ -8,14 +8,11 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util import slugify -from .const import DOMAIN, DATA_ISAPI - +from .const import DATA_ISAPI, DOMAIN from .isapi import ISAPI, 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: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None: """Set up a Hikvision IP Camera.""" config = hass.data[DOMAIN][entry.entry_id] @@ -45,9 +42,7 @@ def __init__( self._attr_device_info = isapi.get_device_info(camera.id) self._attr_name = camera.name if stream_info.type_id == 1 else stream_info.type - self._attr_unique_id = slugify( - f"{isapi.device_info.serial_no.lower()}_{stream_info.id}" - ) + self._attr_unique_id = slugify(f"{isapi.device_info.serial_no.lower()}_{stream_info.id}") self.entity_id = f"camera.{self.unique_id}" self.isapi = isapi self.stream_info = stream_info @@ -56,8 +51,6 @@ async def stream_source(self) -> str | None: """Return the source of the stream.""" return self.isapi.get_stream_source(self.stream_info) - async def async_camera_image( - self, width: int | None = None, height: int | None = None - ) -> bytes | None: + 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) diff --git a/custom_components/hikvision_next/config_flow.py b/custom_components/hikvision_next/config_flow.py index d2c77db..32d28b8 100755 --- a/custom_components/hikvision_next/config_flow.py +++ b/custom_components/hikvision_next/config_flow.py @@ -1,8 +1,8 @@ """Config flow for hikvision_next integration.""" from __future__ import annotations -import asyncio +import asyncio from http import HTTPStatus import logging from typing import Any @@ -32,31 +32,21 @@ async def get_schema(self, user_input: dict[str, Any]): local_ip = await async_get_source_ip(self.hass) return vol.Schema( { - vol.Required( - CONF_HOST, default=user_input.get(CONF_HOST, "http://") - ): str, - vol.Required( - CONF_USERNAME, default=user_input.get(CONF_USERNAME, "") - ): str, - vol.Required( - CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "") - ): str, + vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "http://")): str, + 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), ): bool, vol.Required( DATA_ALARM_SERVER_HOST, - default=user_input.get( - DATA_ALARM_SERVER_HOST, f"http://{local_ip}:8123" - ), + default=user_input.get(DATA_ALARM_SERVER_HOST, f"http://{local_ip}:8123"), ): str, } ) - async def async_step_user( - self, user_input: dict[str, Any] | None = None - ) -> FlowResult: + async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult: """Handle a flow initiated by the user.""" errors = {} @@ -71,14 +61,8 @@ async def async_step_user( await isapi.get_hardware_info() if self._reauth_entry: - self.hass.config_entries.async_update_entry( - self._reauth_entry, data=user_input - ) - self.hass.async_create_task( - self.hass.config_entries.async_reload( - self._reauth_entry.entry_id - ) - ) + self.hass.config_entries.async_update_entry(self._reauth_entry, data=user_input) + self.hass.async_create_task(self.hass.config_entries.async_reload(self._reauth_entry.entry_id)) return self.async_abort(reason="reauth_successful") await self.async_set_unique_id({(DOMAIN, isapi.device_info.serial_no)}) @@ -97,9 +81,7 @@ async def async_step_user( _LOGGER.error("Unexpected exception %s", ex) errors["base"] = "unknown" else: - return self.async_create_entry( - title=isapi.device_info.name, data=user_input - ) + return self.async_create_entry(title=isapi.device_info.name, data=user_input) schema = await self.get_schema(user_input or {}) return self.async_show_form(step_id="user", data_schema=schema, errors=errors) @@ -107,8 +89,6 @@ async def async_step_user( async def async_step_reauth(self, entry_data: dict[str, Any]) -> FlowResult: """Schedule reauth.""" _LOGGER.warning("Attempt to reauth in 120s") - self._reauth_entry = self.hass.config_entries.async_get_entry( - self.context["entry_id"] - ) + self._reauth_entry = self.hass.config_entries.async_get_entry(self.context["entry_id"]) await asyncio.sleep(120) return await self.async_step_user(entry_data) diff --git a/custom_components/hikvision_next/coordinator.py b/custom_components/hikvision_next/coordinator.py index bcb125c..10d120a 100644 --- a/custom_components/hikvision_next/coordinator.py +++ b/custom_components/hikvision_next/coordinator.py @@ -42,13 +42,9 @@ async def _async_update_data(self): for event in camera.supported_events: try: entity_id = ENTITY_ID_FORMAT.format(event.unique_id) - data[entity_id] = await self.isapi.get_event_enabled_state( - event - ) + data[entity_id] = await self.isapi.get_event_enabled_state(event) except Exception as ex: # pylint: disable=broad-except - self.isapi.handle_exception( - ex, f"Cannot fetch state for {event.id}" - ) + self.isapi.handle_exception(ex, f"Cannot fetch state for {event.id}") # Refresh HDD data try: @@ -81,15 +77,11 @@ async def _async_update_data(self): if self.isapi.device_info.support_holiday_mode: data[HOLIDAY_MODE] = await self.isapi.get_holiday_enabled_state() except Exception as ex: # pylint: disable=broad-except - self.isapi.handle_exception( - ex, f"Cannot fetch state for {HOLIDAY_MODE}" - ) + self.isapi.handle_exception(ex, f"Cannot fetch state for {HOLIDAY_MODE}") try: if self.isapi.device_info.support_alarm_server: alarm_server = await self.isapi.get_alarm_server() data[DATA_ALARM_SERVER_HOST] = alarm_server except Exception as ex: # pylint: disable=broad-except - self.isapi.handle_exception( - ex, f"Cannot fetch state for {DATA_ALARM_SERVER_HOST}" - ) + self.isapi.handle_exception(ex, f"Cannot fetch state for {DATA_ALARM_SERVER_HOST}") return data diff --git a/custom_components/hikvision_next/diagnostics.py b/custom_components/hikvision_next/diagnostics.py index b83d21d..131d130 100644 --- a/custom_components/hikvision_next/diagnostics.py +++ b/custom_components/hikvision_next/diagnostics.py @@ -1,8 +1,8 @@ """Diagnostics support for Wiser""" from __future__ import annotations + import inspect import json - from typing import Any from homeassistant.config_entries import ConfigEntry @@ -23,9 +23,7 @@ ] -async def async_get_config_entry_diagnostics( - hass: HomeAssistant, entry: ConfigEntry -) -> dict[str, Any]: +async def async_get_config_entry_diagnostics(hass: HomeAssistant, entry: ConfigEntry) -> dict[str, Any]: """Return diagnostics for a config entry.""" return await _async_get_diagnostics(hass, entry) @@ -61,11 +59,7 @@ async def _async_get_diagnostics( info.update({"Event States": event_states}) # Add raw device info - info.update( - await get_isapi_data( - "RAW Device Info", isapi.isapi.System.deviceInfo, "DeviceInfo" - ) - ) + info.update(await get_isapi_data("RAW Device Info", isapi.isapi.System.deviceInfo, "DeviceInfo")) # Add raw camera info info.update( @@ -84,32 +78,16 @@ async def _async_get_diagnostics( ) # Add raw capabilities - info.update( - await get_isapi_data( - "RAW Capabilities Info", isapi.isapi.System.capabilities, "DeviceCap" - ) - ) + info.update(await get_isapi_data("RAW Capabilities Info", isapi.isapi.System.capabilities, "DeviceCap")) # Add raw supported events - info.update( - await get_isapi_data( - "RAW Events Info", isapi.isapi.Event.triggers, "EventTriggerList" - ) - ) + info.update(await get_isapi_data("RAW Events Info", isapi.isapi.Event.triggers, "EventTriggerList")) # Add raw streams info - info.update( - await get_isapi_data( - "RAW Streams Info", isapi.isapi.Streaming.channels, "StreamingChannelList" - ) - ) + info.update(await get_isapi_data("RAW Streams Info", isapi.isapi.Streaming.channels, "StreamingChannelList")) # Add raw holiday info - info.update( - await get_isapi_data( - "RAW Holiday Info", isapi.isapi.System.Holidays, "HolidayList" - ) - ) + info.update(await get_isapi_data("RAW Holiday Info", isapi.isapi.System.Holidays, "HolidayList")) # Add alarms server info info.update( @@ -130,7 +108,7 @@ async def get_isapi_data(title: str, path: object, filter_key: str = "") -> dict if filter_key: response = response.get(filter_key, {}) return {title: anonymise_data(response)} - except Exception as ex: # pylint: disable=broad-except + except Exception as ex: # pylint: disable=broad-except return {title: ex} @@ -152,6 +130,7 @@ def anonymise_data(data): class ObjectEncoder(json.JSONEncoder): """Class to encode object to json.""" + def default(self, o): if hasattr(o, "to_json"): return self.default(o.to_json()) diff --git a/custom_components/hikvision_next/isapi.py b/custom_components/hikvision_next/isapi.py index 702799d..6cd2376 100644 --- a/custom_components/hikvision_next/isapi.py +++ b/custom_components/hikvision_next/isapi.py @@ -176,20 +176,12 @@ def __init__(self, host: str, username: str, password: str) -> None: 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 - ) + hw_info = (await self.isapi.System.deviceInfo(method=GET)).get("DeviceInfo", {}) + _LOGGER.debug("%s/ISAPI/System/deviceInfo %s", self.isapi.host, hw_info) # Get device capabilities - capabilities = (await self.isapi.System.capabilities(method=GET)).get( - "DeviceCap", {} - ) - _LOGGER.debug( - "%s/ISAPI/System/capabilities %s", self.isapi.host, capabilities - ) + capabilities = (await self.isapi.System.capabilities(method=GET)).get("DeviceCap", {}) + _LOGGER.debug("%s/ISAPI/System/capabilities %s", self.isapi.host, capabilities) # Set DeviceInfo self.device_info = HikDeviceInfo( @@ -201,28 +193,14 @@ async def get_hardware_info(self): 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_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) - ), + 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(), ) @@ -230,11 +208,7 @@ async def get_hardware_info(self): # Set if NVR based on whether more than 1 supported IP or analog cameras # Single IP camera will show 0 supported devices in total - if ( - self.device_info.support_analog_cameras - + self.device_info.support_digital_cameras - > 1 - ): + if self.device_info.support_analog_cameras + self.device_info.support_digital_cameras > 1: self.device_info.is_nvr = True async def get_cameras(self): @@ -264,11 +238,7 @@ async def get_cameras(self): # Get analog and digital cameras attached to NVR if self.device_info.support_digital_cameras > 0: digital_cameras = ( - ( - await self.isapi.ContentMgmt.InputProxy.channels( - method=GET - ) - ) + (await self.isapi.ContentMgmt.InputProxy.channels(method=GET)) .get("InputProxyChannelList", {}) .get("InputProxyChannel", []) ) @@ -293,9 +263,7 @@ async def get_cameras(self): serial_no = source.get("serialNumber") if not serial_no: - serial_no = str(source.get("proxyProtocol")) + str( - source.get("ipAddress", "") - ).replace(".", "") + serial_no = str(source.get("proxyProtocol")) + str(source.get("ipAddress", "")).replace(".", "") self.cameras.append( IPCamera( @@ -335,9 +303,7 @@ async def get_cameras(self): for analog_camera in analog_cameras: camera_id = analog_camera.get("id") - device_serial_no = ( - f"{self.device_info.serial_no}-VI{camera_id}" - ) + device_serial_no = f"{self.device_info.serial_no}-VI{camera_id}" self.cameras.append( AnalogCamera( @@ -388,9 +354,7 @@ async def get_camera_event_capabilities( """Get events support by camera device and integration""" events = [] - camera_supported_events = [ - s for s in supported_events if s.channel_id == int(channel_id) - ] + camera_supported_events = [s for s in supported_events if s.channel_id == int(channel_id)] for event in camera_supported_events: if EVENTS.get(event.event_id): @@ -413,17 +377,11 @@ async def get_supported_events_info(self): events = [] event_triggers = await self.isapi.Event.triggers(method=GET) event_notification = event_triggers.get("EventNotification") - _LOGGER.debug( - "%s/ISAPI/Event/triggers %s", self.isapi.host, event_triggers - ) + _LOGGER.debug("%s/ISAPI/Event/triggers %s", self.isapi.host, event_triggers) if event_notification: - supported_events = deep_get( - event_notification, "EventTriggerList.EventTrigger" - ) + supported_events = deep_get(event_notification, "EventTriggerList.EventTrigger") else: - supported_events = deep_get( - event_triggers, "EventTriggerList.EventTrigger" - ) + supported_events = deep_get(event_triggers, "EventTriggerList.EventTrigger") for support_event in supported_events: event_type = support_event.get("eventType") @@ -431,9 +389,7 @@ async def get_supported_events_info(self): "videoInputChannelID", support_event.get("dynVideoInputChannelID", 0), ) - notifications = support_event.get( - "EventTriggerNotificationList", {} - ) + notifications = support_event.get("EventTriggerNotificationList", {}) # Fix for empty EventTriggerNotificationList in IP camera if not notifications: @@ -452,10 +408,7 @@ async def get_supported_events_info(self): SupportedEventsInfo( channel_id=int(channel), event_id=event_type.lower(), - notifications=[ - notify.get("notificationMethod") - for notify in notifications - ] + notifications=[notify.get("notificationMethod") for notify in notifications] if notifications else [], ) @@ -463,24 +416,16 @@ async def get_supported_events_info(self): return events - def get_event_url( - self, event_id: str, channel_id: int, is_nvr: bool, camera_type: str - ) -> str: + def get_event_url(self, event_id: str, channel_id: int, is_nvr: bool, camera_type: str) -> str: """Get event ISAPI URL.""" event_type = EVENTS[event_id]["type"] slug = EVENTS[event_id]["slug"] - if ( - is_nvr - and camera_type == DEVICE_TYPE_IP_CAMERA - and event_type == EVENT_BASIC - ): + if is_nvr and camera_type == DEVICE_TYPE_IP_CAMERA and event_type == EVENT_BASIC: # ISAPI/ContentMgmt/InputProxy/channels/{channel_id}/video/{event} url = f"ContentMgmt/InputProxy/channels/{channel_id}/video/{slug}" - elif ( - is_nvr and camera_type == DEVICE_TYPE_ANALOG_CAMERA or not is_nvr - ) and event_type == EVENT_BASIC: + elif (is_nvr and camera_type == DEVICE_TYPE_ANALOG_CAMERA or not is_nvr) and event_type == EVENT_BASIC: # ISAPI/System/Video/inputs/channels/{channel_id}/{event} url = f"System/Video/inputs/channels/{channel_id}/{slug}" else: @@ -488,17 +433,13 @@ def get_event_url( url = f"Smart/{slug}/{channel_id}" return url - async def get_camera_streams( - self, channel_id: int - ) -> list[CameraStreamInfo]: + async def get_camera_streams(self, channel_id: int) -> list[CameraStreamInfo]: """Get stream info for all cameras""" streams = [] for stream_type_id, stream_type in STREAM_TYPE.items(): try: stream_id = f"{channel_id}0{stream_type_id}" - stream_info = ( - await self.isapi.Streaming.channels[stream_id](method=GET) - ).get("StreamingChannel") + stream_info = (await self.isapi.Streaming.channels[stream_id](method=GET)).get("StreamingChannel") _LOGGER.debug( "%s/ISAPI/Streaming/channels/%s %s", self.isapi.host, @@ -523,14 +464,10 @@ async def get_camera_streams( continue return streams - def get_camera_by_id( - self, camera_id: int - ) -> IPCamera | AnalogCamera | None: + def get_camera_by_id(self, camera_id: int) -> IPCamera | AnalogCamera | None: """Get camera object by id.""" try: - return [ - camera for camera in self.cameras if camera.id == camera_id - ][0] + return [camera for camera in self.cameras if camera.id == camera_id][0] except IndexError: # Camera id does not exist return None @@ -538,18 +475,12 @@ def get_camera_by_id( async def get_storage_devices(self): """Get HDD storage devices.""" storage_list = [] - storage_info = ( - (await self.isapi.ContentMgmt.Storage(method=GET)) - .get("storage", {}) - .get("hddList", {}) - ) + storage_info = (await self.isapi.ContentMgmt.Storage(method=GET)).get("storage", {}).get("hddList", {}) if not isinstance(storage_info, list): storage_info = [storage_info] - _LOGGER.debug( - "%s/ISAPI/ContentMgmt/Storage %s", self.isapi.host, storage_info - ) + _LOGGER.debug("%s/ISAPI/ContentMgmt/Storage %s", self.isapi.host, storage_info) for storage in storage_info: storage = storage.get("hdd") @@ -571,11 +502,7 @@ async def get_storage_devices(self): def get_storage_device_by_id(self, device_id: int) -> HDDInfo | None: """Get storage object by id.""" try: - return [ - storage_device - for storage_device in self.device_info.storage - if storage_device.id == device_id - ][0] + return [storage_device for storage_device in self.device_info.storage if storage_device.id == device_id][0] except IndexError: # Storage id does not exist return None @@ -586,9 +513,7 @@ def get_device_info(self, device_id: int = 0) -> DeviceInfo: return DeviceInfo( manufacturer=self.device_info.manufacturer, identifiers={(DOMAIN, self.device_info.serial_no)}, - connections={ - (dr.CONNECTION_NETWORK_MAC, self.device_info.mac_address) - }, + connections={(dr.CONNECTION_NETWORK_MAC, self.device_info.mac_address)}, model=self.device_info.model, name=self.device_info.name, sw_version=self.device_info.firmware, @@ -602,12 +527,8 @@ def get_device_info(self, device_id: int = 0) -> DeviceInfo: identifiers={(DOMAIN, camera_info.serial_no)}, model=camera_info.model, name=camera_info.name, - sw_version=self.device_info.firmware - if is_ip_camera - else "Unknown", - via_device=(DOMAIN, self.device_info.serial_no) - if self.device_info.is_nvr - else None, + sw_version=self.device_info.firmware if is_ip_camera else "Unknown", + via_device=(DOMAIN, self.device_info.serial_no) if self.device_info.is_nvr else None, ) async def get_event_enabled_state(self, event: EventInfo) -> bool: @@ -618,9 +539,7 @@ async def get_event_enabled_state(self, event: EventInfo) -> bool: node = slug[0].upper() + slug[1:] return str_to_bool(state[node]["enabled"]) - async def get_event_switch_mutex( - self, event: EventInfo, channel_id: int - ) -> list[MutexIssue]: + async def get_event_switch_mutex(self, event: EventInfo, channel_id: int) -> list[MutexIssue]: """Get if event is mutually exclusive with enabled events""" mutex_issues = [] @@ -635,9 +554,7 @@ async def get_event_switch_mutex( data = {"function": event_id, "channelID": int(channel_id)} url = "System/mutexFunction?format=json" try: - response = await self.request( - POST, url, present="json", data=json.dumps(data) - ) + response = await self.request(POST, url, present="json", data=json.dumps(data)) except HTTPStatusError: return [] @@ -657,9 +574,7 @@ async def get_event_switch_mutex( ) return mutex_issues - async def set_event_enabled_state( - self, channel_id: int, event: EventInfo, is_enabled: bool - ) -> None: + async def set_event_enabled_state(self, channel_id: int, event: EventInfo, is_enabled: bool) -> None: """Set event detection state.""" # Validate that this event switch is not mutually exclusive with another enabled one @@ -678,9 +593,7 @@ async def set_event_enabled_state( data[node]["enabled"] = new_state xml = xmltodict.unparse(data) response = await self.request(PUT, event.url, data=xml) - _LOGGER.debug( - "[PUT] %s/ISAPI/%s %s", self.isapi.host, event.url, response - ) + _LOGGER.debug("[PUT] %s/ISAPI/%s %s", self.isapi.host, event.url, response) else: raise HomeAssistantError( f"You cannot enable {EVENTS[event.id]['label']} events. Please disable {EVENTS[mutex_issues[0].event_id]['label']} on channels {mutex_issues[0].channels} first" @@ -693,9 +606,7 @@ async def get_holiday_enabled_state(self, holiday_index=0) -> bool: holiday = data["HolidayList"]["holiday"][holiday_index] return str_to_bool(holiday["enabled"]["#text"]) - async def set_holiday_enabled_state( - self, is_enabled: bool, holiday_index=0 - ) -> None: + async def set_holiday_enabled_state(self, is_enabled: bool, holiday_index=0) -> None: """Enable or disable holiday, by enable set time span to year starting from today.""" data = await self.isapi.System.Holidays(method=GET) @@ -710,22 +621,16 @@ async def set_holiday_enabled_state( holiday["holidayMode"]["#text"] = "date" holiday["holidayDate"] = { "startDate": today.strftime("%Y-%m-%d"), - "endDate": today.replace(year=today.year + 1).strftime( - "%Y-%m-%d" - ), + "endDate": today.replace(year=today.year + 1).strftime("%Y-%m-%d"), } holiday.pop("holidayWeek", None) holiday.pop("holidayMonth", None) xml = xmltodict.unparse(data) response = await self.isapi.System.Holidays(method=PUT, data=xml) - _LOGGER.debug( - "[PUT] %s/ISAPI/System/Holidays %s", self.isapi.host, response - ) + _LOGGER.debug("[PUT] %s/ISAPI/System/Holidays %s", self.isapi.host, response) def _get_event_notification_host(self, data: Node) -> Node: - hosts = deep_get( - data, "HttpHostNotificationList.HttpHostNotification", {} - ) + hosts = deep_get(data, "HttpHostNotificationList.HttpHostNotification", {}) if isinstance(hosts, list): # return hosts[0] @@ -740,9 +645,7 @@ async def get_alarm_server(self) -> AlarmServer | None: except HTTPStatusError: return None - _LOGGER.debug( - "%s/ISAPI/Event/notification/httpHosts %s", self.isapi.host, data - ) + _LOGGER.debug("%s/ISAPI/Event/notification/httpHosts %s", self.isapi.host, data) host = self._get_event_notification_host(data) @@ -760,9 +663,7 @@ async def set_alarm_server(self, base_url: str, path: str) -> None: address = urlparse(base_url) data = await self.isapi.Event.notification.httpHosts(method=GET) - _LOGGER.debug( - "%s/ISAPI/Event/notification/httpHosts %s", self.isapi.host, data - ) + _LOGGER.debug("%s/ISAPI/Event/notification/httpHosts %s", self.isapi.host, data) host = self._get_event_notification_host(data) if ( host["protocolType"] == address.scheme.upper() @@ -780,25 +681,19 @@ async def set_alarm_server(self, base_url: str, path: str) -> None: host["httpAuthenticationMethod"] = "none" xml = xmltodict.unparse(data) - response = await self.isapi.Event.notification.httpHosts( - method=PUT, data=xml - ) + response = await self.isapi.Event.notification.httpHosts(method=PUT, data=xml) _LOGGER.debug( "[PUT] %s/ISAPI/Event/notification/httpHosts %s", self.isapi.host, response, ) - async def request( - self, method: str, url: str, present: str = "dict", **data - ) -> Any: + async def request(self, method: str, url: str, present: str = "dict", **data) -> Any: """Send request""" full_url = f"{self.isapi.host}/{self.isapi.isapi_prefix}/{url}" try: - return await self.isapi.common_request( - method, full_url, present, self.isapi.timeout, **data - ) + return await self.isapi.common_request(method, full_url, present, self.isapi.timeout, **data) except HTTPStatusError as ex: raise ex @@ -824,9 +719,7 @@ def is_reauth_needed(): return True elif isinstance(ex, (asyncio.TimeoutError, TimeoutException)): - raise HomeAssistantError( - f"Timeout while connecting to {host} {details}" - ) from ex + raise HomeAssistantError(f"Timeout while connecting to {host} {details}") from ex _LOGGER.warning("Unexpected exception | %s | %s", details, ex) return False @@ -842,9 +735,7 @@ def parse_event_notification(xml: str) -> AlertInfo: alert = data["EventNotificationAlert"] - channel_id = int( - alert.get("channelID", alert.get("dynChannelID", "0")) - ) + channel_id = int(alert.get("channelID", alert.get("dynChannelID", "0"))) event_id = alert.get("eventType") if not event_id or event_id == "duration": @@ -881,9 +772,7 @@ async def get_camera_image( "videoResolutionWidth": stream.width, "videoResolutionHeight": stream.height, } - chunks = self.isapi.Streaming.channels[stream.id].picture( - method=GET, type="opaque_data", params=params - ) + chunks = self.isapi.Streaming.channels[stream.id].picture(method=GET, type="opaque_data", params=params) image_bytes = b"".join([chunk async for chunk in chunks]) return image_bytes diff --git a/custom_components/hikvision_next/notifications.py b/custom_components/hikvision_next/notifications.py index f8bf726..6e586f1 100644 --- a/custom_components/hikvision_next/notifications.py +++ b/custom_components/hikvision_next/notifications.py @@ -2,24 +2,20 @@ from __future__ import annotations -import logging from http import HTTPStatus +import logging from urllib.parse import urlparse from aiohttp import web +from requests_toolbelt.multipart import MultipartDecoder + from homeassistant.components.http import HomeAssistantView from homeassistant.const import CONTENT_TYPE_TEXT_PLAIN, STATE_ON from homeassistant.core import HomeAssistant from homeassistant.helpers import device_registry as dr from homeassistant.util import slugify -from requests_toolbelt.multipart import MultipartDecoder -from .const import ( - ALARM_SERVER_PATH, - DATA_ISAPI, - DOMAIN, - HIKVISION_EVENT, -) +from .const import ALARM_SERVER_PATH, DATA_ISAPI, DOMAIN, HIKVISION_EVENT from .isapi import ISAPI, AlertInfo, IPCamera _LOGGER = logging.getLogger(__name__) @@ -57,9 +53,7 @@ async def post(self, request: web.Request): except Exception as ex: # pylint: disable=broad-except _LOGGER.warning("Cannot process incoming event %s", ex) - response = web.Response( - status=HTTPStatus.OK, content_type=CONTENT_TYPE_TEXT_PLAIN - ) + response = web.Response(status=HTTPStatus.OK, content_type=CONTENT_TYPE_TEXT_PLAIN) return response def get_isapi_instance(self, device_ip) -> ISAPI: @@ -69,8 +63,7 @@ def get_isapi_instance(self, device_ip) -> ISAPI: entry = [ entry for entry in self.hass.config_entries.async_entries(DOMAIN) - if not entry.disabled_by - and urlparse(entry.data.get("host")).hostname == device_ip + if not entry.disabled_by and urlparse(entry.data.get("host")).hostname == device_ip ][0] config = self.hass.data[DOMAIN][entry.entry_id] @@ -106,9 +99,7 @@ async def parse_event_request(self, request: web.Request) -> str: _LOGGER.debug("image found") if not xml: - raise ValueError( - f"Unexpected event Content-Type {content_type_header}" - ) + raise ValueError(f"Unexpected event Content-Type {content_type_header}") return xml def get_alert_info(self, xml: str) -> AlertInfo: @@ -124,8 +115,7 @@ def get_alert_info(self, xml: str) -> AlertInfo: alert.channel_id = [ camera.id for camera in self.isapi.cameras - if isinstance(camera, IPCamera) - and camera.input_port == alert.channel_id - 32 + if isinstance(camera, IPCamera) and camera.input_port == alert.channel_id - 32 ][0] except IndexError: alert.channel_id = alert.channel_id - 32 @@ -139,10 +129,7 @@ def trigger_sensor(self, hass: HomeAssistant, xml: str) -> None: _LOGGER.debug("Alert: %s", alert) serial_no = self.isapi.device_info.serial_no.lower() - entity_id = ( - f"binary_sensor.{slugify(serial_no)}_{alert.channel_id}" - f"_{alert.event_id}" - ) + entity_id = f"binary_sensor.{slugify(serial_no)}_{alert.channel_id}" f"_{alert.event_id}" entity = hass.states.get(entity_id) if entity: hass.states.async_set(entity_id, STATE_ON, entity.attributes) diff --git a/custom_components/hikvision_next/sensor.py b/custom_components/hikvision_next/sensor.py index 6832e22..8eb8033 100644 --- a/custom_components/hikvision_next/sensor.py +++ b/custom_components/hikvision_next/sensor.py @@ -1,7 +1,6 @@ """Platform for sensor integration.""" from __future__ import annotations -from .isapi import HDDInfo from homeassistant.components.sensor import ENTITY_ID_FORMAT, SensorEntity from homeassistant.config_entries import ConfigEntry @@ -17,6 +16,7 @@ EVENTS_COORDINATOR, SECONDARY_COORDINATOR, ) +from .isapi import HDDInfo ALARM_SERVER_SETTINGS = { "protocolType": "Protocol", @@ -59,14 +59,10 @@ class AlarmServerSensor(CoordinatorEntity, SensorEntity): def __init__(self, coordinator, key: str) -> None: super().__init__(coordinator) isapi = coordinator.isapi - self._attr_unique_id = ( - f"{isapi.device_info.serial_no}_{DATA_ALARM_SERVER_HOST}_{key}" - ) + 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_name = ALARM_SERVER_SENSOR_LABEL_FORMAT.format( - ALARM_SERVER_SETTINGS[key] - ) + self._attr_name = ALARM_SERVER_SENSOR_LABEL_FORMAT.format(ALARM_SERVER_SETTINGS[key]) self.key = key @property @@ -85,9 +81,7 @@ class HDDSensor(CoordinatorEntity, SensorEntity): def __init__(self, coordinator, hdd: HDDInfo) -> None: super().__init__(coordinator) isapi = coordinator.isapi - self._attr_unique_id = ( - f"{isapi.device_info.serial_no}_{hdd.id}_{hdd.name}" - ) + 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_name = f"HDD {hdd.id}" diff --git a/custom_components/hikvision_next/switch.py b/custom_components/hikvision_next/switch.py index 365e7d7..1a1da71 100644 --- a/custom_components/hikvision_next/switch.py +++ b/custom_components/hikvision_next/switch.py @@ -23,9 +23,7 @@ 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: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None: """Add hikvision_next entities from a config_entry.""" config = hass.data[DOMAIN][entry.entry_id] @@ -64,9 +62,7 @@ def is_on(self) -> bool | None: async def async_turn_on(self, **kwargs: Any) -> None: try: - await self.coordinator.isapi.set_event_enabled_state( - self.camera.id, self.event, True - ) + await self.coordinator.isapi.set_event_enabled_state(self.camera.id, self.event, True) except Exception as ex: raise ex finally: @@ -74,9 +70,7 @@ async def async_turn_on(self, **kwargs: Any) -> None: async def async_turn_off(self, **kwargs: Any) -> None: try: - await self.coordinator.isapi.set_event_enabled_state( - self.camera.id, self.event, False - ) + await self.coordinator.isapi.set_event_enabled_state(self.camera.id, self.event, False) except Exception as ex: raise ex finally: @@ -97,9 +91,7 @@ class HolidaySwitch(CoordinatorEntity, SwitchEntity): def __init__(self, coordinator) -> None: super().__init__(coordinator) - self._attr_unique_id = ( - f"{slugify(coordinator.isapi.device_info.serial_no.lower())}_{HOLIDAY_MODE}" - ) + 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_name = HOLIDAY_MODE_SWITCH_LABEL