diff --git a/homeassistant/components/octoprint/__init__.py b/homeassistant/components/octoprint/__init__.py index 9db4e834571..8f6de2f0a28 100644 --- a/homeassistant/components/octoprint/__init__.py +++ b/homeassistant/components/octoprint/__init__.py @@ -56,7 +56,7 @@ def ensure_valid_path(value): return value -PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.SENSOR] +PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.CAMERA, Platform.SENSOR] DEFAULT_NAME = "OctoPrint" CONF_NUMBER_OF_TOOLS = "number_of_tools" CONF_BED = "bed" diff --git a/homeassistant/components/octoprint/camera.py b/homeassistant/components/octoprint/camera.py new file mode 100644 index 00000000000..ea886c97936 --- /dev/null +++ b/homeassistant/components/octoprint/camera.py @@ -0,0 +1,59 @@ +"""Support for OctoPrint binary camera.""" +from __future__ import annotations + +from pyoctoprintapi import OctoprintClient, WebcamSettings + +from homeassistant.components.mjpeg.camera import MjpegCamera +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from . import OctoprintDataUpdateCoordinator +from .const import DOMAIN + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the available OctoPrint camera.""" + coordinator: OctoprintDataUpdateCoordinator = hass.data[DOMAIN][ + config_entry.entry_id + ]["coordinator"] + client: OctoprintClient = hass.data[DOMAIN][config_entry.entry_id]["client"] + device_id = config_entry.unique_id + + assert device_id is not None + + camera_info = await client.get_webcam_info() + + if not camera_info or not camera_info.enabled: + return + + async_add_entities( + [ + OctoprintCamera( + camera_info, + coordinator.device_info, + device_id, + ) + ] + ) + + +class OctoprintCamera(MjpegCamera): + """Representation of an OctoPrint Camera Stream.""" + + def __init__( + self, camera_settings: WebcamSettings, device_info: DeviceInfo, device_id: str + ) -> None: + """Initialize as a subclass of MjpegCamera.""" + super().__init__( + device_info=device_info, + mjpeg_url=camera_settings.stream_url, + name="OctoPrint Camera", + still_image_url=camera_settings.external_snapshot_url, + unique_id=f"camera-{device_id}", + ) diff --git a/tests/components/octoprint/test_camera.py b/tests/components/octoprint/test_camera.py new file mode 100644 index 00000000000..c95cf924d15 --- /dev/null +++ b/tests/components/octoprint/test_camera.py @@ -0,0 +1,67 @@ +"""The tests for Octoptint camera module.""" + +from unittest.mock import patch + +from pyoctoprintapi import WebcamSettings + +from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN +from homeassistant.helpers import entity_registry as er + +from . import init_integration + + +async def test_camera(hass): + """Test the underlying camera.""" + with patch( + "pyoctoprintapi.OctoprintClient.get_webcam_info", + return_value=WebcamSettings( + base_url="http://fake-octoprint/", + raw={ + "streamUrl": "/webcam/?action=stream", + "snapshotUrl": "http://127.0.0.1:8080/?action=snapshot", + "webcamEnabled": True, + }, + ), + ): + await init_integration(hass, CAMERA_DOMAIN) + + entity_registry = er.async_get(hass) + + entry = entity_registry.async_get("camera.octoprint_camera") + assert entry is not None + assert entry.unique_id == "camera-uuid" + + +async def test_camera_disabled(hass): + """Test that the camera does not load if there is not one configured.""" + with patch( + "pyoctoprintapi.OctoprintClient.get_webcam_info", + return_value=WebcamSettings( + base_url="http://fake-octoprint/", + raw={ + "streamUrl": "/webcam/?action=stream", + "snapshotUrl": "http://127.0.0.1:8080/?action=snapshot", + "webcamEnabled": False, + }, + ), + ): + await init_integration(hass, CAMERA_DOMAIN) + + entity_registry = er.async_get(hass) + + entry = entity_registry.async_get("camera.octoprint_camera") + assert entry is None + + +async def test_no_supported_camera(hass): + """Test that the camera does not load if there is not one configured.""" + with patch( + "pyoctoprintapi.OctoprintClient.get_webcam_info", + return_value=None, + ): + await init_integration(hass, CAMERA_DOMAIN) + + entity_registry = er.async_get(hass) + + entry = entity_registry.async_get("camera.octoprint_camera") + assert entry is None