diff --git a/custom_components/owlet/__init__.py b/custom_components/owlet/__init__.py index a6f93f1..c4bb801 100644 --- a/custom_components/owlet/__init__.py +++ b/custom_components/owlet/__init__.py @@ -30,7 +30,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession from .const import CONF_OWLET_EXPIRY, CONF_OWLET_REFRESH, DOMAIN, SUPPORTED_VERSIONS from .coordinator import OwletCoordinator -PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR, Platform.SWITCH] +PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.CAMERA, Platform.SENSOR, Platform.SWITCH] _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/owlet/camera.py b/custom_components/owlet/camera.py new file mode 100644 index 0000000..476dd46 --- /dev/null +++ b/custom_components/owlet/camera.py @@ -0,0 +1,217 @@ +"""Owlet Camera integration.""" +from __future__ import annotations + +import asyncio +from datetime import timedelta +import logging +from typing import Any + +import aiohttp +import async_timeout + +from homeassistant.components.camera import Camera, CameraEntityFeature +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_API_TOKEN, CONF_REGION, CONF_USERNAME +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import DOMAIN, MANUFACTURER + +_LOGGER = logging.getLogger(__name__) + +# Camera API endpoints based on region +CAMERA_KMS_ENDPOINTS = { + "world": "https://camera-kms.owletdata.com/kms/", + "europe": "https://camera-kms.eu.owletdata.com/kms/", +} + +# AWS Kinesis Video endpoint template +AWS_KINESIS_ENDPOINT_TEMPLATE = "https://kinesisvideo.{region}.amazonaws.com" + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up Owlet cameras.""" + session = async_get_clientsession(hass) + region = config_entry.data[CONF_REGION] + token = config_entry.data[CONF_API_TOKEN] + + # Get cameras from Owlet API + camera_api = OwletCameraAPI(session, region, token) + + try: + cameras = await camera_api.get_cameras() + _LOGGER.info(f"Found {len(cameras)} Owlet camera(s)") + + entities = [ + OwletCamera(hass, camera, camera_api) + for camera in cameras + ] + + async_add_entities(entities) + + except Exception as err: + _LOGGER.error(f"Error setting up Owlet cameras: {err}") + + +class OwletCameraAPI: + """API client for Owlet cameras.""" + + def __init__(self, session: aiohttp.ClientSession, region: str, token: str): + """Initialize the camera API.""" + self.session = session + self.region = region + self.token = token + self.kms_endpoint = CAMERA_KMS_ENDPOINTS.get( + region, CAMERA_KMS_ENDPOINTS["world"] + ) + + async def get_cameras(self) -> list[dict[str, Any]]: + """Get list of cameras from Owlet API.""" + # This will need to call the Owlet devices API to get cameras + # For now, return empty list - needs actual API implementation + headers = { + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + } + + try: + # TODO: Implement actual camera discovery endpoint + # This might be part of the existing get_devices call + # For now, we'll need to check if pyowletapi has camera support + _LOGGER.warning("Camera discovery not yet implemented") + return [] + + except Exception as err: + _LOGGER.error(f"Error getting cameras: {err}") + return [] + + async def get_stream_credentials(self, camera_id: str) -> dict[str, Any]: + """Get AWS Kinesis credentials for a camera.""" + headers = { + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + } + + try: + async with async_timeout.timeout(10): + response = await self.session.post( + self.kms_endpoint, + json={"camera_id": camera_id}, + headers=headers, + ) + response.raise_for_status() + return await response.json() + + except asyncio.TimeoutError: + _LOGGER.error(f"Timeout getting credentials for camera {camera_id}") + raise + except aiohttp.ClientError as err: + _LOGGER.error(f"Error getting credentials for camera {camera_id}: {err}") + raise + + async def get_hls_streaming_url( + self, stream_name: str, aws_credentials: dict[str, Any] + ) -> str: + """Get HLS streaming URL from AWS Kinesis Video Streams.""" + # This would use AWS SDK to: + # 1. Get data endpoint for the stream + # 2. Call GetHLSStreamingSessionURL + # For now, this is a placeholder + _LOGGER.warning("HLS streaming URL generation not yet implemented") + return "" + + +class OwletCamera(Camera): + """Representation of an Owlet Camera.""" + + _attr_has_entity_name = True + _attr_supported_features = CameraEntityFeature.STREAM + + def __init__( + self, + hass: HomeAssistant, + camera_data: dict[str, Any], + api: OwletCameraAPI, + ) -> None: + """Initialize the camera.""" + super().__init__() + self._hass = hass + self._api = api + self._camera_data = camera_data + self._attr_name = camera_data.get("name", "Owlet Camera") + self._attr_unique_id = camera_data.get("device_id") or camera_data.get("dsn") + self._stream_url: str | None = None + self._last_url_refresh = None + + @property + def device_info(self): + """Return device information.""" + return { + "identifiers": {(DOMAIN, self.unique_id)}, + "name": self._attr_name, + "manufacturer": MANUFACTURER, + "model": self._camera_data.get("model", "Owlet Cam"), + } + + @property + def available(self) -> bool: + """Return True if camera is available.""" + return True + + async def async_camera_image( + self, width: int | None = None, height: int | None = None + ) -> bytes | None: + """Return a still image from the camera.""" + # Get the stream URL and extract a frame + # This is optional - HLS streams don't provide easy still images + return None + + async def stream_source(self) -> str | None: + """Return the stream source URL.""" + # Refresh URL if needed (AWS URLs expire) + if self._should_refresh_url(): + await self._refresh_stream_url() + + return self._stream_url + + def _should_refresh_url(self) -> bool: + """Check if stream URL needs to be refreshed.""" + if self._stream_url is None: + return True + + # AWS HLS URLs typically expire after a period + # Refresh every 30 minutes to be safe + if self._last_url_refresh is None: + return True + + from datetime import datetime + age = datetime.now() - self._last_url_refresh + return age > timedelta(minutes=30) + + async def _refresh_stream_url(self) -> None: + """Refresh the HLS streaming URL.""" + try: + camera_id = self._camera_data.get("device_id") or self._camera_data.get("dsn") + + # Step 1: Get AWS credentials from Owlet KMS API + credentials = await self._api.get_stream_credentials(camera_id) + + # Step 2: Use credentials to get HLS URL from AWS Kinesis + stream_name = self._camera_data.get("kinesis_stream_name", camera_id) + self._stream_url = await self._api.get_hls_streaming_url( + stream_name, credentials + ) + + from datetime import datetime + self._last_url_refresh = datetime.now() + + _LOGGER.info(f"Refreshed stream URL for camera {self._attr_name}") + + except Exception as err: + _LOGGER.error(f"Error refreshing stream URL: {err}") + self._stream_url = None diff --git a/custom_components/owlet/kinesis_client.py b/custom_components/owlet/kinesis_client.py new file mode 100644 index 0000000..d724d2b --- /dev/null +++ b/custom_components/owlet/kinesis_client.py @@ -0,0 +1,213 @@ +"""AWS Kinesis Video Streams client for Owlet cameras.""" +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta +import hashlib +import hmac +import json +import logging +from typing import Any +from urllib.parse import quote + +import aiohttp + +_LOGGER = logging.getLogger(__name__) + + +class KinesisVideoClient: + """Client for AWS Kinesis Video Streams API.""" + + def __init__( + self, + session: aiohttp.ClientSession, + region: str = "eu-west-1", + ): + """Initialize the Kinesis client.""" + self.session = session + self.region = region + self.service = "kinesisvideo" + self.control_endpoint = f"https://kinesisvideo.{region}.amazonaws.com" + + async def get_data_endpoint( + self, + stream_name: str, + access_key: str, + secret_key: str, + session_token: str | None = None, + ) -> str: + """Get the data endpoint for a Kinesis Video Stream.""" + endpoint = self.control_endpoint + headers = self._get_signed_headers( + method="POST", + uri="/getDataEndpoint", + access_key=access_key, + secret_key=secret_key, + session_token=session_token, + payload={ + "StreamName": stream_name, + "APIName": "GET_HLS_STREAMING_SESSION_URL", + }, + ) + + try: + async with asyncio.timeout(10): + response = await self.session.post( + f"{endpoint}/getDataEndpoint", + json={ + "StreamName": stream_name, + "APIName": "GET_HLS_STREAMING_SESSION_URL", + }, + headers=headers, + ) + response.raise_for_status() + data = await response.json() + return data.get("DataEndpoint", "") + + except Exception as err: + _LOGGER.error(f"Error getting data endpoint: {err}") + raise + + async def get_hls_streaming_url( + self, + stream_name: str, + data_endpoint: str, + access_key: str, + secret_key: str, + session_token: str | None = None, + expires: int = 43200, # 12 hours + ) -> str: + """Get HLS streaming URL for a Kinesis Video Stream.""" + service = "kinesisvideo" + + headers = self._get_signed_headers( + method="POST", + uri="/getHLSStreamingSessionURL", + access_key=access_key, + secret_key=secret_key, + session_token=session_token, + endpoint=data_endpoint, + payload={ + "StreamName": stream_name, + "PlaybackMode": "LIVE", + "HLSFragmentSelector": { + "FragmentSelectorType": "SERVER_TIMESTAMP" + }, + "ContainerFormat": "MPEG_TS", + "DiscontinuityMode": "ALWAYS", + "DisplayFragmentTimestamp": "NEVER", + "Expires": expires, + }, + ) + + try: + async with asyncio.timeout(10): + response = await self.session.post( + f"{data_endpoint}/getHLSStreamingSessionURL", + json={ + "StreamName": stream_name, + "PlaybackMode": "LIVE", + "HLSFragmentSelector": { + "FragmentSelectorType": "SERVER_TIMESTAMP" + }, + "ContainerFormat": "MPEG_TS", + "DiscontinuityMode": "ALWAYS", + "DisplayFragmentTimestamp": "NEVER", + "Expires": expires, + }, + headers=headers, + ) + response.raise_for_status() + data = await response.json() + return data.get("HLSStreamingSessionURL", "") + + except Exception as err: + _LOGGER.error(f"Error getting HLS URL: {err}") + raise + + def _get_signed_headers( + self, + method: str, + uri: str, + access_key: str, + secret_key: str, + session_token: str | None = None, + endpoint: str | None = None, + payload: dict[str, Any] | None = None, + ) -> dict[str, str]: + """Generate AWS Signature V4 signed headers.""" + if endpoint is None: + endpoint = self.control_endpoint + + # Parse endpoint to get host + host = endpoint.replace("https://", "").replace("http://", "") + + # Current timestamp + now = datetime.utcnow() + amz_date = now.strftime("%Y%m%dT%H%M%SZ") + date_stamp = now.strftime("%Y%m%d") + + # Payload + payload_str = json.dumps(payload) if payload else "" + payload_hash = hashlib.sha256(payload_str.encode("utf-8")).hexdigest() + + # Canonical request + canonical_headers = f"host:{host}\nx-amz-date:{amz_date}\n" + signed_headers = "host;x-amz-date" + + if session_token: + canonical_headers += f"x-amz-security-token:{session_token}\n" + signed_headers += ";x-amz-security-token" + + canonical_request = ( + f"{method}\n" + f"{uri}\n" + f"\n" # Query string (empty) + f"{canonical_headers}\n" + f"{signed_headers}\n" + f"{payload_hash}" + ) + + # String to sign + algorithm = "AWS4-HMAC-SHA256" + credential_scope = f"{date_stamp}/{self.region}/{self.service}/aws4_request" + string_to_sign = ( + f"{algorithm}\n" + f"{amz_date}\n" + f"{credential_scope}\n" + f"{hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()}" + ) + + # Signing key + def sign(key: bytes, msg: str) -> bytes: + return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest() + + k_date = sign(f"AWS4{secret_key}".encode("utf-8"), date_stamp) + k_region = sign(k_date, self.region) + k_service = sign(k_region, self.service) + signing_key = sign(k_service, "aws4_request") + + # Signature + signature = hmac.new( + signing_key, string_to_sign.encode("utf-8"), hashlib.sha256 + ).hexdigest() + + # Authorization header + authorization_header = ( + f"{algorithm} " + f"Credential={access_key}/{credential_scope}, " + f"SignedHeaders={signed_headers}, " + f"Signature={signature}" + ) + + headers = { + "Content-Type": "application/json", + "X-Amz-Date": amz_date, + "Authorization": authorization_header, + "Host": host, + } + + if session_token: + headers["X-Amz-Security-Token"] = session_token + + return headers