Add PrusaLink integration (home-assistant#77429)

Co-authored-by: Martin Hjelmare <[email protected]>

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

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

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,
+        )

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

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

homeassistant/components/prusalink/const.py

@@ -0,0 +1,3 @@
+"""Constants for the PrusaLink integration."""
+
+DOMAIN = "prusalink"

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