Compare commits

...

2 Commits

Author SHA1 Message Date
epenet
9b09146b3c Migrate prusalink to use runtime_data
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-02 08:01:33 +00:00
Mike Degatano
6e567ced92 Wrap hassio import in is_hassio check in get_system_info helper (#167111) 2026-04-02 09:05:09 +02:00
7 changed files with 28 additions and 50 deletions

View File

@@ -24,6 +24,7 @@ from .coordinator import (
InfoUpdateCoordinator,
JobUpdateCoordinator,
LegacyStatusCoordinator,
PrusaLinkConfigEntry,
PrusaLinkUpdateCoordinator,
StatusCoordinator,
)
@@ -36,7 +37,7 @@ PLATFORMS: list[Platform] = [
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: PrusaLinkConfigEntry) -> bool:
"""Set up PrusaLink from a config entry."""
if entry.version == 1 and entry.minor_version < 2:
raise ConfigEntryError("Please upgrade your printer's firmware.")
@@ -57,7 +58,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
for coordinator in coordinators.values():
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinators
entry.runtime_data = coordinators
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@@ -120,9 +121,6 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: PrusaLinkConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -13,12 +13,10 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import PrusaLinkUpdateCoordinator
from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator
from .entity import PrusaLinkEntity
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo, PrinterInfo)
@@ -56,13 +54,11 @@ BINARY_SENSORS: dict[str, tuple[PrusaLinkBinarySensorEntityDescription, ...]] =
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: PrusaLinkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PrusaLink sensor based on a config entry."""
coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][
entry.entry_id
]
coordinators = entry.runtime_data
entities: list[PrusaLinkEntity] = []
for coordinator_type, binary_sensors in BINARY_SENSORS.items():

View File

@@ -10,13 +10,11 @@ from pyprusalink import JobInfo, LegacyPrinterStatus, PrinterStatus, PrusaLink
from pyprusalink.types import Conflict, PrinterState
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import PrusaLinkUpdateCoordinator
from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator
from .entity import PrusaLinkEntity
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
@@ -71,13 +69,11 @@ BUTTONS: dict[str, tuple[PrusaLinkButtonEntityDescription, ...]] = {
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: PrusaLinkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PrusaLink buttons based on a config entry."""
coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][
entry.entry_id
]
coordinators = entry.runtime_data
entities: list[PrusaLinkEntity] = []
@@ -124,9 +120,7 @@ class PrusaLinkButtonEntity(PrusaLinkEntity, ButtonEntity):
"Action conflicts with current printer state"
) from err
coordinators: dict[str, PrusaLinkUpdateCoordinator] = self.hass.data[DOMAIN][
self.coordinator.config_entry.entry_id
]
coordinators = self.coordinator.config_entry.runtime_data
for coordinator in coordinators.values():
coordinator.expect_change()

View File

@@ -5,22 +5,20 @@ from __future__ import annotations
from pyprusalink.types import PrinterState
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import JobUpdateCoordinator
from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator
from .entity import PrusaLinkEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: PrusaLinkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PrusaLink camera."""
coordinator: JobUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]["job"]
coordinator = entry.runtime_data["job"]
async_add_entities([PrusaLinkJobPreviewEntity(coordinator)])
@@ -31,7 +29,7 @@ class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera):
last_image: bytes
_attr_translation_key = "job_preview"
def __init__(self, coordinator: JobUpdateCoordinator) -> None:
def __init__(self, coordinator: PrusaLinkUpdateCoordinator) -> None:
"""Initialize a PrusaLink camera entity."""
super().__init__(coordinator)
Camera.__init__(self)

View File

@@ -35,14 +35,17 @@ _MINIMUM_REFRESH_INTERVAL = 1.0
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
type PrusaLinkConfigEntry = ConfigEntry[dict[str, PrusaLinkUpdateCoordinator]]
class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):
"""Update coordinator for the printer."""
config_entry: ConfigEntry
config_entry: PrusaLinkConfigEntry
expect_change_until = 0.0
def __init__(
self, hass: HomeAssistant, config_entry: ConfigEntry, api: PrusaLink
self, hass: HomeAssistant, config_entry: PrusaLinkConfigEntry, api: PrusaLink
) -> None:
"""Initialize the update coordinator."""
self.api = api

View File

@@ -16,7 +16,6 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
REVOLUTIONS_PER_MINUTE,
@@ -29,8 +28,7 @@ from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow
from homeassistant.util.variance import ignore_variance
from .const import DOMAIN
from .coordinator import PrusaLinkUpdateCoordinator
from .coordinator import PrusaLinkConfigEntry, PrusaLinkUpdateCoordinator
from .entity import PrusaLinkEntity
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo, PrinterInfo)
@@ -204,13 +202,11 @@ SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: PrusaLinkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PrusaLink sensor based on a config entry."""
coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][
entry.entry_id
]
coordinators = entry.runtime_data
entities: list[PrusaLinkEntity] = []

View File

@@ -6,7 +6,7 @@ from functools import cache
from getpass import getuser
import logging
import platform
from typing import TYPE_CHECKING, Any
from typing import Any
from homeassistant.const import __version__ as current_version
from homeassistant.core import HomeAssistant
@@ -15,7 +15,6 @@ from homeassistant.util.package import is_docker_env, is_virtual_env
from homeassistant.util.system_info import is_official_image
from .hassio import is_hassio
from .importlib import async_import_module
from .singleton import singleton
_LOGGER = logging.getLogger(__name__)
@@ -54,15 +53,6 @@ cached_get_user = cache(getuser)
@bind_hass
async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]:
"""Return info about the system."""
# Local import to avoid circular dependencies
# We use the import helper because hassio
# may not be loaded yet and we don't want to
# do blocking I/O in the event loop to import it.
if TYPE_CHECKING:
from homeassistant.components import hassio # noqa: PLC0415
else:
hassio = await async_import_module(hass, "homeassistant.components.hassio")
is_hassio_ = is_hassio(hass)
info_object = {
@@ -105,6 +95,9 @@ async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]:
# Enrich with Supervisor information
if is_hassio_:
# Local import to avoid circular dependencies
from homeassistant.components import hassio # noqa: PLC0415
if not (info := hassio.get_info(hass)):
_LOGGER.warning("No Home Assistant Supervisor info available")
info = {}