From 8b81d683a4abd32f14295adec6316503baa18e7e Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:07:58 +0200 Subject: [PATCH 01/35] prusalink: bump pyprusalink to 3.0.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 3.0.0 changes that motivate this bump: - `get_job()` now returns `JobInfo | None` (returns None on 204 instead of casting an empty dict to JobInfo) — clarifies the no-job case - `PrinterInfo` fields migrated from `T | None` to `NotRequired[T]` to match what the API actually returns - `jobId` parameter renamed to `job_id` on cancel/pause/resume/ continue_job; HA calls these positionally, no impact - `py.typed` marker added (PEP 561) — pyprusalink's typing now applies to HA's mypy run; subsequent commits fix the latent issues this surfaces Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/manifest.json | 2 +- requirements_all.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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/requirements_all.txt b/requirements_all.txt index bf8c3a8d64306e..849880dd1aab48 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 From c039c082e06fd60a7a3d7fd637542b73cd6b81cd Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:20:16 +0200 Subject: [PATCH 02/35] prusalink: coordinator typing for nullable JobInfo and pre-fetch None Two changes that mypy strict now requires after pyprusalink 3.0.0 added its py.typed marker: - `JobUpdateCoordinator` now binds `[JobInfo | None]` to reflect that pyprusalink's `get_job()` returns `None` on HTTP 204 when no job is running. `_fetch_data` return type follows. - `_get_update_interval(data: T)` becomes `data: T | None` because the base class is called once from `__init__` (line 57) with `None` before the first fetch. The parameter is unused by the base today, but kept on the signature for subclasses that may override based on payload state (e.g. a future transfer coordinator polling faster while a transfer is active). Documented in a docstring. The `T` TypeVar gains `JobInfo | None` as a valid binding so `JobUpdateCoordinator[JobInfo | None]` type-checks. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../components/prusalink/coordinator.py | 29 +++++++++++++++---- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/homeassistant/components/prusalink/coordinator.py b/homeassistant/components/prusalink/coordinator.py index 9f97a39c7e6963..50ffe09076342f 100644 --- a/homeassistant/components/prusalink/coordinator.py +++ b/homeassistant/components/prusalink/coordinator.py @@ -31,7 +31,12 @@ # 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. +T = TypeVar( + "T", PrinterStatus, LegacyPrinterStatus, JobInfo | None, PrinterInfo, VersionInfo +) type PrusaLinkConfigEntry = ConfigEntry[dict[str, PrusaLinkUpdateCoordinator]] @@ -85,8 +90,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 +121,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() From da57938e6edf2589a6f2a316bdec7e8492df9662 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:23:00 +0200 Subject: [PATCH 03/35] prusalink: cast undocumented `original` version field for MK3/MK2.5 workaround `version.get("original", "")` looks up a field that is not declared on the `VersionInfo` TypedDict because `original` is an undocumented field returned only by older standalone PrusaLink 0.7.2 builds on MK3 and MK2.5 printers. With pyprusalink 3.0.0's py.typed marker active, mypy narrows the .get() default-fallback return type to `object | str` and correctly objects to `.startswith` on `object`. Cast to `str` since we know what the printer actually returns when the field is present. The runtime behaviour is unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/config_flow.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py index 29e762a823dd35..9b270b8ced2809 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 @@ -40,8 +40,11 @@ def ensure_printer_is_supported(version: VersionInfo) -> None: return # 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( + # the 2.0.0 API, but doesn't advertise it yet. `original` is an + # undocumented field returned by older standalone PrusaLink builds; + # it is not part of VersionInfo, hence the cast. + original = cast(str, version.get("original", "")) + if original.startswith( ("PrusaLink I3MK3", "PrusaLink I3MK2") ) and AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]): return From 7c6a8a3ea3bcf27fa48f0539896d290c1db46dd1 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:34:54 +0200 Subject: [PATCH 04/35] prusalink: skip available_fn when coordinator data is None pyprusalink 3.0.0 changes `get_job()` to return `None` on HTTP 204 when no job is running, instead of casting an empty dict to JobInfo. The job coordinator's `data` therefore becomes `None` whenever the printer is not running a job. Every entity's `available_fn` lambda expects a dict (uses `.get()` or direct indexing). Without this guard, accessing the job-coordinator- backed entities while the printer is idle raised AttributeError. Short-circuit at the base entity so the lambdas never see `None`. Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/entity.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) 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 From a9df9383db73c7b037fc49c5a528952fbee023ba Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:35:36 +0200 Subject: [PATCH 05/35] prusalink tests: mock get_job() returning None for the idle case MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `mock_job_api_idle` fixture was returning `{}` — the value 2.x's `get_job()` produced on HTTP 204. With pyprusalink 3.0.0 the method returns `None` instead, so the mock needs to match. With the new mock, the test suite exercises the `coordinator.data is None` path in `PrusaLinkEntity.available` introduced in the previous commit. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/components/prusalink/conftest.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index c94b98f3ce0926..96113f82552b30 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 From e0f7fee68bf107042cdd05060777c2954e4dc77d Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:38:09 +0200 Subject: [PATCH 06/35] prusalink: address typing surfaced by pyprusalink 3.0.0's py.typed marker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyprusalink 3.0.0 ships the PEP 561 `py.typed` marker so the library's declared types now apply to HA's mypy run. Five issues surfaced — two are real fixes, two are pre-existing debt routed through `# type: ignore` for a focused followup, and one is a redundant cast we can just drop. - printer.state sensor: rearrange `cast(str, …).lower()` parentheses so the cast wraps the value before `.lower()` is called. Runtime is unchanged; mypy now sees `str.lower()` instead of `PrinterState.lower()`. - printer.telemetry.material sensor: the `LegacyPrinterStatus.telemetry` field is `LegacyPrinterTelemetry | None`, so `data["telemetry"] ["material"]` could KeyError at runtime when `telemetry` was None. Real latent bug: add `available_fn` that gates on `telemetry`, and cast the inner Optional to satisfy mypy. - job.filename and job.finish sensors: `available_fn` already gates against None / Optional, but mypy doesn't see through the guarantee into the lambdas. `# type: ignore` with a TODO comment so a focused followup PR can replace them with proper narrowing. - info.min_extrusion_temp sensor: `min_extrusion_temp` was migrated from `int | None` to `NotRequired[int]` in pyprusalink 3.0.0; the cast that previously narrowed `int | None` to `int` is now redundant and can be dropped. Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/sensor.py | 23 +++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index ce38101dcdcabc..3de913f281b6f9 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -6,7 +6,7 @@ from typing import cast from pyprusalink.types import JobInfo, PrinterInfo, PrinterState, PrinterStatus -from pyprusalink.types_legacy import LegacyPrinterStatus +from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry from homeassistant.components.sensor import ( SensorDeviceClass, @@ -47,7 +47,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 +149,12 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[LegacyPrinterStatus]( key="printer.telemetry.material", translation_key="material", - value_fn=lambda data: cast(str, data["telemetry"]["material"]), + # `available_fn` guarantees `telemetry` is not None at this + # point; the inner cast narrows the Optional for the index. + value_fn=lambda data: cast( + str, cast(LegacyPrinterTelemetry, data["telemetry"])["material"] + ), + available_fn=lambda data: data.get("telemetry") is not None, ), ), "job": ( @@ -166,7 +171,10 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[JobInfo]( key="job.filename", translation_key="filename", - value_fn=lambda data: cast(str, data["file"]["display_name"]), + # available_fn ensures `file` is not None; mypy doesn't follow + # that guarantee into the lambda. TODO: tighten in a followup + # PR (latent typing debt surfaced by pyprusalink 3.0.0 py.typed). + value_fn=lambda data: cast(str, data["file"]["display_name"]), # type: ignore[index] available_fn=lambda data: ( data.get("file") is not None and data.get("state") != PrinterState.IDLE.value @@ -189,8 +197,11 @@ class PrusaLinkSensorEntityDescription[ key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, + # available_fn ensures `time_remaining` is not None; mypy doesn't + # follow the guarantee. TODO: tighten in a followup PR (latent + # typing debt surfaced by pyprusalink 3.0.0 py.typed). value_fn=ignore_variance( - lambda data: utcnow() + timedelta(seconds=data["time_remaining"]), + lambda data: utcnow() + timedelta(seconds=data["time_remaining"]), # type: ignore[arg-type] timedelta(minutes=2), ), available_fn=lambda data: ( @@ -213,7 +224,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, ), From 60d4585b09baf691d060290417aa7b90042ee665 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:47:54 +0200 Subject: [PATCH 07/35] prusalink: replace type:ignore on job sensors with proper narrowing Follow-up to #170480, addressing the two `# type: ignore` comments that PR added to the `job.filename` and `job.finish` sensors. Both lambdas are guarded at runtime by `available_fn` (verifying `data["file"]` and `data["time_remaining"]` are not None respectively), but mypy doesn't follow the guarantee from `available_fn` into `value_fn`. Switch to the inner-cast pattern already used on the `printer.telemetry.material` sensor: - `job.filename`: `cast(JobFilePrint, data["file"])["display_name"]` narrows `JobFilePrint | None` so the index is type-safe. - `job.finish`: `cast(int, data["time_remaining"])` narrows `int | None` so it can be passed to `timedelta(seconds=...)`. Runtime behaviour is unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/sensor.py | 26 +++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 3de913f281b6f9..8c007145481729 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -5,7 +5,13 @@ from datetime import datetime, timedelta from typing import cast -from pyprusalink.types import JobInfo, PrinterInfo, PrinterState, PrinterStatus +from pyprusalink.types import ( + JobFilePrint, + JobInfo, + PrinterInfo, + PrinterState, + PrinterStatus, +) from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry from homeassistant.components.sensor import ( @@ -171,10 +177,11 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[JobInfo]( key="job.filename", translation_key="filename", - # available_fn ensures `file` is not None; mypy doesn't follow - # that guarantee into the lambda. TODO: tighten in a followup - # PR (latent typing debt surfaced by pyprusalink 3.0.0 py.typed). - value_fn=lambda data: cast(str, data["file"]["display_name"]), # type: ignore[index] + # `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 @@ -197,11 +204,12 @@ class PrusaLinkSensorEntityDescription[ key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, - # available_fn ensures `time_remaining` is not None; mypy doesn't - # follow the guarantee. TODO: tighten in a followup PR (latent - # typing debt surfaced by pyprusalink 3.0.0 py.typed). + # `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"]), # type: ignore[arg-type] + lambda data: ( + utcnow() + timedelta(seconds=cast(int, data["time_remaining"])) + ), timedelta(minutes=2), ), available_fn=lambda data: ( From 355cba5edb137c262010fb40b18bf0b7a840870d Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 13:52:35 +0200 Subject: [PATCH 08/35] prusalink tests: type mock_api parameter to match idle-job mock yield `mock_job_api_idle` was updated to yield `None` (matching pyprusalink 3.0.0's `get_job()`), but the dependent `mock_api` fixture's parameter annotation still said `dict[str, Any]`. Tighten it to `None` so the type contract matches. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/components/prusalink/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index 96113f82552b30..53632e34270b0f 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -209,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.""" From d82490947710113e78465c27a786232420e645f6 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Wed, 13 May 2026 14:29:37 +0200 Subject: [PATCH 09/35] prusalink: switch coordinator TypeVar from constraint to bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Copilot's review: a constraint TypeVar with `JobInfo | None` as one of the members is an unusual pattern — it binds T to the union as a whole, drops the ability to parameterise on `JobInfo` alone, and is inconsistent with how HA core typically handles unions (`bound=` over a union, not union-as-constraint-member). Switch to `bound=` over the same six types. Functionally equivalent for our subclasses (each still parameterises with its concrete type), removes the asymmetry-of-shape in the constraint list, and aligns with the HA-core idiom. Co-Authored-By: Claude Opus 4.7 (1M context) --- homeassistant/components/prusalink/coordinator.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/prusalink/coordinator.py b/homeassistant/components/prusalink/coordinator.py index 50ffe09076342f..24b54032e2061d 100644 --- a/homeassistant/components/prusalink/coordinator.py +++ b/homeassistant/components/prusalink/coordinator.py @@ -33,9 +33,17 @@ # 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. +# 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", PrinterStatus, LegacyPrinterStatus, JobInfo | None, PrinterInfo, VersionInfo + "T", + bound=PrinterStatus + | LegacyPrinterStatus + | JobInfo + | None + | PrinterInfo + | VersionInfo, ) From 647895ab0644edf447acc723a139a49ea7cd754d Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 14:29:42 +0200 Subject: [PATCH 10/35] prusalink: fix mock_job_api_idle Generator type annotation --- tests/components/prusalink/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index 53632e34270b0f..aa72b4fc608e56 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -123,7 +123,7 @@ def mock_get_status_printing() -> Generator[dict[str, Any]]: @pytest.fixture -def mock_job_api_idle() -> Generator[None]: +def mock_job_api_idle() -> Generator[None, None, None]: """Mock PrusaLink job API having no job. pyprusalink >= 3.0.0 returns `None` from `get_job()` on HTTP 204 when From 086b43e42ae425196b24cdfb17eafb37cc888083 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 14:49:00 +0200 Subject: [PATCH 11/35] prusalink: use unknown state for job sensors when idle or no job Job sensors (progress, filename, print_start, print_finish) now return None native_value when printer is idle or has no active job, which Home Assistant interprets as 'unknown' state. This differentiates from 'unavailable' which is reserved for when coordinator.data is None (connectivity issues). Changes: - Refactor entity.available to call available_fn with coordinator.data (which may be None), allowing available_fn to implement None-safe logic - Add helper functions (_has_active_job, _job_progress, _job_filename, _job_start, _job_finish) that handle None data gracefully - Job sensors use available_fn=lambda _: True to remain available while returning None values via value_fn when idle - Update test assertions to expect 'unknown' instead of 'unavailable' for idle job sensors - Update sensor entity description generic type to accept JobInfo | None --- homeassistant/components/prusalink/camera.py | 3 +- homeassistant/components/prusalink/entity.py | 11 +-- homeassistant/components/prusalink/sensor.py | 94 ++++++++++---------- tests/components/prusalink/test_sensor.py | 16 ++-- 4 files changed, 59 insertions(+), 65 deletions(-) diff --git a/homeassistant/components/prusalink/camera.py b/homeassistant/components/prusalink/camera.py index 6b21c28220065a..bf9dcacf22e1e2 100644 --- a/homeassistant/components/prusalink/camera.py +++ b/homeassistant/components/prusalink/camera.py @@ -36,7 +36,8 @@ class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera): key="job_preview", translation_key="job_preview", available_fn=lambda data: bool( - data.get("state") != PrinterState.IDLE.value + data is not None + and data.get("state") != PrinterState.IDLE.value and (file := data.get("file")) and file.get("refs", {}).get("thumbnail") ), diff --git a/homeassistant/components/prusalink/entity.py b/homeassistant/components/prusalink/entity.py index b36b455cbbdae4..c0003c0e7e1f51 100644 --- a/homeassistant/components/prusalink/entity.py +++ b/homeassistant/components/prusalink/entity.py @@ -29,15 +29,8 @@ class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]): @property def available(self) -> bool: """Return if entity is available.""" - # `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) + return super().available and self.entity_description.available_fn( + self.coordinator.data ) @property diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 8c007145481729..909ad4e326775b 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -5,13 +5,7 @@ from datetime import datetime, timedelta from typing import cast -from pyprusalink.types import ( - JobFilePrint, - JobInfo, - PrinterInfo, - PrinterState, - PrinterStatus, -) +from pyprusalink.types import JobInfo, PrinterInfo, PrinterState, PrinterStatus from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry from homeassistant.components.sensor import ( @@ -30,15 +24,47 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.typing import StateType from homeassistant.util.dt import utcnow -from homeassistant.util.variance import ignore_variance from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator from .entity import PrusaLinkEntity, PrusaLinkEntityDescription +def _job_progress(data: JobInfo | None) -> float | None: + """Return job progress or None if no active job is running.""" + if data is None or data.get("state") == PrinterState.IDLE.value: + return None + return data["progress"] + + +def _job_filename(data: JobInfo | None) -> str | None: + """Return job filename or None if no active job is running.""" + if data is None or data.get("state") == PrinterState.IDLE.value: + return None + file_data = data["file"] + if file_data is None: + return None + return file_data["display_name"] + + +def _job_start(data: JobInfo | None) -> datetime | None: + """Return print start timestamp or None if no active job is running.""" + if data is None or data.get("state") == PrinterState.IDLE.value: + return None + return utcnow() - timedelta(seconds=data["time_printing"]) + + +def _job_finish(data: JobInfo | None) -> datetime | None: + """Return print finish timestamp or None if no active job is running.""" + if data is None or data.get("state") == PrinterState.IDLE.value: + return None + time_remaining = data["time_remaining"] + if time_remaining is None: + return None + return utcnow() + timedelta(seconds=time_remaining) + @dataclass(frozen=True, kw_only=True) class PrusaLinkSensorEntityDescription[ - T: (PrinterStatus, LegacyPrinterStatus, JobInfo, PrinterInfo) + T: PrinterStatus | LegacyPrinterStatus | JobInfo | None | PrinterInfo ]( SensorEntityDescription, PrusaLinkEntityDescription, @@ -164,58 +190,32 @@ class PrusaLinkSensorEntityDescription[ ), ), "job": ( - PrusaLinkSensorEntityDescription[JobInfo]( + PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.progress", translation_key="progress", native_unit_of_measurement=PERCENTAGE, - value_fn=lambda data: cast(float, data["progress"]), - available_fn=lambda data: ( - data.get("progress") is not None - and data.get("state") != PrinterState.IDLE.value - ), + value_fn=_job_progress, + available_fn=lambda _: True, ), - PrusaLinkSensorEntityDescription[JobInfo]( + PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.filename", translation_key="filename", - # `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 - ), + value_fn=_job_filename, + available_fn=lambda _: True, ), - PrusaLinkSensorEntityDescription[JobInfo]( + PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.start", translation_key="print_start", device_class=SensorDeviceClass.TIMESTAMP, - value_fn=ignore_variance( - lambda data: utcnow() - timedelta(seconds=data["time_printing"]), - timedelta(minutes=2), - ), - available_fn=lambda data: ( - data.get("time_printing") is not None - and data.get("state") != PrinterState.IDLE.value - ), + value_fn=_job_start, + available_fn=lambda _: True, ), - PrusaLinkSensorEntityDescription[JobInfo]( + PrusaLinkSensorEntityDescription[JobInfo | None]( 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=cast(int, data["time_remaining"])) - ), - timedelta(minutes=2), - ), - available_fn=lambda data: ( - data.get("time_remaining") is not None - and data.get("state") != PrinterState.IDLE.value - ), + value_fn=_job_finish, + available_fn=lambda _: True, ), ), "info": ( diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index 5c4ee7f87f8d16..30f41c06fe74b9 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -110,21 +110,21 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api) state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") @@ -219,21 +219,21 @@ async def test_sensors_idle_job_mk3( state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == "unavailable" + assert state.state == "unknown" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") From 529432f0a8a44ba98ffbbdf39f7351f34e19751e Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 16:19:41 +0200 Subject: [PATCH 12/35] prusalink tests: use Generator[None] for idle job fixture --- tests/components/prusalink/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index aa72b4fc608e56..53632e34270b0f 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -123,7 +123,7 @@ def mock_get_status_printing() -> Generator[dict[str, Any]]: @pytest.fixture -def mock_job_api_idle() -> Generator[None, None, None]: +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 From 8a5ace6b2e9ae0ef1dfec1f1a7446f3b97774cda Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 16:37:13 +0200 Subject: [PATCH 13/35] prusalink: restore variance suppression for job timestamp sensors --- homeassistant/components/prusalink/sensor.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 909ad4e326775b..40b898e4338668 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -24,10 +24,20 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.typing import StateType from homeassistant.util.dt import utcnow +from homeassistant.util.variance import ignore_variance from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator from .entity import PrusaLinkEntity, PrusaLinkEntityDescription +_stable_job_start = ignore_variance( + lambda printing_seconds: utcnow() - timedelta(seconds=printing_seconds), + timedelta(minutes=2), +) +_stable_job_finish = ignore_variance( + lambda remaining_seconds: utcnow() + timedelta(seconds=remaining_seconds), + timedelta(minutes=2), +) + def _job_progress(data: JobInfo | None) -> float | None: """Return job progress or None if no active job is running.""" @@ -50,7 +60,7 @@ def _job_start(data: JobInfo | None) -> datetime | None: """Return print start timestamp or None if no active job is running.""" if data is None or data.get("state") == PrinterState.IDLE.value: return None - return utcnow() - timedelta(seconds=data["time_printing"]) + return _stable_job_start(data["time_printing"]) def _job_finish(data: JobInfo | None) -> datetime | None: @@ -60,7 +70,8 @@ def _job_finish(data: JobInfo | None) -> datetime | None: time_remaining = data["time_remaining"] if time_remaining is None: return None - return utcnow() + timedelta(seconds=time_remaining) + return _stable_job_finish(time_remaining) + @dataclass(frozen=True, kw_only=True) class PrusaLinkSensorEntityDescription[ From b04678498773d2facf384d4fd8d2ed2592ac3b17 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 17:05:25 +0200 Subject: [PATCH 14/35] prusalink tests: use Iterator for idle job fixture type --- tests/components/prusalink/conftest.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index 53632e34270b0f..ad1b31915917a0 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -1,6 +1,6 @@ """Fixtures for PrusaLink.""" -from collections.abc import Generator +from collections.abc import Generator, Iterator from typing import Any from unittest.mock import patch @@ -123,7 +123,7 @@ def mock_get_status_printing() -> Generator[dict[str, Any]]: @pytest.fixture -def mock_job_api_idle() -> Generator[None]: +def mock_job_api_idle() -> Iterator[None]: """Mock PrusaLink job API having no job. pyprusalink >= 3.0.0 returns `None` from `get_job()` on HTTP 204 when From ebdb1d5d7612258e8883a019896a860136f560d5 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 17:20:52 +0200 Subject: [PATCH 15/35] prusalink: refactor job-sensor helpers to use _has_active_job predicate --- homeassistant/components/prusalink/sensor.py | 23 +++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 40b898e4338668..6a3a9095151321 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -39,18 +39,25 @@ ) +def _has_active_job(data: JobInfo | None) -> JobInfo | None: + """Return job payload if there is an active job, otherwise None.""" + if data is None or data.get("state") == PrinterState.IDLE.value: + return None + return data + + def _job_progress(data: JobInfo | None) -> float | None: """Return job progress or None if no active job is running.""" - if data is None or data.get("state") == PrinterState.IDLE.value: + if (active_job := _has_active_job(data)) is None: return None - return data["progress"] + return active_job["progress"] def _job_filename(data: JobInfo | None) -> str | None: """Return job filename or None if no active job is running.""" - if data is None or data.get("state") == PrinterState.IDLE.value: + if (active_job := _has_active_job(data)) is None: return None - file_data = data["file"] + file_data = active_job["file"] if file_data is None: return None return file_data["display_name"] @@ -58,16 +65,16 @@ def _job_filename(data: JobInfo | None) -> str | None: def _job_start(data: JobInfo | None) -> datetime | None: """Return print start timestamp or None if no active job is running.""" - if data is None or data.get("state") == PrinterState.IDLE.value: + if (active_job := _has_active_job(data)) is None: return None - return _stable_job_start(data["time_printing"]) + return _stable_job_start(active_job["time_printing"]) def _job_finish(data: JobInfo | None) -> datetime | None: """Return print finish timestamp or None if no active job is running.""" - if data is None or data.get("state") == PrinterState.IDLE.value: + if (active_job := _has_active_job(data)) is None: return None - time_remaining = data["time_remaining"] + time_remaining = active_job["time_remaining"] if time_remaining is None: return None return _stable_job_finish(time_remaining) From 4dda9725a9e105d109ef12ae1248ed6d904f7e15 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 17:35:16 +0200 Subject: [PATCH 16/35] prusalink tests: add test for active job with nullable fields --- tests/components/prusalink/conftest.py | 22 +++++++++++ tests/components/prusalink/test_sensor.py | 47 +++++++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index ad1b31915917a0..f2a4cf83a8e209 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -203,6 +203,28 @@ def mock_job_api_attention( mock_get_status_printing["printer"]["state"] = "ATTENTION" +@pytest.fixture +def mock_job_api_printing_with_nullable_fields() -> Generator[dict[str, Any]]: + """Mock PrusaLink printing with missing/None nullable fields. + + Tests that job helpers handle active jobs where file or time_remaining + are None (nullable fields), returning None without raising and allowing + sensors to show unknown state. + """ + resp = { + "id": 129, + "state": "PRINTING", + "progress": 45.00, + "time_remaining": None, # Missing finish time + "time_printing": 43987, + "file": None, # No file info + "serial_print": False, + "inaccurate_estimates": True, + } + with patch("pyprusalink.PrusaLink.get_job", return_value=resp): + yield resp + + @pytest.fixture def mock_api( mock_version_api: dict[str, str], diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index 30f41c06fe74b9..962286fde00d15 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -296,6 +296,53 @@ async def test_sensors_active_job( assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_sensors_active_job_with_nullable_fields( + hass: HomeAssistant, + mock_config_entry, + mock_api, + mock_get_status_printing, + mock_job_api_printing_with_nullable_fields, +) -> None: + """Test job sensors with active job but missing nullable fields. + + Verifies that job helpers handle None values for nullable fields (file, + time_remaining) without raising, returning None to indicate unknown state. + """ + with patch( + "homeassistant.components.prusalink.sensor.utcnow", + return_value=datetime(2022, 8, 27, 14, 0, 0, tzinfo=UTC), + ): + assert await async_setup_component(hass, "prusalink", {}) + + state = hass.states.get("sensor.mock_title") + assert state is not None + assert state.state == "printing" + + # progress is required, should always have a value + state = hass.states.get("sensor.mock_title_progress") + assert state is not None + assert state.state == "45.0" + assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" + + # file is None, filename helper should return None -> unknown + state = hass.states.get("sensor.mock_title_filename") + assert state is not None + assert state.state == "unknown" + + # time_printing is required, print_start should always have a value + state = hass.states.get("sensor.mock_title_print_start") + assert state is not None + assert state.state == "2022-08-27T01:46:53+00:00" + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP + + # time_remaining is None, print_finish helper should return None -> unknown + state = hass.states.get("sensor.mock_title_print_finish") + assert state is not None + assert state.state == "unknown" + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP + + @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_axis_x_y_sensors( hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: None From 873d3bbd4627ce28e8c3b8c69c3105d8510c1052 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:02:11 +0200 Subject: [PATCH 17/35] prusalink: add review-driven docs and telemetry regression test --- homeassistant/components/prusalink/sensor.py | 4 ++++ tests/components/prusalink/conftest.py | 3 +++ tests/components/prusalink/test_sensor.py | 17 +++++++++++++++++ 3 files changed, 24 insertions(+) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 6a3a9095151321..db4513b686de24 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -50,6 +50,8 @@ def _job_progress(data: JobInfo | None) -> float | None: """Return job progress or None if no active job is running.""" if (active_job := _has_active_job(data)) is None: return None + # Required JobInfo fields are intentionally accessed directly so upstream + # contract violations fail fast instead of being silently masked. return active_job["progress"] @@ -213,6 +215,8 @@ class PrusaLinkSensorEntityDescription[ translation_key="progress", native_unit_of_measurement=PERCENTAGE, value_fn=_job_progress, + # Job sensors stay available when idle/no-job so `None` values are + # represented as `unknown` instead of `unavailable`. available_fn=lambda _: True, ), PrusaLinkSensorEntityDescription[JobInfo | None]( diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index f2a4cf83a8e209..f050978d74747b 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -128,6 +128,9 @@ def mock_job_api_idle() -> Iterator[None]: 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. + + Iterator is intentional for this yield fixture: it matches pytest usage + while avoiding unnecessary Generator send/return type parameters. """ with patch("pyprusalink.PrusaLink.get_job", return_value=None): yield None diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index 962286fde00d15..b946435a9c1959 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -343,6 +343,23 @@ async def test_sensors_active_job_with_nullable_fields( assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_material_sensor_unavailable_when_legacy_telemetry_missing( + hass: HomeAssistant, + mock_config_entry, + mock_api, + mock_get_legacy_printer: dict[str, Any], +) -> None: + """Material sensor is unavailable when legacy telemetry is missing.""" + mock_get_legacy_printer["telemetry"] = None + + assert await async_setup_component(hass, "prusalink", {}) + + state = hass.states.get("sensor.mock_title_material") + assert state is not None + assert state.state == "unavailable" + + @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_axis_x_y_sensors( hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: None From 4cd88c598bcf414fcaa83c3f9636a4e4cc4c3741 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:17:24 +0200 Subject: [PATCH 18/35] prusalink: harden mk3 original-version workaround --- .../components/prusalink/config_flow.py | 7 +++--- .../components/prusalink/test_config_flow.py | 25 +++++++++++++++++++ 2 files changed, 29 insertions(+), 3 deletions(-) diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py index 9b270b8ced2809..6313c1a67f0459 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, cast +from typing import Any from awesomeversion import AwesomeVersion, AwesomeVersionException from httpx import HTTPError, InvalidURL @@ -42,8 +42,9 @@ 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. `original` is an # undocumented field returned by older standalone PrusaLink builds; - # it is not part of VersionInfo, hence the cast. - original = cast(str, version.get("original", "")) + # it is not part of VersionInfo. + original_val = version.get("original") + original = original_val if isinstance(original_val, str) else "" if original.startswith( ("PrusaLink I3MK3", "PrusaLink I3MK2") ) and AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]): diff --git a/tests/components/prusalink/test_config_flow.py b/tests/components/prusalink/test_config_flow.py index cc66d25b35d036..5a71fb1551839b 100644 --- a/tests/components/prusalink/test_config_flow.py +++ b/tests/components/prusalink/test_config_flow.py @@ -182,6 +182,31 @@ async def test_form_invalid_mk3_server_version( assert result2["errors"] == {"base": "not_supported"} +async def test_form_mk3_original_none( + hass: HomeAssistant, mock_version_api +) -> None: + """Test MK2/MK3 workaround path handles original=None safely.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + mock_version_api["api"] = "0.9.0-legacy" + mock_version_api["server"] = "0.7.2" + mock_version_api["original"] = None + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "1.1.1.1", + "username": "abcdefg", + "password": "abcdefg", + }, + ) + + assert result2["type"] is FlowResultType.FORM + assert result2["errors"] == {"base": "not_supported"} + + async def test_form_cannot_connect(hass: HomeAssistant) -> None: """Test we handle cannot connect error.""" result = await hass.config_entries.flow.async_init( From 1533a4fdfd6abd5039169c51b8c66669894da178 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:25:01 +0200 Subject: [PATCH 19/35] prusalink: harden legacy material sensor value extraction --- homeassistant/components/prusalink/sensor.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index db4513b686de24..1f72a9162f64d5 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -6,7 +6,7 @@ from typing import cast from pyprusalink.types import JobInfo, PrinterInfo, PrinterState, PrinterStatus -from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry +from pyprusalink.types_legacy import LegacyPrinterStatus from homeassistant.components.sensor import ( SensorDeviceClass, @@ -82,6 +82,14 @@ def _job_finish(data: JobInfo | None) -> datetime | None: return _stable_job_finish(time_remaining) +def _legacy_material(data: LegacyPrinterStatus) -> str | None: + """Return material name or None when legacy telemetry is missing.""" + telemetry = data.get("telemetry") + if telemetry is None: + return None + return telemetry["material"] + + @dataclass(frozen=True, kw_only=True) class PrusaLinkSensorEntityDescription[ T: PrinterStatus | LegacyPrinterStatus | JobInfo | None | PrinterInfo @@ -201,11 +209,7 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[LegacyPrinterStatus]( key="printer.telemetry.material", translation_key="material", - # `available_fn` guarantees `telemetry` is not None at this - # point; the inner cast narrows the Optional for the index. - value_fn=lambda data: cast( - str, cast(LegacyPrinterTelemetry, data["telemetry"])["material"] - ), + value_fn=_legacy_material, available_fn=lambda data: data.get("telemetry") is not None, ), ), From 7301badb267d505c715c3cbd8a4119a354387b69 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:32:47 +0200 Subject: [PATCH 20/35] prusalink tests: use state constants in sensor assertions --- tests/components/prusalink/test_sensor.py | 24 ++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index b946435a9c1959..c749ea6ee5a6b1 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -17,6 +17,8 @@ ATTR_UNIT_OF_MEASUREMENT, PERCENTAGE, REVOLUTIONS_PER_MINUTE, + STATE_UNAVAILABLE, + STATE_UNKNOWN, Platform, UnitOfLength, UnitOfTemperature, @@ -110,21 +112,21 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api) state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") @@ -219,21 +221,21 @@ async def test_sensors_idle_job_mk3( state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") @@ -328,7 +330,7 @@ async def test_sensors_active_job_with_nullable_fields( # file is None, filename helper should return None -> unknown state = hass.states.get("sensor.mock_title_filename") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN # time_printing is required, print_start should always have a value state = hass.states.get("sensor.mock_title_print_start") @@ -339,7 +341,7 @@ async def test_sensors_active_job_with_nullable_fields( # time_remaining is None, print_finish helper should return None -> unknown state = hass.states.get("sensor.mock_title_print_finish") assert state is not None - assert state.state == "unknown" + assert state.state == STATE_UNKNOWN assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP @@ -357,7 +359,7 @@ async def test_material_sensor_unavailable_when_legacy_telemetry_missing( state = hass.states.get("sensor.mock_title_material") assert state is not None - assert state.state == "unavailable" + assert state.state == STATE_UNAVAILABLE @pytest.mark.usefixtures("entity_registry_enabled_by_default") From 29ed1eaf68fb33e592f24de60cbe5d4c619d1aca Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:35:27 +0200 Subject: [PATCH 21/35] prusalink: deduplicate always-available job sensor callable --- homeassistant/components/prusalink/sensor.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 1f72a9162f64d5..c3f1ca0e7bee9a 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -90,6 +90,11 @@ def _legacy_material(data: LegacyPrinterStatus) -> str | None: return telemetry["material"] +def _always_available(_: object) -> bool: + """Keep entities available while coordinator is connected.""" + return True + + @dataclass(frozen=True, kw_only=True) class PrusaLinkSensorEntityDescription[ T: PrinterStatus | LegacyPrinterStatus | JobInfo | None | PrinterInfo @@ -221,27 +226,27 @@ class PrusaLinkSensorEntityDescription[ value_fn=_job_progress, # Job sensors stay available when idle/no-job so `None` values are # represented as `unknown` instead of `unavailable`. - available_fn=lambda _: True, + available_fn=_always_available, ), PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.filename", translation_key="filename", value_fn=_job_filename, - available_fn=lambda _: True, + available_fn=_always_available, ), PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.start", translation_key="print_start", device_class=SensorDeviceClass.TIMESTAMP, value_fn=_job_start, - available_fn=lambda _: True, + available_fn=_always_available, ), PrusaLinkSensorEntityDescription[JobInfo | None]( key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, value_fn=_job_finish, - available_fn=lambda _: True, + available_fn=_always_available, ), ), "info": ( From 2a3e06e2dbbc1f17c86681b6fddbf2edf3fee224 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 18:49:49 +0200 Subject: [PATCH 22/35] prusalink tests: apply ruff formatting in config flow test --- tests/components/prusalink/test_config_flow.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/components/prusalink/test_config_flow.py b/tests/components/prusalink/test_config_flow.py index 5a71fb1551839b..c069f800619686 100644 --- a/tests/components/prusalink/test_config_flow.py +++ b/tests/components/prusalink/test_config_flow.py @@ -182,9 +182,7 @@ async def test_form_invalid_mk3_server_version( assert result2["errors"] == {"base": "not_supported"} -async def test_form_mk3_original_none( - hass: HomeAssistant, mock_version_api -) -> None: +async def test_form_mk3_original_none(hass: HomeAssistant, mock_version_api) -> None: """Test MK2/MK3 workaround path handles original=None safely.""" result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} From 77af260993c9b4043be6007fa637781d1b34c346 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 19:09:09 +0200 Subject: [PATCH 23/35] prusalink: make job timestamp stabilization per-entity --- homeassistant/components/prusalink/sensor.py | 62 ++++++++++++++++---- 1 file changed, 50 insertions(+), 12 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index c3f1ca0e7bee9a..59f282adaa0a18 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -29,15 +29,6 @@ from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator from .entity import PrusaLinkEntity, PrusaLinkEntityDescription -_stable_job_start = ignore_variance( - lambda printing_seconds: utcnow() - timedelta(seconds=printing_seconds), - timedelta(minutes=2), -) -_stable_job_finish = ignore_variance( - lambda remaining_seconds: utcnow() + timedelta(seconds=remaining_seconds), - timedelta(minutes=2), -) - def _has_active_job(data: JobInfo | None) -> JobInfo | None: """Return job payload if there is an active job, otherwise None.""" @@ -69,7 +60,7 @@ def _job_start(data: JobInfo | None) -> datetime | None: """Return print start timestamp or None if no active job is running.""" if (active_job := _has_active_job(data)) is None: return None - return _stable_job_start(active_job["time_printing"]) + return utcnow() - timedelta(seconds=active_job["time_printing"]) def _job_finish(data: JobInfo | None) -> datetime | None: @@ -79,7 +70,40 @@ def _job_finish(data: JobInfo | None) -> datetime | None: time_remaining = active_job["time_remaining"] if time_remaining is None: return None - return _stable_job_finish(time_remaining) + return utcnow() + timedelta(seconds=time_remaining) + + +def _make_stable_job_start() -> Callable[[JobInfo | None], datetime | None]: + """Return a per-entity stable job-start value function.""" + stable_job_start = ignore_variance( + lambda printing_seconds: utcnow() - timedelta(seconds=printing_seconds), + timedelta(minutes=2), + ) + + def _value_fn(data: JobInfo | None) -> datetime | None: + if (active_job := _has_active_job(data)) is None: + return None + return stable_job_start(active_job["time_printing"]) + + return _value_fn + + +def _make_stable_job_finish() -> Callable[[JobInfo | None], datetime | None]: + """Return a per-entity stable job-finish value function.""" + stable_job_finish = ignore_variance( + lambda remaining_seconds: utcnow() + timedelta(seconds=remaining_seconds), + timedelta(minutes=2), + ) + + def _value_fn(data: JobInfo | None) -> datetime | None: + if (active_job := _has_active_job(data)) is None: + return None + time_remaining = active_job["time_remaining"] + if time_remaining is None: + return None + return stable_job_finish(time_remaining) + + return _value_fn def _legacy_material(data: LegacyPrinterStatus) -> str | None: @@ -105,6 +129,7 @@ class PrusaLinkSensorEntityDescription[ """Describes PrusaLink sensor entity.""" value_fn: Callable[[T], datetime | StateType] + value_fn_factory: Callable[[], Callable[[T], datetime | StateType]] | None = None SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = { @@ -239,6 +264,7 @@ class PrusaLinkSensorEntityDescription[ translation_key="print_start", device_class=SensorDeviceClass.TIMESTAMP, value_fn=_job_start, + value_fn_factory=_make_stable_job_start, available_fn=_always_available, ), PrusaLinkSensorEntityDescription[JobInfo | None]( @@ -246,6 +272,7 @@ class PrusaLinkSensorEntityDescription[ translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, value_fn=_job_finish, + value_fn_factory=_make_stable_job_finish, available_fn=_always_available, ), ), @@ -306,8 +333,19 @@ def __init__( super().__init__(coordinator=coordinator) self.entity_description = description self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{description.key}" + # Some sensors need a fresh callable per entity to avoid shared state + # across multiple entries (e.g., variance cache for timestamps). + if description.value_fn_factory is not None: + self._value_fn = cast( + Callable[[object], datetime | StateType], + description.value_fn_factory(), + ) + else: + self._value_fn = cast( + Callable[[object], datetime | StateType], description.value_fn + ) @property def native_value(self) -> datetime | StateType: """Return the state of the sensor.""" - return self.entity_description.value_fn(self.coordinator.data) + return self._value_fn(self.coordinator.data) From 03542e7d5d394cb3687eda264bc82dd930fb3d37 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 20:22:01 +0200 Subject: [PATCH 24/35] prusalink: clarify nullable-field behavior in job helper docstrings --- homeassistant/components/prusalink/sensor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 59f282adaa0a18..517df62438014c 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -47,7 +47,7 @@ def _job_progress(data: JobInfo | None) -> float | None: def _job_filename(data: JobInfo | None) -> str | None: - """Return job filename or None if no active job is running.""" + """Return job filename, or None if there is no active job or the active job has no file metadata.""" if (active_job := _has_active_job(data)) is None: return None file_data = active_job["file"] @@ -64,7 +64,7 @@ def _job_start(data: JobInfo | None) -> datetime | None: def _job_finish(data: JobInfo | None) -> datetime | None: - """Return print finish timestamp or None if no active job is running.""" + """Return print finish timestamp, or None if there is no active job or the active job has no remaining-time estimate.""" if (active_job := _has_active_job(data)) is None: return None time_remaining = active_job["time_remaining"] From d67329779459350db821dd7ff641832f2f331a54 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 20:24:33 +0200 Subject: [PATCH 25/35] prusalink: enforce exclusive value_fn and value_fn_factory --- homeassistant/components/prusalink/sensor.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 517df62438014c..9d6944306aa123 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -128,9 +128,16 @@ class PrusaLinkSensorEntityDescription[ ): """Describes PrusaLink sensor entity.""" - value_fn: Callable[[T], datetime | StateType] + value_fn: Callable[[T], datetime | StateType] | None = None value_fn_factory: Callable[[], Callable[[T], datetime | StateType]] | None = None + def __post_init__(self) -> None: + """Ensure exactly one value callable source is configured.""" + if (self.value_fn is None) == (self.value_fn_factory is None): + raise ValueError( + "Exactly one of value_fn or value_fn_factory must be provided" + ) + SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = { "status": ( @@ -263,7 +270,6 @@ class PrusaLinkSensorEntityDescription[ key="job.start", translation_key="print_start", device_class=SensorDeviceClass.TIMESTAMP, - value_fn=_job_start, value_fn_factory=_make_stable_job_start, available_fn=_always_available, ), @@ -271,7 +277,6 @@ class PrusaLinkSensorEntityDescription[ key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, - value_fn=_job_finish, value_fn_factory=_make_stable_job_finish, available_fn=_always_available, ), @@ -340,10 +345,14 @@ def __init__( Callable[[object], datetime | StateType], description.value_fn_factory(), ) - else: + elif description.value_fn is not None: self._value_fn = cast( Callable[[object], datetime | StateType], description.value_fn ) + else: + raise ValueError( + f"Missing value callable for sensor description: {description.key}" + ) @property def native_value(self) -> datetime | StateType: From e14183661eef715fc5f6af839da2c274f2ffe45c Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 20:25:45 +0200 Subject: [PATCH 26/35] prusalink tests: add two-config-entry timestamp isolation regression --- tests/components/prusalink/test_sensor.py | 48 ++++++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index c749ea6ee5a6b1..06100d1ba7ff63 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -2,10 +2,11 @@ from datetime import UTC, datetime from typing import Any -from unittest.mock import patch +from unittest.mock import Mock, patch import pytest +from homeassistant.components.prusalink import sensor as prusalink_sensor from homeassistant.components.sensor import ( ATTR_OPTIONS, ATTR_STATE_CLASS, @@ -298,6 +299,51 @@ async def test_sensors_active_job( assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE +async def test_job_start_stabilization_isolated_per_config_entry() -> None: + """Ensure job.start variance cache is not shared across config entries.""" + now = datetime(2022, 8, 27, 14, 0, 0, tzinfo=UTC) + + job_start_description = next( + description + for description in prusalink_sensor.SENSORS["job"] + if description.key == "job.start" + ) + + coordinator_one = Mock() + coordinator_one.config_entry = Mock(entry_id="entry_one") + coordinator_one.data = { + "state": "PRINTING", + "time_printing": 600, + "time_remaining": 0, + "progress": 0.0, + "file": None, + } + + coordinator_two = Mock() + coordinator_two.config_entry = Mock(entry_id="entry_two") + coordinator_two.data = { + "state": "PRINTING", + "time_printing": 660, + "time_remaining": 0, + "progress": 0.0, + "file": None, + } + + with patch("homeassistant.components.prusalink.sensor.utcnow", return_value=now): + entity_one = prusalink_sensor.PrusaLinkSensorEntity( + coordinator_one, job_start_description + ) + entity_two = prusalink_sensor.PrusaLinkSensorEntity( + coordinator_two, job_start_description + ) + + state_one = entity_one.native_value + state_two = entity_two.native_value + + assert state_one == datetime(2022, 8, 27, 13, 50, 0, tzinfo=UTC) + assert state_two == datetime(2022, 8, 27, 13, 49, 0, tzinfo=UTC) + + @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_sensors_active_job_with_nullable_fields( hass: HomeAssistant, From afa387024c788149d3bb4ef6a52279220f7c5bbf Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 21:12:08 +0200 Subject: [PATCH 27/35] prusalink: wrap job helper docstrings --- homeassistant/components/prusalink/sensor.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index 9d6944306aa123..b3bba0d2c3b9e6 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -47,7 +47,10 @@ def _job_progress(data: JobInfo | None) -> float | None: def _job_filename(data: JobInfo | None) -> str | None: - """Return job filename, or None if there is no active job or the active job has no file metadata.""" + """Return job filename. + + Return None if there is no active job or the active job has no file metadata. + """ if (active_job := _has_active_job(data)) is None: return None file_data = active_job["file"] @@ -64,7 +67,11 @@ def _job_start(data: JobInfo | None) -> datetime | None: def _job_finish(data: JobInfo | None) -> datetime | None: - """Return print finish timestamp, or None if there is no active job or the active job has no remaining-time estimate.""" + """Return print finish timestamp. + + Return None if there is no active job or the active job has no + remaining-time estimate. + """ if (active_job := _has_active_job(data)) is None: return None time_remaining = active_job["time_remaining"] From 3ea8d6ac297557b1b874f6119bd53bd724c5b2c7 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Sat, 16 May 2026 21:24:46 +0200 Subject: [PATCH 28/35] Remove unused job timestamp helpers in prusalink --- homeassistant/components/prusalink/sensor.py | 21 -------------------- 1 file changed, 21 deletions(-) diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py index b3bba0d2c3b9e6..00079251fcc5e2 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -59,27 +59,6 @@ def _job_filename(data: JobInfo | None) -> str | None: return file_data["display_name"] -def _job_start(data: JobInfo | None) -> datetime | None: - """Return print start timestamp or None if no active job is running.""" - if (active_job := _has_active_job(data)) is None: - return None - return utcnow() - timedelta(seconds=active_job["time_printing"]) - - -def _job_finish(data: JobInfo | None) -> datetime | None: - """Return print finish timestamp. - - Return None if there is no active job or the active job has no - remaining-time estimate. - """ - if (active_job := _has_active_job(data)) is None: - return None - time_remaining = active_job["time_remaining"] - if time_remaining is None: - return None - return utcnow() + timedelta(seconds=time_remaining) - - def _make_stable_job_start() -> Callable[[JobInfo | None], datetime | None]: """Return a per-entity stable job-start value function.""" stable_job_start = ignore_variance( From 3902167a6fbe33a10e7025734f5d518e6cd4b932 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 12:47:49 +0200 Subject: [PATCH 29/35] prusalink: finalize minimal pyprusalink 3.0 bump adjustments --- homeassistant/components/prusalink/camera.py | 3 +- homeassistant/components/prusalink/entity.py | 11 +- homeassistant/components/prusalink/sensor.py | 172 ++++++------------- tests/components/prusalink/conftest.py | 29 +--- tests/components/prusalink/test_sensor.py | 130 +------------- 5 files changed, 75 insertions(+), 270 deletions(-) diff --git a/homeassistant/components/prusalink/camera.py b/homeassistant/components/prusalink/camera.py index bf9dcacf22e1e2..6b21c28220065a 100644 --- a/homeassistant/components/prusalink/camera.py +++ b/homeassistant/components/prusalink/camera.py @@ -36,8 +36,7 @@ class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera): key="job_preview", translation_key="job_preview", available_fn=lambda data: bool( - data is not None - and data.get("state") != PrinterState.IDLE.value + data.get("state") != PrinterState.IDLE.value and (file := data.get("file")) and file.get("refs", {}).get("thumbnail") ), 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/sensor.py b/homeassistant/components/prusalink/sensor.py index 00079251fcc5e2..8c007145481729 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, @@ -30,99 +36,16 @@ from .entity import PrusaLinkEntity, PrusaLinkEntityDescription -def _has_active_job(data: JobInfo | None) -> JobInfo | None: - """Return job payload if there is an active job, otherwise None.""" - if data is None or data.get("state") == PrinterState.IDLE.value: - return None - return data - - -def _job_progress(data: JobInfo | None) -> float | None: - """Return job progress or None if no active job is running.""" - if (active_job := _has_active_job(data)) is None: - return None - # Required JobInfo fields are intentionally accessed directly so upstream - # contract violations fail fast instead of being silently masked. - return active_job["progress"] - - -def _job_filename(data: JobInfo | None) -> str | None: - """Return job filename. - - Return None if there is no active job or the active job has no file metadata. - """ - if (active_job := _has_active_job(data)) is None: - return None - file_data = active_job["file"] - if file_data is None: - return None - return file_data["display_name"] - - -def _make_stable_job_start() -> Callable[[JobInfo | None], datetime | None]: - """Return a per-entity stable job-start value function.""" - stable_job_start = ignore_variance( - lambda printing_seconds: utcnow() - timedelta(seconds=printing_seconds), - timedelta(minutes=2), - ) - - def _value_fn(data: JobInfo | None) -> datetime | None: - if (active_job := _has_active_job(data)) is None: - return None - return stable_job_start(active_job["time_printing"]) - - return _value_fn - - -def _make_stable_job_finish() -> Callable[[JobInfo | None], datetime | None]: - """Return a per-entity stable job-finish value function.""" - stable_job_finish = ignore_variance( - lambda remaining_seconds: utcnow() + timedelta(seconds=remaining_seconds), - timedelta(minutes=2), - ) - - def _value_fn(data: JobInfo | None) -> datetime | None: - if (active_job := _has_active_job(data)) is None: - return None - time_remaining = active_job["time_remaining"] - if time_remaining is None: - return None - return stable_job_finish(time_remaining) - - return _value_fn - - -def _legacy_material(data: LegacyPrinterStatus) -> str | None: - """Return material name or None when legacy telemetry is missing.""" - telemetry = data.get("telemetry") - if telemetry is None: - return None - return telemetry["material"] - - -def _always_available(_: object) -> bool: - """Keep entities available while coordinator is connected.""" - return True - - @dataclass(frozen=True, kw_only=True) class PrusaLinkSensorEntityDescription[ - T: PrinterStatus | LegacyPrinterStatus | JobInfo | None | PrinterInfo + T: (PrinterStatus, LegacyPrinterStatus, JobInfo, PrinterInfo) ]( SensorEntityDescription, PrusaLinkEntityDescription, ): """Describes PrusaLink sensor entity.""" - value_fn: Callable[[T], datetime | StateType] | None = None - value_fn_factory: Callable[[], Callable[[T], datetime | StateType]] | None = None - - def __post_init__(self) -> None: - """Ensure exactly one value callable source is configured.""" - if (self.value_fn is None) == (self.value_fn_factory is None): - raise ValueError( - "Exactly one of value_fn or value_fn_factory must be provided" - ) + value_fn: Callable[[T], datetime | StateType] SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = { @@ -232,39 +155,67 @@ def __post_init__(self) -> None: PrusaLinkSensorEntityDescription[LegacyPrinterStatus]( key="printer.telemetry.material", translation_key="material", - value_fn=_legacy_material, + # `available_fn` guarantees `telemetry` is not None at this + # point; the inner cast narrows the Optional for the index. + value_fn=lambda data: cast( + str, cast(LegacyPrinterTelemetry, data["telemetry"])["material"] + ), available_fn=lambda data: data.get("telemetry") is not None, ), ), "job": ( - PrusaLinkSensorEntityDescription[JobInfo | None]( + PrusaLinkSensorEntityDescription[JobInfo]( key="job.progress", translation_key="progress", native_unit_of_measurement=PERCENTAGE, - value_fn=_job_progress, - # Job sensors stay available when idle/no-job so `None` values are - # represented as `unknown` instead of `unavailable`. - available_fn=_always_available, + value_fn=lambda data: cast(float, data["progress"]), + available_fn=lambda data: ( + data.get("progress") is not None + and data.get("state") != PrinterState.IDLE.value + ), ), - PrusaLinkSensorEntityDescription[JobInfo | None]( + PrusaLinkSensorEntityDescription[JobInfo]( key="job.filename", translation_key="filename", - value_fn=_job_filename, - available_fn=_always_available, + # `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 + ), ), - PrusaLinkSensorEntityDescription[JobInfo | None]( + PrusaLinkSensorEntityDescription[JobInfo]( key="job.start", translation_key="print_start", device_class=SensorDeviceClass.TIMESTAMP, - value_fn_factory=_make_stable_job_start, - available_fn=_always_available, + value_fn=ignore_variance( + lambda data: utcnow() - timedelta(seconds=data["time_printing"]), + timedelta(minutes=2), + ), + available_fn=lambda data: ( + data.get("time_printing") is not None + and data.get("state") != PrinterState.IDLE.value + ), ), - PrusaLinkSensorEntityDescription[JobInfo | None]( + PrusaLinkSensorEntityDescription[JobInfo]( key="job.finish", translation_key="print_finish", device_class=SensorDeviceClass.TIMESTAMP, - value_fn_factory=_make_stable_job_finish, - available_fn=_always_available, + # `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=cast(int, data["time_remaining"])) + ), + timedelta(minutes=2), + ), + available_fn=lambda data: ( + data.get("time_remaining") is not None + and data.get("state") != PrinterState.IDLE.value + ), ), ), "info": ( @@ -324,23 +275,8 @@ def __init__( super().__init__(coordinator=coordinator) self.entity_description = description self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{description.key}" - # Some sensors need a fresh callable per entity to avoid shared state - # across multiple entries (e.g., variance cache for timestamps). - if description.value_fn_factory is not None: - self._value_fn = cast( - Callable[[object], datetime | StateType], - description.value_fn_factory(), - ) - elif description.value_fn is not None: - self._value_fn = cast( - Callable[[object], datetime | StateType], description.value_fn - ) - else: - raise ValueError( - f"Missing value callable for sensor description: {description.key}" - ) @property def native_value(self) -> datetime | StateType: """Return the state of the sensor.""" - return self._value_fn(self.coordinator.data) + return self.entity_description.value_fn(self.coordinator.data) diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py index f050978d74747b..53632e34270b0f 100644 --- a/tests/components/prusalink/conftest.py +++ b/tests/components/prusalink/conftest.py @@ -1,6 +1,6 @@ """Fixtures for PrusaLink.""" -from collections.abc import Generator, Iterator +from collections.abc import Generator from typing import Any from unittest.mock import patch @@ -123,14 +123,11 @@ def mock_get_status_printing() -> Generator[dict[str, Any]]: @pytest.fixture -def mock_job_api_idle() -> Iterator[None]: +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. - - Iterator is intentional for this yield fixture: it matches pytest usage - while avoiding unnecessary Generator send/return type parameters. """ with patch("pyprusalink.PrusaLink.get_job", return_value=None): yield None @@ -206,28 +203,6 @@ def mock_job_api_attention( mock_get_status_printing["printer"]["state"] = "ATTENTION" -@pytest.fixture -def mock_job_api_printing_with_nullable_fields() -> Generator[dict[str, Any]]: - """Mock PrusaLink printing with missing/None nullable fields. - - Tests that job helpers handle active jobs where file or time_remaining - are None (nullable fields), returning None without raising and allowing - sensors to show unknown state. - """ - resp = { - "id": 129, - "state": "PRINTING", - "progress": 45.00, - "time_remaining": None, # Missing finish time - "time_printing": 43987, - "file": None, # No file info - "serial_print": False, - "inaccurate_estimates": True, - } - with patch("pyprusalink.PrusaLink.get_job", return_value=resp): - yield resp - - @pytest.fixture def mock_api( mock_version_api: dict[str, str], diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py index 06100d1ba7ff63..5c4ee7f87f8d16 100644 --- a/tests/components/prusalink/test_sensor.py +++ b/tests/components/prusalink/test_sensor.py @@ -2,11 +2,10 @@ from datetime import UTC, datetime from typing import Any -from unittest.mock import Mock, patch +from unittest.mock import patch import pytest -from homeassistant.components.prusalink import sensor as prusalink_sensor from homeassistant.components.sensor import ( ATTR_OPTIONS, ATTR_STATE_CLASS, @@ -18,8 +17,6 @@ ATTR_UNIT_OF_MEASUREMENT, PERCENTAGE, REVOLUTIONS_PER_MINUTE, - STATE_UNAVAILABLE, - STATE_UNKNOWN, Platform, UnitOfLength, UnitOfTemperature, @@ -113,21 +110,21 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api) state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") @@ -222,21 +219,21 @@ async def test_sensors_idle_job_mk3( state = hass.states.get("sensor.workshop_mock_title_progress") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" state = hass.states.get("sensor.workshop_mock_title_filename") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" state = hass.states.get("sensor.workshop_mock_title_print_start") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_print_finish") assert state is not None - assert state.state == STATE_UNKNOWN + assert state.state == "unavailable" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP state = hass.states.get("sensor.workshop_mock_title_hotend_fan") @@ -299,115 +296,6 @@ async def test_sensors_active_job( assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE -async def test_job_start_stabilization_isolated_per_config_entry() -> None: - """Ensure job.start variance cache is not shared across config entries.""" - now = datetime(2022, 8, 27, 14, 0, 0, tzinfo=UTC) - - job_start_description = next( - description - for description in prusalink_sensor.SENSORS["job"] - if description.key == "job.start" - ) - - coordinator_one = Mock() - coordinator_one.config_entry = Mock(entry_id="entry_one") - coordinator_one.data = { - "state": "PRINTING", - "time_printing": 600, - "time_remaining": 0, - "progress": 0.0, - "file": None, - } - - coordinator_two = Mock() - coordinator_two.config_entry = Mock(entry_id="entry_two") - coordinator_two.data = { - "state": "PRINTING", - "time_printing": 660, - "time_remaining": 0, - "progress": 0.0, - "file": None, - } - - with patch("homeassistant.components.prusalink.sensor.utcnow", return_value=now): - entity_one = prusalink_sensor.PrusaLinkSensorEntity( - coordinator_one, job_start_description - ) - entity_two = prusalink_sensor.PrusaLinkSensorEntity( - coordinator_two, job_start_description - ) - - state_one = entity_one.native_value - state_two = entity_two.native_value - - assert state_one == datetime(2022, 8, 27, 13, 50, 0, tzinfo=UTC) - assert state_two == datetime(2022, 8, 27, 13, 49, 0, tzinfo=UTC) - - -@pytest.mark.usefixtures("entity_registry_enabled_by_default") -async def test_sensors_active_job_with_nullable_fields( - hass: HomeAssistant, - mock_config_entry, - mock_api, - mock_get_status_printing, - mock_job_api_printing_with_nullable_fields, -) -> None: - """Test job sensors with active job but missing nullable fields. - - Verifies that job helpers handle None values for nullable fields (file, - time_remaining) without raising, returning None to indicate unknown state. - """ - with patch( - "homeassistant.components.prusalink.sensor.utcnow", - return_value=datetime(2022, 8, 27, 14, 0, 0, tzinfo=UTC), - ): - assert await async_setup_component(hass, "prusalink", {}) - - state = hass.states.get("sensor.mock_title") - assert state is not None - assert state.state == "printing" - - # progress is required, should always have a value - state = hass.states.get("sensor.mock_title_progress") - assert state is not None - assert state.state == "45.0" - assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" - - # file is None, filename helper should return None -> unknown - state = hass.states.get("sensor.mock_title_filename") - assert state is not None - assert state.state == STATE_UNKNOWN - - # time_printing is required, print_start should always have a value - state = hass.states.get("sensor.mock_title_print_start") - assert state is not None - assert state.state == "2022-08-27T01:46:53+00:00" - assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP - - # time_remaining is None, print_finish helper should return None -> unknown - state = hass.states.get("sensor.mock_title_print_finish") - assert state is not None - assert state.state == STATE_UNKNOWN - assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP - - -@pytest.mark.usefixtures("entity_registry_enabled_by_default") -async def test_material_sensor_unavailable_when_legacy_telemetry_missing( - hass: HomeAssistant, - mock_config_entry, - mock_api, - mock_get_legacy_printer: dict[str, Any], -) -> None: - """Material sensor is unavailable when legacy telemetry is missing.""" - mock_get_legacy_printer["telemetry"] = None - - assert await async_setup_component(hass, "prusalink", {}) - - state = hass.states.get("sensor.mock_title_material") - assert state is not None - assert state.state == STATE_UNAVAILABLE - - @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_axis_x_y_sensors( hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_api: None From 41a9d7cd727ae358bab89dbc61e8557c40bb667d Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 13:07:18 +0200 Subject: [PATCH 30/35] Move improved handling of legacyprinter to followup-PR --- .../components/prusalink/config_flow.py | 8 ++----- .../components/prusalink/test_config_flow.py | 23 ------------------- 2 files changed, 2 insertions(+), 29 deletions(-) diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py index 6313c1a67f0459..29e762a823dd35 100644 --- a/homeassistant/components/prusalink/config_flow.py +++ b/homeassistant/components/prusalink/config_flow.py @@ -40,12 +40,8 @@ def ensure_printer_is_supported(version: VersionInfo) -> None: return # 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. `original` is an - # undocumented field returned by older standalone PrusaLink builds; - # it is not part of VersionInfo. - original_val = version.get("original") - original = original_val if isinstance(original_val, str) else "" - if original.startswith( + # 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"]): return diff --git a/tests/components/prusalink/test_config_flow.py b/tests/components/prusalink/test_config_flow.py index c069f800619686..cc66d25b35d036 100644 --- a/tests/components/prusalink/test_config_flow.py +++ b/tests/components/prusalink/test_config_flow.py @@ -182,29 +182,6 @@ async def test_form_invalid_mk3_server_version( assert result2["errors"] == {"base": "not_supported"} -async def test_form_mk3_original_none(hass: HomeAssistant, mock_version_api) -> None: - """Test MK2/MK3 workaround path handles original=None safely.""" - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER} - ) - - mock_version_api["api"] = "0.9.0-legacy" - mock_version_api["server"] = "0.7.2" - mock_version_api["original"] = None - - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], - { - "host": "1.1.1.1", - "username": "abcdefg", - "password": "abcdefg", - }, - ) - - assert result2["type"] is FlowResultType.FORM - assert result2["errors"] == {"base": "not_supported"} - - async def test_form_cannot_connect(hass: HomeAssistant) -> None: """Test we handle cannot connect error.""" result = await hass.config_entries.flow.async_init( From 71a1aad2e8d84cf1085cbbf17c040e80449ac1bd Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 13:11:00 +0200 Subject: [PATCH 31/35] prusalink: keep bump scope and fix binary_sensor mypy --- homeassistant/components/prusalink/binary_sensor.py | 8 +++++++- homeassistant/components/prusalink/sensor.py | 9 ++------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/homeassistant/components/prusalink/binary_sensor.py b/homeassistant/components/prusalink/binary_sensor.py index 6a255959204072..c6be3a91c12acc 100644 --- a/homeassistant/components/prusalink/binary_sensor.py +++ b/homeassistant/components/prusalink/binary_sensor.py @@ -30,12 +30,18 @@ class PrusaLinkBinarySensorEntityDescription[ value_fn: Callable[[T], bool] +def _status_connect_ok(data: PrinterStatus) -> bool: + """Return whether the printer connection status is healthy.""" + status_connect = data["printer"].get("status_connect") + return bool(status_connect and status_connect.get("ok")) + + BINARY_SENSORS: dict[str, tuple[PrusaLinkBinarySensorEntityDescription, ...]] = { "status": ( PrusaLinkBinarySensorEntityDescription[PrinterStatus]( key="printer.status_connect", device_class=BinarySensorDeviceClass.CONNECTIVITY, - value_fn=lambda data: data["printer"]["status_connect"]["ok"], + value_fn=_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/sensor.py b/homeassistant/components/prusalink/sensor.py index 8c007145481729..f5c731af721b90 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -12,7 +12,7 @@ PrinterState, PrinterStatus, ) -from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry +from pyprusalink.types_legacy import LegacyPrinterStatus from homeassistant.components.sensor import ( SensorDeviceClass, @@ -155,12 +155,7 @@ class PrusaLinkSensorEntityDescription[ PrusaLinkSensorEntityDescription[LegacyPrinterStatus]( key="printer.telemetry.material", translation_key="material", - # `available_fn` guarantees `telemetry` is not None at this - # point; the inner cast narrows the Optional for the index. - value_fn=lambda data: cast( - str, cast(LegacyPrinterTelemetry, data["telemetry"])["material"] - ), - available_fn=lambda data: data.get("telemetry") is not None, + value_fn=lambda data: cast(str, data["telemetry"]["material"]), ), ), "job": ( From 31ff0ede4d1479a4a2f27e3951157bdfe3143668 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 13:20:20 +0200 Subject: [PATCH 32/35] prusalink: fix mypy in config_flow and sensor --- homeassistant/components/prusalink/config_flow.py | 8 +++++--- homeassistant/components/prusalink/sensor.py | 7 +++++-- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py index 29e762a823dd35..b10aa1bd471407 100644 --- a/homeassistant/components/prusalink/config_flow.py +++ b/homeassistant/components/prusalink/config_flow.py @@ -41,9 +41,11 @@ 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"]): + if ( + isinstance(original := version.get("original"), str) + and 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/sensor.py b/homeassistant/components/prusalink/sensor.py index f5c731af721b90..6bb3330a6eab09 100644 --- a/homeassistant/components/prusalink/sensor.py +++ b/homeassistant/components/prusalink/sensor.py @@ -12,7 +12,7 @@ PrinterState, PrinterStatus, ) -from pyprusalink.types_legacy import LegacyPrinterStatus +from pyprusalink.types_legacy import LegacyPrinterStatus, LegacyPrinterTelemetry from homeassistant.components.sensor import ( SensorDeviceClass, @@ -155,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": ( From d0b43a9c78f9109a0f43a585c817144c1de9fe57 Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 18:46:11 +0200 Subject: [PATCH 33/35] prusalink: rollback fixes for previous local typing-errors --- homeassistant/components/prusalink/binary_sensor.py | 8 +------- homeassistant/components/prusalink/config_flow.py | 8 +++----- 2 files changed, 4 insertions(+), 12 deletions(-) diff --git a/homeassistant/components/prusalink/binary_sensor.py b/homeassistant/components/prusalink/binary_sensor.py index c6be3a91c12acc..6a255959204072 100644 --- a/homeassistant/components/prusalink/binary_sensor.py +++ b/homeassistant/components/prusalink/binary_sensor.py @@ -30,18 +30,12 @@ class PrusaLinkBinarySensorEntityDescription[ value_fn: Callable[[T], bool] -def _status_connect_ok(data: PrinterStatus) -> bool: - """Return whether the printer connection status is healthy.""" - status_connect = data["printer"].get("status_connect") - return bool(status_connect and status_connect.get("ok")) - - BINARY_SENSORS: dict[str, tuple[PrusaLinkBinarySensorEntityDescription, ...]] = { "status": ( PrusaLinkBinarySensorEntityDescription[PrinterStatus]( key="printer.status_connect", device_class=BinarySensorDeviceClass.CONNECTIVITY, - value_fn=_status_connect_ok, + value_fn=lambda data: 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 b10aa1bd471407..29e762a823dd35 100644 --- a/homeassistant/components/prusalink/config_flow.py +++ b/homeassistant/components/prusalink/config_flow.py @@ -41,11 +41,9 @@ 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 ( - isinstance(original := version.get("original"), str) - and original.startswith(("PrusaLink I3MK3", "PrusaLink I3MK2")) - and AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]) - ): + if version.get("original", "").startswith( + ("PrusaLink I3MK3", "PrusaLink I3MK2") + ) and AwesomeVersion("0.7.2") <= AwesomeVersion(version["server"]): return except AwesomeVersionException as err: From 90bc2dff47d1bb971186702288e98c721245581a Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 19:24:43 +0200 Subject: [PATCH 34/35] prusalink: align typing fixes with CI mypy checks --- homeassistant/components/prusalink/binary_sensor.py | 3 ++- homeassistant/components/prusalink/config_flow.py | 9 +++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/homeassistant/components/prusalink/binary_sensor.py b/homeassistant/components/prusalink/binary_sensor.py index 6a255959204072..855b81a580c74c 100644 --- a/homeassistant/components/prusalink/binary_sensor.py +++ b/homeassistant/components/prusalink/binary_sensor.py @@ -2,6 +2,7 @@ from collections.abc import Callable from dataclasses import dataclass +from typing import cast from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus from pyprusalink.types_legacy import LegacyPrinterStatus @@ -35,7 +36,7 @@ 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, 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: From 68488618bd19c48ed069c797e78d1fbc4a33d99a Mon Sep 17 00:00:00 2001 From: Heikki Henriksen Date: Thu, 21 May 2026 19:37:04 +0200 Subject: [PATCH 35/35] prusalink: narrow status_connect typing for mypy with 3.0.0 --- homeassistant/components/prusalink/binary_sensor.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/prusalink/binary_sensor.py b/homeassistant/components/prusalink/binary_sensor.py index 855b81a580c74c..7e12e9253d5992 100644 --- a/homeassistant/components/prusalink/binary_sensor.py +++ b/homeassistant/components/prusalink/binary_sensor.py @@ -4,7 +4,7 @@ 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 ( @@ -36,7 +36,9 @@ class PrusaLinkBinarySensorEntityDescription[ PrusaLinkBinarySensorEntityDescription[PrinterStatus]( key="printer.status_connect", device_class=BinarySensorDeviceClass.CONNECTIVITY, - value_fn=lambda data: cast(bool, 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