From 73c6aa701ff582ad6e7fc1fffa42d204c82a7eee Mon Sep 17 00:00:00 2001 From: RenierM26 <66512715+RenierM26@users.noreply.github.com> Date: Sun, 4 Jul 2021 13:16:27 +0200 Subject: [PATCH] Add basic typing to ezviz camera platform (#52492) * Add basic typing to camera platform. Small cleanups. * Add full typing to all functions --- homeassistant/components/ezviz/camera.py | 101 ++++++++++++++--------- homeassistant/components/ezviz/const.py | 2 +- 2 files changed, 62 insertions(+), 41 deletions(-) diff --git a/homeassistant/components/ezviz/camera.py b/homeassistant/components/ezviz/camera.py index b09e5cdd901..76fbaee3757 100644 --- a/homeassistant/components/ezviz/camera.py +++ b/homeassistant/components/ezviz/camera.py @@ -1,4 +1,6 @@ """Support ezviz camera devices.""" +from __future__ import annotations + import asyncio import logging @@ -8,10 +10,17 @@ import voluptuous as vol from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_STREAM, Camera from homeassistant.components.ffmpeg import DATA_FFMPEG -from homeassistant.config_entries import SOURCE_DISCOVERY, SOURCE_IGNORE, SOURCE_IMPORT +from homeassistant.config_entries import ( + SOURCE_DISCOVERY, + SOURCE_IGNORE, + SOURCE_IMPORT, + ConfigEntry, +) from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME +from homeassistant.core import HomeAssistant from homeassistant.helpers import config_validation as cv, entity_platform -from homeassistant.helpers.restore_state import RestoreEntity +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import ( @@ -39,6 +48,7 @@ from .const import ( SERVICE_PTZ, SERVICE_WAKE_DEVICE, ) +from .coordinator import EzvizDataUpdateCoordinator CAMERA_SCHEMA = vol.Schema( {vol.Required(CONF_USERNAME): cv.string, vol.Required(CONF_PASSWORD): cv.string} @@ -55,7 +65,12 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( _LOGGER = logging.getLogger(__name__) -async def async_setup_platform(hass, config, async_add_entities, discovery_info=None): +async def async_setup_platform( + hass: HomeAssistant, + config: ConfigType, + async_add_entities: entity_platform.AddEntitiesCallback, + discovery_info: DiscoveryInfoType | None = None, +) -> None: """Set up a Ezviz IP Camera from platform config.""" _LOGGER.warning( "Loading ezviz via platform config is deprecated, it will be automatically imported. Please remove it afterwards" @@ -91,10 +106,16 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info= ) -async def async_setup_entry(hass, entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: entity_platform.AddEntitiesCallback, +) -> None: """Set up Ezviz cameras based on a config entry.""" - coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR] + coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][ + DATA_COORDINATOR + ] camera_config_entries = hass.config_entries.async_entries(DOMAIN) camera_entities = [] @@ -169,7 +190,7 @@ async def async_setup_entry(hass, entry, async_add_entities): async_add_entities(camera_entities) - platform = entity_platform.current_platform.get() + platform = entity_platform.async_get_current_platform() platform.async_register_entity_service( SERVICE_PTZ, @@ -210,20 +231,22 @@ async def async_setup_entry(hass, entry, async_add_entities): ) -class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): +class EzvizCamera(CoordinatorEntity, Camera): """An implementation of a Ezviz security camera.""" + coordinator: EzvizDataUpdateCoordinator + def __init__( self, - hass, - coordinator, - idx, - camera_username, - camera_password, - camera_rtsp_stream, - local_rtsp_port, - ffmpeg_arguments, - ): + hass: HomeAssistant, + coordinator: EzvizDataUpdateCoordinator, + idx: int, + camera_username: str, + camera_password: str, + camera_rtsp_stream: str | None, + local_rtsp_port: int | None, + ffmpeg_arguments: str | None, + ) -> None: """Initialize a Ezviz security camera.""" super().__init__(coordinator) Camera.__init__(self) @@ -240,51 +263,48 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): self._local_ip = self.coordinator.data[self._idx]["local_ip"] @property - def available(self): + def available(self) -> bool: """Return True if entity is available.""" - if self.coordinator.data[self._idx]["status"] == 2: - return False - - return True + return self.coordinator.data[self._idx]["status"] != 2 @property - def supported_features(self): + def supported_features(self) -> int: """Return supported features.""" if self._rtsp_stream: return SUPPORT_STREAM return 0 @property - def name(self): + def name(self) -> str: """Return the name of this device.""" return self._name @property - def model(self): + def model(self) -> str: """Return the model of this device.""" return self.coordinator.data[self._idx]["device_sub_category"] @property - def brand(self): + def brand(self) -> str: """Return the manufacturer of this device.""" return MANUFACTURER @property - def is_on(self): + def is_on(self) -> bool: """Return true if on.""" return bool(self.coordinator.data[self._idx]["status"]) @property - def is_recording(self): + def is_recording(self) -> bool: """Return true if the device is recording.""" return self.coordinator.data[self._idx]["alarm_notify"] @property - def motion_detection_enabled(self): + def motion_detection_enabled(self) -> bool: """Camera Motion Detection Status.""" return self.coordinator.data[self._idx]["alarm_notify"] - def enable_motion_detection(self): + def enable_motion_detection(self) -> None: """Enable motion detection in camera.""" try: self.coordinator.ezviz_client.set_camera_defence(self._serial, 1) @@ -292,7 +312,7 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): except InvalidHost as err: raise InvalidHost("Error enabling motion detection") from err - def disable_motion_detection(self): + def disable_motion_detection(self) -> None: """Disable motion detection.""" try: self.coordinator.ezviz_client.set_camera_defence(self._serial, 0) @@ -301,11 +321,11 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): raise InvalidHost("Error disabling motion detection") from err @property - def unique_id(self): + def unique_id(self) -> str: """Return the name of this camera.""" return self._serial - async def async_camera_image(self): + async def async_camera_image(self) -> bytes | None: """Return a frame from the camera stream.""" ffmpeg = ImageFrame(self._ffmpeg.binary) @@ -315,7 +335,7 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): return image @property - def device_info(self): + def device_info(self) -> DeviceInfo: """Return the device_info of the device.""" return { "identifiers": {(DOMAIN, self._serial)}, @@ -325,7 +345,7 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): "sw_version": self.coordinator.data[self._idx]["version"], } - async def stream_source(self): + async def stream_source(self) -> str | None: """Return the stream source.""" local_ip = self.coordinator.data[self._idx]["local_ip"] if self._local_rtsp_port: @@ -340,9 +360,8 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): return rtsp_stream_source return None - def perform_ptz(self, direction, speed): + def perform_ptz(self, direction: str, speed: int) -> None: """Perform a PTZ action on the camera.""" - _LOGGER.debug("PTZ action '%s' on %s", direction, self._name) try: self.coordinator.ezviz_client.ptz_control( str(direction).upper(), self._serial, "START", speed @@ -354,21 +373,21 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): except HTTPError as err: raise HTTPError("Cannot perform PTZ") from err - def perform_sound_alarm(self, enable): + def perform_sound_alarm(self, enable: int) -> None: """Sound the alarm on a camera.""" try: self.coordinator.ezviz_client.sound_alarm(self._serial, enable) except HTTPError as err: raise HTTPError("Cannot sound alarm") from err - def perform_wake_device(self): + def perform_wake_device(self) -> None: """Basically wakes the camera by querying the device.""" try: self.coordinator.ezviz_client.get_detection_sensibility(self._serial) except (HTTPError, PyEzvizError) as err: raise PyEzvizError("Cannot wake device") from err - def perform_alarm_sound(self, level): + def perform_alarm_sound(self, level: int) -> None: """Enable/Disable movement sound alarm.""" try: self.coordinator.ezviz_client.alarm_sound(self._serial, level, 1) @@ -377,7 +396,9 @@ class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity): "Cannot set alarm sound level for on movement detected" ) from err - def perform_set_alarm_detection_sensibility(self, level, type_value): + def perform_set_alarm_detection_sensibility( + self, level: int, type_value: int + ) -> None: """Set camera detection sensibility level service.""" try: self.coordinator.ezviz_client.detection_sensibility( diff --git a/homeassistant/components/ezviz/const.py b/homeassistant/components/ezviz/const.py index e3e2cae712c..ec1471d8bc4 100644 --- a/homeassistant/components/ezviz/const.py +++ b/homeassistant/components/ezviz/const.py @@ -34,7 +34,7 @@ SERVICE_DETECTION_SENSITIVITY = "set_alarm_detection_sensibility" EU_URL = "apiieu.ezvizlife.com" RUSSIA_URL = "apirus.ezvizru.com" DEFAULT_CAMERA_USERNAME = "admin" -DEFAULT_RTSP_PORT = "554" +DEFAULT_RTSP_PORT = 554 DEFAULT_TIMEOUT = 25 DEFAULT_FFMPEG_ARGUMENTS = ""