sorted imports and max line length 120 (#66)

maciej-or · Jun 28, 2023 · 3fa5f88 · 3fa5f88
1 parent b7050f8
commit 3fa5f88
Show file tree

Hide file tree

Showing 10 changed files with 106 additions and 332 deletions.
diff --git 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
@@ -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
@@ -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
@@ -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,18 +81,14 @@ 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)
 
     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
@@ -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
@@ -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())