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

Pass width and height when requesting camera snapshot (#53835)

Browse source
This commit is contained in:
J. Nick Koston 2021-08-10 19:33:06 -05:00 • committed by GitHub
parent 390023a576
commit e99576c094
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
53 changed files with 418 additions and 113 deletions
Download patch file Download diff file Expand all files Collapse all files

6
homeassistant/components/abode/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Abode Security System cameras."""
from __future__ import annotations
from datetime import timedelta
import abodepy.helpers.constants as CONST
@ -73,7 +75,9 @@ class AbodeCamera(AbodeDevice, Camera):
else:
self._response = None
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Get a camera image."""
self.refresh_image()

6
homeassistant/components/amcrest/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Amcrest IP cameras."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from functools import partial
@ -181,7 +183,9 @@ class AmcrestCam(Camera):
finally:
self._snapshot_task = None
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
_LOGGER.debug("Take snapshot from %s", self._name)
try:

6
homeassistant/components/arlo/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Netgear Arlo IP cameras."""
from __future__ import annotations
import logging
from haffmpeg.camera import CameraMjpeg
@ -62,7 +64,9 @@ class ArloCam(Camera):
self._last_refresh = None
self.attrs = {}
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
return self._camera.last_image_from_cache

5
homeassistant/components/august/camera.py
View file

@ -1,4 +1,5 @@
"""Support for August doorbell camera."""
from __future__ import annotations
from yalexs.activity import ActivityType
from yalexs.util import update_doorbell_image_from_activity
@ -68,7 +69,9 @@ class AugustCamera(AugustEntityMixin, Camera):
if doorbell_activity is not None:
update_doorbell_image_from_activity(self._detail, doorbell_activity)
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
self._update_from_data()

6
homeassistant/components/blink/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Blink system camera."""
from __future__ import annotations
import logging
from homeassistant.components.camera import Camera
@ -65,6 +67,8 @@ class BlinkCamera(Camera):
self._camera.snap_picture()
self.data.refresh()
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
return self._camera.image_from_cache.content

6
homeassistant/components/bloomsky/camera.py
View file

@ -1,4 +1,6 @@
"""Support for a camera of a BloomSky weather station."""
from __future__ import annotations
import logging
import requests
@ -37,7 +39,9 @@ class BloomSkyCamera(Camera):
self._logger = logging.getLogger(__name__)
self._attr_unique_id = self._id
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Update the camera's image if it has changed."""
try:
self._url = self._bloomsky.devices[self._id]["Data"]["ImageURL"]

4
homeassistant/components/buienradar/camera.py
View file

@ -143,7 +143,9 @@ class BuienradarCam(Camera):
_LOGGER.error("Failed to fetch image, %s", type(err))
return False
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""
Return a still image response from the camera.

108
homeassistant/components/camera/__init__.py
View file

@ -8,7 +8,9 @@ from collections.abc import Awaitable, Mapping
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import partial
import hashlib
import inspect
import logging
import os
from random import SystemRandom
@ -62,6 +64,7 @@ from .const import (
DOMAIN,
SERVICE_RECORD,
)
from .img_util import scale_jpeg_camera_image
from .prefs import CameraPreferences
# mypy: allow-untyped-calls
@ -138,23 +141,72 @@ async def async_request_stream(hass: HomeAssistant, entity_id: str, fmt: str) ->
return await _async_stream_endpoint_url(hass, camera, fmt)
@bind_hass
async def async_get_image(
hass: HomeAssistant, entity_id: str, timeout: int = 10
async def _async_get_image(
camera: Camera,
timeout: int = 10,
width: int | None = None,
height: int | None = None,
) -> Image:
"""Fetch an image from a camera entity."""
camera = _get_camera_from_entity_id(hass, entity_id)
"""Fetch a snapshot image from a camera.
If width and height are passed, an attempt to scale
the image will be made on a best effort basis.
Not all cameras can scale images or return jpegs
that we can scale, however the majority of cases
are handled.
"""
with suppress(asyncio.CancelledError, asyncio.TimeoutError):
async with async_timeout.timeout(timeout):
image = await camera.async_camera_image()
# Calling inspect will be removed in 2022.1 after all
# custom components have had a chance to change their signature
sig = inspect.signature(camera.async_camera_image)
if "height" in sig.parameters and "width" in sig.parameters:
image_bytes = await camera.async_camera_image(
width=width, height=height
)
else:
_LOGGER.warning(
"The camera entity %s does not support requesting width and height, please open an issue with the integration author",
camera.entity_id,
)
image_bytes = await camera.async_camera_image()
if image:
return Image(camera.content_type, image)
if image_bytes:
content_type = camera.content_type
image = Image(content_type, image_bytes)
if (
width is not None
and height is not None
and "jpeg" in content_type
or "jpg" in content_type
):
assert width is not None
assert height is not None
return Image(
content_type, scale_jpeg_camera_image(image, width, height)
)
return image
raise HomeAssistantError("Unable to get image")
@bind_hass
async def async_get_image(
hass: HomeAssistant,
entity_id: str,
timeout: int = 10,
width: int | None = None,
height: int | None = None,
) -> Image:
"""Fetch an image from a camera entity.
width and height will be passed to the underlying camera.
"""
camera = _get_camera_from_entity_id(hass, entity_id)
return await _async_get_image(camera, timeout, width, height)
@bind_hass
async def async_get_stream_source(hass: HomeAssistant, entity_id: str) -> str | None:
"""Fetch the stream source for a camera entity."""
@ -387,12 +439,27 @@ class Camera(Entity):
"""Return the source of the stream."""
return None
def camera_image(self) -> bytes | None:
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
raise NotImplementedError()
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
sig = inspect.signature(self.camera_image)
# Calling inspect will be removed in 2022.1 after all
# custom components have had a chance to change their signature
if "height" in sig.parameters and "width" in sig.parameters:
return await self.hass.async_add_executor_job(
partial(self.camera_image, width=width, height=height)
)
_LOGGER.warning(
"The camera entity %s does not support requesting width and height, please open an issue with the integration author",
self.entity_id,
)
return await self.hass.async_add_executor_job(self.camera_image)
async def handle_async_still_stream(
@ -529,14 +596,19 @@ class CameraImageView(CameraView):
async def handle(self, request: web.Request, camera: Camera) -> web.Response:
"""Serve camera image."""
with suppress(asyncio.CancelledError, asyncio.TimeoutError):
async with async_timeout.timeout(CAMERA_IMAGE_TIMEOUT):
image = await camera.async_camera_image()
if image:
return web.Response(body=image, content_type=camera.content_type)
raise web.HTTPInternalServerError()
width = request.query.get("width")
height = request.query.get("height")
try:
image = await _async_get_image(
camera,
CAMERA_IMAGE_TIMEOUT,
int(width) if width else None,
int(height) if height else None,
)
except (HomeAssistantError, ValueError) as ex:
raise web.HTTPInternalServerError() from ex
else:
return web.Response(body=image.content, content_type=image.content_type)
class CameraMjpegStream(CameraView):

30
homeassistant/components/homekit/img_util.py → homeassistant/components/camera/img_util.py
View file

@ -1,19 +1,32 @@
"""Image processing for HomeKit component."""
"""Image processing for cameras."""
import logging
from typing import TYPE_CHECKING, cast
SUPPORTED_SCALING_FACTORS = [(7, 8), (3, 4), (5, 8), (1, 2), (3, 8), (1, 4), (1, 8)]
_LOGGER = logging.getLogger(__name__)
JPEG_QUALITY = 75
def scale_jpeg_camera_image(cam_image, width, height):
if TYPE_CHECKING:
from turbojpeg import TurboJPEG
from . import Image
def scale_jpeg_camera_image(cam_image: "Image", width: int, height: int) -> bytes:
"""Scale a camera image as close as possible to one of the supported scaling factors."""
turbo_jpeg = TurboJPEGSingleton.instance()
if not turbo_jpeg:
return cam_image.content
(current_width, current_height, _, _) = turbo_jpeg.decode_header(cam_image.content)
try:
(current_width, current_height, _, _) = turbo_jpeg.decode_header(
cam_image.content
)
except OSError:
return cam_image.content
if current_width <= width or current_height <= height:
return cam_image.content
@ -26,10 +39,13 @@ def scale_jpeg_camera_image(cam_image, width, height):
scaling_factor = supported_sf
break
return turbo_jpeg.scale_with_quality(
return cast(
bytes,
turbo_jpeg.scale_with_quality(
cam_image.content,
scaling_factor=scaling_factor,
quality=75,
quality=JPEG_QUALITY,
),
)
@ -45,13 +61,13 @@ class TurboJPEGSingleton:
__instance = None
@staticmethod
def instance():
def instance() -> "TurboJPEG":
"""Singleton for TurboJPEG."""
if TurboJPEGSingleton.__instance is None:
TurboJPEGSingleton()
return TurboJPEGSingleton.__instance
def __init__(self):
def __init__(self) -> None:
"""Try to create TurboJPEG only once."""
# pylint: disable=unused-private-member
# https://github.com/PyCQA/pylint/issues/4681

1
homeassistant/components/camera/manifest.json
View file

@ -3,6 +3,7 @@
"name": "Camera",
"documentation": "https://www.home-assistant.io/integrations/camera",
"dependencies": ["http"],
"requirements": ["PyTurboJPEG==1.5.0"],
"after_dependencies": ["media_player"],
"codeowners": [],
"quality_scale": "internal"

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

@ -123,7 +123,9 @@ class CanaryCamera(CoordinatorEntity, Camera):
"""Return the camera motion detection status."""
return not self.location.is_recording
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> 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(

6
homeassistant/components/demo/camera.py
View file

@ -1,4 +1,6 @@
"""Demo camera platform that has a fake camera."""
from __future__ import annotations
from pathlib import Path
from homeassistant.components.camera import SUPPORT_ON_OFF, Camera
@ -25,7 +27,9 @@ class DemoCamera(Camera):
self.is_streaming = True
self._images_index = 0
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes:
"""Return a faked still image response."""
self._images_index = (self._images_index + 1) % 4
image_path = Path(__file__).parent / f"demo_{self._images_index}.jpg"

6
homeassistant/components/doorbird/camera.py
View file

@ -1,4 +1,6 @@
"""Support for viewing the camera feed from a DoorBird video doorbell."""
from __future__ import annotations
import asyncio
import datetime
import logging
@ -112,7 +114,9 @@ class DoorBirdCamera(DoorBirdEntity, Camera):
"""Get the name of the camera."""
return self._name
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Pull a still image from the camera."""
now = dt_util.utcnow()

6
homeassistant/components/environment_canada/camera.py
View file

@ -1,4 +1,6 @@
"""Support for the Environment Canada radar imagery."""
from __future__ import annotations
import datetime
from env_canada import ECRadar
@ -68,7 +70,9 @@ class ECCamera(Camera):
self.image = None
self.timestamp = None
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
self.update()
return self.image

4
homeassistant/components/esphome/camera.py
View file

@ -50,7 +50,9 @@ class EsphomeCamera(Camera, EsphomeEntity[CameraInfo, CameraState]):
async with self._image_cond:
self._image_cond.notify_all()
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return single camera image bytes."""
if not self.available:
return None

4
homeassistant/components/ezviz/camera.py
View file

@ -325,7 +325,9 @@ class EzvizCamera(CoordinatorEntity, Camera):
"""Return the name of this camera."""
return self._serial
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a frame from the camera stream."""
ffmpeg = ImageFrame(self._ffmpeg.binary)

6
homeassistant/components/familyhub/camera.py
View file

@ -1,4 +1,6 @@
"""Family Hub camera for Samsung Refrigerators."""
from __future__ import annotations
from pyfamilyhublocal import FamilyHubCam
import voluptuous as vol
@ -38,7 +40,9 @@ class FamilyHubCamera(Camera):
self._name = name
self.family_hub_cam = family_hub_cam
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response."""
return await self.family_hub_cam.async_get_cam_image()

5
homeassistant/components/ffmpeg/camera.py
View file

@ -1,4 +1,5 @@
"""Support for Cameras with FFmpeg as decoder."""
from __future__ import annotations
from haffmpeg.camera import CameraMjpeg
from haffmpeg.tools import IMAGE_JPEG
@ -49,7 +50,9 @@ class FFmpegCamera(Camera):
"""Return the stream source."""
return self._input.split(" ")[-1]
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
return await async_get_image(
self.hass,

6
homeassistant/components/foscam/camera.py
View file

@ -1,4 +1,6 @@
"""This component provides basic support for Foscam IP cameras."""
from __future__ import annotations
import asyncio
from libpyfoscam import FoscamCamera
@ -172,7 +174,9 @@ class HassFoscamCamera(Camera):
"""Return the entity unique ID."""
return self._unique_id
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
# Send the request to snap a picture and return raw jpg data
# Handle exception if host is not reachable or url failed

10
homeassistant/components/generic/camera.py
View file

@ -1,4 +1,6 @@
"""Support for IP Cameras."""
from __future__ import annotations
import asyncio
import logging
@ -118,13 +120,17 @@ class GenericCamera(Camera):
"""Return the interval between frames of the mjpeg stream."""
return self._frame_interval
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
return asyncio.run_coroutine_threadsafe(
self.async_camera_image(), self.hass.loop
).result()
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
try:
url = self._still_image_url.async_render(parse_result=False)

3
homeassistant/components/homekit/manifest.json
View file

@ -6,8 +6,7 @@
"HAP-python==4.0.0",
"fnvhash==0.1.0",
"PyQRCode==1.2.1",
"base36==0.1.1",
"PyTurboJPEG==1.5.0"
"base36==0.1.1"
],
"dependencies": ["http", "camera", "ffmpeg", "network"],
"after_dependencies": ["zeroconf"],

10
homeassistant/components/homekit/type_cameras.py
View file

@ -55,7 +55,6 @@ from .const import (
SERV_SPEAKER,
SERV_STATELESS_PROGRAMMABLE_SWITCH,
)
from .img_util import scale_jpeg_camera_image
from .util import pid_is_alive
_LOGGER = logging.getLogger(__name__)
@ -467,8 +466,9 @@ class Camera(HomeAccessory, PyhapCamera):
async def async_get_snapshot(self, image_size):
"""Return a jpeg of a snapshot from the camera."""
return scale_jpeg_camera_image(
await self.hass.components.camera.async_get_image(self.entity_id),
image_size["image-width"],
image_size["image-height"],
image = await self.hass.components.camera.async_get_image(
self.entity_id,
width=image_size["image-width"],
height=image_size["image-height"],
)
return image.content

10
homeassistant/components/homekit_controller/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Homekit cameras."""
from __future__ import annotations
from aiohomekit.model.services import ServicesTypes
from homeassistant.components.camera import Camera
@ -21,12 +23,14 @@ class HomeKitCamera(AccessoryEntity, Camera):
"""Return the current state of the camera."""
return "idle"
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a jpeg with the current camera snapshot."""
return await self._accessory.pairing.image(
self._aid,
640,
480,
width or 640,
height or 480,
)

4
homeassistant/components/hyperion/camera.py
View file

@ -210,7 +210,9 @@ class HyperionCamera(Camera):
finally:
await self._stop_image_streaming_for_client()
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return single camera image bytes."""
async with self._image_streaming() as is_streaming:
if is_streaming:

7
homeassistant/components/local_file/camera.py
View file

@ -1,4 +1,6 @@
"""Camera that loads a picture from a local file."""
from __future__ import annotations
import logging
import mimetypes
import os
@ -73,7 +75,9 @@ class LocalFile(Camera):
if content is not None:
self.content_type = content
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
try:
with open(self._file_path, "rb") as file:
@ -84,6 +88,7 @@ class LocalFile(Camera):
self._name,
self._file_path,
)
return None
def check_file_path_access(self, file_path):
"""Check that filepath given is readable."""

6
homeassistant/components/logi_circle/camera.py
View file

@ -1,4 +1,6 @@
"""Support to the Logi Circle cameras."""
from __future__ import annotations
from datetime import timedelta
import logging
@ -142,7 +144,9 @@ class LogiCam(Camera):
return state
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image from the camera."""
return await self._camera.live_stream.download_jpeg()

16
homeassistant/components/mjpeg/camera.py
View file

@ -1,4 +1,6 @@
"""Support for IP Cameras."""
from __future__ import annotations
import asyncio
from contextlib import closing
import logging
@ -106,7 +108,9 @@ class MjpegCamera(Camera):
self._auth = aiohttp.BasicAuth(self._username, password=self._password)
self._verify_ssl = device_info.get(CONF_VERIFY_SSL)
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
# DigestAuth is not supported
if (
@ -130,11 +134,17 @@ class MjpegCamera(Camera):
except aiohttp.ClientError as err:
_LOGGER.error("Error getting new camera image from %s: %s", self._name, err)
def camera_image(self):
return None
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
if self._username and self._password:
if self._authentication == HTTP_DIGEST_AUTHENTICATION:
auth = HTTPDigestAuth(self._username, self._password)
auth: HTTPDigestAuth | HTTPBasicAuth = HTTPDigestAuth(
self._username, self._password
)
else:
auth = HTTPBasicAuth(self._username, self._password)
req = requests.get(

6
homeassistant/components/mqtt/camera.py
View file

@ -1,4 +1,6 @@
"""Camera that loads a picture from an MQTT topic."""
from __future__ import annotations
import functools
import voluptuous as vol
@ -98,6 +100,8 @@ class MqttCamera(MqttEntity, Camera):
},
)
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
return self._last_image

4
homeassistant/components/neato/camera.py
View file

@ -66,7 +66,9 @@ class NeatoCleaningMap(Camera):
self._image_url: str | None = None
self._image: bytes | None = None
def camera_image(self) -> bytes | None:
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
self.update()
return self._image

4
homeassistant/components/nest/camera_sdm.py
View file

@ -180,7 +180,9 @@ class NestCamera(Camera):
self._device.add_update_listener(self.async_write_ha_state)
)
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
# Returns the snapshot of the last event for ~30 seconds after the event
active_event_image = await self._async_active_event_image()

6
homeassistant/components/nest/legacy/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Nest Cameras."""
from __future__ import annotations
from datetime import timedelta
import logging
@ -131,7 +133,9 @@ class NestCamera(Camera):
def _ready_for_snapshot(self, now):
return self._next_snapshot_at is None or now > self._next_snapshot_at
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
now = utcnow()
if self._ready_for_snapshot(now):

8
homeassistant/components/netatmo/camera.py
View file

@ -194,10 +194,14 @@ class NetatmoCamera(NetatmoBase, Camera):
self.data_handler.data[self._data_classes[0]["name"]],
)
async def async_camera_image(self) -> bytes | None:
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
try:
return await self._data.async_get_live_snapshot(camera_id=self._id)
return cast(
bytes, await self._data.async_get_live_snapshot(camera_id=self._id)
)
except (
aiohttp.ClientPayloadError,
aiohttp.ContentTypeError,

6
homeassistant/components/onvif/camera.py
View file

@ -1,4 +1,6 @@
"""Support for ONVIF Cameras with FFmpeg as decoder."""
from __future__ import annotations
import asyncio
from haffmpeg.camera import CameraMjpeg
@ -120,7 +122,9 @@ class ONVIFCameraEntity(ONVIFBaseEntity, Camera):
"""Return the stream source."""
return self._stream_uri
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
image = None

16
homeassistant/components/proxy/camera.py
View file

@ -1,4 +1,6 @@
"""Proxy camera platform that enables image processing of camera data."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import io
@ -219,13 +221,17 @@ class ProxyCamera(Camera):
self._last_image = None
self._mode = config.get(CONF_MODE)
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return camera image."""
return asyncio.run_coroutine_threadsafe(
self.async_camera_image(), self.hass.loop
).result()
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
now = dt_util.utcnow()
@ -244,13 +250,13 @@ class ProxyCamera(Camera):
job = _resize_image
else:
job = _crop_image
image = await self.hass.async_add_executor_job(
image_bytes: bytes = await self.hass.async_add_executor_job(
job, image.content, self._image_opts
)
if self._cache_images:
self._last_image = image
return image
self._last_image = image_bytes
return image_bytes
async def handle_async_mjpeg_stream(self, request):
"""Generate an HTTP MJPEG stream from camera images."""

6
homeassistant/components/push/camera.py
View file

@ -1,4 +1,6 @@
"""Camera platform that receives images through HTTP POST."""
from __future__ import annotations
import asyncio
from collections import deque
from datetime import timedelta
@ -155,7 +157,9 @@ class PushCamera(Camera):
self.async_write_ha_state()
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response."""
if self.queue:
if self._state == STATE_IDLE:

5
homeassistant/components/qvr_pro/camera.py
View file

@ -1,4 +1,5 @@
"""Support for QVR Pro streams."""
from __future__ import annotations
import logging
@ -88,7 +89,9 @@ class QVRProCamera(Camera):
return attrs
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Get image bytes from camera."""
try:
return self._client.get_snapshot(self.guid)

6
homeassistant/components/ring/camera.py
View file

@ -1,4 +1,6 @@
"""This component provides support to the Ring Door Bell camera."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from itertools import chain
@ -101,7 +103,9 @@ class RingCam(RingEntityMixin, Camera):
"last_video_id": self._last_video_id,
}
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
ffmpeg = ImageFrame(self._ffmpeg.binary)

6
homeassistant/components/rpi_camera/camera.py
View file

@ -1,4 +1,6 @@
"""Camera platform that has a Raspberry Pi camera."""
from __future__ import annotations
import logging
import os
import shutil
@ -122,7 +124,9 @@ class RaspberryCamera(Camera):
):
pass
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return raspistill image response."""
with open(self._config[CONF_FILE_PATH], "rb") as file:
return file.read()

6
homeassistant/components/skybell/camera.py
View file

@ -1,4 +1,6 @@
"""Camera support for the Skybell HD Doorbell."""
from __future__ import annotations
from datetime import timedelta
import logging
@ -75,7 +77,9 @@ class SkybellCamera(SkybellDevice, Camera):
return self._device.activity_image
return self._device.image
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Get the latest camera image."""
super().update()

4
homeassistant/components/synology_dsm/camera.py
View file

@ -123,7 +123,9 @@ class SynoDSMCamera(SynologyDSMBaseEntity, Camera):
"""Return the camera motion detection status."""
return self.camera_data.is_motion_detection_enabled # type: ignore[no-any-return]
def camera_image(self) -> bytes | None:
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
_LOGGER.debug(
"SynoDSMCamera.camera_image(%s)",

6
homeassistant/components/uvc/camera.py
View file

@ -194,10 +194,12 @@ class UnifiVideoCamera(Camera):
self._caminfo = caminfo
return True
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return the image of this camera."""
if not self._camera and not self._login():
return
return None
def _get_image(retry=True):
try:

4
homeassistant/components/verisure/camera.py
View file

@ -79,7 +79,9 @@ class VerisureSmartcam(CoordinatorEntity, Camera):
"via_device": (DOMAIN, self.coordinator.entry.data[CONF_GIID]),
}
def camera_image(self) -> bytes | None:
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
self.check_imagelist()
if not self._image:

6
homeassistant/components/vivotek/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Vivotek IP Cameras."""
from __future__ import annotations
from libpyvivotek import VivotekCamera
import voluptuous as vol
@ -87,7 +89,9 @@ class VivotekCam(Camera):
"""Return the interval between frames of the mjpeg stream."""
return self._frame_interval
def camera_image(self):
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image."""
return self._cam.snapshot()

6
homeassistant/components/xeoma/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Xeoma Cameras."""
from __future__ import annotations
import logging
from pyxeoma.xeoma import Xeoma, XeomaError
@ -109,7 +111,9 @@ class XeomaCamera(Camera):
self._password = password
self._last_image = None
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
try:

6
homeassistant/components/xiaomi/camera.py
View file

@ -1,4 +1,6 @@
"""This component provides support for Xiaomi Cameras."""
from __future__ import annotations
import asyncio
from ftplib import FTP, error_perm
import logging
@ -138,7 +140,9 @@ class XiaomiCamera(Camera):
return f"ftp://{self.user}:{self.passwd}@{host}:{self.port}{ftp.pwd()}/{video}"
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
try:

6
homeassistant/components/yi/camera.py
View file

@ -1,4 +1,6 @@
"""Support for Xiaomi Cameras (HiSilicon Hi3518e V200)."""
from __future__ import annotations
import asyncio
import logging
@ -119,7 +121,9 @@ class YiCamera(Camera):
self._is_on = False
return None
async def async_camera_image(self):
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return a still image response from the camera."""
url = await self._get_latest_video_url()
if url and url != self._last_url:

2
requirements_all.txt
View file

@ -57,7 +57,7 @@ PySocks==1.7.1
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1
# homeassistant.components.homekit
# homeassistant.components.camera
PyTurboJPEG==1.5.0
# homeassistant.components.vicare

2
requirements_test_all.txt
View file

@ -26,7 +26,7 @@ PyRMVtransport==0.3.2
# homeassistant.components.transport_nsw
PyTransportNSW==0.1.1
# homeassistant.components.homekit
# homeassistant.components.camera
PyTurboJPEG==1.5.0
# homeassistant.components.xiaomi_aqara

17
tests/components/camera/common.py
View file

@ -3,8 +3,12 @@
All containing methods are legacy helpers that should not be used by new
components. Instead call the service directly.
"""
from unittest.mock import Mock
from homeassistant.components.camera.const import DATA_CAMERA_PREFS, PREF_PRELOAD_STREAM
EMPTY_8_6_JPEG = b"empty_8_6"
def mock_camera_prefs(hass, entity_id, prefs=None):
"""Fixture for cloud component."""
@ -13,3 +17,16 @@ def mock_camera_prefs(hass, entity_id, prefs=None):
prefs_to_set.update(prefs)
hass.data[DATA_CAMERA_PREFS]._prefs[entity_id] = prefs_to_set
return prefs_to_set
def mock_turbo_jpeg(
first_width=None, second_width=None, first_height=None, second_height=None
):
"""Mock a TurboJPEG instance."""
mocked_turbo_jpeg = Mock()
mocked_turbo_jpeg.decode_header.side_effect = [
(first_width, first_height, 0, 0),
(second_width, second_height, 0, 0),
]
mocked_turbo_jpeg.scale_with_quality.return_value = EMPTY_8_6_JPEG
return mocked_turbo_jpeg

28
tests/components/homekit/test_img_util.py → tests/components/camera/test_img_util.py
View file

@ -1,8 +1,10 @@
"""Test HomeKit img_util module."""
"""Test img_util module."""
from unittest.mock import patch
from turbojpeg import TurboJPEG
from homeassistant.components.camera import Image
from homeassistant.components.homekit.img_util import (
from homeassistant.components.camera.img_util import (
TurboJPEGSingleton,
scale_jpeg_camera_image,
)
@ -12,13 +14,23 @@ from .common import EMPTY_8_6_JPEG, mock_turbo_jpeg
EMPTY_16_12_JPEG = b"empty_16_12"
def _clear_turbojpeg_singleton():
TurboJPEGSingleton.__instance = None
def _reset_turbojpeg_singleton():
TurboJPEGSingleton.__instance = TurboJPEG()
def test_turbojpeg_singleton():
"""Verify the instance always gives back the same."""
_clear_turbojpeg_singleton()
assert TurboJPEGSingleton.instance() == TurboJPEGSingleton.instance()
def test_scale_jpeg_camera_image():
"""Test we can scale a jpeg image."""
_clear_turbojpeg_singleton()
camera_image = Image("image/jpeg", EMPTY_16_12_JPEG)
@ -27,6 +39,12 @@ def test_scale_jpeg_camera_image():
TurboJPEGSingleton()
assert scale_jpeg_camera_image(camera_image, 16, 12) == camera_image.content
turbo_jpeg = mock_turbo_jpeg(first_width=16, first_height=12)
turbo_jpeg.decode_header.side_effect = OSError
with patch("turbojpeg.TurboJPEG", return_value=turbo_jpeg):
TurboJPEGSingleton()
assert scale_jpeg_camera_image(camera_image, 16, 12) == camera_image.content
turbo_jpeg = mock_turbo_jpeg(first_width=16, first_height=12)
with patch("turbojpeg.TurboJPEG", return_value=turbo_jpeg):
TurboJPEGSingleton()
@ -44,11 +62,11 @@ def test_scale_jpeg_camera_image():
def test_turbojpeg_load_failure():
"""Handle libjpegturbo not being installed."""
_clear_turbojpeg_singleton()
with patch("turbojpeg.TurboJPEG", side_effect=Exception):
TurboJPEGSingleton()
assert TurboJPEGSingleton.instance() is False
with patch("turbojpeg.TurboJPEG"):
_clear_turbojpeg_singleton()
TurboJPEGSingleton()
assert TurboJPEGSingleton.instance()
assert TurboJPEGSingleton.instance() is not None

47
tests/components/camera/test_init.py
View file

@ -20,6 +20,8 @@ from homeassistant.const import (
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from .common import EMPTY_8_6_JPEG, mock_turbo_jpeg
from tests.components.camera import common
@ -75,6 +77,51 @@ async def test_get_image_from_camera(hass, image_mock_url):
assert image.content == b"Test"
async def test_get_image_from_camera_with_width_height(hass, image_mock_url):
"""Grab an image from camera entity with width and height."""
turbo_jpeg = mock_turbo_jpeg(
first_width=16, first_height=12, second_width=300, second_height=200
)
with patch(
"homeassistant.components.camera.img_util.TurboJPEGSingleton.instance",
return_value=turbo_jpeg,
), patch(
"homeassistant.components.demo.camera.Path.read_bytes",
autospec=True,
return_value=b"Test",
) as mock_camera:
image = await camera.async_get_image(
hass, "camera.demo_camera", width=640, height=480
)
assert mock_camera.called
assert image.content == b"Test"
async def test_get_image_from_camera_with_width_height_scaled(hass, image_mock_url):
"""Grab an image from camera entity with width and height and scale it."""
turbo_jpeg = mock_turbo_jpeg(
first_width=16, first_height=12, second_width=300, second_height=200
)
with patch(
"homeassistant.components.camera.img_util.TurboJPEGSingleton.instance",
return_value=turbo_jpeg,
), patch(
"homeassistant.components.demo.camera.Path.read_bytes",
autospec=True,
return_value=b"Valid jpeg",
) as mock_camera:
image = await camera.async_get_image(
hass, "camera.demo_camera", width=4, height=3
)
assert mock_camera.called
assert image.content_type == "image/jpeg"
assert image.content == EMPTY_8_6_JPEG
async def test_get_stream_source_from_camera(hass, mock_camera):
"""Fetch stream source from camera entity."""

17
tests/components/homekit/common.py
View file

@ -1,17 +0,0 @@
"""Collection of fixtures and functions for the HomeKit tests."""
from unittest.mock import Mock
EMPTY_8_6_JPEG = b"empty_8_6"
def mock_turbo_jpeg(
first_width=None, second_width=None, first_height=None, second_height=None
):
"""Mock a TurboJPEG instance."""
mocked_turbo_jpeg = Mock()
mocked_turbo_jpeg.decode_header.side_effect = [
(first_width, first_height, 0, 0),
(second_width, second_height, 0, 0),
]
mocked_turbo_jpeg.scale_with_quality.return_value = EMPTY_8_6_JPEG
return mocked_turbo_jpeg

4
tests/components/homekit/test_type_cameras.py
View file

@ -7,6 +7,7 @@ from pyhap.accessory_driver import AccessoryDriver
import pytest
from homeassistant.components import camera, ffmpeg
from homeassistant.components.camera.img_util import TurboJPEGSingleton
from homeassistant.components.homekit.accessories import HomeBridge
from homeassistant.components.homekit.const import (
AUDIO_CODEC_COPY,
@ -26,14 +27,13 @@ from homeassistant.components.homekit.const import (
VIDEO_CODEC_COPY,
VIDEO_CODEC_H264_OMX,
)
from homeassistant.components.homekit.img_util import TurboJPEGSingleton
from homeassistant.components.homekit.type_cameras import Camera
from homeassistant.components.homekit.type_switches import Switch
from homeassistant.const import ATTR_DEVICE_CLASS, STATE_OFF, STATE_ON
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from .common import mock_turbo_jpeg
from tests.components.camera.common import mock_turbo_jpeg
MOCK_START_STREAM_TLV = "ARUCAQEBEDMD1QMXzEaatnKSQ2pxovYCNAEBAAIJAQECAgECAwEAAwsBAgAFAgLQAgMBHgQXAQFjAgQ768/RAwIrAQQEAAAAPwUCYgUDLAEBAwIMAQEBAgEAAwECBAEUAxYBAW4CBCzq28sDAhgABAQAAKBABgENBAEA"
MOCK_END_POINTS_TLV = "ARAzA9UDF8xGmrZykkNqcaL2AgEAAxoBAQACDTE5Mi4xNjguMjA4LjUDAi7IBAKkxwQlAQEAAhDN0+Y0tZ4jzoO0ske9UsjpAw6D76oVXnoi7DbawIG4CwUlAQEAAhCyGcROB8P7vFRDzNF2xrK1Aw6NdcLugju9yCfkWVSaVAYEDoAsAAcEpxV8AA=="