diff --git a/homeassistant/components/octoprint/__init__.py b/homeassistant/components/octoprint/__init__.py index 6c1eb62831c..f92cf0c8d30 100644 --- a/homeassistant/components/octoprint/__init__.py +++ b/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" diff --git a/homeassistant/components/octoprint/button.py b/homeassistant/components/octoprint/button.py new file mode 100644 index 00000000000..97676592f47 --- /dev/null +++ b/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.""" diff --git a/tests/components/octoprint/test_button.py b/tests/components/octoprint/test_button.py new file mode 100644 index 00000000000..603739159af --- /dev/null +++ b/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