Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add basic typing to ezviz camera platform (#52492) 

* Add basic typing to camera platform. Small cleanups.

* Add full typing to all functions
pull/53445/head
RenierM26 2021-07-04 13:16:27 +02:00 committed by GitHub
parent 77a06e01f7
commit 73c6aa701f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 62 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
homeassistant/components/ezviz/camera.py
View File

@ -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(

2
homeassistant/components/ezviz/const.py
View File

@ -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 = ""