base camera func

.github/workflows/validate.yml

@@ -33,5 +33,3 @@ jobs:
           uses: "hacs/action@main"
           with:
             category: "integration"
-            # Remove this 'ignore' key when you have added brand images for your integration to https://github.com/home-assistant/brands
-            ignore: "brands"

custom_components/starling_home_hub/api.py

@@ -1,13 +1,12 @@
 """Starling Home Hub Developer Connect API Client."""
 from __future__ import annotations
-from .models import Device, Devices, Status, SpecificDevice
+from .models import Device, Devices, Status, SpecificDevice, StartStream, StreamStatus
 
 import asyncio
 import socket
-
 import aiohttp
 import async_timeout
-
+import base64
 
 class StarlingHomeHubApiClientError(Exception):
     """Exception to indicate a general API error."""
@@ -67,14 +66,38 @@ async def async_get_devices(self) -> list[Device]:
 
         return Devices(**devices_response).devices
 
-    # async def async_set_title(self, value: str) -> any:
-    #     """Get data from the API."""
-    #     return await self._api_wrapper(
-    #         method="patch",
-    #         url="https://jsonplaceholder.typicode.com/posts/1",
-    #         data={"title": value},
-    #         headers={"Content-type": "application/json; charset=UTF-8"},
-    #     )
+    async def async_start_stream(self, device_id: str, sdp_offer: str) -> StartStream:
+        """Start a WebRTC Stream."""
+        start_stream_response = await self._api_wrapper(
+            method="post",
+            url=self.get_api_url_for_endpoint(f"devices/{device_id}/stream"),
+            data={"offer": base64.b64encode(sdp_offer.encode()).decode()},
+            headers={"Content-type": "application/json; charset=UTF-8"}
+        )
+
+        return StartStream(**start_stream_response)
+
+    async def async_stop_stream(self, device_id: str, stream_id: str) -> StreamStatus:
+        """Stop a WebRTC Stream."""
+        stop_stream_response = await self._api_wrapper(
+            method="post",
+            url=self.get_api_url_for_endpoint(f"devices/{device_id}/stream/{stream_id}/stop"),
+            headers={"Content-type": "application/json; charset=UTF-8"},
+            data={}
+        )
+
+        return StreamStatus(**stop_stream_response)
+
+    async def async_extend_stream(self, device_id: str, stream_id: str) -> StreamStatus:
+        """Extend a WebRTC Stream."""
+        extend_stream_response = await self._api_wrapper(
+            method="post",
+            url=self.get_api_url_for_endpoint(f"devices/{device_id}/stream/{stream_id}/extend"),
+            headers={"Content-type": "application/json; charset=UTF-8"},
+            data={}
+        )
+
+        return StreamStatus(**extend_stream_response)
 
     async def _api_wrapper(
         self,

custom_components/starling_home_hub/binary_sensor.py

@@ -12,7 +12,7 @@
 from .const import DOMAIN, LOGGER
 from .coordinator import StarlingHomeHubDataUpdateCoordinator
 from .entity import StarlingHomeHubEntity
-from .models import CoordinatorData, SpecificDevice, Device
+from .models import CoordinatorData, Device
 
 from dataclasses import dataclass
 
@@ -35,6 +35,12 @@ class StarlingHomeHubNestProtectBinarySensorDescription(BinarySensorEntityDescri
         value_fn=lambda device: device["coDetected"],
         device_class=BinarySensorDeviceClass.CO
     ),
+    StarlingHomeHubNestProtectBinarySensorDescription(
+        key="occupancy_detected",
+        name="Occupancy Detected",
+        value_fn=lambda device: device["occupancyDetected"],
+        device_class=BinarySensorDeviceClass.OCCUPANCY
+    ),
     StarlingHomeHubNestProtectBinarySensorDescription(
         key="battery_status",
         name="Battery Status",
@@ -77,13 +83,10 @@ def __init__(
 
         self.device_id = device_id
         self.coordinator = coordinator
-
-        super().__init__(coordinator, self.get_device(), entity_description)
         self.entity_description = entity_description
+        self._attr_unique_id = f"{device_id}-{self.entity_description.key}"
 
-    def get_device(self) -> SpecificDevice:
-        """Get the actual device data from coordinator."""
-        return self.coordinator.data.devices.get(self.device_id)
+        super().__init__(coordinator)
 
     @property
     def is_on(self) -> bool:

custom_components/starling_home_hub/camera.py

@@ -0,0 +1,184 @@
+"""Camera platform for starling_home_hub."""
+from __future__ import annotations
+from collections.abc import Callable
+from pathlib import Path
+from datetime import timedelta, datetime, timezone
+import asyncio
+import functools
+import base64
+
+from homeassistant.components.camera import Camera, CameraEntityFeature, StreamType
+from homeassistant.components.stream import CONF_EXTRA_PART_WAIT_TIME
+from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.config_entries import ConfigEntry
+
+from .const import DOMAIN, LOGGER
+from .coordinator import StarlingHomeHubDataUpdateCoordinator
+from .entity import StarlingHomeHubEntity
+from .models import CoordinatorData, StartStream
+
+PLACEHOLDER = Path(__file__).parent / "placeholder.png"
+STREAM_EXPIRATION_BUFFER = timedelta(seconds=60)
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None:
+    """Set up the sensor platform."""
+
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    entities: list[StarlingHomeHubNestCamera] = []
+    data: CoordinatorData = coordinator.data
+
+    for device in filter(lambda device: device[1].properties["type"] == "cam", data.devices.items()):
+        entities.append(
+            StarlingHomeHubNestCamera(
+                device_id=device[0],
+                coordinator=coordinator
+            )
+        )
+
+    async_add_entities(entities, True)
+
+class StarlingHomeHubNestCamera(StarlingHomeHubEntity, Camera):
+    """Starling Home Hub Nest Camera class."""
+
+    def __init__(
+        self,
+        device_id: str,
+        coordinator: StarlingHomeHubDataUpdateCoordinator
+    ) -> None:
+        """Initialize the Nest Protect Sensor class."""
+
+        self.device_id = device_id
+        self.coordinator = coordinator
+        self._attr_unique_id = f"{device_id}-camera"
+
+        super().__init__(coordinator)
+        Camera.__init__(self)
+
+        self._stream: StartStream | None = None
+        self._create_stream_url_lock = asyncio.Lock()
+        self._stream_refresh_unsub: Callable[[], None] | None = None
+        self._attr_is_streaming = False
+        self._attr_supported_features = CameraEntityFeature(0)
+        self.stream_options[CONF_EXTRA_PART_WAIT_TIME] = 3
+
+        device = self.get_device()
+
+        if device.properties["supportsStreaming"]:
+            self._attr_is_streaming = True
+            self._attr_supported_features |= CameraEntityFeature.STREAM
+
+    @property
+    def use_stream_for_stills(self) -> bool:
+        """Whether or not to use stream to generate stills."""
+
+        device = self.get_device()
+        supports_streaming = device.properties["supportsStreaming"]
+        LOGGER.debug(f"supports streaming: {supports_streaming}")
+
+        return supports_streaming
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+
+        # Cameras are marked unavailable on stream errors in #54659 however nest
+        # streams have a high error rate (#60353). Given nest streams are so flaky,
+        # marking the stream unavailable has other side effects like not showing
+        # the camera image which sometimes are still able to work. Until the
+        # streams are fixed, just leave the streams as available.
+
+        return True
+
+    @property
+    def frontend_stream_type(self) -> StreamType | None:
+        """Return the type of stream supported by this camera."""
+
+        device = self.get_device()
+        if device.properties["supportsStreaming"]:
+            return StreamType.WEB_RTC
+        else:
+            return None
+
+    async def stream_source(self) -> str | None:
+        """Return stream source for the camera."""
+
+        return None
+
+    def _schedule_stream_refresh(self) -> None:
+        """Schedules an alarm to refresh the stream url before expiration."""
+
+        if not self._stream:
+            return
+
+        refresh_time = datetime.now(timezone.utc) + STREAM_EXPIRATION_BUFFER
+        LOGGER.debug("New stream url expires at %s", refresh_time)
+
+        if self._stream_refresh_unsub is not None:
+            self._stream_refresh_unsub()
+
+        self._stream_refresh_unsub = async_track_point_in_utc_time(
+            self.hass,
+            self._handle_stream_refresh,
+            refresh_time,
+        )
+
+    async def _handle_stream_refresh(self, now: datetime) -> None:
+        """Alarm that fires to check if the stream should be refreshed."""
+
+        if not self._stream:
+            return
+
+        try:
+            await self.coordinator.extend_stream(self.device_id, self._stream.streamId)
+        except Exception as err:
+            LOGGER.debug("Failed to extend stream: %s", err)
+            self._stream = None
+            if self.stream:
+                await self.stream.stop()
+                self.stream = None
+            return
+
+        self._schedule_stream_refresh()
+
+    async def async_will_remove_from_hass(self) -> None:
+        """Invalidate the RTSP token when unloaded."""
+
+        if self._stream:
+            LOGGER.debug("Invalidating stream")
+            await self.coordinator.stop_stream(self.device_id, self._stream.streamId)
+
+        if self._stream_refresh_unsub:
+            self._stream_refresh_unsub()
+
+    async def async_camera_image(
+        self, width: int | None = None, height: int | None = None
+    ) -> bytes | None:
+        """Return bytes of camera image."""
+
+        return await self.hass.async_add_executor_job(self.placeholder_image)
+
+    async def async_handle_web_rtc_offer(self, offer_sdp: str) -> str | None:
+        """Return the source of the stream."""
+
+        device = self.get_device()
+
+        if not device.properties["supportsStreaming"]:
+            return await super().async_handle_web_rtc_offer(offer_sdp)
+
+        self._stream = await self.coordinator.start_stream(device_id=self.device_id, sdp_offer=offer_sdp)
+        self._schedule_stream_refresh()
+
+        return self.decode_stream_answer(self._stream.answer)
+
+    def decode_stream_answer(self, stream_answer: str) -> str:
+        """Decode the stream RTC offer back to a string."""
+
+        return base64.decodebytes(stream_answer.encode()).decode()
+
+    @classmethod
+    @functools.cache
+    def placeholder_image(cls) -> bytes:
+        """Return placeholder image to use when no stream is available."""
+        return PLACEHOLDER.read_bytes()

custom_components/starling_home_hub/const.py

@@ -11,5 +11,6 @@
 ATTRIBUTION = "Based on the Starling Home Hub Developer Connect API"
 
 PLATFORMS: list[Platform] = [
-    Platform.BINARY_SENSOR
+    Platform.BINARY_SENSOR,
+    Platform.CAMERA
 ]

custom_components/starling_home_hub/coordinator.py

@@ -17,7 +17,7 @@
     StarlingHomeHubApiClientError,
 )
 from .const import DOMAIN, LOGGER
-from .models import CoordinatorData, SpecificDevice
+from .models import CoordinatorData, SpecificDevice, StartStream, StreamStatus
 
 # https://developers.home-assistant.io/docs/integration_fetching_data#coordinated-single-api-poll-for-data-for-all-entities
 class StarlingHomeHubDataUpdateCoordinator(DataUpdateCoordinator):
@@ -40,8 +40,21 @@ def __init__(
         )
         self.client = client
 
+    async def start_stream(self, device_id: str, sdp_offer: str) -> StartStream:
+        """Start a stream."""
+        return await self.client.async_start_stream(device_id=device_id, sdp_offer=sdp_offer)
+
+    async def stop_stream(self, device_id: str, stream_id: str) -> StreamStatus:
+        """Stop a stream."""
+        return await self.client.async_stop_stream(device_id=device_id, stream_id=stream_id)
+
+    async def extend_stream(self, device_id: str, stream_id: str) -> StreamStatus:
+        """Extend a stream."""
+        return await self.client.async_extend_stream(device_id=device_id, stream_id=stream_id)
+
     async def fetch_data(self) -> CoordinatorData:
         """Fetch data for the devices."""
+
         devices = await self.client.async_get_devices()
         status = await self.client.async_get_status()
 
@@ -56,6 +69,7 @@ async def fetch_data(self) -> CoordinatorData:
 
     async def _async_update_data(self) -> CoordinatorData:
         """Update data via library."""
+
         try:
             return await self.fetch_data()
         except StarlingHomeHubApiClientAuthenticationError as exception:

custom_components/starling_home_hub/entity.py

@@ -1,7 +1,7 @@
 """StarlingHomeHubEntity class."""
 from __future__ import annotations
 
-from homeassistant.helpers.entity import DeviceInfo, EntityDescription
+from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import ATTRIBUTION, DOMAIN
@@ -11,28 +11,36 @@
 class StarlingHomeHubEntity(CoordinatorEntity):
     """StarlingHomeHubEntity class."""
 
+    device_id: str
+
     _attr_attribution = ATTRIBUTION
 
-    def __init__(self, coordinator: StarlingHomeHubDataUpdateCoordinator, device: SpecificDevice, entity_description: EntityDescription) -> None:
+    def __init__(self, coordinator: StarlingHomeHubDataUpdateCoordinator) -> None:
         """Initialize."""
 
         super().__init__(coordinator)
 
-        self.entity_description = entity_description
-
+        device = self.get_device()
         device_properties = device.properties
         device_id = device_properties["id"]
-        self._attr_unique_id = f"{device_id}-{self.entity_description.key}"
 
         model = "Unknown"
 
         if device_properties["type"] == "protect":
             model = "Nest Protect"
 
+        if device_properties["type"] == "cam":
+            model = device_properties["cameraModel"]
+
         self._attr_device_info = DeviceInfo(
             identifiers={(DOMAIN, device_id)},
             name=device_properties["name"],
             model=model,
             manufacturer="Google",
-            suggested_area=device_properties["where"]
+            suggested_area=device_properties["where"],
+            serial_number=device_properties["serialNumber"]
         )
+
+    def get_device(self) -> SpecificDevice:
+        """Get the actual device data from coordinator."""
+        return self.coordinator.data.devices.get(self.device_id)