Allow controlling PrusaLink print jobs (#78720)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Paulus Schoutsen 2022-09-28 04:42:19 -04:00 committed by GitHub
parent 87dfe82613
commit 010430780e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 300 additions and 30 deletions

View file

@ -4,6 +4,7 @@ from __future__ import annotations
from abc import abstractmethod
from datetime import timedelta
import logging
from time import monotonic
from typing import Generic, TypeVar
import async_timeout
@ -11,7 +12,7 @@ from pyprusalink import InvalidAuth, JobInfo, PrinterInfo, PrusaLink, PrusaLinkE
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
@ -22,7 +23,7 @@ from homeassistant.helpers.update_coordinator import (
from .const import DOMAIN
PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.CAMERA]
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.CAMERA, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
@ -63,30 +64,46 @@ class PrusaLinkUpdateCoordinator(DataUpdateCoordinator, Generic[T]):
"""Update coordinator for the printer."""
config_entry: ConfigEntry
expect_change_until = 0.0
def __init__(self, hass: HomeAssistant, api: PrusaLink) -> None:
"""Initialize the update coordinator."""
self.api = api
super().__init__(
hass, _LOGGER, name=DOMAIN, update_interval=timedelta(seconds=30)
hass, _LOGGER, name=DOMAIN, update_interval=self._get_update_interval(None)
)
async def _async_update_data(self) -> T:
"""Update the data."""
try:
with async_timeout.timeout(5):
return await self._fetch_data()
data = await self._fetch_data()
except InvalidAuth:
raise UpdateFailed("Invalid authentication") from None
except PrusaLinkError as err:
raise UpdateFailed(str(err)) from err
self.update_interval = self._get_update_interval(data)
return data
@abstractmethod
async def _fetch_data(self) -> T:
"""Fetch the actual data."""
raise NotImplementedError
@callback
def expect_change(self) -> None:
"""Expect a change."""
self.expect_change_until = monotonic() + 30
def _get_update_interval(self, data: T) -> timedelta:
"""Get new update interval."""
if self.expect_change_until > monotonic():
return timedelta(seconds=5)
return timedelta(seconds=30)
class PrinterUpdateCoordinator(PrusaLinkUpdateCoordinator[PrinterInfo]):
"""Printer update coordinator."""
@ -95,6 +112,15 @@ class PrinterUpdateCoordinator(PrusaLinkUpdateCoordinator[PrinterInfo]):
"""Fetch the printer data."""
return await self.api.get_printer()
def _get_update_interval(self, data: T) -> timedelta:
"""Get new update interval."""
if data and any(
data["state"]["flags"][key] for key in ("pausing", "cancelling")
):
return timedelta(seconds=5)
return super()._get_update_interval(data)
class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
"""Job update coordinator."""

View file

@ -0,0 +1,126 @@
"""PrusaLink sensors."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, Generic, TypeVar, cast
from pyprusalink import Conflict, JobInfo, PrinterInfo, PrusaLink
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
T = TypeVar("T", PrinterInfo, JobInfo)
@dataclass
class PrusaLinkButtonEntityDescriptionMixin(Generic[T]):
"""Mixin for required keys."""
press_fn: Callable[[PrusaLink], Coroutine[Any, Any, None]]
@dataclass
class PrusaLinkButtonEntityDescription(
ButtonEntityDescription, PrusaLinkButtonEntityDescriptionMixin[T], Generic[T]
):
"""Describes PrusaLink button entity."""
available_fn: Callable[[T], bool] = lambda _: True
BUTTONS: dict[str, tuple[PrusaLinkButtonEntityDescription, ...]] = {
"printer": (
PrusaLinkButtonEntityDescription[PrinterInfo](
key="printer.cancel_job",
name="Cancel Job",
press_fn=lambda api: cast(Coroutine, api.cancel_job()),
available_fn=lambda data: any(
data["state"]["flags"][flag]
for flag in ("printing", "pausing", "paused")
),
),
PrusaLinkButtonEntityDescription[PrinterInfo](
key="job.pause_job",
name="Pause Job",
press_fn=lambda api: cast(Coroutine, api.pause_job()),
available_fn=lambda data: (
data["state"]["flags"]["printing"]
and not data["state"]["flags"]["paused"]
),
),
PrusaLinkButtonEntityDescription[PrinterInfo](
key="job.resume_job",
name="Resume Job",
press_fn=lambda api: cast(Coroutine, api.resume_job()),
available_fn=lambda data: cast(bool, data["state"]["flags"]["paused"]),
),
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up PrusaLink buttons based on a config entry."""
coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][
entry.entry_id
]
entities: list[PrusaLinkEntity] = []
for coordinator_type, sensors in BUTTONS.items():
coordinator = coordinators[coordinator_type]
entities.extend(
PrusaLinkButtonEntity(coordinator, sensor_description)
for sensor_description in sensors
)
async_add_entities(entities)
class PrusaLinkButtonEntity(PrusaLinkEntity, ButtonEntity):
"""Defines a PrusaLink button."""
entity_description: PrusaLinkButtonEntityDescription
def __init__(
self,
coordinator: PrusaLinkUpdateCoordinator,
description: PrusaLinkButtonEntityDescription,
) -> None:
"""Initialize a PrusaLink sensor entity."""
super().__init__(coordinator=coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{description.key}"
@property
def available(self) -> bool:
"""Return if sensor is available."""
return super().available and self.entity_description.available_fn(
self.coordinator.data
)
async def async_press(self) -> None:
"""Press the button."""
try:
await self.entity_description.press_fn(self.coordinator.api)
except Conflict as err:
raise HomeAssistantError(
"Action conflicts with current printer state"
) from err
coordinators: dict[str, PrusaLinkUpdateCoordinator] = self.hass.data[DOMAIN][
self.coordinator.config_entry.entry_id
]
for coordinator in coordinators.values():
coordinator.expect_change()
await coordinator.async_request_refresh()

View file

@ -3,7 +3,7 @@
"name": "PrusaLink",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/prusalink",
"requirements": ["pyprusalink==1.0.1"],
"requirements": ["pyprusalink==1.1.0"],
"dhcp": [
{
"macaddress": "109C70*"

View file

@ -1813,7 +1813,7 @@ pyprof2calltree==1.4.5
pyprosegur==0.0.5
# homeassistant.components.prusalink
pyprusalink==1.0.1
pyprusalink==1.1.0
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

View file

@ -1284,7 +1284,7 @@ pyprof2calltree==1.4.5
pyprosegur==0.0.5
# homeassistant.components.prusalink
pyprusalink==1.0.1
pyprusalink==1.1.0
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

View file

@ -68,23 +68,23 @@ def mock_printer_api(hass):
@pytest.fixture
def mock_job_api_idle(hass):
"""Mock PrusaLink job API having no job."""
with patch(
"pyprusalink.PrusaLink.get_job",
return_value={
"state": "Operational",
"job": None,
"progress": None,
},
):
yield
resp = {
"state": "Operational",
"job": None,
"progress": None,
}
with patch("pyprusalink.PrusaLink.get_job", return_value=resp):
yield resp
@pytest.fixture
def mock_job_api_active(hass):
"""Mock PrusaLink job API having no job."""
with patch(
"pyprusalink.PrusaLink.get_job",
return_value={
def mock_job_api_printing(hass, mock_printer_api, mock_job_api_idle):
"""Mock PrusaLink printing."""
mock_printer_api["state"]["text"] = "Printing"
mock_printer_api["state"]["flags"]["printing"] = True
mock_job_api_idle.update(
{
"state": "Printing",
"job": {
"estimatedPrintTime": 117007,
@ -99,9 +99,18 @@ def mock_job_api_active(hass):
"printTime": 43987,
"printTimeLeft": 73020,
},
},
):
yield
}
)
@pytest.fixture
def mock_job_api_paused(hass, mock_printer_api, mock_job_api_idle):
"""Mock PrusaLink paused printing."""
mock_printer_api["state"]["text"] = "Paused"
mock_printer_api["state"]["flags"]["printing"] = False
mock_printer_api["state"]["flags"]["paused"] = True
mock_job_api_idle["state"] = "Paused"
@pytest.fixture

View file

@ -0,0 +1,107 @@
"""Test Prusalink buttons."""
from unittest.mock import patch
from pyprusalink import Conflict
import pytest
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
@pytest.fixture(autouse=True)
def setup_button_platform_only():
"""Only setup button platform."""
with patch("homeassistant.components.prusalink.PLATFORMS", [Platform.BUTTON]):
yield
@pytest.mark.parametrize(
"object_id, method",
(
("mock_title_cancel_job", "cancel_job"),
("mock_title_pause_job", "pause_job"),
),
)
async def test_button_pause_cancel(
hass: HomeAssistant,
mock_config_entry,
mock_api,
hass_client,
mock_job_api_printing,
object_id,
method,
) -> None:
"""Test cancel and pause button."""
entity_id = f"button.{object_id}"
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "unknown"
with patch(f"pyprusalink.PrusaLink.{method}") as mock_meth:
await hass.services.async_call(
"button",
"press",
{"entity_id": entity_id},
blocking=True,
)
assert len(mock_meth.mock_calls) == 1
# Verify it calls correct method + does error handling
with pytest.raises(HomeAssistantError), patch(
f"pyprusalink.PrusaLink.{method}", side_effect=Conflict
):
await hass.services.async_call(
"button",
"press",
{"entity_id": entity_id},
blocking=True,
)
@pytest.mark.parametrize(
"object_id, method",
(("mock_title_resume_job", "resume_job"),),
)
async def test_button_resume(
hass: HomeAssistant,
mock_config_entry,
mock_api,
hass_client,
mock_job_api_paused,
object_id,
method,
) -> None:
"""Test resume button."""
entity_id = f"button.{object_id}"
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "unknown"
with patch(f"pyprusalink.PrusaLink.{method}") as mock_meth, patch(
"homeassistant.components.prusalink.PrusaLinkUpdateCoordinator._fetch_data"
):
await hass.services.async_call(
"button",
"press",
{"entity_id": entity_id},
blocking=True,
)
assert len(mock_meth.mock_calls) == 1
# Verify it calls correct method + does error handling
with pytest.raises(HomeAssistantError), patch(
f"pyprusalink.PrusaLink.{method}", side_effect=Conflict
):
await hass.services.async_call(
"button",
"press",
{"entity_id": entity_id},
blocking=True,
)

View file

@ -22,7 +22,7 @@ async def test_camera_no_job(
mock_api,
hass_client,
) -> None:
"""Test sensors while no job active."""
"""Test camera while no job active."""
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("camera.mock_title_job_preview")
assert state is not None
@ -34,9 +34,13 @@ async def test_camera_no_job(
async def test_camera_active_job(
hass: HomeAssistant, mock_config_entry, mock_api, mock_job_api_active, hass_client
hass: HomeAssistant,
mock_config_entry,
mock_api,
mock_job_api_printing,
hass_client,
):
"""Test sensors while no job active."""
"""Test camera while job active."""
assert await async_setup_component(hass, "prusalink", {})
state = hass.states.get("camera.mock_title_job_preview")
assert state is not None

View file

@ -79,11 +79,9 @@ async def test_sensors_active_job(
mock_config_entry,
mock_api,
mock_printer_api,
mock_job_api_active,
mock_job_api_printing,
):
"""Test sensors while active job."""
mock_printer_api["state"]["flags"]["printing"] = True
with patch(
"homeassistant.components.prusalink.sensor.utcnow",
return_value=datetime(2022, 8, 27, 14, 0, 0, tzinfo=timezone.utc),