jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-19 14:57:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Replace Camera STREAM_ constants with StreamType enum (#69871)

Browse Source
This commit is contained in:
Franck Nijhof 2022-04-12 01:27:27 +02:00 committed by GitHub
parent 75fce1f036
commit c93c7e8eff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 56 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
homeassistant/components/camera/__init__.py
View File

@ -54,7 +54,7 @@ from homeassistant.helpers.network import get_url
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import bind_hass
from .const import (
from .const import ( # noqa: F401
CAMERA_IMAGE_TIMEOUT,
CAMERA_STREAM_SOURCE_TIMEOUT,
CONF_DURATION,
@ -65,6 +65,7 @@ from .const import (
SERVICE_RECORD,
STREAM_TYPE_HLS,
STREAM_TYPE_WEB_RTC,
StreamType,
)
from .img_util import scale_jpeg_camera_image
from .prefs import CameraPreferences
@ -436,7 +437,7 @@ class Camera(Entity):
# Entity Properties
_attr_brand: str | None = None
_attr_frame_interval: float = MIN_STREAM_INTERVAL
_attr_frontend_stream_type: str | None
_attr_frontend_stream_type: StreamType | None
_attr_is_on: bool = True
_attr_is_recording: bool = False
_attr_is_streaming: bool = False
@ -500,7 +501,7 @@ class Camera(Entity):
return self._attr_frame_interval
@property
def frontend_stream_type(self) -> str | None:
def frontend_stream_type(self) -> StreamType | None:
"""Return the type of stream supported by this camera.
A camera may have a single stream type which is used to inform the
@ -512,8 +513,8 @@ class Camera(Entity):
if not self.supported_features & CameraEntityFeature.STREAM:
return None
if self._rtsp_to_webrtc:
return STREAM_TYPE_WEB_RTC
return STREAM_TYPE_HLS
return StreamType.WEB_RTC
return StreamType.HLS
@property
def available(self) -> bool:
@ -546,7 +547,7 @@ class Camera(Entity):
"""Return the source of the stream.
This is used by cameras with CameraEntityFeature.STREAM
and STREAM_TYPE_HLS.
and StreamType.HLS.
"""
# pylint: disable=no-self-use
return None
@ -555,7 +556,7 @@ class Camera(Entity):
"""Handle the WebRTC offer and return an answer.
This is used by cameras with CameraEntityFeature.STREAM
and STREAM_TYPE_WEB_RTC.
and StreamType.WEB_RTC.
Integrations can override with a native WebRTC implementation.
"""
@ -870,7 +871,7 @@ async def ws_camera_web_rtc_offer(
entity_id = msg["entity_id"]
offer = msg["offer"]
camera = _get_camera_from_entity_id(hass, entity_id)
if camera.frontend_stream_type != STREAM_TYPE_WEB_RTC:
if camera.frontend_stream_type != StreamType.WEB_RTC:
connection.send_error(
msg["id"],
"web_rtc_offer_failed",

26
homeassistant/components/camera/const.py
View File

@ -1,6 +1,8 @@
"""Constants for Camera component."""
from typing import Final
from homeassistant.backports.enum import StrEnum
DOMAIN: Final = "camera"
DATA_CAMERA_PREFS: Final = "camera_prefs"
@ -16,11 +18,23 @@ CONF_DURATION: Final = "duration"
CAMERA_STREAM_SOURCE_TIMEOUT: Final = 10
CAMERA_IMAGE_TIMEOUT: Final = 10
# A camera that supports CAMERA_SUPPORT_STREAM may have a single stream
# type which is used to inform the frontend which player to use.
# Streams with RTSP sources typically use the stream component which uses
# HLS for display. WebRTC streams use the home assistant core for a signal
# path to initiate a stream, but the stream itself is between the client and
# device.
class StreamType(StrEnum):
"""Camera stream type.
A camera that supports CAMERA_SUPPORT_STREAM may have a single stream
type which is used to inform the frontend which player to use.
Streams with RTSP sources typically use the stream component which uses
HLS for display. WebRTC streams use the home assistant core for a signal
path to initiate a stream, but the stream itself is between the client and
device.
"""
HLS = "hls"
WEB_RTC = "web_rtc"
# These constants are deprecated as of Home Assistant 2022.5
# Please use the StreamType enum instead.
STREAM_TYPE_HLS = "hls"
STREAM_TYPE_WEB_RTC = "web_rtc"

6
homeassistant/components/camera/media_source.py
View File

@ -21,7 +21,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_component import EntityComponent
from . import Camera, _async_stream_endpoint_url
from .const import DOMAIN, STREAM_TYPE_HLS
from .const import DOMAIN, StreamType
async def async_get_media_source(hass: HomeAssistant) -> CameraMediaSource:
@ -52,7 +52,7 @@ class CameraMediaSource(MediaSource):
f"/api/camera_proxy_stream/{camera.entity_id}", camera.content_type
)
if stream_type != STREAM_TYPE_HLS:
if stream_type != StreamType.HLS:
raise Unresolvable("Camera does not support MJPEG or HLS streaming.")
if "stream" not in self.hass.config.components:
@ -86,7 +86,7 @@ class CameraMediaSource(MediaSource):
if stream_type is None:
content_type = camera.content_type
elif can_stream_hls and stream_type == STREAM_TYPE_HLS:
elif can_stream_hls and stream_type == StreamType.HLS:
content_type = FORMAT_CONTENT_TYPE[HLS_PROVIDER]
else:

6
homeassistant/components/nest/camera_sdm.py
View File

@ -18,7 +18,7 @@ from google_nest_sdm.device import Device
from google_nest_sdm.exceptions import ApiException
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.components.camera.const import STREAM_TYPE_WEB_RTC
from homeassistant.components.camera.const import StreamType
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, PlatformNotReady
@ -114,13 +114,13 @@ class NestCamera(Camera):
return supported_features
@property
def frontend_stream_type(self) -> str | None:
def frontend_stream_type(self) -> StreamType | None:
"""Return the type of stream supported by this camera."""
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
if StreamingProtocol.WEB_RTC in trait.supported_protocols:
return STREAM_TYPE_WEB_RTC
return StreamType.WEB_RTC
return super().frontend_stream_type
@property

10
pylint/plugins/hass_imports.py
View File

@ -42,6 +42,16 @@ _OBSOLETE_IMPORT: dict[str, list[ObsoleteImportMatch]] = {
reason="replaced by CameraEntityFeature enum",
constant=re.compile(r"^SUPPORT_(\w*)$"),
),
ObsoleteImportMatch(
reason="replaced by StreamType enum",
constant=re.compile(r"^STREAM_TYPE_(\w*)$"),
),
],
"homeassistant.components.camera.const": [
ObsoleteImportMatch(
reason="replaced by StreamType enum",
constant=re.compile(r"^STREAM_TYPE_(\w*)$"),
),
],
"homeassistant.components.climate": [
ObsoleteImportMatch(

6
tests/components/camera/conftest.py
View File

@ -4,7 +4,7 @@ from unittest.mock import PropertyMock, patch
import pytest
from homeassistant.components import camera
from homeassistant.components.camera.const import STREAM_TYPE_HLS, STREAM_TYPE_WEB_RTC
from homeassistant.components.camera.const import StreamType
from homeassistant.setup import async_setup_component
from .common import WEBRTC_ANSWER
@ -30,7 +30,7 @@ async def mock_camera_hls_fixture(mock_camera):
"""Initialize a demo camera platform with HLS."""
with patch(
"homeassistant.components.camera.Camera.frontend_stream_type",
new_callable=PropertyMock(return_value=STREAM_TYPE_HLS),
new_callable=PropertyMock(return_value=StreamType.HLS),
):
yield
@ -45,7 +45,7 @@ async def mock_camera_web_rtc_fixture(hass):
with patch(
"homeassistant.components.camera.Camera.frontend_stream_type",
new_callable=PropertyMock(return_value=STREAM_TYPE_WEB_RTC),
new_callable=PropertyMock(return_value=StreamType.WEB_RTC),
), patch(
"homeassistant.components.camera.Camera.async_handle_web_rtc_offer",
return_value=WEBRTC_ANSWER,

4
tests/components/camera/test_media_source.py
View File

@ -4,7 +4,7 @@ from unittest.mock import PropertyMock, patch
import pytest
from homeassistant.components import media_source
from homeassistant.components.camera.const import STREAM_TYPE_WEB_RTC
from homeassistant.components.camera.const import StreamType
from homeassistant.components.stream.const import FORMAT_CONTENT_TYPE
from homeassistant.setup import async_setup_component
@ -88,7 +88,7 @@ async def test_resolving_errors(hass, mock_camera_hls):
with pytest.raises(media_source.Unresolvable) as exc_info, patch(
"homeassistant.components.camera.Camera.frontend_stream_type",
new_callable=PropertyMock(return_value=STREAM_TYPE_WEB_RTC),
new_callable=PropertyMock(return_value=StreamType.WEB_RTC),
):
await media_source.async_resolve_media(
hass, "media-source://camera/camera.demo_camera"

17
tests/components/nest/test_camera_sdm.py
View File

@ -14,12 +14,7 @@ from google_nest_sdm.event import EventMessage
import pytest
from homeassistant.components import camera
from homeassistant.components.camera import (
STATE_IDLE,
STATE_STREAMING,
STREAM_TYPE_HLS,
STREAM_TYPE_WEB_RTC,
)
from homeassistant.components.camera import STATE_IDLE, STATE_STREAMING, StreamType
from homeassistant.components.nest.const import DOMAIN
from homeassistant.components.websocket_api.const import TYPE_RESULT
from homeassistant.core import HomeAssistant
@ -243,7 +238,7 @@ async def test_camera_stream(
cam = hass.states.get("camera.my_camera")
assert cam is not None
assert cam.state == STATE_STREAMING
assert cam.attributes["frontend_stream_type"] == STREAM_TYPE_HLS
assert cam.attributes["frontend_stream_type"] == StreamType.HLS
stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
assert stream_source == "rtsp://some/url?auth=g.0.streamingToken"
@ -267,7 +262,7 @@ async def test_camera_ws_stream(
cam = hass.states.get("camera.my_camera")
assert cam is not None
assert cam.state == STATE_STREAMING
assert cam.attributes["frontend_stream_type"] == STREAM_TYPE_HLS
assert cam.attributes["frontend_stream_type"] == StreamType.HLS
client = await hass_ws_client(hass)
await client.send_json(
@ -591,7 +586,7 @@ async def test_camera_web_rtc(
cam = hass.states.get("camera.my_camera")
assert cam is not None
assert cam.state == STATE_STREAMING
assert cam.attributes["frontend_stream_type"] == STREAM_TYPE_WEB_RTC
assert cam.attributes["frontend_stream_type"] == StreamType.WEB_RTC
client = await hass_ws_client(hass)
await client.send_json(
@ -624,7 +619,7 @@ async def test_camera_web_rtc_unsupported(
cam = hass.states.get("camera.my_camera")
assert cam is not None
assert cam.state == STATE_STREAMING
assert cam.attributes["frontend_stream_type"] == STREAM_TYPE_HLS
assert cam.attributes["frontend_stream_type"] == StreamType.HLS
client = await hass_ws_client(hass)
await client.send_json(
@ -718,7 +713,7 @@ async def test_camera_multiple_streams(
assert cam is not None
assert cam.state == STATE_STREAMING
# Prefer WebRTC over RTSP/HLS
assert cam.attributes["frontend_stream_type"] == STREAM_TYPE_WEB_RTC
assert cam.attributes["frontend_stream_type"] == StreamType.WEB_RTC
# RTSP stream
stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")