Enable strict type checks for camera platform (#50395)

This commit is contained in:
Ruslan Sayfutdinov 2021-05-10 14:12:15 +01:00 committed by GitHub
parent 0cdb8ad892
commit 8e2b3aab44
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 171 additions and 107 deletions

View File

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

View File

@ -4,13 +4,14 @@ from __future__ import annotations
import asyncio
import base64
import collections
from collections.abc import Awaitable, Mapping
from contextlib import suppress
from datetime import timedelta
from datetime import datetime, timedelta
import hashlib
import logging
import os
from random import SystemRandom
from typing import cast, final
from typing import Callable, Final, cast, final
from aiohttp import web
import async_timeout
@ -28,6 +29,8 @@ from homeassistant.components.media_player.const import (
)
from homeassistant.components.stream import Stream, create_stream
from homeassistant.components.stream.const import FORMAT_CONTENT_TYPE, OUTPUT_FORMATS
from homeassistant.components.websocket_api import ActiveConnection
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_FILENAME,
@ -36,7 +39,7 @@ from homeassistant.const import (
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.core import callback
from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.config_validation import ( # noqa: F401
@ -46,6 +49,7 @@ from homeassistant.helpers.config_validation import ( # noqa: F401
from homeassistant.helpers.entity import Entity, entity_sources
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.network import get_url
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from .const import (
@ -59,53 +63,53 @@ from .const import (
)
from .prefs import CameraPreferences
# mypy: allow-untyped-calls, allow-untyped-defs
# mypy: allow-untyped-calls
_LOGGER = logging.getLogger(__name__)
SERVICE_ENABLE_MOTION = "enable_motion_detection"
SERVICE_DISABLE_MOTION = "disable_motion_detection"
SERVICE_SNAPSHOT = "snapshot"
SERVICE_PLAY_STREAM = "play_stream"
SERVICE_ENABLE_MOTION: Final = "enable_motion_detection"
SERVICE_DISABLE_MOTION: Final = "disable_motion_detection"
SERVICE_SNAPSHOT: Final = "snapshot"
SERVICE_PLAY_STREAM: Final = "play_stream"
SCAN_INTERVAL = timedelta(seconds=30)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
SCAN_INTERVAL: Final = timedelta(seconds=30)
ENTITY_ID_FORMAT: Final = DOMAIN + ".{}"
ATTR_FILENAME = "filename"
ATTR_MEDIA_PLAYER = "media_player"
ATTR_FORMAT = "format"
ATTR_FILENAME: Final = "filename"
ATTR_MEDIA_PLAYER: Final = "media_player"
ATTR_FORMAT: Final = "format"
STATE_RECORDING = "recording"
STATE_STREAMING = "streaming"
STATE_IDLE = "idle"
STATE_RECORDING: Final = "recording"
STATE_STREAMING: Final = "streaming"
STATE_IDLE: Final = "idle"
# Bitfield of features supported by the camera entity
SUPPORT_ON_OFF = 1
SUPPORT_STREAM = 2
SUPPORT_ON_OFF: Final = 1
SUPPORT_STREAM: Final = 2
DEFAULT_CONTENT_TYPE = "image/jpeg"
ENTITY_IMAGE_URL = "/api/camera_proxy/{0}?token={1}"
DEFAULT_CONTENT_TYPE: Final = "image/jpeg"
ENTITY_IMAGE_URL: Final = "/api/camera_proxy/{0}?token={1}"
TOKEN_CHANGE_INTERVAL = timedelta(minutes=5)
_RND = SystemRandom()
TOKEN_CHANGE_INTERVAL: Final = timedelta(minutes=5)
_RND: Final = SystemRandom()
MIN_STREAM_INTERVAL = 0.5 # seconds
MIN_STREAM_INTERVAL: Final = 0.5 # seconds
CAMERA_SERVICE_SNAPSHOT = {vol.Required(ATTR_FILENAME): cv.template}
CAMERA_SERVICE_SNAPSHOT: Final = {vol.Required(ATTR_FILENAME): cv.template}
CAMERA_SERVICE_PLAY_STREAM = {
CAMERA_SERVICE_PLAY_STREAM: Final = {
vol.Required(ATTR_MEDIA_PLAYER): cv.entities_domain(DOMAIN_MP),
vol.Optional(ATTR_FORMAT, default="hls"): vol.In(OUTPUT_FORMATS),
}
CAMERA_SERVICE_RECORD = {
CAMERA_SERVICE_RECORD: Final = {
vol.Required(CONF_FILENAME): cv.template,
vol.Optional(CONF_DURATION, default=30): vol.Coerce(int),
vol.Optional(CONF_LOOKBACK, default=0): vol.Coerce(int),
}
WS_TYPE_CAMERA_THUMBNAIL = "camera_thumbnail"
SCHEMA_WS_CAMERA_THUMBNAIL = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
WS_TYPE_CAMERA_THUMBNAIL: Final = "camera_thumbnail"
SCHEMA_WS_CAMERA_THUMBNAIL: Final = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
{
vol.Required("type"): WS_TYPE_CAMERA_THUMBNAIL,
vol.Required("entity_id"): cv.entity_id,
@ -122,14 +126,16 @@ class Image:
@bind_hass
async def async_request_stream(hass, entity_id, fmt):
async def async_request_stream(hass: HomeAssistant, entity_id: str, fmt: str) -> str:
"""Request a stream for a camera entity."""
camera = _get_camera_from_entity_id(hass, entity_id)
return await _async_stream_endpoint_url(hass, camera, fmt)
@bind_hass
async def async_get_image(hass, entity_id, timeout=10):
async def async_get_image(
hass: HomeAssistant, entity_id: str, timeout: int = 10
) -> Image:
"""Fetch an image from a camera entity."""
camera = _get_camera_from_entity_id(hass, entity_id)
@ -144,7 +150,7 @@ async def async_get_image(hass, entity_id, timeout=10):
@bind_hass
async def async_get_stream_source(hass, entity_id):
async def async_get_stream_source(hass: HomeAssistant, entity_id: str) -> str | None:
"""Fetch the stream source for a camera entity."""
camera = _get_camera_from_entity_id(hass, entity_id)
@ -152,14 +158,21 @@ async def async_get_stream_source(hass, entity_id):
@bind_hass
async def async_get_mjpeg_stream(hass, request, entity_id):
async def async_get_mjpeg_stream(
hass: HomeAssistant, request: web.Request, entity_id: str
) -> web.StreamResponse:
"""Fetch an mjpeg stream from a camera entity."""
camera = _get_camera_from_entity_id(hass, entity_id)
return await camera.handle_async_mjpeg_stream(request)
async def async_get_still_stream(request, image_cb, content_type, interval):
async def async_get_still_stream(
request: web.Request,
image_cb: Callable[[], Awaitable[bytes | None]],
content_type: str,
interval: float,
) -> web.StreamResponse:
"""Generate an HTTP MJPEG stream from camera images.
This method must be run in the event loop.
@ -168,7 +181,7 @@ async def async_get_still_stream(request, image_cb, content_type, interval):
response.content_type = CONTENT_TYPE_MULTIPART.format("--frameboundary")
await response.prepare(request)
async def write_to_mjpeg_stream(img_bytes):
async def write_to_mjpeg_stream(img_bytes: bytes) -> None:
"""Write image to stream."""
await response.write(
bytes(
@ -202,7 +215,7 @@ async def async_get_still_stream(request, image_cb, content_type, interval):
return response
def _get_camera_from_entity_id(hass, entity_id):
def _get_camera_from_entity_id(hass: HomeAssistant, entity_id: str) -> Camera:
"""Get camera component from entity_id."""
component = hass.data.get(DOMAIN)
@ -217,10 +230,10 @@ def _get_camera_from_entity_id(hass, entity_id):
if not camera.is_on:
raise HomeAssistantError("Camera is off")
return camera
return cast(Camera, camera)
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the camera component."""
component = hass.data[DOMAIN] = EntityComponent(
_LOGGER, DOMAIN, hass, SCAN_INTERVAL
@ -241,8 +254,9 @@ async def async_setup(hass, config):
await component.async_setup(config)
async def preload_stream(_):
async def preload_stream(_event: Event) -> None:
for camera in component.entities:
camera = cast(Camera, camera)
camera_prefs = prefs.get(camera.entity_id)
if not camera_prefs.preload_stream:
continue
@ -256,9 +270,10 @@ async def async_setup(hass, config):
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, preload_stream)
@callback
def update_tokens(time):
def update_tokens(time: datetime) -> None:
"""Update tokens of the entities."""
for entity in component.entities:
entity = cast(Camera, entity)
entity.async_update_token()
entity.async_write_ha_state()
@ -287,67 +302,69 @@ async def async_setup(hass, config):
return True
async def async_setup_entry(hass, entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
return await hass.data[DOMAIN].async_setup_entry(entry)
component: EntityComponent = hass.data[DOMAIN]
return await component.async_setup_entry(entry)
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.data[DOMAIN].async_unload_entry(entry)
component: EntityComponent = hass.data[DOMAIN]
return await component.async_unload_entry(entry)
class Camera(Entity):
"""The base class for camera entities."""
def __init__(self):
def __init__(self) -> None:
"""Initialize a camera."""
self.is_streaming = False
self.stream = None
self.stream_options = {}
self.content_type = DEFAULT_CONTENT_TYPE
self.is_streaming: bool = False
self.stream: Stream | None = None
self.stream_options: dict[str, str] = {}
self.content_type: str = DEFAULT_CONTENT_TYPE
self.access_tokens: collections.deque = collections.deque([], 2)
self.async_update_token()
@property
def should_poll(self):
def should_poll(self) -> bool:
"""No need to poll cameras."""
return False
@property
def entity_picture(self):
def entity_picture(self) -> str:
"""Return a link to the camera feed as entity picture."""
return ENTITY_IMAGE_URL.format(self.entity_id, self.access_tokens[-1])
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
return 0
@property
def is_recording(self):
def is_recording(self) -> bool:
"""Return true if the device is recording."""
return False
@property
def brand(self):
def brand(self) -> str | None:
"""Return the camera brand."""
return None
@property
def motion_detection_enabled(self):
def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status."""
return None
return False
@property
def model(self):
def model(self) -> str | None:
"""Return the camera model."""
return None
@property
def frame_interval(self):
def frame_interval(self) -> float:
"""Return the interval between frames of the mjpeg stream."""
return 0.5
return MIN_STREAM_INTERVAL
async def create_stream(self) -> Stream | None:
"""Create a Stream for stream_source."""
@ -360,25 +377,29 @@ class Camera(Entity):
self.stream = create_stream(self.hass, source, options=self.stream_options)
return self.stream
async def stream_source(self):
async def stream_source(self) -> str | None:
"""Return the source of the stream."""
return None
def camera_image(self):
def camera_image(self) -> bytes | None:
"""Return bytes of camera image."""
raise NotImplementedError()
async def async_camera_image(self):
async def async_camera_image(self) -> bytes | None:
"""Return bytes of camera image."""
return await self.hass.async_add_executor_job(self.camera_image)
async def handle_async_still_stream(self, request, interval):
async def handle_async_still_stream(
self, request: web.Request, interval: float
) -> web.StreamResponse:
"""Generate an HTTP MJPEG stream from camera images."""
return await async_get_still_stream(
request, self.async_camera_image, self.content_type, interval
)
async def handle_async_mjpeg_stream(self, request):
async def handle_async_mjpeg_stream(
self, request: web.Request
) -> web.StreamResponse:
"""Serve an HTTP MJPEG stream from the camera.
This method can be overridden by camera platforms to proxy
@ -387,7 +408,7 @@ class Camera(Entity):
return await self.handle_async_still_stream(request, self.frame_interval)
@property
def state(self):
def state(self) -> str:
"""Return the camera state."""
if self.is_recording:
return STATE_RECORDING
@ -396,45 +417,45 @@ class Camera(Entity):
return STATE_IDLE
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if on."""
return True
def turn_off(self):
def turn_off(self) -> None:
"""Turn off camera."""
raise NotImplementedError()
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn off camera."""
await self.hass.async_add_executor_job(self.turn_off)
def turn_on(self):
def turn_on(self) -> None:
"""Turn off camera."""
raise NotImplementedError()
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn off camera."""
await self.hass.async_add_executor_job(self.turn_on)
def enable_motion_detection(self):
def enable_motion_detection(self) -> None:
"""Enable motion detection in the camera."""
raise NotImplementedError()
async def async_enable_motion_detection(self):
async def async_enable_motion_detection(self) -> None:
"""Call the job and enable motion detection."""
await self.hass.async_add_executor_job(self.enable_motion_detection)
def disable_motion_detection(self):
def disable_motion_detection(self) -> None:
"""Disable motion detection in camera."""
raise NotImplementedError()
async def async_disable_motion_detection(self):
async def async_disable_motion_detection(self) -> None:
"""Call the job and disable motion detection."""
await self.hass.async_add_executor_job(self.disable_motion_detection)
@final
@property
def state_attributes(self):
def state_attributes(self) -> dict[str, str | None]:
"""Return the camera state attributes."""
attrs = {"access_token": self.access_tokens[-1]}
@ -450,7 +471,7 @@ class Camera(Entity):
return attrs
@callback
def async_update_token(self):
def async_update_token(self) -> None:
"""Update the used token."""
self.access_tokens.append(
hashlib.sha256(_RND.getrandbits(256).to_bytes(32, "little")).hexdigest()
@ -466,7 +487,7 @@ class CameraView(HomeAssistantView):
"""Initialize a basic camera view."""
self.component = component
async def get(self, request: web.Request, entity_id: str) -> web.Response:
async def get(self, request: web.Request, entity_id: str) -> web.StreamResponse:
"""Start a GET request."""
camera = self.component.get_entity(entity_id)
@ -489,7 +510,7 @@ class CameraView(HomeAssistantView):
return await self.handle(request, camera)
async def handle(self, request, camera):
async def handle(self, request: web.Request, camera: Camera) -> web.StreamResponse:
"""Handle the camera request."""
raise NotImplementedError()
@ -518,7 +539,7 @@ class CameraMjpegStream(CameraView):
url = "/api/camera_proxy_stream/{entity_id}"
name = "api:camera:stream"
async def handle(self, request: web.Request, camera: Camera) -> web.Response:
async def handle(self, request: web.Request, camera: Camera) -> web.StreamResponse:
"""Serve camera stream, possibly with interval."""
interval_str = request.query.get("interval")
if interval_str is None:
@ -535,7 +556,9 @@ class CameraMjpegStream(CameraView):
@websocket_api.async_response
async def websocket_camera_thumbnail(hass, connection, msg):
async def websocket_camera_thumbnail(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get camera thumbnail websocket command.
Async friendly.
@ -566,7 +589,9 @@ async def websocket_camera_thumbnail(hass, connection, msg):
}
)
@websocket_api.async_response
async def ws_camera_stream(hass, connection, msg):
async def ws_camera_stream(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get camera stream websocket command.
Async friendly.
@ -590,7 +615,9 @@ async def ws_camera_stream(hass, connection, msg):
{vol.Required("type"): "camera/get_prefs", vol.Required("entity_id"): cv.entity_id}
)
@websocket_api.async_response
async def websocket_get_prefs(hass, connection, msg):
async def websocket_get_prefs(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS].get(msg["entity_id"])
connection.send_result(msg["id"], prefs.as_dict())
@ -604,7 +631,9 @@ async def websocket_get_prefs(hass, connection, msg):
}
)
@websocket_api.async_response
async def websocket_update_prefs(hass, connection, msg):
async def websocket_update_prefs(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle request for account info."""
prefs = hass.data[DATA_CAMERA_PREFS]
@ -617,10 +646,12 @@ async def websocket_update_prefs(hass, connection, msg):
connection.send_result(msg["id"], prefs.get(entity_id).as_dict())
async def async_handle_snapshot_service(camera, service):
async def async_handle_snapshot_service(
camera: Camera, service_call: ServiceCall
) -> None:
"""Handle snapshot services calls."""
hass = camera.hass
filename = service.data[ATTR_FILENAME]
filename = service_call.data[ATTR_FILENAME]
filename.hass = hass
snapshot_file = filename.async_render(variables={ATTR_ENTITY_ID: camera})
@ -632,8 +663,10 @@ async def async_handle_snapshot_service(camera, service):
image = await camera.async_camera_image()
def _write_image(to_file, image_data):
def _write_image(to_file: str, image_data: bytes | None) -> None:
"""Executor helper to write image."""
if image_data is None:
return
if not os.path.exists(os.path.dirname(to_file)):
os.makedirs(os.path.dirname(to_file), exist_ok=True)
with open(to_file, "wb") as img_file:
@ -645,13 +678,15 @@ async def async_handle_snapshot_service(camera, service):
_LOGGER.error("Can't write image to file: %s", err)
async def async_handle_play_stream_service(camera, service_call):
async def async_handle_play_stream_service(
camera: Camera, service_call: ServiceCall
) -> None:
"""Handle play stream services calls."""
fmt = service_call.data[ATTR_FORMAT]
url = await _async_stream_endpoint_url(camera.hass, camera, fmt)
hass = camera.hass
data = {
data: Mapping[str, str] = {
ATTR_MEDIA_CONTENT_ID: f"{get_url(hass)}{url}",
ATTR_MEDIA_CONTENT_TYPE: FORMAT_CONTENT_TYPE[fmt],
}
@ -696,7 +731,9 @@ async def async_handle_play_stream_service(camera, service_call):
)
async def _async_stream_endpoint_url(hass, camera, fmt):
async def _async_stream_endpoint_url(
hass: HomeAssistant, camera: Camera, fmt: str
) -> str:
stream = await camera.create_stream()
if not stream:
raise HomeAssistantError(
@ -712,7 +749,9 @@ async def _async_stream_endpoint_url(hass, camera, fmt):
return stream.endpoint_url(fmt)
async def async_handle_record_service(camera, call):
async def async_handle_record_service(
camera: Camera, service_call: ServiceCall
) -> None:
"""Handle stream recording service calls."""
stream = await camera.create_stream()
@ -720,10 +759,12 @@ async def async_handle_record_service(camera, call):
raise HomeAssistantError(f"{camera.entity_id} does not support record service")
hass = camera.hass
filename = call.data[CONF_FILENAME]
filename = service_call.data[CONF_FILENAME]
filename.hass = hass
video_path = filename.async_render(variables={ATTR_ENTITY_ID: camera})
await stream.async_record(
video_path, duration=call.data[CONF_DURATION], lookback=call.data[CONF_LOOKBACK]
video_path,
duration=service_call.data[CONF_DURATION],
lookback=service_call.data[CONF_LOOKBACK],
)

View File

@ -1,27 +1,30 @@
"""Preference management for camera component."""
from homeassistant.helpers.typing import UNDEFINED
from __future__ import annotations
from typing import Final
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from .const import DOMAIN, PREF_PRELOAD_STREAM
# mypy: allow-untyped-defs, no-check-untyped-defs
STORAGE_KEY = DOMAIN
STORAGE_VERSION = 1
STORAGE_KEY: Final = DOMAIN
STORAGE_VERSION: Final = 1
class CameraEntityPreferences:
"""Handle preferences for camera entity."""
def __init__(self, prefs):
def __init__(self, prefs: dict[str, bool]) -> None:
"""Initialize prefs."""
self._prefs = prefs
def as_dict(self):
def as_dict(self) -> dict[str, bool]:
"""Return dictionary version."""
return self._prefs
@property
def preload_stream(self):
def preload_stream(self) -> bool:
"""Return if stream is loaded on hass start."""
return self._prefs.get(PREF_PRELOAD_STREAM, False)
@ -29,13 +32,13 @@ class CameraEntityPreferences:
class CameraPreferences:
"""Handle camera preferences."""
def __init__(self, hass):
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize camera prefs."""
self._hass = hass
self._store = hass.helpers.storage.Store(STORAGE_VERSION, STORAGE_KEY)
self._prefs = None
self._prefs: dict[str, dict[str, bool]] | None = None
async def async_initialize(self):
async def async_initialize(self) -> None:
"""Finish initializing the preferences."""
prefs = await self._store.async_load()
@ -45,9 +48,15 @@ class CameraPreferences:
self._prefs = prefs
async def async_update(
self, entity_id, *, preload_stream=UNDEFINED, stream_options=UNDEFINED
):
self,
entity_id: str,
*,
preload_stream: bool | UndefinedType = UNDEFINED,
stream_options: dict[str, str] | UndefinedType = UNDEFINED,
) -> None:
"""Update camera preferences."""
# Prefs already initialized.
assert self._prefs is not None
if not self._prefs.get(entity_id):
self._prefs[entity_id] = {}
@ -57,6 +66,8 @@ class CameraPreferences:
await self._store.async_save(self._prefs)
def get(self, entity_id):
def get(self, entity_id: str) -> CameraEntityPreferences:
"""Get preferences for an entity."""
# Prefs are already initialized.
assert self._prefs is not None
return CameraEntityPreferences(self._prefs.get(entity_id, {}))

View File

@ -124,7 +124,7 @@ class Stream:
if self.options is None:
self.options = {}
def endpoint_url(self, fmt):
def endpoint_url(self, fmt: str) -> str:
"""Start the stream and returns a url for the output format."""
if fmt not in self._outputs:
raise ValueError(f"Stream is not configured for format '{fmt}'")

View File

@ -78,7 +78,7 @@ class SynoDSMCamera(SynologyDSMBaseEntity, Camera):
},
coordinator,
)
Camera.__init__(self) # type: ignore[no-untyped-call]
Camera.__init__(self)
self._camera_id = camera_id
@property

View File

@ -110,6 +110,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.camera.*]
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