Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade Prusa Link to new Digest Authentication and /v1/ API (#103396) 

Co-authored-by: Robert Resch <robert@resch.dev>
pull/106073/head
Niklas Wagner 2023-12-19 18:07:27 +01:00 committed by GitHub
parent c226d793d4
commit 91f8d3faef
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 466 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS
View File

@ -993,8 +993,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/proximity/ @mib1185
/tests/components/proximity/ @mib1185
/homeassistant/components/proxmoxve/ @jhollowe @Corbeno
/homeassistant/components/prusalink/ @balloob
/tests/components/prusalink/ @balloob
/homeassistant/components/prusalink/ @balloob @Skaronator
/tests/components/prusalink/ @balloob @Skaronator
/homeassistant/components/ps4/ @ktnrg45
/tests/components/ps4/ @ktnrg45
/homeassistant/components/pure_energie/ @klaasnicolaas

110
homeassistant/components/prusalink/__init__.py
View File

@ -6,13 +6,21 @@ import asyncio
from datetime import timedelta
import logging
from time import monotonic
from typing import Generic, TypeVar
from typing import TypeVar
from pyprusalink import InvalidAuth, JobInfo, PrinterInfo, PrusaLink, PrusaLinkError
from pyprusalink import JobInfo, LegacyPrinterStatus, PrinterStatus, PrusaLink
from pyprusalink.types import InvalidAuth, PrusaLinkError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.const import (
CONF_API_KEY,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import (
@ -27,16 +35,71 @@ PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.CAMERA, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up PrusaLink from a config entry."""
async def _migrate_to_version_2(
hass: HomeAssistant, entry: ConfigEntry
) -> PrusaLink | None:
"""Migrate to Version 2."""
_LOGGER.debug("Migrating entry to version 2")
data = dict(entry.data)
# "maker" is currently hardcoded in the firmware
# https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/bfb0ffc745ee6546e7efdba618d0e7c0f4c909cd/lib/WUI/wui_api.h#L19
data = {
**entry.data,
CONF_USERNAME: "maker",
CONF_PASSWORD: entry.data[CONF_API_KEY],
}
data.pop(CONF_API_KEY)
api = PrusaLink(
async_get_clientsession(hass),
entry.data["host"],
entry.data["api_key"],
data[CONF_HOST],
data[CONF_USERNAME],
data[CONF_PASSWORD],
)
try:
await api.get_info()
except InvalidAuth:
# We are unable to reach the new API which usually means
# that the user is running an outdated firmware version
ir.async_create_issue(
hass,
DOMAIN,
"firmware_5_1_required",
is_fixable=False,
severity=ir.IssueSeverity.ERROR,
translation_key="firmware_5_1_required",
translation_placeholders={
"entry_title": entry.title,
"prusa_mini_firmware_update": "https://help.prusa3d.com/article/firmware-updating-mini-mini_124784",
"prusa_mk4_xl_firmware_update": "https://help.prusa3d.com/article/how-to-update-firmware-mk4-xl_453086",
},
)
return None
entry.version = 2
hass.config_entries.async_update_entry(entry, data=data)
_LOGGER.info("Migrated config entry to version %d", entry.version)
return api
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up PrusaLink from a config entry."""
if entry.version == 1:
if (api := await _migrate_to_version_2(hass, entry)) is None:
return False
ir.async_delete_issue(hass, DOMAIN, "firmware_5_1_required")
else:
api = PrusaLink(
async_get_clientsession(hass),
entry.data[CONF_HOST],
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
)
coordinators = {
"printer": PrinterUpdateCoordinator(hass, api),
"legacy_status": LegacyStatusCoordinator(hass, api),
"status": StatusCoordinator(hass, api),
"job": JobUpdateCoordinator(hass, api),
}
for coordinator in coordinators.values():
@ -49,6 +112,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate old entry."""
# Version 1->2 migration are handled in async_setup_entry.
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
@ -57,10 +126,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
T = TypeVar("T", PrinterInfo, JobInfo)
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
class PrusaLinkUpdateCoordinator(DataUpdateCoordinator, Generic[T], ABC):
class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):
"""Update coordinator for the printer."""
config_entry: ConfigEntry
@ -105,21 +174,20 @@ class PrusaLinkUpdateCoordinator(DataUpdateCoordinator, Generic[T], ABC):
return timedelta(seconds=30)
class PrinterUpdateCoordinator(PrusaLinkUpdateCoordinator[PrinterInfo]):
class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]):
"""Printer update coordinator."""
async def _fetch_data(self) -> PrinterInfo:
async def _fetch_data(self) -> PrinterStatus:
"""Fetch the printer data."""
return await self.api.get_printer()
return await self.api.get_status()
def _get_update_interval(self, data: T) -> timedelta:
"""Get new update interval."""
if data and any(
data["state"]["flags"][key] for key in ("pausing", "cancelling")
):
return timedelta(seconds=5)
return super()._get_update_interval(data)
class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]):
"""Printer legacy update coordinator."""
async def _fetch_data(self) -> LegacyPrinterStatus:
"""Fetch the printer data."""
return await self.api.get_legacy_printer()
class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
@ -142,5 +210,5 @@ class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]):
identifiers={(DOMAIN, self.coordinator.config_entry.entry_id)},
name=self.coordinator.config_entry.title,
manufacturer="Prusa",
configuration_url=self.coordinator.api.host,
configuration_url=self.coordinator.api.client.host,
)

40
homeassistant/components/prusalink/button.py
View File

@ -5,7 +5,8 @@ from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, Generic, TypeVar, cast
from pyprusalink import Conflict, JobInfo, PrinterInfo, PrusaLink
from 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
@ -15,14 +16,14 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
T = TypeVar("T", PrinterInfo, JobInfo)
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
@dataclass(frozen=True)
class PrusaLinkButtonEntityDescriptionMixin(Generic[T]):
"""Mixin for required keys."""
press_fn: Callable[[PrusaLink], Coroutine[Any, Any, None]]
press_fn: Callable[[PrusaLink], Callable[[int], Coroutine[Any, Any, None]]]
@dataclass(frozen=True)
@ -35,33 +36,34 @@ class PrusaLinkButtonEntityDescription(
BUTTONS: dict[str, tuple[PrusaLinkButtonEntityDescription, ...]] = {
"printer": (
PrusaLinkButtonEntityDescription[PrinterInfo](
"status": (
PrusaLinkButtonEntityDescription[PrinterStatus](
key="printer.cancel_job",
translation_key="cancel_job",
icon="mdi:cancel",
press_fn=lambda api: cast(Coroutine, api.cancel_job()),
available_fn=lambda data: any(
data["state"]["flags"][flag]
for flag in ("printing", "pausing", "paused")
press_fn=lambda api: api.cancel_job,
available_fn=lambda data: (
data["printer"]["state"]
in [PrinterState.PRINTING.value, PrinterState.PAUSED.value]
),
),
PrusaLinkButtonEntityDescription[PrinterInfo](
PrusaLinkButtonEntityDescription[PrinterStatus](
key="job.pause_job",
translation_key="pause_job",
icon="mdi:pause",
press_fn=lambda api: cast(Coroutine, api.pause_job()),
available_fn=lambda data: (
data["state"]["flags"]["printing"]
and not data["state"]["flags"]["paused"]
press_fn=lambda api: api.pause_job,
available_fn=lambda data: cast(
bool, data["printer"]["state"] == PrinterState.PRINTING.value
),
),
PrusaLinkButtonEntityDescription[PrinterInfo](
PrusaLinkButtonEntityDescription[PrinterStatus](
key="job.resume_job",
translation_key="resume_job",
icon="mdi:play",
press_fn=lambda api: cast(Coroutine, api.resume_job()),
available_fn=lambda data: cast(bool, data["state"]["flags"]["paused"]),
press_fn=lambda api: api.resume_job,
available_fn=lambda data: cast(
bool, data["printer"]["state"] == PrinterState.PAUSED.value
),
),
),
}
@ -113,8 +115,10 @@ class PrusaLinkButtonEntity(PrusaLinkEntity, ButtonEntity):
async def async_press(self) -> None:
"""Press the button."""
job_id = self.coordinator.data["job"]["id"]
func = self.entity_description.press_fn(self.coordinator.api)
try:
await self.entity_description.press_fn(self.coordinator.api)
await func(job_id)
except Conflict as err:
raise HomeAssistantError(
"Action conflicts with current printer state"

10
homeassistant/components/prusalink/camera.py
View File

@ -35,7 +35,11 @@ class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera):
@property
def available(self) -> bool:
"""Get if camera is available."""
return super().available and self.coordinator.data.get("job") is not None
return (
super().available
and (file := self.coordinator.data.get("file"))
and file.get("refs", {}).get("thumbnail")
)
async def async_camera_image(
self, width: int | None = None, height: int | None = None
@ -44,11 +48,11 @@ class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera):
if not self.available:
return None
path = self.coordinator.data["job"]["file"]["path"]
path = self.coordinator.data["file"]["refs"]["thumbnail"]
if self.last_path == path:
return self.last_image
self.last_image = await self.coordinator.api.get_large_thumbnail(path)
self.last_image = await self.coordinator.api.get_file(path)
self.last_path = path
return self.last_image

22
homeassistant/components/prusalink/config_flow.py
View File

@ -7,11 +7,12 @@ from typing import Any
from aiohttp import ClientError
from awesomeversion import AwesomeVersion, AwesomeVersionException
from pyprusalink import InvalidAuth, PrusaLink
from pyprusalink import PrusaLink
from pyprusalink.types import InvalidAuth
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError
@ -25,7 +26,10 @@ _LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_API_KEY): str,
# "maker" is currently hardcoded in the firmware
# https://github.com/prusa3d/Prusa-Firmware-Buddy/blob/bfb0ffc745ee6546e7efdba618d0e7c0f4c909cd/lib/WUI/wui_api.h#L19
vol.Required(CONF_USERNAME, default="maker"): str,
vol.Required(CONF_PASSWORD): str,
}
)
@ -35,7 +39,12 @@ async def validate_input(hass: HomeAssistant, data: dict[str, str]) -> dict[str,
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
api = PrusaLink(async_get_clientsession(hass), data[CONF_HOST], data[CONF_API_KEY])
api = PrusaLink(
async_get_clientsession(hass),
data[CONF_HOST],
data[CONF_USERNAME],
data[CONF_PASSWORD],
)
try:
async with asyncio.timeout(5):
@ -57,7 +66,7 @@ async def validate_input(hass: HomeAssistant, data: dict[str, str]) -> dict[str,
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for PrusaLink."""
VERSION = 1
VERSION = 2
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@ -74,7 +83,8 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data = {
CONF_HOST: host,
CONF_API_KEY: user_input[CONF_API_KEY],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
}
errors = {}

4
homeassistant/components/prusalink/manifest.json
View File

@ -1,7 +1,7 @@
{
"domain": "prusalink",
"name": "PrusaLink",
"codeowners": ["@balloob"],
"codeowners": ["@balloob", "@Skaronator"],
"config_flow": true,
"dhcp": [
{
@ -10,5 +10,5 @@
],
"documentation": "https://www.home-assistant.io/integrations/prusalink",
"iot_class": "local_polling",
"requirements": ["pyprusalink==1.1.0"]
"requirements": ["pyprusalink==2.0.0"]
}

97
homeassistant/components/prusalink/sensor.py
View File

@ -6,7 +6,8 @@ from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Generic, TypeVar, cast
from pyprusalink import JobInfo, PrinterInfo
from pyprusalink.types import JobInfo, PrinterState, PrinterStatus
from pyprusalink.types_legacy import LegacyPrinterStatus
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -15,7 +16,12 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, UnitOfLength, UnitOfTemperature
from homeassistant.const import (
PERCENTAGE,
REVOLUTIONS_PER_MINUTE,
UnitOfLength,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
@ -24,7 +30,7 @@ from homeassistant.util.variance import ignore_variance
from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator
T = TypeVar("T", PrinterInfo, JobInfo)
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
@dataclass(frozen=True)
@ -44,78 +50,91 @@ class PrusaLinkSensorEntityDescription(
SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
"printer": (
PrusaLinkSensorEntityDescription[PrinterInfo](
"status": (
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.state",
name=None,
icon="mdi:printer-3d",
value_fn=lambda data: (
"pausing"
if (flags := data["state"]["flags"])["pausing"]
else "cancelling"
if flags["cancelling"]
else "paused"
if flags["paused"]
else "printing"
if flags["printing"]
else "idle"
),
value_fn=lambda data: (cast(str, data["printer"]["state"].lower())),
device_class=SensorDeviceClass.ENUM,
options=["cancelling", "idle", "paused", "pausing", "printing"],
options=[state.value.lower() for state in PrinterState],
translation_key="printer_state",
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.temp-bed",
translation_key="heatbed_temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["telemetry"]["temp-bed"]),
value_fn=lambda data: cast(float, data["printer"]["temp_bed"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.temp-nozzle",
translation_key="nozzle_temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["telemetry"]["temp-nozzle"]),
value_fn=lambda data: cast(float, data["printer"]["temp_nozzle"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.temp-bed.target",
translation_key="heatbed_target_temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["temperature"]["bed"]["target"]),
value_fn=lambda data: cast(float, data["printer"]["target_bed"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.temp-nozzle.target",
translation_key="nozzle_target_temperature",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["temperature"]["tool0"]["target"]),
value_fn=lambda data: cast(float, data["printer"]["target_nozzle"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.z-height",
translation_key="z_height",
native_unit_of_measurement=UnitOfLength.MILLIMETERS,
device_class=SensorDeviceClass.DISTANCE,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: cast(float, data["telemetry"]["z-height"]),
value_fn=lambda data: cast(float, data["printer"]["axis_z"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.print-speed",
translation_key="print_speed",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda data: cast(float, data["telemetry"]["print-speed"]),
value_fn=lambda data: cast(float, data["printer"]["speed"]),
),
PrusaLinkSensorEntityDescription[PrinterInfo](
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.print-flow",
translation_key="print_flow",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda data: cast(float, data["printer"]["flow"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.fan-hotend",
translation_key="fan_hotend",
native_unit_of_measurement=REVOLUTIONS_PER_MINUTE,
value_fn=lambda data: cast(float, data["printer"]["fan_hotend"]),
entity_registry_enabled_default=False,
),
PrusaLinkSensorEntityDescription[PrinterStatus](
key="printer.telemetry.fan-print",
translation_key="fan_print",
native_unit_of_measurement=REVOLUTIONS_PER_MINUTE,
value_fn=lambda data: cast(float, data["printer"]["fan_print"]),
entity_registry_enabled_default=False,
),
),
"legacy_status": (
PrusaLinkSensorEntityDescription[LegacyPrinterStatus](
key="printer.telemetry.material",
translation_key="material",
icon="mdi:palette-swatch-variant",
@ -128,15 +147,15 @@ SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
translation_key="progress",
icon="mdi:progress-clock",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda data: cast(float, data["progress"]["completion"]) * 100,
value_fn=lambda data: cast(float, data["progress"]),
available_fn=lambda data: data.get("progress") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.filename",
translation_key="filename",
icon="mdi:file-image-outline",
value_fn=lambda data: cast(str, data["job"]["file"]["display"]),
available_fn=lambda data: data.get("job") is not None,
value_fn=lambda data: cast(str, data["file"]["display_name"]),
available_fn=lambda data: data.get("file") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.start",
@ -144,12 +163,10 @@ SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:clock-start",
value_fn=ignore_variance(
lambda data: (
utcnow() - timedelta(seconds=data["progress"]["printTime"])
),
lambda data: (utcnow() - timedelta(seconds=data["time_printing"])),
timedelta(minutes=2),
),
available_fn=lambda data: data.get("progress") is not None,
available_fn=lambda data: data.get("time_printing") is not None,
),
PrusaLinkSensorEntityDescription[JobInfo](
key="job.finish",
@ -157,12 +174,10 @@ SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = {
icon="mdi:clock-end",
device_class=SensorDeviceClass.TIMESTAMP,
value_fn=ignore_variance(
lambda data: (
utcnow() + timedelta(seconds=data["progress"]["printTimeLeft"])
),
lambda data: (utcnow() + timedelta(seconds=data["time_remaining"])),
timedelta(minutes=2),
),
available_fn=lambda data: data.get("progress") is not None,
available_fn=lambda data: data.get("time_remaining") is not None,
),
),
}

28
homeassistant/components/prusalink/strings.json
View File

@ -4,7 +4,8 @@
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"api_key": "[%key:common::config_flow::data::api_key%]"
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
@ -15,15 +16,25 @@
"not_supported": "Only PrusaLink API v2 is supported"
}
},
"issues": {
"firmware_5_1_required": {
"description": "The PrusaLink integration has been updated to utilize the latest v1 API endpoints, which require firmware version 4.7.0 or later. If you own a Prusa Mini, please make sure your printer is running firmware 5.1.0 or a more recent version, as firmware versions 4.7.x and 5.0.x are not available for this model.\n\nFollow the guide below to update your {entry_title}.\n* [Prusa Mini Firmware Update]({prusa_mini_firmware_update})\n* [Prusa MK4/XL Firmware Update]({prusa_mk4_xl_firmware_update})\n\nAfter you've updated your printer's firmware, make sure to reload the config entry to fix this issue.",
"title": "Firmware update required"
}
},
"entity": {
"sensor": {
"printer_state": {
"state": {
"cancelling": "Cancelling",
"idle": "[%key:common::state::idle%]",
"busy": "Busy",
"printing": "Printing",
"paused": "[%key:common::state::paused%]",
"pausing": "Pausing",
"printing": "Printing"
"finished": "Finished",
"stopped": "Stopped",
"error": "Error",
"attention": "Attention",
"ready": "Ready"
}
},
"heatbed_temperature": {
@ -56,6 +67,15 @@
"print_speed": {
"name": "Print speed"
},
"print_flow": {
"name": "Print flow"
},
"fan_hotend": {
"name": "Hotend fan"
},
"fan_print": {
"name": "Print fan"
},
"z_height": {
"name": "Z-Height"
}

2
requirements_all.txt
View File

@ -2005,7 +2005,7 @@ pyprof2calltree==1.4.5
pyprosegur==0.0.9
# homeassistant.components.prusalink
pyprusalink==1.1.0
pyprusalink==2.0.0
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

2
requirements_test_all.txt
View File

@ -1525,7 +1525,7 @@ pyprof2calltree==1.4.5
pyprosegur==0.0.9
# homeassistant.components.prusalink
pyprusalink==1.1.0
pyprusalink==2.0.0
# homeassistant.components.ps4
pyps4-2ndscreen==1.3.1

177
tests/components/prusalink/conftest.py
View File

@ -1,9 +1,10 @@
"""Fixtures for PrusaLink."""
from unittest.mock import patch
import pytest
from homeassistant.components.prusalink import DOMAIN
from tests.common import MockConfigEntry
@ -11,7 +12,9 @@ from tests.common import MockConfigEntry
def mock_config_entry(hass):
"""Mock a PrusaLink config entry."""
entry = MockConfigEntry(
domain="prusalink", data={"host": "http://example.com", "api_key": "abcdefgh"}
domain=DOMAIN,
data={"host": "http://example.com", "username": "dummy", "password": "dummypw"},
version=2,
)
entry.add_to_hass(hass)
return entry
@ -23,96 +26,138 @@ def mock_version_api(hass):
resp = {
"api": "2.0.0",
"server": "2.1.2",
"text": "PrusaLink MINI",
"hostname": "PrusaMINI",
"text": "PrusaLink",
"hostname": "PrusaXL",
}
with patch("pyprusalink.PrusaLink.get_version", return_value=resp):
yield resp
@pytest.fixture
def mock_printer_api(hass):
def mock_info_api(hass):
"""Mock PrusaLink info API."""
resp = {
"nozzle_diameter": 0.40,
"mmu": False,
"serial": "serial-1337",
"hostname": "PrusaXL",
"min_extrusion_temp": 170,
}
with patch("pyprusalink.PrusaLink.get_info", return_value=resp):
yield resp
@pytest.fixture
def mock_get_legacy_printer(hass):
"""Mock PrusaLink printer API."""
resp = {"telemetry": {"material": "PLA"}}
with patch("pyprusalink.PrusaLink.get_legacy_printer", return_value=resp):
yield resp
@pytest.fixture
def mock_get_status_idle(hass):
"""Mock PrusaLink printer API."""
resp = {
"telemetry": {
"temp-bed": 41.9,
"temp-nozzle": 47.8,
"print-speed": 100,
"z-height": 1.8,
"material": "PLA",
"storage": {
"path": "/usb/",
"name": "usb",
"read_only": False,
},
"temperature": {
"tool0": {"actual": 47.8, "target": 210.1, "display": 0.0, "offset": 0},
"bed": {"actual": 41.9, "target": 60.5, "offset": 0},
},
"state": {
"text": "Operational",
"flags": {
"operational": True,
"paused": False,
"printing": False,
"cancelling": False,
"pausing": False,
"sdReady": False,
"error": False,
"closedOnError": False,
"ready": True,
"busy": False,
},
"printer": {
"state": "IDLE",
"temp_bed": 41.9,
"target_bed": 60.5,
"temp_nozzle": 47.8,
"target_nozzle": 210.1,
"axis_z": 1.8,
"axis_x": 7.9,
"axis_y": 8.4,
"flow": 100,
"speed": 100,
"fan_hotend": 100,
"fan_print": 75,
},
}
with patch("pyprusalink.PrusaLink.get_printer", return_value=resp):
with patch("pyprusalink.PrusaLink.get_status", return_value=resp):
yield resp
@pytest.fixture
def mock_get_status_printing(hass):
"""Mock PrusaLink printer API."""
resp = {
"job": {
"id": 129,
"progress": 37.00,
"time_remaining": 73020,
"time_printing": 43987,
},
"storage": {"path": "/usb/", "name": "usb", "read_only": False},
"printer": {
"state": "PRINTING",
"temp_bed": 53.9,
"target_bed": 85.0,
"temp_nozzle": 6.0,
"target_nozzle": 0.0,
"axis_z": 5.0,
"flow": 100,
"speed": 100,
"fan_hotend": 5000,
"fan_print": 2500,
},
}
with patch("pyprusalink.PrusaLink.get_status", return_value=resp):
yield resp
@pytest.fixture
def mock_job_api_idle(hass):
"""Mock PrusaLink job API having no job."""
resp = {}
with patch("pyprusalink.PrusaLink.get_job", return_value=resp):
yield resp
@pytest.fixture
def mock_job_api_printing(hass):
"""Mock PrusaLink printing."""
resp = {
"state": "Operational",
"job": None,
"progress": None,
"id": 129,
"state": "PRINTING",
"progress": 37.00,
"time_remaining": 73020,
"time_printing": 43987,
"file": {
"refs": {
"icon": "/thumb/s/usb/TabletStand3~4.BGC",
"thumbnail": "/thumb/l/usb/TabletStand3~4.BGC",
"download": "/usb/TabletStand3~4.BGC",
},
"name": "TabletStand3~4.BGC",
"display_name": "TabletStand3.bgcode",
"path": "/usb",
"size": 754535,
"m_timestamp": 1698686881,
},
}
with patch("pyprusalink.PrusaLink.get_job", return_value=resp):
yield resp
@pytest.fixture
def mock_job_api_printing(hass, mock_printer_api, mock_job_api_idle):
"""Mock PrusaLink printing."""
mock_printer_api["state"]["text"] = "Printing"
mock_printer_api["state"]["flags"]["printing"] = True
mock_job_api_idle.update(
{
"state": "Printing",
"job": {
"estimatedPrintTime": 117007,
"file": {
"name": "TabletStand3.gcode",
"path": "/usb/TABLET~1.GCO",
"display": "TabletStand3.gcode",
},
},
"progress": {
"completion": 0.37,
"printTime": 43987,
"printTimeLeft": 73020,
},
}
)
@pytest.fixture
def mock_job_api_paused(hass, mock_printer_api, mock_job_api_idle):
def mock_job_api_paused(hass, mock_get_status_printing, mock_job_api_printing):
"""Mock PrusaLink paused printing."""
mock_printer_api["state"]["text"] = "Paused"
mock_printer_api["state"]["flags"]["printing"] = False
mock_printer_api["state"]["flags"]["paused"] = True
mock_job_api_idle["state"] = "Paused"
mock_job_api_printing["state"] = "PAUSED"
mock_get_status_printing["printer"]["state"] = "PAUSED"
@pytest.fixture
def mock_api(mock_version_api, mock_printer_api, mock_job_api_idle):
def mock_api(
mock_version_api,
mock_info_api,
mock_get_legacy_printer,
mock_get_status_idle,
mock_job_api_idle,
):
"""Mock PrusaLink API."""

10
tests/components/prusalink/test_button.py
View File

@ -1,7 +1,7 @@
"""Test Prusalink buttons."""
from unittest.mock import patch
from pyprusalink import Conflict
from pyprusalink.types import Conflict
import pytest
from homeassistant.const import Platform
@ -32,6 +32,7 @@ async def test_button_pause_cancel(
mock_api,
hass_client: ClientSessionGenerator,
mock_job_api_printing,
mock_get_status_printing,
object_id,
method,
) -> None:
@ -66,9 +67,12 @@ async def test_button_pause_cancel(
@pytest.mark.parametrize(
("object_id", "method"),
(("mock_title_resume_job", "resume_job"),),
(
("mock_title_cancel_job", "cancel_job"),
("mock_title_resume_job", "resume_job"),
),
)
async def test_button_resume(
async def test_button_resume_cancel(
hass: HomeAssistant,
mock_config_entry,
mock_api,

4
tests/components/prusalink/test_camera.py
View File

@ -49,13 +49,13 @@ async def test_camera_active_job(
client = await hass_client()
with patch("pyprusalink.PrusaLink.get_large_thumbnail", return_value=b"hello"):
with patch("pyprusalink.PrusaLink.get_file", return_value=b"hello"):
resp = await client.get("/api/camera_proxy/camera.mock_title_preview")
assert resp.status == 200
assert await resp.read() == b"hello"
# Make sure we hit cached value.
with patch("pyprusalink.PrusaLink.get_large_thumbnail", side_effect=ValueError):
with patch("pyprusalink.PrusaLink.get_file", side_effect=ValueError):
resp = await client.get("/api/camera_proxy/camera.mock_title_preview")
assert resp.status == 200
assert await resp.read() == b"hello"

23
tests/components/prusalink/test_config_flow.py
View File

@ -25,16 +25,18 @@ async def test_form(hass: HomeAssistant, mock_version_api) -> None:
result["flow_id"],
{
"host": "http://1.1.1.1/",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "PrusaMINI"
assert result2["title"] == "PrusaXL"
assert result2["data"] == {
"host": "http://1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
}
assert len(mock_setup_entry.mock_calls) == 1
@ -53,7 +55,8 @@ async def test_form_invalid_auth(hass: HomeAssistant) -> None:
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)
@ -75,7 +78,8 @@ async def test_form_unknown(hass: HomeAssistant) -> None:
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)
@ -95,7 +99,8 @@ async def test_form_too_low_version(hass: HomeAssistant, mock_version_api) -> No
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)
@ -115,7 +120,8 @@ async def test_form_invalid_version_2(hass: HomeAssistant, mock_version_api) ->
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)
@ -137,7 +143,8 @@ async def test_form_cannot_connect(hass: HomeAssistant) -> None:
result["flow_id"],
{
"host": "1.1.1.1",
"api_key": "abcdefg",
"username": "abcdefg",
"password": "abcdefg",
},
)

79
tests/components/prusalink/test_init.py
View File

@ -2,14 +2,17 @@
from datetime import timedelta
from unittest.mock import patch
from pyprusalink import InvalidAuth, PrusaLinkError
from pyprusalink.types import InvalidAuth, PrusaLinkError
import pytest
from homeassistant.components.prusalink import DOMAIN
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.util.dt import utcnow
from tests.common import async_fire_time_changed
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_unloading(
@ -39,7 +42,13 @@ async def test_failed_update(
assert mock_config_entry.state == ConfigEntryState.LOADED
with patch(
"homeassistant.components.prusalink.PrusaLink.get_printer",
"homeassistant.components.prusalink.PrusaLink.get_version",
side_effect=exception,
), patch(
"homeassistant.components.prusalink.PrusaLink.get_status",
side_effect=exception,
), patch(
"homeassistant.components.prusalink.PrusaLink.get_legacy_printer",
side_effect=exception,
), patch(
"homeassistant.components.prusalink.PrusaLink.get_job",
@ -50,3 +59,67 @@ async def test_failed_update(
for state in hass.states.async_all():
assert state.state == "unavailable"
async def test_migration_1_2(
hass: HomeAssistant, issue_registry: ir.IssueRegistry, mock_api
) -> None:
"""Test migrating from version 1 to 2."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "http://prusaxl.local",
CONF_API_KEY: "api-key",
},
version=1,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
config_entries = hass.config_entries.async_entries(DOMAIN)
# Ensure that we have username, password after migration
assert len(config_entries) == 1
assert config_entries[0].data == {
CONF_HOST: "http://prusaxl.local",
CONF_USERNAME: "maker",
CONF_PASSWORD: "api-key",
}
# Make sure that we don't have any issues
assert len(issue_registry.issues) == 0
async def test_outdated_firmware_migration_1_2(
hass: HomeAssistant, issue_registry: ir.IssueRegistry, mock_api
) -> None:
"""Test migrating from version 1 to 2."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "http://prusaxl.local",
CONF_API_KEY: "api-key",
},
version=1,
)
entry.add_to_hass(hass)
with patch(
"pyprusalink.PrusaLink.get_info",
side_effect=InvalidAuth,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
# Make sure that we don't have thrown the issues
assert len(issue_registry.issues) == 1
# Reloading the integration with a working API (e.g. User updated firmware)
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
# Integration should be running now, the issue should be gone
assert entry.state == ConfigEntryState.LOADED
assert len(issue_registry.issues) == 0

40
tests/components/prusalink/test_sensor.py
View File

@ -15,6 +15,7 @@ from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
PERCENTAGE,
REVOLUTIONS_PER_MINUTE,
Platform,
UnitOfLength,
UnitOfTemperature,
@ -44,11 +45,15 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api)
assert state.state == "idle"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.ENUM
assert state.attributes[ATTR_OPTIONS] == [
"cancelling",
"idle",
"paused",
"pausing",
"busy",
"printing",
"paused",
"finished",
"stopped",
"error",
"attention",
"ready",
]
state = hass.states.get("sensor.mock_title_heatbed_temperature")
@ -95,6 +100,11 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api)
assert state is not None
assert state.state == "PLA"
state = hass.states.get("sensor.mock_title_print_flow")
assert state is not None
assert state.state == "100"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == PERCENTAGE
state = hass.states.get("sensor.mock_title_progress")
assert state is not None
assert state.state == "unavailable"
@ -114,12 +124,22 @@ async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api)
assert state.state == "unavailable"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP
state = hass.states.get("sensor.mock_title_hotend_fan")
assert state is not None
assert state.state == "100"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE
state = hass.states.get("sensor.mock_title_print_fan")
assert state is not None
assert state.state == "75"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE
async def test_sensors_active_job(
hass: HomeAssistant,
mock_config_entry,
mock_api,
mock_printer_api,
mock_get_status_printing,
mock_job_api_printing,
) -> None:
"""Test sensors while active job."""
@ -140,7 +160,7 @@ async def test_sensors_active_job(
state = hass.states.get("sensor.mock_title_filename")
assert state is not None
assert state.state == "TabletStand3.gcode"
assert state.state == "TabletStand3.bgcode"
state = hass.states.get("sensor.mock_title_print_start")
assert state is not None
@ -151,3 +171,13 @@ async def test_sensors_active_job(
assert state is not None
assert state.state == "2022-08-28T10:17:00+00:00"
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP
state = hass.states.get("sensor.mock_title_hotend_fan")
assert state is not None
assert state.state == "5000"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE
state = hass.states.get("sensor.mock_title_print_fan")
assert state is not None
assert state.state == "2500"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == REVOLUTIONS_PER_MINUTE