diff --git a/.strict-typing b/.strict-typing index 1e6a47f508b..f36439a66d3 100644 --- a/.strict-typing +++ b/.strict-typing @@ -201,6 +201,7 @@ homeassistant.components.persistent_notification.* homeassistant.components.pi_hole.* homeassistant.components.powerwall.* homeassistant.components.proximity.* +homeassistant.components.prusalink.* homeassistant.components.pvoutput.* homeassistant.components.pure_energie.* homeassistant.components.qnap_qsw.* diff --git a/CODEOWNERS b/CODEOWNERS index fe634225124..9a6578d8fd5 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -846,6 +846,8 @@ build.json @home-assistant/supervisor /homeassistant/components/prosegur/ @dgomes /tests/components/prosegur/ @dgomes /homeassistant/components/proxmoxve/ @jhollowe @Corbeno +/homeassistant/components/prusalink/ @balloob +/tests/components/prusalink/ @balloob /homeassistant/components/ps4/ @ktnrg45 /tests/components/ps4/ @ktnrg45 /homeassistant/components/pure_energie/ @klaasnicolaas diff --git a/homeassistant/components/prusalink/__init__.py b/homeassistant/components/prusalink/__init__.py new file mode 100644 index 00000000000..cbc77b92e8a --- /dev/null +++ b/homeassistant/components/prusalink/__init__.py @@ -0,0 +1,120 @@ +"""The PrusaLink integration.""" +from __future__ import annotations + +from abc import abstractmethod +from datetime import timedelta +import logging +from typing import Generic, TypeVar + +import async_timeout +from pyprusalink import InvalidAuth, JobInfo, PrinterInfo, PrusaLink, PrusaLinkError + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, + UpdateFailed, +) + +from .const import DOMAIN + +PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.CAMERA] +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up PrusaLink from a config entry.""" + api = PrusaLink( + async_get_clientsession(hass), + entry.data["host"], + entry.data["api_key"], + ) + + coordinators = { + "printer": PrinterUpdateCoordinator(hass, api), + "job": JobUpdateCoordinator(hass, api), + } + for coordinator in coordinators.values(): + await coordinator.async_config_entry_first_refresh() + + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinators + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + 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): + hass.data[DOMAIN].pop(entry.entry_id) + + return unload_ok + + +T = TypeVar("T", PrinterInfo, JobInfo) + + +class PrusaLinkUpdateCoordinator(DataUpdateCoordinator, Generic[T]): + """Update coordinator for the printer.""" + + config_entry: ConfigEntry + + def __init__(self, hass: HomeAssistant, api: PrusaLink) -> None: + """Initialize the update coordinator.""" + self.api = api + + super().__init__( + hass, _LOGGER, name=DOMAIN, update_interval=timedelta(seconds=30) + ) + + async def _async_update_data(self) -> T: + """Update the data.""" + try: + with async_timeout.timeout(5): + return await self._fetch_data() + except InvalidAuth: + raise UpdateFailed("Invalid authentication") from None + except PrusaLinkError as err: + raise UpdateFailed(str(err)) from err + + @abstractmethod + async def _fetch_data(self) -> T: + """Fetch the actual data.""" + raise NotImplementedError + + +class PrinterUpdateCoordinator(PrusaLinkUpdateCoordinator[PrinterInfo]): + """Printer update coordinator.""" + + async def _fetch_data(self) -> PrinterInfo: + """Fetch the printer data.""" + return await self.api.get_printer() + + +class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]): + """Job update coordinator.""" + + async def _fetch_data(self) -> JobInfo: + """Fetch the printer data.""" + return await self.api.get_job() + + +class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]): + """Defines a base PrusaLink entity.""" + + _attr_has_entity_name = True + + @property + def device_info(self) -> DeviceInfo: + """Return device information about this PrusaLink device.""" + return DeviceInfo( + identifiers={(DOMAIN, self.coordinator.config_entry.entry_id)}, + name=self.coordinator.config_entry.title, + manufacturer="Prusa", + configuration_url=self.coordinator.api.host, + ) diff --git a/homeassistant/components/prusalink/camera.py b/homeassistant/components/prusalink/camera.py new file mode 100644 index 00000000000..a6c16e2f5f2 --- /dev/null +++ b/homeassistant/components/prusalink/camera.py @@ -0,0 +1,54 @@ +"""Camera entity for PrusaLink.""" +from __future__ import annotations + +from homeassistant.components.camera import Camera +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from . import DOMAIN, JobUpdateCoordinator, PrusaLinkEntity + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up PrusaLink camera.""" + coordinator: JobUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]["job"] + async_add_entities([PrusaLinkJobPreviewEntity(coordinator)]) + + +class PrusaLinkJobPreviewEntity(PrusaLinkEntity, Camera): + """Defines a PrusaLink camera.""" + + last_path = "" + last_image: bytes + _attr_name = "Job Preview" + + def __init__(self, coordinator: JobUpdateCoordinator) -> None: + """Initialize a PrusaLink camera entity.""" + super().__init__(coordinator) + Camera.__init__(self) + self._attr_unique_id = f"{self.coordinator.config_entry.entry_id}_job_preview" + + @property + def available(self) -> bool: + """Get if camera is available.""" + return super().available and self.coordinator.data.get("job") is not None + + async def async_camera_image( + self, width: int | None = None, height: int | None = None + ) -> bytes | None: + """Return a still image from the camera.""" + if not self.available: + return None + + path = self.coordinator.data["job"]["file"]["path"] + + if self.last_path == path: + return self.last_image + + self.last_image = await self.coordinator.api.get_large_thumbnail(path) + self.last_path = path + return self.last_image diff --git a/homeassistant/components/prusalink/config_flow.py b/homeassistant/components/prusalink/config_flow.py new file mode 100644 index 00000000000..da21cca99de --- /dev/null +++ b/homeassistant/components/prusalink/config_flow.py @@ -0,0 +1,106 @@ +"""Config flow for PrusaLink integration.""" +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +from aiohttp import ClientError +import async_timeout +from awesomeversion import AwesomeVersion, AwesomeVersionException +from pyprusalink import InvalidAuth, PrusaLink +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResult +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +def add_protocol(value: str) -> str: + """Validate URL has a scheme.""" + value = value.rstrip("/") + if value.startswith(("http://", "https://")): + return value + + return f"http://{value}" + + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Required("host"): vol.All(str, add_protocol), + vol.Required("api_key"): str, + } +) + + +async def validate_input(hass: HomeAssistant, data: dict[str, str]) -> dict[str, str]: + """Validate the user input allows us to connect. + + Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. + """ + api = PrusaLink(async_get_clientsession(hass), data["host"], data["api_key"]) + + try: + async with async_timeout.timeout(5): + version = await api.get_version() + + except (asyncio.TimeoutError, ClientError) as err: + _LOGGER.error("Could not connect to PrusaLink: %s", err) + raise CannotConnect from err + + try: + if AwesomeVersion(version["api"]) < AwesomeVersion("2.0.0"): + raise NotSupported + except AwesomeVersionException as err: + raise NotSupported from err + + return {"title": version["hostname"]} + + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for PrusaLink.""" + + VERSION = 1 + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle the initial step.""" + if user_input is None: + return self.async_show_form( + step_id="user", data_schema=STEP_USER_DATA_SCHEMA + ) + + errors = {} + + try: + info = await validate_input(self.hass, user_input) + except CannotConnect: + errors["base"] = "cannot_connect" + except NotSupported: + errors["base"] = "not_supported" + except InvalidAuth: + errors["base"] = "invalid_auth" + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected exception") + errors["base"] = "unknown" + else: + return self.async_create_entry(title=info["title"], data=user_input) + + return self.async_show_form( + step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors + ) + + +class CannotConnect(HomeAssistantError): + """Error to indicate we cannot connect.""" + + +class NotSupported(HomeAssistantError): + """Error to indicate we cannot connect.""" diff --git a/homeassistant/components/prusalink/const.py b/homeassistant/components/prusalink/const.py new file mode 100644 index 00000000000..76f8d9f2693 --- /dev/null +++ b/homeassistant/components/prusalink/const.py @@ -0,0 +1,3 @@ +"""Constants for the PrusaLink integration.""" + +DOMAIN = "prusalink" diff --git a/homeassistant/components/prusalink/manifest.json b/homeassistant/components/prusalink/manifest.json new file mode 100644 index 00000000000..9efed0be74a --- /dev/null +++ b/homeassistant/components/prusalink/manifest.json @@ -0,0 +1,14 @@ +{ + "domain": "prusalink", + "name": "PrusaLink", + "config_flow": true, + "documentation": "https://www.home-assistant.io/integrations/prusalink", + "requirements": ["pyprusalink==1.0.1"], + "dhcp": [ + { + "macaddress": "109C70*" + } + ], + "codeowners": ["@balloob"], + "iot_class": "local_polling" +} diff --git a/homeassistant/components/prusalink/sensor.py b/homeassistant/components/prusalink/sensor.py new file mode 100644 index 00000000000..f2a4f2fec81 --- /dev/null +++ b/homeassistant/components/prusalink/sensor.py @@ -0,0 +1,173 @@ +"""PrusaLink sensors.""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Generic, TypeVar, cast + +from pyprusalink import JobInfo, PrinterInfo + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, + SensorStateClass, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import PERCENTAGE, TEMP_CELSIUS +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import StateType +from homeassistant.util.dt import utcnow +from homeassistant.util.variance import ignore_variance + +from . import DOMAIN, PrusaLinkEntity, PrusaLinkUpdateCoordinator + +T = TypeVar("T", PrinterInfo, JobInfo) + + +@dataclass +class PrusaLinkSensorEntityDescriptionMixin(Generic[T]): + """Mixin for required keys.""" + + value_fn: Callable[[T], datetime | StateType] + + +@dataclass +class PrusaLinkSensorEntityDescription( + SensorEntityDescription, PrusaLinkSensorEntityDescriptionMixin[T], Generic[T] +): + """Describes PrusaLink sensor entity.""" + + available_fn: Callable[[T], bool] = lambda _: True + + +SENSORS: dict[str, tuple[PrusaLinkSensorEntityDescription, ...]] = { + "printer": ( + PrusaLinkSensorEntityDescription[PrinterInfo]( + key="printer.state", + 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" + ), + device_class="prusalink__printer_state", + ), + PrusaLinkSensorEntityDescription[PrinterInfo]( + key="printer.telemetry.temp-bed", + name="Heatbed", + native_unit_of_measurement=TEMP_CELSIUS, + device_class=SensorDeviceClass.TEMPERATURE, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda data: cast(float, data["telemetry"]["temp-bed"]), + entity_registry_enabled_default=False, + ), + PrusaLinkSensorEntityDescription[PrinterInfo]( + key="printer.telemetry.temp-nozzle", + name="Nozzle Temperature", + native_unit_of_measurement=TEMP_CELSIUS, + device_class=SensorDeviceClass.TEMPERATURE, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda data: cast(float, data["telemetry"]["temp-nozzle"]), + entity_registry_enabled_default=False, + ), + ), + "job": ( + PrusaLinkSensorEntityDescription[JobInfo]( + key="job.progress", + name="Progress", + native_unit_of_measurement=PERCENTAGE, + value_fn=lambda data: cast(float, data["progress"]["completion"]) * 100, + available_fn=lambda data: data.get("progress") is not None, + ), + PrusaLinkSensorEntityDescription[JobInfo]( + key="job.filename", + name="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, + ), + PrusaLinkSensorEntityDescription[JobInfo]( + key="job.start", + name="Print Start", + device_class=SensorDeviceClass.TIMESTAMP, + value_fn=ignore_variance( + lambda data: ( + utcnow() - timedelta(seconds=data["progress"]["printTime"]) + ), + timedelta(minutes=2), + ), + available_fn=lambda data: data.get("progress") is not None, + ), + PrusaLinkSensorEntityDescription[JobInfo]( + key="job.finish", + name="Print Finish", + device_class=SensorDeviceClass.TIMESTAMP, + value_fn=ignore_variance( + lambda data: ( + utcnow() + timedelta(seconds=data["progress"]["printTimeLeft"]) + ), + timedelta(minutes=2), + ), + available_fn=lambda data: data.get("progress") is not None, + ), + ), +} + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up PrusaLink sensor based on a config entry.""" + coordinators: dict[str, PrusaLinkUpdateCoordinator] = hass.data[DOMAIN][ + entry.entry_id + ] + + entities: list[PrusaLinkEntity] = [] + + for coordinator_type, sensors in SENSORS.items(): + coordinator = coordinators[coordinator_type] + entities.extend( + PrusaLinkSensorEntity(coordinator, sensor_description) + for sensor_description in sensors + ) + + async_add_entities(entities) + + +class PrusaLinkSensorEntity(PrusaLinkEntity, SensorEntity): + """Defines a PrusaLink sensor.""" + + entity_description: PrusaLinkSensorEntityDescription + + def __init__( + self, + coordinator: PrusaLinkUpdateCoordinator, + description: PrusaLinkSensorEntityDescription, + ) -> None: + """Initialize a PrusaLink sensor entity.""" + super().__init__(coordinator=coordinator) + self.entity_description = description + self._attr_unique_id = f"{coordinator.config_entry.entry_id}_{description.key}" + + @property + def native_value(self) -> datetime | StateType: + """Return the state of the sensor.""" + return self.entity_description.value_fn(self.coordinator.data) + + @property + def available(self) -> bool: + """Return if sensor is available.""" + return super().available and self.entity_description.available_fn( + self.coordinator.data + ) diff --git a/homeassistant/components/prusalink/strings.json b/homeassistant/components/prusalink/strings.json new file mode 100644 index 00000000000..24835324e18 --- /dev/null +++ b/homeassistant/components/prusalink/strings.json @@ -0,0 +1,18 @@ +{ + "config": { + "step": { + "user": { + "data": { + "host": "[%key:common::config_flow::data::host%]", + "api_key": "[%key:common::config_flow::data::api_key%]" + } + } + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "unknown": "[%key:common::config_flow::error::unknown%]", + "not_supported": "Only PrusaLink API v2 is supported" + } + } +} diff --git a/homeassistant/components/prusalink/strings.sensor.json b/homeassistant/components/prusalink/strings.sensor.json new file mode 100644 index 00000000000..6e1fe62e7f5 --- /dev/null +++ b/homeassistant/components/prusalink/strings.sensor.json @@ -0,0 +1,11 @@ +{ + "state": { + "prusalink__printer_state": { + "pausing": "Pausing", + "cancelling": "Cancelling", + "paused": "Paused", + "printing": "Printing", + "idle": "Idle" + } + } +} diff --git a/homeassistant/components/prusalink/translations/en.json b/homeassistant/components/prusalink/translations/en.json new file mode 100644 index 00000000000..e9be6d8d96e --- /dev/null +++ b/homeassistant/components/prusalink/translations/en.json @@ -0,0 +1,17 @@ +{ + "config": { + "error": { + "cannot_connect": "Failed to connect", + "invalid_auth": "Invalid authentication", + "unknown": "Unexpected error" + }, + "step": { + "user": { + "data": { + "api_key": "API Key", + "host": "Host" + } + } + } + } +} \ No newline at end of file diff --git a/homeassistant/components/prusalink/translations/sensor.en.json b/homeassistant/components/prusalink/translations/sensor.en.json new file mode 100644 index 00000000000..98b9c3a9265 --- /dev/null +++ b/homeassistant/components/prusalink/translations/sensor.en.json @@ -0,0 +1,11 @@ +{ + "state": { + "prusalink__printer_state": { + "cancelling": "Cancelling", + "idle": "Idle", + "paused": "Paused", + "pausing": "Pausing", + "printing": "Printing" + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index a9b303eabea..133c02fd210 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -290,6 +290,7 @@ FLOWS = { "profiler", "progettihwsw", "prosegur", + "prusalink", "ps4", "pure_energie", "pushover", diff --git a/homeassistant/generated/dhcp.py b/homeassistant/generated/dhcp.py index 77255577cc2..8ced5265136 100644 --- a/homeassistant/generated/dhcp.py +++ b/homeassistant/generated/dhcp.py @@ -80,6 +80,7 @@ DHCP: list[dict[str, str | bool]] = [ {'domain': 'oncue', 'hostname': 'kohlergen*', 'macaddress': '00146F*'}, {'domain': 'overkiz', 'hostname': 'gateway*', 'macaddress': 'F8811A*'}, {'domain': 'powerwall', 'hostname': '1118431-*'}, + {'domain': 'prusalink', 'macaddress': '109C70*'}, {'domain': 'qnap_qsw', 'macaddress': '245EBE*'}, {'domain': 'rachio', 'hostname': 'rachio-*', 'macaddress': '009D6B*'}, {'domain': 'rachio', 'hostname': 'rachio-*', 'macaddress': 'F0038C*'}, diff --git a/homeassistant/util/variance.py b/homeassistant/util/variance.py new file mode 100644 index 00000000000..626b111817f --- /dev/null +++ b/homeassistant/util/variance.py @@ -0,0 +1,49 @@ +"""Util functions to help filter out similar results.""" +from __future__ import annotations + +from collections.abc import Callable +from datetime import datetime, timedelta +import functools +from typing import Any, TypeVar, overload + +T = TypeVar("T", int, float, datetime) + + +@overload +def ignore_variance( + func: Callable[..., int], ignored_variance: int +) -> Callable[..., int]: + ... + + +@overload +def ignore_variance( + func: Callable[..., float], ignored_variance: float +) -> Callable[..., float]: + ... + + +@overload +def ignore_variance( + func: Callable[..., datetime], ignored_variance: timedelta +) -> Callable[..., datetime]: + ... + + +def ignore_variance(func: Callable[..., T], ignored_variance: Any) -> Callable[..., T]: + """Wrap a function that returns old result if new result does not vary enough.""" + last_value: T | None = None + + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> T: + nonlocal last_value + + value = func(*args, **kwargs) + + if last_value is not None and abs(value - last_value) < ignored_variance: + return last_value + + last_value = value + return value + + return wrapper diff --git a/mypy.ini b/mypy.ini index 863e673401c..d6665cb40c8 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1769,6 +1769,16 @@ disallow_untyped_defs = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.prusalink.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.pvoutput.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/requirements_all.txt b/requirements_all.txt index a17bac8c905..4be9e231781 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1780,6 +1780,9 @@ pyprof2calltree==1.4.5 # homeassistant.components.prosegur pyprosegur==0.0.5 +# homeassistant.components.prusalink +pyprusalink==1.0.1 + # homeassistant.components.ps4 pyps4-2ndscreen==1.3.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 553dfd7eb58..08e4ef72be9 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1251,6 +1251,9 @@ pyprof2calltree==1.4.5 # homeassistant.components.prosegur pyprosegur==0.0.5 +# homeassistant.components.prusalink +pyprusalink==1.0.1 + # homeassistant.components.ps4 pyps4-2ndscreen==1.3.1 diff --git a/script/scaffold/templates/config_flow/integration/__init__.py b/script/scaffold/templates/config_flow/integration/__init__.py index 704292a2e9b..4d74c5d41ff 100644 --- a/script/scaffold/templates/config_flow/integration/__init__.py +++ b/script/scaffold/templates/config_flow/integration/__init__.py @@ -21,7 +21,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # TODO 3. Store an API object for your platforms to access # hass.data[DOMAIN][entry.entry_id] = MyApi(...) - hass.config_entries.async_setup_platforms(entry, PLATFORMS) + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True diff --git a/tests/components/prusalink/__init__.py b/tests/components/prusalink/__init__.py new file mode 100644 index 00000000000..a34a40b107f --- /dev/null +++ b/tests/components/prusalink/__init__.py @@ -0,0 +1 @@ +"""Tests for the PrusaLink integration.""" diff --git a/tests/components/prusalink/conftest.py b/tests/components/prusalink/conftest.py new file mode 100644 index 00000000000..9d968d615aa --- /dev/null +++ b/tests/components/prusalink/conftest.py @@ -0,0 +1,109 @@ +"""Fixtures for PrusaLink.""" + +from unittest.mock import patch + +import pytest + +from tests.common import MockConfigEntry + + +@pytest.fixture +def mock_config_entry(hass): + """Mock a PrusaLink config entry.""" + entry = MockConfigEntry( + domain="prusalink", data={"host": "http://example.com", "api_key": "abcdefgh"} + ) + entry.add_to_hass(hass) + return entry + + +@pytest.fixture +def mock_version_api(hass): + """Mock PrusaLink version API.""" + resp = { + "api": "2.0.0", + "server": "2.1.2", + "text": "PrusaLink MINI", + "hostname": "PrusaMINI", + } + with patch("pyprusalink.PrusaLink.get_version", return_value=resp): + yield resp + + +@pytest.fixture +def mock_printer_api(hass): + """Mock PrusaLink printer API.""" + resp = { + "telemetry": { + "temp-bed": 41.9, + "temp-nozzle": 47.8, + "print-speed": 100, + "z-height": 1.8, + "material": "PLA", + }, + "temperature": { + "tool0": {"actual": 47.8, "target": 0.0, "display": 0.0, "offset": 0}, + "bed": {"actual": 41.9, "target": 0.0, "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, + }, + }, + } + with patch("pyprusalink.PrusaLink.get_printer", return_value=resp): + yield resp + + +@pytest.fixture +def mock_job_api_idle(hass): + """Mock PrusaLink job API having no job.""" + with patch( + "pyprusalink.PrusaLink.get_job", + return_value={ + "state": "Operational", + "job": None, + "progress": None, + }, + ): + yield + + +@pytest.fixture +def mock_job_api_active(hass): + """Mock PrusaLink job API having no job.""" + with patch( + "pyprusalink.PrusaLink.get_job", + return_value={ + "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, + }, + }, + ): + yield + + +@pytest.fixture +def mock_api(mock_version_api, mock_printer_api, mock_job_api_idle): + """Mock PrusaLink API.""" diff --git a/tests/components/prusalink/test_camera.py b/tests/components/prusalink/test_camera.py new file mode 100644 index 00000000000..36ec8ec3700 --- /dev/null +++ b/tests/components/prusalink/test_camera.py @@ -0,0 +1,51 @@ +"""Test Prusalink camera.""" + +from unittest.mock import patch + +import pytest + +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + + +@pytest.fixture(autouse=True) +def setup_camera_platform_only(): + """Only setup camera platform.""" + with patch("homeassistant.components.prusalink.PLATFORMS", [Platform.CAMERA]): + yield + + +async def test_camera_no_job( + hass: HomeAssistant, + mock_config_entry, + mock_api, +) -> None: + """Test sensors while no job active.""" + assert await async_setup_component(hass, "prusalink", {}) + state = hass.states.get("camera.mock_title_job_preview") + assert state is not None + assert state.state == "unavailable" + + +async def test_camera_active_job( + hass: HomeAssistant, mock_config_entry, mock_api, mock_job_api_active, hass_client +): + """Test sensors while no job active.""" + assert await async_setup_component(hass, "prusalink", {}) + state = hass.states.get("camera.mock_title_job_preview") + assert state is not None + assert state.state == "idle" + + client = await hass_client() + + with patch("pyprusalink.PrusaLink.get_large_thumbnail", return_value=b"hello"): + resp = await client.get("/api/camera_proxy/camera.mock_title_job_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): + resp = await client.get("/api/camera_proxy/camera.mock_title_job_preview") + assert resp.status == 200 + assert await resp.read() == b"hello" diff --git a/tests/components/prusalink/test_config_flow.py b/tests/components/prusalink/test_config_flow.py new file mode 100644 index 00000000000..78cd652f5eb --- /dev/null +++ b/tests/components/prusalink/test_config_flow.py @@ -0,0 +1,125 @@ +"""Test the PrusaLink config flow.""" +import asyncio +from unittest.mock import patch + +from homeassistant import config_entries +from homeassistant.components.prusalink.config_flow import InvalidAuth +from homeassistant.components.prusalink.const import DOMAIN +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + + +async def test_form(hass: HomeAssistant, mock_version_api) -> None: + """Test we get the form.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + with patch( + "homeassistant.components.prusalink.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "http://1.1.1.1/", + "api_key": "abcdefg", + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "PrusaMINI" + assert result2["data"] == { + "host": "http://1.1.1.1", + "api_key": "abcdefg", + } + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_form_invalid_auth(hass: HomeAssistant) -> None: + """Test we handle invalid auth.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.prusalink.config_flow.PrusaLink.get_version", + side_effect=InvalidAuth, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "1.1.1.1", + "api_key": "abcdefg", + }, + ) + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "invalid_auth"} + + +async def test_form_unknown(hass: HomeAssistant) -> None: + """Test we handle unknown error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.prusalink.config_flow.PrusaLink.get_version", + side_effect=ValueError, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "1.1.1.1", + "api_key": "abcdefg", + }, + ) + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "unknown"} + + +async def test_form_invalid_version(hass: HomeAssistant, mock_version_api) -> None: + """Test we handle invalid auth.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + mock_version_api["api"] = "1.2.0" + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "1.1.1.1", + "api_key": "abcdefg", + }, + ) + + assert result2["type"] == 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( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + + with patch( + "homeassistant.components.prusalink.config_flow.PrusaLink.get_version", + side_effect=asyncio.TimeoutError, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + "host": "1.1.1.1", + "api_key": "abcdefg", + }, + ) + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "cannot_connect"} diff --git a/tests/components/prusalink/test_init.py b/tests/components/prusalink/test_init.py new file mode 100644 index 00000000000..a36c70bb882 --- /dev/null +++ b/tests/components/prusalink/test_init.py @@ -0,0 +1,23 @@ +"""Test setting up and unloading PrusaLink.""" + + +from homeassistant.config_entries import ConfigEntry, ConfigEntryState +from homeassistant.core import HomeAssistant + + +async def test_sensors_no_job( + hass: HomeAssistant, + mock_config_entry: ConfigEntry, + mock_api, +): + """Test sensors while no job active.""" + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + assert mock_config_entry.state == ConfigEntryState.LOADED + + assert hass.states.async_entity_ids_count() > 0 + + assert await hass.config_entries.async_unload(mock_config_entry.entry_id) + assert mock_config_entry.state == ConfigEntryState.NOT_LOADED + + for state in hass.states.async_all(): + assert state.state == "unavailable" diff --git a/tests/components/prusalink/test_sensor.py b/tests/components/prusalink/test_sensor.py new file mode 100644 index 00000000000..2ce62cf3990 --- /dev/null +++ b/tests/components/prusalink/test_sensor.py @@ -0,0 +1,114 @@ +"""Test Prusalink sensors.""" + +from datetime import datetime, timezone +from unittest.mock import PropertyMock, patch + +import pytest + +from homeassistant.components.sensor import ( + ATTR_STATE_CLASS, + SensorDeviceClass, + SensorStateClass, +) +from homeassistant.const import ( + ATTR_DEVICE_CLASS, + ATTR_UNIT_OF_MEASUREMENT, + TEMP_CELSIUS, + Platform, +) +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + + +@pytest.fixture(autouse=True) +def setup_sensor_platform_only(): + """Only setup sensor platform.""" + with patch( + "homeassistant.components.prusalink.PLATFORMS", [Platform.SENSOR] + ), patch( + "homeassistant.helpers.entity.Entity.entity_registry_enabled_default", + PropertyMock(return_value=True), + ): + yield + + +async def test_sensors_no_job(hass: HomeAssistant, mock_config_entry, mock_api): + """Test sensors while no job active.""" + assert await async_setup_component(hass, "prusalink", {}) + + state = hass.states.get("sensor.mock_title") + assert state is not None + assert state.state == "idle" + + state = hass.states.get("sensor.mock_title_heatbed") + assert state is not None + assert state.state == "41.9" + assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE + assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT + + state = hass.states.get("sensor.mock_title_nozzle_temperature") + assert state is not None + assert state.state == "47.8" + assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == TEMP_CELSIUS + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TEMPERATURE + assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT + + state = hass.states.get("sensor.mock_title_progress") + assert state is not None + assert state.state == "unavailable" + assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" + + state = hass.states.get("sensor.mock_title_filename") + assert state is not None + assert state.state == "unavailable" + + state = hass.states.get("sensor.mock_title_print_start") + assert state is not None + assert state.state == "unavailable" + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP + + state = hass.states.get("sensor.mock_title_print_finish") + assert state is not None + assert state.state == "unavailable" + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP + + +async def test_sensors_active_job( + hass: HomeAssistant, + mock_config_entry, + mock_api, + mock_printer_api, + mock_job_api_active, +): + """Test sensors while active job.""" + mock_printer_api["state"]["flags"]["printing"] = True + + with patch( + "homeassistant.components.prusalink.sensor.utcnow", + return_value=datetime(2022, 8, 27, 14, 0, 0, tzinfo=timezone.utc), + ): + assert await async_setup_component(hass, "prusalink", {}) + + state = hass.states.get("sensor.mock_title") + assert state is not None + assert state.state == "printing" + + state = hass.states.get("sensor.mock_title_progress") + assert state is not None + assert state.state == "37.0" + assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "%" + + state = hass.states.get("sensor.mock_title_filename") + assert state is not None + assert state.state == "TabletStand3.gcode" + + 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 + + state = hass.states.get("sensor.mock_title_print_finish") + assert state is not None + assert state.state == "2022-08-28T10:17:00+00:00" + assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.TIMESTAMP diff --git a/tests/util/test_variance.py b/tests/util/test_variance.py new file mode 100644 index 00000000000..c100d8b04d0 --- /dev/null +++ b/tests/util/test_variance.py @@ -0,0 +1,40 @@ +"""Test variance method.""" +from datetime import datetime, timedelta + +import pytest + +from homeassistant.util.variance import ignore_variance + + +@pytest.mark.parametrize( + "value_1, value_2, variance, expected", + [ + (1, 1, 1, 1), + (1, 2, 2, 1), + (1, 2, 0, 2), + (2, 1, 0, 1), + ( + datetime(2020, 1, 1, 0, 0), + datetime(2020, 1, 2, 0, 0), + timedelta(days=2), + datetime(2020, 1, 1, 0, 0), + ), + ( + datetime(2020, 1, 2, 0, 0), + datetime(2020, 1, 1, 0, 0), + timedelta(days=2), + datetime(2020, 1, 2, 0, 0), + ), + ( + datetime(2020, 1, 1, 0, 0), + datetime(2020, 1, 2, 0, 0), + timedelta(days=1), + datetime(2020, 1, 2, 0, 0), + ), + ], +) +def test_ignore_variance(value_1, value_2, variance, expected): + """Test ignore_variance.""" + with_ignore = ignore_variance(lambda x: x, variance) + assert with_ignore(value_1) == value_1 + assert with_ignore(value_2) == expected