Initial commit

.github/workflows/validate.yml

@@ -0,0 +1,19 @@
+name: Validate
+
+on:
+  push:
+  pull_request:
+  schedule:
+    - cron: "0 0 * * *"
+  workflow_dispatch:
+
+jobs:
+  validate-hacs:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: HACS validation
+        uses: hacs/action@main
+        with:
+          category: integration

README.md

@@ -0,0 +1,8 @@
+# Immich integration for Home Assistant
+
+This custom integration for Home Assistant allows you to display random pictures from your Immich instance inside Home Assistant.
+
+The pictures will be provided under an `image` entity, which should work with other Home Assistant components, like the `picture` card.
+
+A random image will be fetched every 5 minutes.
+The images are selected from your Favorites on your Immich instance.

custom_components/immich/__init__.py

@@ -0,0 +1,39 @@
+"""The immich integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.const import CONF_HOST, CONF_API_KEY
+from homeassistant.core import HomeAssistant
+from datetime import timedelta
+
+from .const import DOMAIN
+from .hub import ImmichHub, InvalidAuth
+
+PLATFORMS: list[Platform] = [Platform.IMAGE]
+SCAN_INTERVAL = timedelta(minutes=5)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up immich from a config entry."""
+
+    hass.data.setdefault(DOMAIN, {})
+
+    hub = ImmichHub(host=entry.data[CONF_HOST], api_key=entry.data[CONF_API_KEY])
+
+    if not await hub.authenticate():
+        raise InvalidAuth
+
+    hass.data[DOMAIN][entry.entry_id] = hub
+
+    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

custom_components/immich/config_flow.py

@@ -0,0 +1,76 @@
+"""Config flow for Immich integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+from url_normalize import url_normalize
+from urllib.parse import urljoin
+from urllib.parse import urlparse
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST, CONF_API_KEY
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN
+from .hub import ImmichHub, InvalidAuth, CannotConnect
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_API_KEY): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+
+    url = url_normalize(data[CONF_HOST])
+    api_key = data[CONF_API_KEY]
+
+    hub = ImmichHub(host=url, api_key=api_key)
+
+    if not await hub.authenticate():
+        raise InvalidAuth
+
+    # Return info that you want to store in the config entry.
+    return {
+        "title": urlparse(url).hostname,
+        "data": {CONF_HOST: url, CONF_API_KEY: api_key},
+    }
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for immich."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            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
+        )

custom_components/immich/const.py

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

custom_components/immich/hub.py

@@ -0,0 +1,110 @@
+"""Hub for Immich integration."""
+from __future__ import annotations
+
+import aiohttp
+import logging
+from urllib.parse import urljoin
+import random
+
+from homeassistant.exceptions import HomeAssistantError
+
+_HEADER_API_KEY = "x-api-key"
+_LOGGER = logging.getLogger(__name__)
+
+
+class ImmichHub:
+    """Immich API hub."""
+
+    def __init__(self, host: str, api_key: str) -> None:
+        """Initialize."""
+        self.host = host
+        self.api_key = api_key
+
+    async def authenticate(self) -> bool:
+        """Test if we can authenticate with the host."""
+        try:
+            async with aiohttp.ClientSession() as session:
+                url = urljoin(self.host, "/api/auth/validateToken")
+                headers = {"Accept": "application/json", _HEADER_API_KEY: self.api_key}
+
+                async with session.post(url=url, headers=headers) as response:
+                    if response.status != 200:
+                        raw_result = await response.text()
+                        _LOGGER.error("Error from API: body=%s", raw_result)
+                        return False
+
+                    json_result = await response.json()
+
+                    if not json_result.get("authStatus"):
+                        raw_result = await response.text()
+                        _LOGGER.error("Error from API: body=%s", raw_result)
+                        return False
+
+                    return True
+        except aiohttp.ClientError as exception:
+            _LOGGER.error("Error connecting to the API: %s", exception)
+            raise CannotConnect from exception
+
+    async def get_random_picture(self) -> dict | None:
+        """Get a random picture from the API."""
+        assets = [
+            asset for asset in await self._list_favorites() if asset["type"] == "IMAGE"
+        ]
+
+        if not assets:
+            _LOGGER.error("No assets found in favorites")
+            return None
+
+        # Select random item in list
+        random_asset = random.choice(assets)
+
+        _LOGGER.debug("Random asset: %s", random_asset)
+        return random_asset
+
+    async def download_asset(self, asset_id: str) -> bytes:
+        """Download the asset."""
+        try:
+            async with aiohttp.ClientSession() as session:
+                url = urljoin(self.host, f"/api/asset/file/{asset_id}")
+                headers = {_HEADER_API_KEY: self.api_key}
+
+                async with session.get(url=url, headers=headers) as response:
+                    if response.status != 200:
+                        _LOGGER.error("Error from API: status=%d", response.status)
+                        raise ApiError()
+
+                    return await response.read()
+        except aiohttp.ClientError as exception:
+            _LOGGER.error("Error connecting to the API: %s", exception)
+            raise CannotConnect from exception
+
+    async def _list_favorites(self) -> list[dict]:
+        try:
+            async with aiohttp.ClientSession() as session:
+                url = urljoin(self.host, "/api/asset?isFavorite=true")
+                headers = {"Accept": "application/json", _HEADER_API_KEY: self.api_key}
+
+                async with session.get(url=url, headers=headers) as response:
+                    if response.status != 200:
+                        raw_result = await response.text()
+                        _LOGGER.error("Error from API: body=%s", raw_result)
+                        raise ApiError()
+
+                    json_result = await response.json()
+
+                    return json_result
+        except aiohttp.ClientError as exception:
+            _LOGGER.error("Error connecting to the API: %s", exception)
+            raise CannotConnect from exception
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class ApiError(HomeAssistantError):
+    """Error to indicate that the API returned an error."""

custom_components/immich/image.py

@@ -0,0 +1,71 @@
+"""Image device for Immich integration."""
+from __future__ import annotations
+
+from datetime import datetime
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from .hub import ImmichHub
+from homeassistant.components.image import ImageEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_API_KEY
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+_LOGGER = logging.getLogger(__name__)
+
+REFRESH_INTERVAL = 10
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Immich image platform."""
+
+    hub = ImmichHub(
+        host=config_entry.data[CONF_HOST], api_key=config_entry.data[CONF_API_KEY]
+    )
+
+    async_add_entities([ImmichImage(hass, hub)])
+
+
+class ImmichImage(ImageEntity):
+    """A class to let you visualize the map."""
+
+    _attr_unique_id = "favorite_image"
+    _attr_has_entity_name = True
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    # We want to get a new image every so often, as defined by the refresh interval
+    _attr_should_poll = True
+
+    _cached_bytes = None
+
+    def __init__(self, hass: HomeAssistant, hub: ImmichHub) -> None:
+        """Initialize the Immich image entity."""
+        super().__init__(hass=hass, verify_ssl=True)
+        self.hub = hub
+        self.hass = hass
+
+    async def async_update(self) -> None:
+        """Update the image entity data."""
+        await self._load_and_cache_image()
+
+    async def async_image(self) -> bytes | None:
+        """Return a random image from the Immich API."""
+        if not self._cached_bytes:
+            await self._load_and_cache_image()
+
+        return self._cached_bytes
+
+    async def _load_and_cache_image(self) -> None:
+        random_asset = await self.hub.get_random_picture()
+
+        if random_asset:
+            asset_bytes = await self.hub.download_asset(random_asset["id"])
+            self._cached_bytes = asset_bytes
+            self._attr_image_last_updated = datetime.now()

custom_components/immich/manifest.json

@@ -0,0 +1,18 @@
+{
+  "domain": "immich",
+  "name": "Immich",
+  "version": "0.0.1",
+  "codeowners": ["@outadoc"],
+  "config_flow": true,
+  "dependencies": [],
+  "documentation": "https://www.home-assistant.io/integrations/immich",
+  "homekit": {},
+  "iot_class": "cloud_polling",
+  "requirements": [
+    "url-normalize==1.4.3",
+    "aiohttp==3.9.3",
+    "urllib3>=1.26.5,<2"
+  ],
+  "ssdp": [],
+  "zeroconf": []
+}

custom_components/immich/strings.json

@@ -0,0 +1,20 @@
+{
+  "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%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

custom_components/immich/translations/en.json

@@ -0,0 +1,20 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_auth": "Invalid authentication",
+            "unknown": "Unexpected error"
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "api_key": "API key",
+                    "host": "Host"
+                }
+            }
+        }
+    }
+}