Bump pyprusalink to 3.0.0 (#170480)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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 @@ BINARY_SENSORS: dict[str, tuple[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

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:

homeassistant/components/prusalink/coordinator.py

@@ -31,7 +31,20 @@ _LOGGER = logging.getLogger(__name__)
 # 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 @@ class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):
         """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 @@ class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]):
         return await self.api.get_legacy_printer()
 
 
-class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
-    """Job update coordinator."""
+class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo | None]):
+    """Job update coordinator.
 
-    async def _fetch_data(self) -> JobInfo:
+    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 | None:
         """Fetch the printer data."""
         return await self.api.get_job()
 

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

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"]
 }

homeassistant/components/prusalink/sensor.py

@@ -5,8 +5,14 @@ from dataclasses import dataclass
 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 @@ SENSORS: dict[str, tuple[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 @@ SENSORS: dict[str, tuple[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 @@ SENSORS: dict[str, tuple[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 @@ SENSORS: dict[str, tuple[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 @@ SENSORS: dict[str, tuple[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,
         ),

requirements_all.txt

@@ -2455,7 +2455,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

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."""