evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add strict type annotations to canary (#50943)

Browse source 
* Add strict type annotations

* Add missing futur import

* Apply suggestions

* Apply suggestions
This commit is contained in:
Michael 2021-05-22 10:14:59 +02:00 • committed by GitHub
parent 15e2c6d7dc
commit 2e316f6fd5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 175 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -18,6 +18,7 @@ homeassistant.components.bond.*
homeassistant.components.brother.*
homeassistant.components.calendar.*
homeassistant.components.camera.*
homeassistant.components.canary.*
homeassistant.components.cover.*
homeassistant.components.device_automation.*
homeassistant.components.elgato.*

16
homeassistant/components/canary/__init__.py
View file

@ -1,9 +1,12 @@
"""Support for Canary devices."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Final
from canary.api import Api
from requests import ConnectTimeout, HTTPError
from requests.exceptions import ConnectTimeout, HTTPError
import voluptuous as vol
from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN
@ -12,6 +15,7 @@ from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_FFMPEG_ARGUMENTS,
@ -23,11 +27,11 @@ from .const import (
)
from .coordinator import CanaryDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
MIN_TIME_BETWEEN_UPDATES: Final = timedelta(seconds=30)
CONFIG_SCHEMA = vol.Schema(
CONFIG_SCHEMA: Final = vol.Schema(
vol.All(
cv.deprecated(DOMAIN),
{
@ -45,10 +49,10 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["alarm_control_panel", "camera", "sensor"]
PLATFORMS: Final[list[str]] = ["alarm_control_panel", "camera", "sensor"]
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Canary integration."""
hass.data.setdefault(DOMAIN, {})

37
homeassistant/components/canary/alarm_control_panel.py
View file

@ -1,7 +1,14 @@
"""Support for Canary alarm."""
from __future__ import annotations
from canary.api import LOCATION_MODE_AWAY, LOCATION_MODE_HOME, LOCATION_MODE_NIGHT
from typing import Any
from canary.api import (
LOCATION_MODE_AWAY,
LOCATION_MODE_HOME,
LOCATION_MODE_NIGHT,
Location,
)
from homeassistant.components.alarm_control_panel import AlarmControlPanelEntity
from homeassistant.components.alarm_control_panel.const import (
@ -44,29 +51,33 @@ async def async_setup_entry(
class CanaryAlarm(CoordinatorEntity, AlarmControlPanelEntity):
"""Representation of a Canary alarm control panel."""
def __init__(self, coordinator, location):
coordinator: CanaryDataUpdateCoordinator
def __init__(
self, coordinator: CanaryDataUpdateCoordinator, location: Location
) -> None:
"""Initialize a Canary security camera."""
super().__init__(coordinator)
self._location_id = location.location_id
self._location_name = location.name
self._location_id: str = location.location_id
self._location_name: str = location.name
@property
def location(self):
def location(self) -> Location:
"""Return information about the location."""
return self.coordinator.data["locations"][self._location_id]
@property
def name(self):
def name(self) -> str:
"""Return the name of the alarm."""
return self._location_name
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique ID of the alarm."""
return str(self._location_id)
@property
def state(self):
def state(self) -> str | None:
"""Return the state of the device."""
if self.location.is_private:
return STATE_ALARM_DISARMED
@ -87,25 +98,25 @@ class CanaryAlarm(CoordinatorEntity, AlarmControlPanelEntity):
return SUPPORT_ALARM_ARM_HOME | SUPPORT_ALARM_ARM_AWAY | SUPPORT_ALARM_ARM_NIGHT
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
return {"private": self.location.is_private}
def alarm_disarm(self, code=None):
def alarm_disarm(self, code: str | None = None) -> None:
"""Send disarm command."""
self.coordinator.canary.set_location_mode(
self._location_id, self.location.mode.name, True
)
def alarm_arm_home(self, code=None):
def alarm_arm_home(self, code: str | None = None) -> None:
"""Send arm home command."""
self.coordinator.canary.set_location_mode(self._location_id, LOCATION_MODE_HOME)
def alarm_arm_away(self, code=None):
def alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command."""
self.coordinator.canary.set_location_mode(self._location_id, LOCATION_MODE_AWAY)
def alarm_arm_night(self, code=None):
def alarm_arm_night(self, code: str | None = None) -> None:
"""Send arm night command."""
self.coordinator.canary.set_location_mode(
self._location_id, LOCATION_MODE_NIGHT

68
homeassistant/components/canary/camera.py
View file

@ -3,17 +3,25 @@ from __future__ import annotations
import asyncio
from datetime import timedelta
from typing import Final
from aiohttp.web import Request, StreamResponse
from canary.api import Device, Location
from canary.live_stream_api import LiveStreamSession
from haffmpeg.camera import CameraMjpeg
from haffmpeg.tools import IMAGE_JPEG, ImageFrame
import voluptuous as vol
from homeassistant.components.camera import PLATFORM_SCHEMA, Camera
from homeassistant.components.ffmpeg import DATA_FFMPEG
from homeassistant.components.camera import (
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
Camera,
)
from homeassistant.components.ffmpeg import DATA_FFMPEG, FFmpegManager
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_aiohttp_proxy_stream
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import Throttle
@ -28,11 +36,11 @@ from .const import (
)
from .coordinator import CanaryDataUpdateCoordinator
MIN_TIME_BETWEEN_SESSION_RENEW = timedelta(seconds=90)
MIN_TIME_BETWEEN_SESSION_RENEW: Final = timedelta(seconds=90)
PLATFORM_SCHEMA = vol.All(
PLATFORM_SCHEMA: Final = vol.All(
cv.deprecated(CONF_FFMPEG_ARGUMENTS),
PLATFORM_SCHEMA.extend(
PARENT_PLATFORM_SCHEMA.extend(
{
vol.Optional(
CONF_FFMPEG_ARGUMENTS, default=DEFAULT_FFMPEG_ARGUMENTS
@ -51,10 +59,10 @@ async def async_setup_entry(
coordinator: CanaryDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
ffmpeg_arguments = entry.options.get(
ffmpeg_arguments: str = entry.options.get(
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
)
cameras = []
cameras: list[CanaryCamera] = []
for location_id, location in coordinator.data["locations"].items():
for device in location.devices:
@ -76,37 +84,47 @@ async def async_setup_entry(
class CanaryCamera(CoordinatorEntity, Camera):
"""An implementation of a Canary security camera."""
def __init__(self, hass, coordinator, location_id, device, timeout, ffmpeg_args):
coordinator: CanaryDataUpdateCoordinator
def __init__(
self,
hass: HomeAssistant,
coordinator: CanaryDataUpdateCoordinator,
location_id: str,
device: Device,
timeout: int,
ffmpeg_args: str,
) -> None:
"""Initialize a Canary security camera."""
super().__init__(coordinator)
Camera.__init__(self)
self._ffmpeg = hass.data[DATA_FFMPEG]
self._ffmpeg: FFmpegManager = hass.data[DATA_FFMPEG]
self._ffmpeg_arguments = ffmpeg_args
self._location_id = location_id
self._device = device
self._device_id = device.device_id
self._device_name = device.name
self._device_id: str = device.device_id
self._device_name: str = device.name
self._device_type_name = device.device_type["name"]
self._timeout = timeout
self._live_stream_session = None
self._live_stream_session: LiveStreamSession | None = None
@property
def location(self):
def location(self) -> Location:
"""Return information about the location."""
return self.coordinator.data["locations"][self._location_id]
@property
def name(self):
def name(self) -> str:
"""Return the name of this device."""
return self._device_name
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique ID of this camera."""
return str(self._device_id)
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, str(self._device_id))},
@ -116,16 +134,16 @@ class CanaryCamera(CoordinatorEntity, Camera):
}
@property
def is_recording(self):
def is_recording(self) -> bool:
"""Return true if the device is recording."""
return self.location.is_recording
return self.location.is_recording # type: ignore[no-any-return]
@property
def motion_detection_enabled(self):
def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status."""
return not self.location.is_recording
async def async_camera_image(self):
async def async_camera_image(self) -> bytes | None:
"""Return a still image response from the camera."""
await self.hass.async_add_executor_job(self.renew_live_stream_session)
live_stream_url = await self.hass.async_add_executor_job(
@ -133,7 +151,7 @@ class CanaryCamera(CoordinatorEntity, Camera):
)
ffmpeg = ImageFrame(self._ffmpeg.binary)
image = await asyncio.shield(
image: bytes | None = await asyncio.shield(
ffmpeg.get_image(
live_stream_url,
output_format=IMAGE_JPEG,
@ -142,10 +160,12 @@ class CanaryCamera(CoordinatorEntity, Camera):
)
return image
async def handle_async_mjpeg_stream(self, request):
async def handle_async_mjpeg_stream(
self, request: Request
) -> StreamResponse | None:
"""Generate an HTTP MJPEG stream from the camera."""
if self._live_stream_session is None:
return
return None
stream = CameraMjpeg(self._ffmpeg.binary)
await stream.open_camera(
@ -164,7 +184,7 @@ class CanaryCamera(CoordinatorEntity, Camera):
await stream.close()
@Throttle(MIN_TIME_BETWEEN_SESSION_RENEW)
def renew_live_stream_session(self):
def renew_live_stream_session(self) -> None:
"""Renew live stream session."""
self._live_stream_session = self.coordinator.canary.get_live_stream_session(
self._device

16
homeassistant/components/canary/config_flow.py
View file

@ -2,13 +2,13 @@
from __future__ import annotations
import logging
from typing import Any
from typing import Final
from canary.api import Api
from requests import ConnectTimeout, HTTPError
from requests.exceptions import ConnectTimeout, HTTPError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, OptionsFlow
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
@ -21,10 +21,10 @@ from .const import (
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
def validate_input(hass: HomeAssistant, data: dict) -> dict[str, Any]:
def validate_input(hass: HomeAssistant, data: ConfigType) -> bool:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -46,7 +46,7 @@ class CanaryConfigFlow(ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Get the options flow for this handler."""
return CanaryOptionsFlowHandler(config_entry)
@ -100,11 +100,11 @@ class CanaryConfigFlow(ConfigFlow, domain=DOMAIN):
class CanaryOptionsFlowHandler(OptionsFlow):
"""Handle Canary client options."""
def __init__(self, config_entry):
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: ConfigType | None = None):
async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
"""Manage Canary options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

16
homeassistant/components/canary/const.py
View file

@ -1,16 +1,18 @@
"""Constants for the Canary integration."""
DOMAIN = "canary"
from typing import Final
MANUFACTURER = "Canary Connect, Inc"
DOMAIN: Final = "canary"
MANUFACTURER: Final = "Canary Connect, Inc"
# Configuration
CONF_FFMPEG_ARGUMENTS = "ffmpeg_arguments"
CONF_FFMPEG_ARGUMENTS: Final = "ffmpeg_arguments"
# Data
DATA_COORDINATOR = "coordinator"
DATA_UNDO_UPDATE_LISTENER = "undo_update_listener"
DATA_COORDINATOR: Final = "coordinator"
DATA_UNDO_UPDATE_LISTENER: Final = "undo_update_listener"
# Defaults
DEFAULT_FFMPEG_ARGUMENTS = "-pred 1"
DEFAULT_TIMEOUT = 10
DEFAULT_FFMPEG_ARGUMENTS: Final = "-pred 1"
DEFAULT_TIMEOUT: Final = 10

16
homeassistant/components/canary/coordinator.py
View file

@ -1,15 +1,19 @@
"""Provides the Canary DataUpdateCoordinator."""
from __future__ import annotations
from collections.abc import ValuesView
from datetime import timedelta
import logging
from async_timeout import timeout
from canary.api import Api
from requests import ConnectTimeout, HTTPError
from canary.api import Api, Location
from requests.exceptions import ConnectTimeout, HTTPError
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
from .model import CanaryData
_LOGGER = logging.getLogger(__name__)
@ -29,10 +33,10 @@ class CanaryDataUpdateCoordinator(DataUpdateCoordinator):
update_interval=update_interval,
)
def _update_data(self) -> dict:
def _update_data(self) -> CanaryData:
"""Fetch data from Canary via sync functions."""
locations_by_id = {}
readings_by_device_id = {}
locations_by_id: dict[str, Location] = {}
readings_by_device_id: dict[str, ValuesView] = {}
for location in self.canary.get_locations():
location_id = location.location_id
@ -49,7 +53,7 @@ class CanaryDataUpdateCoordinator(DataUpdateCoordinator):
"readings": readings_by_device_id,
}
async def _async_update_data(self) -> dict:
async def _async_update_data(self) -> CanaryData:
"""Fetch data from Canary."""
try:

18
homeassistant/components/canary/model.py Normal file
View file

@ -0,0 +1,18 @@
"""Constants for the Canary integration."""
from __future__ import annotations
from collections.abc import ValuesView
from typing import List, Optional, Tuple, TypedDict
from canary.api import Location
class CanaryData(TypedDict):
"""TypedDict for Canary Coordinator Data."""
locations: dict[str, Location]
readings: dict[str, ValuesView]
SensorTypeItem = Tuple[str, Optional[str], Optional[str], Optional[str], List[str]]

67
homeassistant/components/canary/sensor.py
View file

@ -1,7 +1,9 @@
"""Support for Canary sensors."""
from __future__ import annotations
from canary.api import SensorType
from typing import Final
from canary.api import Device, Location, SensorType
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
@ -15,41 +17,43 @@ from homeassistant.const import (
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
from .coordinator import CanaryDataUpdateCoordinator
from .model import SensorTypeItem
SENSOR_VALUE_PRECISION = 2
ATTR_AIR_QUALITY = "air_quality"
SENSOR_VALUE_PRECISION: Final = 2
ATTR_AIR_QUALITY: Final = "air_quality"
# Define variables to store the device names, as referred to by the Canary API.
# Note: If Canary change the name of any of their devices (which they have done),
# then these variables will need updating, otherwise the sensors will stop working
# and disappear in Home Assistant.
CANARY_PRO = "Canary Pro"
CANARY_FLEX = "Canary Flex"
CANARY_PRO: Final = "Canary Pro"
CANARY_FLEX: Final = "Canary Flex"
# Sensor types are defined like so:
# sensor type name, unit_of_measurement, icon, device class, products supported
SENSOR_TYPES = [
["temperature", TEMP_CELSIUS, None, DEVICE_CLASS_TEMPERATURE, [CANARY_PRO]],
["humidity", PERCENTAGE, None, DEVICE_CLASS_HUMIDITY, [CANARY_PRO]],
["air_quality", None, "mdi:weather-windy", None, [CANARY_PRO]],
[
SENSOR_TYPES: Final[list[SensorTypeItem]] = [
("temperature", TEMP_CELSIUS, None, DEVICE_CLASS_TEMPERATURE, [CANARY_PRO]),
("humidity", PERCENTAGE, None, DEVICE_CLASS_HUMIDITY, [CANARY_PRO]),
("air_quality", None, "mdi:weather-windy", None, [CANARY_PRO]),
(
"wifi",
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
None,
DEVICE_CLASS_SIGNAL_STRENGTH,
[CANARY_FLEX],
],
["battery", PERCENTAGE, None, DEVICE_CLASS_BATTERY, [CANARY_FLEX]],
),
("battery", PERCENTAGE, None, DEVICE_CLASS_BATTERY, [CANARY_FLEX]),
]
STATE_AIR_QUALITY_NORMAL = "normal"
STATE_AIR_QUALITY_ABNORMAL = "abnormal"
STATE_AIR_QUALITY_VERY_ABNORMAL = "very_abnormal"
STATE_AIR_QUALITY_NORMAL: Final = "normal"
STATE_AIR_QUALITY_ABNORMAL: Final = "abnormal"
STATE_AIR_QUALITY_VERY_ABNORMAL: Final = "very_abnormal"
async def async_setup_entry(
@ -61,7 +65,7 @@ async def async_setup_entry(
coordinator: CanaryDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
DATA_COORDINATOR
]
sensors = []
sensors: list[CanarySensor] = []
for location in coordinator.data["locations"].values():
for device in location.devices:
@ -79,8 +83,17 @@ async def async_setup_entry(
class CanarySensor(CoordinatorEntity, SensorEntity):
"""Representation of a Canary sensor."""
def __init__(self, coordinator, sensor_type, location, device):
coordinator: CanaryDataUpdateCoordinator
def __init__(
self,
coordinator: CanaryDataUpdateCoordinator,
sensor_type: SensorTypeItem,
location: Location,
device: Device,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._sensor_type = sensor_type
self._device_id = device.device_id
@ -105,7 +118,7 @@ class CanarySensor(CoordinatorEntity, SensorEntity):
self._canary_type = canary_sensor_type
@property
def reading(self):
def reading(self) -> float | None:
"""Return the device sensor reading."""
readings = self.coordinator.data["readings"][self._device_id]
@ -124,22 +137,22 @@ class CanarySensor(CoordinatorEntity, SensorEntity):
return None
@property
def name(self):
def name(self) -> str:
"""Return the name of the Canary sensor."""
return self._name
@property
def state(self):
def state(self) -> float | None:
"""Return the state of the sensor."""
return self.reading
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return the unique ID of this sensor."""
return f"{self._device_id}_{self._sensor_type[0]}"
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, str(self._device_id))},
@ -149,22 +162,22 @@ class CanarySensor(CoordinatorEntity, SensorEntity):
}
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str | None:
"""Return the unit of measurement."""
return self._sensor_type[1]
@property
def device_class(self):
def device_class(self) -> str | None:
"""Device class for the sensor."""
return self._sensor_type[3]
@property
def icon(self):
def icon(self) -> str | None:
"""Icon for the sensor."""
return self._sensor_type[2]
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, str] | None:
"""Return the state attributes."""
reading = self.reading
@ -174,7 +187,7 @@ class CanarySensor(CoordinatorEntity, SensorEntity):
air_quality = STATE_AIR_QUALITY_VERY_ABNORMAL
elif reading <= 0.59:
air_quality = STATE_AIR_QUALITY_ABNORMAL
elif reading <= 1.0:
else:
air_quality = STATE_AIR_QUALITY_NORMAL
return {ATTR_AIR_QUALITY: air_quality}

14
mypy.ini
View file

@ -209,6 +209,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.canary.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.cover.*]
check_untyped_defs = true
disallow_incomplete_defs = true
@ -799,9 +810,6 @@ ignore_errors = true
[mypy-homeassistant.components.bsblan.*]
ignore_errors = true
[mypy-homeassistant.components.canary.*]
ignore_errors = true
[mypy-homeassistant.components.cast.*]
ignore_errors = true

1
script/hassfest/mypy_config.py
View file

@ -31,7 +31,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.bluetooth_tracker.*",
"homeassistant.components.bmw_connected_drive.*",
"homeassistant.components.bsblan.*",
"homeassistant.components.canary.*",
"homeassistant.components.cast.*",
"homeassistant.components.cert_expiry.*",
"homeassistant.components.climacell.*",