Octoprint buttons (#66368)

homeassistant/components/octoprint/__init__.py

@@ -52,7 +52,7 @@ def ensure_valid_path(value):
     return value
 
 
-PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.SENSOR]
 DEFAULT_NAME = "OctoPrint"
 CONF_NUMBER_OF_TOOLS = "number_of_tools"
 CONF_BED = "bed"

homeassistant/components/octoprint/button.py

@@ -0,0 +1,133 @@
+"""Support for Octoprint buttons."""
+from pyoctoprintapi import OctoprintClient, OctoprintPrinterInfo
+
+from homeassistant.components.button import ButtonEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import OctoprintDataUpdateCoordinator
+from .const import DOMAIN
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Octoprint control buttons."""
+    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
+
+    async_add_entities(
+        [
+            OctoprintResumeJobButton(coordinator, device_id, client),
+            OctoprintPauseJobButton(coordinator, device_id, client),
+            OctoprintStopJobButton(coordinator, device_id, client),
+        ]
+    )
+
+
+class OctoprintButton(CoordinatorEntity, ButtonEntity):
+    """Represent an OctoPrint binary sensor."""
+
+    coordinator: OctoprintDataUpdateCoordinator
+    client: OctoprintClient
+
+    def __init__(
+        self,
+        coordinator: OctoprintDataUpdateCoordinator,
+        button_type: str,
+        device_id: str,
+        client: OctoprintClient,
+    ) -> None:
+        """Initialize a new OctoPrint button."""
+        super().__init__(coordinator)
+        self.client = client
+        self._device_id = device_id
+        self._attr_name = f"OctoPrint {button_type}"
+        self._attr_unique_id = f"{button_type}-{device_id}"
+
+    @property
+    def device_info(self):
+        """Device info."""
+        return self.coordinator.device_info
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return self.coordinator.last_update_success and self.coordinator.data["printer"]
+
+
+class OctoprintPauseJobButton(OctoprintButton):
+    """Pause the active job."""
+
+    def __init__(
+        self,
+        coordinator: OctoprintDataUpdateCoordinator,
+        device_id: str,
+        client: OctoprintClient,
+    ) -> None:
+        """Initialize a new OctoPrint button."""
+        super().__init__(coordinator, "Pause Job", device_id, client)
+
+    async def async_press(self) -> None:
+        """Handle the button press."""
+        printer: OctoprintPrinterInfo = self.coordinator.data["printer"]
+
+        if printer.state.flags.printing:
+            await self.client.pause_job()
+        elif not printer.state.flags.paused and not printer.state.flags.pausing:
+            raise InvalidPrinterState("Printer is not printing")
+
+
+class OctoprintResumeJobButton(OctoprintButton):
+    """Resume the active job."""
+
+    def __init__(
+        self,
+        coordinator: OctoprintDataUpdateCoordinator,
+        device_id: str,
+        client: OctoprintClient,
+    ) -> None:
+        """Initialize a new OctoPrint button."""
+        super().__init__(coordinator, "Resume Job", device_id, client)
+
+    async def async_press(self) -> None:
+        """Handle the button press."""
+        printer: OctoprintPrinterInfo = self.coordinator.data["printer"]
+
+        if printer.state.flags.paused:
+            await self.client.resume_job()
+        elif not printer.state.flags.printing and not printer.state.flags.resuming:
+            raise InvalidPrinterState("Printer is not currently paused")
+
+
+class OctoprintStopJobButton(OctoprintButton):
+    """Resume the active job."""
+
+    def __init__(
+        self,
+        coordinator: OctoprintDataUpdateCoordinator,
+        device_id: str,
+        client: OctoprintClient,
+    ) -> None:
+        """Initialize a new OctoPrint button."""
+        super().__init__(coordinator, "Stop Job", device_id, client)
+
+    async def async_press(self) -> None:
+        """Handle the button press."""
+        printer: OctoprintPrinterInfo = self.coordinator.data["printer"]
+
+        if printer.state.flags.printing or printer.state.flags.paused:
+            await self.client.cancel_job()
+
+
+class InvalidPrinterState(HomeAssistantError):
+    """Service attempted in invalid state."""

tests/components/octoprint/test_button.py

@@ -0,0 +1,195 @@
+"""Test the OctoPrint buttons."""
+from unittest.mock import patch
+
+from pyoctoprintapi import OctoprintPrinterInfo
+import pytest
+
+from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN
+from homeassistant.components.button.const import SERVICE_PRESS
+from homeassistant.components.octoprint import OctoprintDataUpdateCoordinator
+from homeassistant.components.octoprint.button import InvalidPrinterState
+from homeassistant.components.octoprint.const import DOMAIN
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import HomeAssistant
+
+from . import init_integration
+
+
+async def test_pause_job(hass: HomeAssistant):
+    """Test the pause job button."""
+    await init_integration(hass, BUTTON_DOMAIN)
+
+    corrdinator: OctoprintDataUpdateCoordinator = hass.data[DOMAIN]["uuid"][
+        "coordinator"
+    ]
+
+    # Test pausing the printer when it is printing
+    with patch("pyoctoprintapi.OctoprintClient.pause_job") as pause_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": True}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_pause_job",
+            },
+            blocking=True,
+        )
+
+        assert len(pause_command.mock_calls) == 1
+
+    # Test pausing the printer when it is paused
+    with patch("pyoctoprintapi.OctoprintClient.pause_job") as pause_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": False, "paused": True}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_pause_job",
+            },
+            blocking=True,
+        )
+
+        assert len(pause_command.mock_calls) == 0
+
+    # Test pausing the printer when it is stopped
+    with patch(
+        "pyoctoprintapi.OctoprintClient.pause_job"
+    ) as pause_command, pytest.raises(InvalidPrinterState):
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {
+                "state": {"flags": {"printing": False, "paused": False}},
+                "temperature": [],
+            }
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_pause_job",
+            },
+            blocking=True,
+        )
+
+
+async def test_resume_job(hass: HomeAssistant):
+    """Test the resume job button."""
+    await init_integration(hass, BUTTON_DOMAIN)
+
+    corrdinator: OctoprintDataUpdateCoordinator = hass.data[DOMAIN]["uuid"][
+        "coordinator"
+    ]
+
+    # Test resuming the printer when it is paused
+    with patch("pyoctoprintapi.OctoprintClient.resume_job") as resume_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": False, "paused": True}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_resume_job",
+            },
+            blocking=True,
+        )
+
+        assert len(resume_command.mock_calls) == 1
+
+    # Test resuming the printer when it is printing
+    with patch("pyoctoprintapi.OctoprintClient.resume_job") as resume_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": True, "paused": False}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_resume_job",
+            },
+            blocking=True,
+        )
+
+        assert len(resume_command.mock_calls) == 0
+
+    # Test resuming the printer when it is stopped
+    with patch(
+        "pyoctoprintapi.OctoprintClient.resume_job"
+    ) as resume_command, pytest.raises(InvalidPrinterState):
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {
+                "state": {"flags": {"printing": False, "paused": False}},
+                "temperature": [],
+            }
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_resume_job",
+            },
+            blocking=True,
+        )
+
+
+async def test_stop_job(hass: HomeAssistant):
+    """Test the stop job button."""
+    await init_integration(hass, BUTTON_DOMAIN)
+
+    corrdinator: OctoprintDataUpdateCoordinator = hass.data[DOMAIN]["uuid"][
+        "coordinator"
+    ]
+
+    # Test stopping the printer when it is paused
+    with patch("pyoctoprintapi.OctoprintClient.cancel_job") as stop_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": False, "paused": True}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_stop_job",
+            },
+            blocking=True,
+        )
+
+        assert len(stop_command.mock_calls) == 1
+
+    # Test stopping the printer when it is printing
+    with patch("pyoctoprintapi.OctoprintClient.cancel_job") as stop_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {"state": {"flags": {"printing": True, "paused": False}}, "temperature": []}
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_stop_job",
+            },
+            blocking=True,
+        )
+
+        assert len(stop_command.mock_calls) == 1
+
+    # Test stopping the printer when it is stopped
+    with patch("pyoctoprintapi.OctoprintClient.cancel_job") as stop_command:
+        corrdinator.data["printer"] = OctoprintPrinterInfo(
+            {
+                "state": {"flags": {"printing": False, "paused": False}},
+                "temperature": [],
+            }
+        )
+        await hass.services.async_call(
+            BUTTON_DOMAIN,
+            SERVICE_PRESS,
+            {
+                ATTR_ENTITY_ID: "button.octoprint_stop_job",
+            },
+            blocking=True,
+        )
+
+        assert len(stop_command.mock_calls) == 0