Add type annotations to init and coordinator. Minor cleanups. (#52506)

homeassistant/components/ezviz/__init__.py

@@ -1,10 +1,10 @@
 """Support for Ezviz camera."""
-from datetime import timedelta
 import logging
 
 from pyezviz.client import EzvizClient
 from pyezviz.exceptions import HTTPError, InvalidURL, PyEzvizError
 
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import (
     CONF_PASSWORD,
     CONF_TIMEOUT,
@@ -12,6 +12,7 @@ from homeassistant.const import (
     CONF_URL,
     CONF_USERNAME,
 )
+from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 
 from .const import (
@@ -28,8 +29,6 @@ from .coordinator import EzvizDataUpdateCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
-MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
-
 PLATFORMS = [
     "binary_sensor",
     "camera",
@@ -38,17 +37,16 @@ PLATFORMS = [
 ]
 
 
-async def async_setup_entry(hass, entry):
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Ezviz from a config entry."""
     hass.data.setdefault(DOMAIN, {})
 
     if not entry.options:
         options = {
-            CONF_FFMPEG_ARGUMENTS: entry.data.get(
+            CONF_FFMPEG_ARGUMENTS: DEFAULT_FFMPEG_ARGUMENTS,
-                CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
+            CONF_TIMEOUT: DEFAULT_TIMEOUT,
-            ),
-            CONF_TIMEOUT: entry.data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
         }
+
         hass.config_entries.async_update_entry(entry, options=options)
 
     if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA:
@@ -70,7 +68,9 @@ async def async_setup_entry(hass, entry):
         _LOGGER.error("Unable to connect to Ezviz service: %s", str(error))
         raise ConfigEntryNotReady from error
 
-    coordinator = EzvizDataUpdateCoordinator(hass, api=ezviz_client)
+    coordinator = EzvizDataUpdateCoordinator(
+        hass, api=ezviz_client, api_timeout=entry.options[CONF_TIMEOUT]
+    )
     await coordinator.async_refresh()
 
     if not coordinator.last_update_success:
@@ -87,7 +87,7 @@ async def async_setup_entry(hass, entry):
     return True
 
 
-async def async_unload_entry(hass, entry):
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
 
     if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA:
@@ -100,12 +100,12 @@ async def async_unload_entry(hass, entry):
     return unload_ok
 
 
-async def _async_update_listener(hass, entry):
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
     """Handle options update."""
     await hass.config_entries.async_reload(entry.entry_id)
 
 
-def _get_ezviz_client_instance(entry):
+def _get_ezviz_client_instance(entry: ConfigEntry) -> EzvizClient:
     """Initialize a new instance of EzvizClientApi."""
     ezviz_client = EzvizClient(
         entry.data[CONF_USERNAME],

homeassistant/components/ezviz/coordinator.py

@@ -3,8 +3,10 @@ from datetime import timedelta
 import logging
 
 from async_timeout import timeout
+from pyezviz.client import EzvizClient
 from pyezviz.exceptions import HTTPError, InvalidURL, PyEzvizError
 
+from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import DOMAIN
@@ -15,23 +17,24 @@ _LOGGER = logging.getLogger(__name__)
 class EzvizDataUpdateCoordinator(DataUpdateCoordinator):
     """Class to manage fetching Ezviz data."""
 
-    def __init__(self, hass, *, api):
+    def __init__(
+        self, hass: HomeAssistant, *, api: EzvizClient, api_timeout: int
+    ) -> None:
         """Initialize global Ezviz data updater."""
         self.ezviz_client = api
+        self._api_timeout = api_timeout
         update_interval = timedelta(seconds=30)
 
         super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
 
-    def _update_data(self):
+    def _update_data(self) -> dict:
         """Fetch data from Ezviz via camera load function."""
-        cameras = self.ezviz_client.load_cameras()
+        return self.ezviz_client.load_cameras()
 
-        return cameras
+    async def _async_update_data(self) -> dict:
-
-    async def _async_update_data(self):
         """Fetch data from Ezviz."""
         try:
-            async with timeout(35):
+            async with timeout(self._api_timeout):
                 return await self.hass.async_add_executor_job(self._update_data)
 
         except (InvalidURL, HTTPError, PyEzvizError) as error: