diff --git a/homeassistant/components/prusalink/binary_sensor.py b/homeassistant/components/prusalink/binary_sensor.py index 6a255959204072..7e12e9253d5992 100644 --- a/homeassistant/components/prusalink/binary_sensor.py +++ b/homeassistant/components/prusalink/binary_sensor.py @@ -2,8 +2,9 @@ from collections.abc import Callable from dataclasses import dataclass +from typing import cast -from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus +from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus, StatusInfo from pyprusalink.types_legacy import LegacyPrinterStatus from homeassistant.components.binary_sensor import ( @@ -35,7 +36,9 @@ class PrusaLinkBinarySensorEntityDescription[ PrusaLinkBinarySensorEntityDescription[PrinterStatus]( key="printer.status_connect", device_class=BinarySensorDeviceClass.CONNECTIVITY, - value_fn=lambda data: data["printer"]["status_connect"]["ok"], + value_fn=lambda data: cast( + bool, cast(StatusInfo, data["printer"]["status_connect"])["ok"] + ), supported_fn=lambda data: ( data["printer"].get("status_connect") is not None and data["printer"]["status_connect"].get("ok") is not None diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py index 29e762a823dd35..c9b763e9ab1658 100644 --- a/homeassistant/components/prusalink/config_flow.py +++ b/homeassistant/components/prusalink/config_flow.py @@ -2,7 +2,7 @@ import asyncio import logging -from typing import Any +from typing import Any, cast from awesomeversion import AwesomeVersion, AwesomeVersionException from httpx import HTTPError, InvalidURL @@ -41,9 +41,10 @@ def ensure_printer_is_supported(version: VersionInfo) -> None: # Workaround to allow PrusaLink 0.7.2 on MK3 and MK2.5 that supports # the 2.0.0 API, but doesn't advertise it yet - if version.get("original", "").startswith( - ("PrusaLink I3MK3", "PrusaLink I3MK2") - ) and AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]): + original = cast(str, version.get("original", "")) + if original.startswith(("PrusaLink I3MK3", "PrusaLink I3MK2")) and ( + AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]) + ): return except AwesomeVersionException as err: diff --git a/homeassistant/components/prusalink/coordinator.py b/homeassistant/components/prusalink/coordinator.py index 9f97a39c7e6963..24b54032e2061d 100644 --- a/homeassistant/components/prusalink/coordinator.py +++ b/homeassistant/components/prusalink/coordinator.py @@ -31,7 +31,20 @@ # rapidly-changing metrics. _MINIMUM_REFRESH_INTERVAL = 1.0 -T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo, PrinterInfo, VersionInfo) +# Job is the only coordinator whose payload can be None — pyprusalink's +# get_job() returns None on HTTP 204 when no job is running. The other +# endpoints always return data or raise on failure. Using `bound=` rather +# than constraint members so `JobInfo | None` fits without forcing a union +# into the constraint list. +T = TypeVar( + "T", + bound=PrinterStatus + | LegacyPrinterStatus + | JobInfo + | None + | PrinterInfo + | VersionInfo, +) type PrusaLinkConfigEntry = ConfigEntry[dict[str, PrusaLinkUpdateCoordinator]] @@ -85,8 +98,15 @@ 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.""" + def _get_update_interval(self, data: T | None) -> timedelta: + """Get new update interval. + + `data` is unused by the base implementation today, but kept on the + signature so subclasses can override based on payload state — e.g. a + future transfer coordinator that polls faster while a transfer is + active. The base class is called once from `__init__` with `None` + before the first fetch, hence `T | None`. + """ if self.expect_change_until > monotonic(): return timedelta(seconds=5) @@ -109,10 +129,15 @@ async def _fetch_data(self) -> LegacyPrinterStatus: return await self.api.get_legacy_printer() -class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]): - """Job update coordinator.""" +class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo | None]): + """Job update coordinator. + + The job endpoint returns nothing (HTTP 204) when no job is running, so + `data` can legitimately be `None` here. Entity code that reads from this + coordinator's data must be `None`-aware. + """ - async def _fetch_data(self) -> JobInfo: + async def _fetch_data(self) -> JobInfo | None: """Fetch the printer data.""" return await self.api.get_job() diff --git a/homeassistant/components/prusalink/entity.py b/homeassistant/components/prusalink/entity.py index c0003c0e7e1f51..b36b455cbbdae4 100644 --- a/homeassistant/components/prusalink/entity.py +++ b/homeassistant/components/prusalink/entity.py @@ -29,8 +29,15 @@ class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]): @property def available(self) -> bool: """Return if entity is available.""" - return super().available and self.entity_description.available_fn( - self.coordinator.data + # `coordinator.data` can be None when the underlying endpoint + # returns no payload — e.g. the job coordinator yields None when + # no job is running on pyprusalink >= 3.0.0. Short-circuit to + # avoid passing None into `available_fn` lambdas that assume a + # dict (.get(), index, etc.). + return ( + super().available + and self.coordinator.data is not None + and self.entity_description.available_fn(self.coordinator.data) ) @property diff --git a/homeassistant/components/prusalink/manifest.json b/homeassistant/components/prusalink/manifest.json index f1c22950fa6d29..4430280ed28943 100644 --- a/homeassistant/components/prusalink/manifest.json +++ b/homeassistant/components/prusalink/manifest.json @@ -11,5 +11,5 @@ "documentation": "https://www.home-assistant.io/integrations/prusalink", "integration_type": "device", "iot_class": "local_polling", - "requirements": ["pyprusalink==2.2.0"] + "requirements": ["pyprusalink==3.0.0"] } diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index ce38101dcdcabc..6bb3330a6eab09 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -5,8 +5,14 @@ from datetime import datetime, timedelta from typing import cast -from pyprusalink.types import JobInfo, PrinterInfo, PrinterState, PrinterStatus -from pyprusalink.types_legacy import LegacyPrinterStatus +from pyprusalink.types import ( + JobFilePrint, + JobInfo, + PrinterInfo, + PrinterState, + PrinterStatus, +) +from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry from homeassistant.components.sensor import ( SensorDeviceClass, @@ -47,7 +53,7 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[PrinterStatus]( key="printer.state", name=None, - value_fn=lambda data: cast(str, data["printer"]["state"].lower()), + value_fn=lambda data: cast(str, data["printer"]["state"]).lower(), device_class=SensorDeviceClass.ENUM, options=[state.value.lower() for state in PrinterState], translation_key="printer_state", @@ -149,7 +155,10 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[LegacyPrinterStatus]( key="printer.telemetry.material", translation_key="material", - value_fn=lambda data: cast(str, data["telemetry"]["material"]), + value_fn=lambda data: cast( + str, cast(LegacyPrinterTelemetry, data["telemetry"])["material"] + ), + available_fn=lambda data: data.get("telemetry") is not None, ), ), "job": ( @@ -166,7 +175,11 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[JobInfo]( key="job.filename", translation_key="filename", - value_fn=lambda data: cast(str, data["file"]["display_name"]), + # `available_fn` guarantees `file` is not None at this point; + # the inner cast narrows the Optional for the index. + value_fn=lambda data: cast( + str, cast(JobFilePrint, data["file"])["display_name"] + ), available_fn=lambda data: ( data.get("file") is not None and data.get("state") != PrinterState.IDLE.value @@ -189,8 +202,12 @@ class PrusaLinkSensorEntityDescription[ key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, + # `available_fn` guarantees `time_remaining` is not None at this + # point; the cast narrows the Optional for `timedelta`. value_fn=ignore_variance( - lambda data: utcnow() + timedelta(seconds=data["time_remaining"]), + lambda data: ( + utcnow() + timedelta(seconds=cast(int, data["time_remaining"])) + ), timedelta(minutes=2), ), available_fn=lambda data: ( @@ -213,7 +230,7 @@ class PrusaLinkSensorEntityDescription[ translation_key="min_extrusion_temp", native_unit_of_measurement=UnitOfTemperature.CELSIUS, device_class=SensorDeviceClass.TEMPERATURE, - value_fn=lambda data: cast(int, data["min_extrusion_temp"]), + value_fn=lambda data: data["min_extrusion_temp"], supported_fn=lambda data: data.get("min_extrusion_temp") is not None, entity_registry_enabled_default=False, ), diff --git a/requirements_all.txt b/requirements_all.txt index a9f81c28da9730..f036466d863593 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2452,7 +2452,7 @@ pyprof2calltree==1.4.5 pyprosegur==0.0.14 # homeassistant.components.prusalink -pyprusalink==2.2.0 +pyprusalink==3.0.0 # homeassistant.components.ps4 pyps4-2ndscreen==1.3.1 diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index c94b98f3ce0926..53632e34270b0f 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -123,11 +123,14 @@ def mock_get_status_printing() -> Generator[dict[str, Any]]: @pytest.fixture -def mock_job_api_idle() -> Generator[dict[str, Any]]: - """Mock PrusaLink job API having no job.""" - resp = {} - with patch("pyprusalink.PrusaLink.get_job", return_value=resp): - yield resp +def mock_job_api_idle() -> Generator[None]: + """Mock PrusaLink job API having no job. + + pyprusalink >= 3.0.0 returns `None` from `get_job()` on HTTP 204 when + no job is running, rather than an empty dict as in 2.x. + """ + with patch("pyprusalink.PrusaLink.get_job", return_value=None): + yield None @pytest.fixture @@ -206,6 +209,6 @@ def mock_api( mock_info_api: dict[str, Any], mock_get_legacy_printer: dict[str, Any], mock_get_status_idle: dict[str, Any], - mock_job_api_idle: dict[str, Any], + mock_job_api_idle: None, ) -> None: """Mock PrusaLink API."""