diff --git a/homeassistant/components/opendisplay/services.py b/homeassistant/components/opendisplay/services.py index 971d0076860d..984abce80e96 100644 --- a/homeassistant/components/opendisplay/services.py +++ b/homeassistant/components/opendisplay/services.py @@ -21,6 +21,7 @@ from opendisplay import ( from PIL import Image as PILImage, ImageOps import probatio +from homeassistant.components import camera, image as image_component from homeassistant.components.bluetooth import ( BluetoothReachabilityIntent, async_address_reachability_diagnostics, @@ -128,6 +129,31 @@ async def _async_download_image(hass: HomeAssistant, url: str) -> PILImage.Image return await hass.async_add_executor_job(_load_image_from_bytes, data) +async def _async_get_pil_image( + hass: HomeAssistant, image_data: dict[str, Any] +) -> PILImage.Image: + """Return a PIL Image for the selected media.""" + media_content_id: str = image_data["media_content_id"] + + # Camera and image entities resolve to never-ending MJPEG streams, + # so fetch a single snapshot from the entity instead. + for integration in camera, image_component: + media_source_prefix = f"media-source://{integration.DOMAIN}/" + if not media_content_id.startswith(media_source_prefix): + continue + + entity_id = media_content_id.removeprefix(media_source_prefix) + snapshot = await integration.async_get_image(hass, entity_id) + return await hass.async_add_executor_job( + _load_image_from_bytes, snapshot.content + ) + + media = await async_resolve_media(hass, media_content_id, None) + if media.path is not None: + return await hass.async_add_executor_job(_load_image, str(media.path)) + return await _async_download_image(hass, media.url) + + async def _async_upload_image(call: ServiceCall) -> None: """Handle the upload_image service call.""" entry = _get_entry_for_device(call) @@ -166,16 +192,7 @@ async def _async_upload_image(call: ServiceCall) -> None: entry.runtime_data.upload_task = current try: - media = await async_resolve_media( - call.hass, image_data["media_content_id"], None - ) - - if media.path is not None: - pil_image = await call.hass.async_add_executor_job( - _load_image, str(media.path) - ) - else: - pil_image = await _async_download_image(call.hass, media.url) + pil_image = await _async_get_pil_image(call.hass, image_data) raw_key = entry.data.get(CONF_ENCRYPTION_KEY) if raw_key is not None and len(raw_key) != 32: diff --git a/tests/components/opendisplay/test_services.py b/tests/components/opendisplay/test_services.py index 903533a91d1d..e3b8a3fdac7f 100644 --- a/tests/components/opendisplay/test_services.py +++ b/tests/components/opendisplay/test_services.py @@ -17,6 +17,8 @@ import probatio import pytest from homeassistant import config_entries +from homeassistant.components.camera import Image as CameraImage +from homeassistant.components.image import Image as ImageEntityImage from homeassistant.components.opendisplay.const import CONF_ENCRYPTION_KEY, DOMAIN from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError, ServiceValidationError @@ -128,6 +130,66 @@ async def test_upload_image_remote_url( mock_upload_device.upload_image.assert_called_once() +@pytest.mark.parametrize( + ("media_content_id", "patch_target", "entity_id", "image_cls"), + [ + pytest.param( + "media-source://image/image.collection", + "homeassistant.components.image.async_get_image", + "image.collection", + ImageEntityImage, + id="image_entity", + ), + pytest.param( + "media-source://camera/camera.front_door", + "homeassistant.components.camera.async_get_image", + "camera.front_door", + CameraImage, + id="camera_entity", + ), + ], +) +async def test_upload_image_from_entity( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_upload_device: MagicMock, + media_content_id: str, + patch_target: str, + entity_id: str, + image_cls: type[ImageEntityImage | CameraImage], +) -> None: + """Test uploading a snapshot from an image or camera entity.""" + device_id = _device_id(hass, mock_config_entry) + + buf = io.BytesIO() + PILImage.new("RGB", (10, 10)).save(buf, format="PNG") + + with ( + patch( + patch_target, return_value=image_cls("image/png", buf.getvalue()) + ) as mock_get_image, + patch( + "homeassistant.components.opendisplay.services.async_resolve_media" + ) as mock_resolve, + ): + await hass.services.async_call( + DOMAIN, + "upload_image", + { + "device_id": device_id, + "image": { + "media_content_id": media_content_id, + "media_content_type": "image/png", + }, + }, + blocking=True, + ) + + mock_get_image.assert_awaited_once_with(hass, entity_id) + mock_resolve.assert_not_called() + mock_upload_device.upload_image.assert_called_once() + + async def test_upload_image_invalid_device_id( hass: HomeAssistant, ) -> None: