Unofficial spoolman home assistant integration
"""Spoolman home assistant integration."""
import logging

import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_NAME, Platform
from homeassistant.core import HomeAssistant

from .const import CONF_URL, DEFAULT_NAME, DOMAIN, SPOOLMAN_API_WRAPPER
from .coordinator import SpoolManCoordinator

_LOGGER = logging.getLogger(__name__)

PLATFORMS = [Platform.SENSOR]

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
    {
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,  # type: ignore
        vol.Required(CONF_URL): cv.string,
    }
)


async def async_setup_platform(
    hass: HomeAssistant, config, add_devices, discovery_info=None
):
    """Set up Spoolman sensor."""
    _LOGGER.debug("__init__.async_setup_platform")


async def async_setup(hass: HomeAssistant, config):
    """Set up the Spoolman component."""
    _LOGGER.debug("__init__.async_setup")
    return True


async def async_setup_entry(hass: HomeAssistant, entry):
    """Set up the Spoolman component from a config entry."""
    _LOGGER.debug("__init__.async_setup_entry")

    coordinator = SpoolManCoordinator(hass, entry)
    await coordinator.async_refresh()
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
    return True


async def async_unload_entry(hass: HomeAssistant, entry):
    """Unload a config entry."""
    _LOGGER.debug("__init__.async_unload_entry")
    unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "sensor")
    if unload_ok:
        hass.data.pop(entry.domain)
    return unload_ok


async def async_get_data(hass: HomeAssistant):
    """Get the latest data from the Spoolman API."""
    _LOGGER.debug("__init__.async_get_data")
    return await hass.data[DOMAIN][SPOOLMAN_API_WRAPPER].get_spool(
        {"allow_archived": False}
    )

custom_components/spoolman/__init__.py
Disane87/spoolman-homeassistant
"""Class for interacting with the Spoolman API."""
import aiohttp


class SpoolmanAPI:
    """Class for interacting with the Spoolman API."""

    def __init__(self, base_url, api_version="v1"):
        """Initialize the Spoolman API."""
        self.base_url = f"{base_url}api/{api_version}"

    async def info(self):
        """Return information about the API."""
        url = f"{self.base_url}/info"
        async with aiohttp.ClientSession() as session, session.get(url) as response:
            response.raise_for_status()
            return await response.json()

    async def health(self):
        """Return the health status of the API."""
        url = f"{self.base_url}/health"
        async with aiohttp.ClientSession() as session, session.get(
            url, timeout=10
        ) as response:
            response.raise_for_status()
            return await response.json()

    async def backup(self):
        """Initiate a backup of the API."""
        url = f"{self.base_url}/backup"
        async with aiohttp.ClientSession() as session, session.post(url) as response:
            response.raise_for_status()
            return await response.json()

    async def get_spool(self, params):
        """Return a list of all spools."""
        url = f"{self.base_url}/spool"
        if len(params) > 0:
            url = f"{url}?{self.string_from_dictionary(params)}"
        async with aiohttp.ClientSession() as session, session.get(url) as response:
            response.raise_for_status()
            return await response.json()

    async def get_spool_by_id(self, spool_id):
        """Return the spool with the specified ID."""
        url = f"{self.base_url}/spool/{spool_id}"
        async with aiohttp.ClientSession() as session, session.get(url) as response:
            response.raise_for_status()
            return await response.json()

    def string_from_dictionary(self, params_dict):
        """Initialize an empty string to hold the result."""
        result_string = ""

        # Iterate through the dictionary and concatenate key-value pairs
        for key, value in params_dict.items():
            # Convert the value to a string if it's not already
            if not isinstance(value, str):
                value = str(value)

            # Concatenate the key and value in the desired format
            result_string += f"{key}={value} "

        # Remove the trailing space if needed
        result_string = result_string.strip()

        # Return the result string
        return result_string

custom_components/spoolman/classes/spoolman_api.py
Disane87/spoolman-homeassistant
"""Config flow for spoolman integration."""
from __future__ import annotations

from typing import Any

import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError

from .classes.spoolman_api import SpoolmanAPI
from .const import (
    CONF_NOTIFICATION_THRESHOLD_CRITICAL,
    CONF_NOTIFICATION_THRESHOLD_INFO,
    CONF_NOTIFICATION_THRESHOLD_WARNING,
    CONF_SHOW_ARCHIVED,
    CONF_UPDATE_INTERVAL,
    CONF_URL,
    DOMAIN,
    NOTIFICATION_THRESHOLDS,
    SPOOLMAN_INFO_PROPERTY,
)


class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for spoolman."""

    VERSION = 1

    def add_trailing_slash(self, input_string):
        """Add traling slashed when not present."""
        if not input_string.endswith("/"):
            input_string += "/"
        return input_string

    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:
            url = self.add_trailing_slash(user_input.get(CONF_URL))

            # Test the API key and URLs here if necessary
            # If valid, create an entry
            # If invalid, set errors
            if not errors:
                spoolman_api = SpoolmanAPI(url)
                try:
                    healty = await spoolman_api.health()
                    if healty.get("status", None) == "healthy":
                        info = await spoolman_api.info()

                        return self.async_create_entry(
                            title=DOMAIN,
                            data={
                                **user_input,
                                CONF_URL: url,
                                SPOOLMAN_INFO_PROPERTY: info,
                            },
                        )

                except Exception as error_message:
                    errors[CONF_URL] = f"Error testing URL: {str(error_message)}"

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {
                    vol.Required(CONF_URL): str,
                    vol.Optional(CONF_UPDATE_INTERVAL, default=15): vol.All(  # type: ignore
                        vol.Coerce(int), vol.Range(min=1)
                    ),
                    vol.Required(
                        CONF_NOTIFICATION_THRESHOLD_INFO,
                        default=NOTIFICATION_THRESHOLDS.get("info", 0),  # type: ignore
                    ): vol.All(int, vol.Range(min=0, max=100)),
                    vol.Required(
                        CONF_NOTIFICATION_THRESHOLD_WARNING,
                        default=NOTIFICATION_THRESHOLDS.get("warning", 0),  # type: ignore
                    ): vol.All(int, vol.Range(min=0, max=100)),
                    vol.Required(
                        CONF_NOTIFICATION_THRESHOLD_CRITICAL,
                        default=NOTIFICATION_THRESHOLDS.get("critical", 0),  # type: ignore
                    ): vol.All(int, vol.Range(min=0, max=100)),
                    vol.Required(CONF_SHOW_ARCHIVED): bool,
                }
            ),
            errors=errors,
        )


class CannotConnect(HomeAssistantError):
    """Error to indicate we cannot connect."""


class InvalidAuth(HomeAssistantError):
    """Error to indicate there is invalid auth."""

custom_components/spoolman/config_flow.py
Disane87/spoolman-homeassistant
"""Constants for the spoolman integration."""

DOMAIN = "spoolman"
DEFAULT_NAME = "Spoolman"
SPOOLMAN_INFO_PROPERTY = "spoolman_info"

CONF_URL = "spoolman_url"
CONF_SHOW_ARCHIVED = "show_archived"
CONF_UPDATE_INTERVAL = "update-interval"
CONF_NOTIFICATION_THRESHOLD_INFO = "notification_threshold_info"
CONF_NOTIFICATION_THRESHOLD_WARNING = "notification_threshold_warning"
CONF_NOTIFICATION_THRESHOLD_CRITICAL = "notification_threshold_critical"

PUBLIC_IMAGE_PATH = "www/spoolman_images"
LOCAL_IMAGE_PATH = "/local/spoolman_images"

EVENT_THRESHOLD_EXCEEDED = "spoolman_spool_threshold_exceeded"

NOTIFICATION_THRESHOLDS = {"critical": 95, "warning": 75, "info": 50}
SPOOLMAN_API_WRAPPER = "spoolman_api_wrapper"

custom_components/spoolman/const.py
Disane87/spoolman-homeassistant
"""Spoolman home assistant data coordinator."""
import logging
from datetime import timedelta

from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .classes.spoolman_api import SpoolmanAPI
from .const import (
    CONF_SHOW_ARCHIVED,
    CONF_UPDATE_INTERVAL,
    CONF_URL,
    DOMAIN,
    SPOOLMAN_API_WRAPPER,
)

_LOGGER = logging.getLogger(__name__)


class SpoolManCoordinator(DataUpdateCoordinator):
    """My custom coordinator."""

    def __init__(self, hass: HomeAssistant, entry) -> None:
        """Initialize my coordinator."""
        _LOGGER.debug("SpoolManCoordinator.__init__")
        url = entry.data[CONF_URL]
        update_interval = entry.data[CONF_UPDATE_INTERVAL]

        super().__init__(
            hass,
            _LOGGER,
            name="Spoolman data",
            update_interval=timedelta(minutes=update_interval),
        )
        self.my_api = url
        self.hass = hass
        self.spoolman_api = SpoolmanAPI(entry.data[CONF_URL])

        hass.data[DOMAIN] = {
            **entry.data,
            SPOOLMAN_API_WRAPPER: self.spoolman_api,
            "coordinator": self,
        }

    async def _async_update_data(self):
        _LOGGER.debug("SpoolManCoordinator._async_update_data")
        config = self.hass.data[DOMAIN]

        show_archived = config.get(CONF_SHOW_ARCHIVED, False)

        try:
            spools = await self.spoolman_api.get_spool(
                {"allow_archived": show_archived}
            )
            return spools

        except Exception as exception:
            raise UpdateFailed(
                f"Error fetching data from Spoolman API: {exception}"
            ) from exception

custom_components/spoolman/coordinator.py
Disane87/spoolman-homeassistant
"""Spoolman home assistant sensor."""
import logging
import os

from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.components.sensor.const import SensorStateClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    UnitOfMass,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from PIL import Image

from .const import (
    CONF_URL,
    DOMAIN,
    EVENT_THRESHOLD_EXCEEDED,
    LOCAL_IMAGE_PATH,
    NOTIFICATION_THRESHOLDS,
    PUBLIC_IMAGE_PATH,
    SPOOLMAN_INFO_PROPERTY,
)
from .coordinator import SpoolManCoordinator

_LOGGER = logging.getLogger(__name__)


ICON = "mdi:printer-3d-nozzle"


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
):
    """Spoolman home assistant sensor init."""
    coordinator = SpoolManCoordinator(hass, config_entry)
    await coordinator.async_config_entry_first_refresh()

    if coordinator.data:
        spool_entities = []
        for idx, spool_data in enumerate(coordinator.data):
            spool_device = Spool(hass, coordinator, spool_data, idx, config_entry)
            spool_entities.append(spool_device)
        async_add_entities(spool_entities)


class Spool(CoordinatorEntity, SensorEntity):
    """Representation of a Spoolman Sensor."""

    def __init__(self, hass, coordinator, spool_data, idx, config_entry) -> None:
        """Spoolman home assistant spool sensor init."""
        super().__init__(coordinator)

        self.config = hass.data[DOMAIN]
        conf_url = self.config[CONF_URL]
        spoolman_info = self.config[SPOOLMAN_INFO_PROPERTY]

        self._spool = spool_data
        self.handled_threshold_events = []
        self._filament = self._spool["filament"]
        self._entry = config_entry
        self._attr_name = f"{self._filament['vendor']['name']} {self._filament['name']} {self._filament['material']}"
        self._attr_unique_id = f"{self._attr_name}_{spool_data['id']}"
        self._attr_has_entity_name = False
        self._attr_device_class = SensorDeviceClass.WEIGHT
        self._attr_state_class = SensorStateClass.MEASUREMENT
        self._attr_native_unit_of_measurement = UnitOfMass.GRAMS
        self._attr_icon = ICON

        location_name = (
            self._spool["location"] if spool_data["archived"] is False else "Archived"
        )

        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, conf_url, location_name)},  # type: ignore
            name=location_name,
            manufacturer="https://github.com/Donkie/Spoolman",
            model="Spoolman",
            configuration_url=conf_url,
            suggested_area=location_name,
            sw_version=f"{spoolman_info.get('version', 'unknown')} ({spoolman_info.get('git_commit', 'unknown')})",
        )
        self.idx = idx

    @callback
    def _handle_coordinator_update(self) -> None:
        """Handle updated data from the coordinator."""
        self._spool = self.coordinator.data[self.idx]

        self._filament = self._spool["filament"]
        self._spool["used_percentage"] = (
            round(self._spool["used_weight"] / self._filament["weight"], 3) * 100
        )

        if self._spool["archived"] is False:
            self.check_for_threshold(self._spool, self._spool["used_percentage"])

        self.async_write_ha_state()

    @property
    def extra_state_attributes(self):
        """Return the attributes of the sensor."""
        spool = self._spool

        return self.flatten_dict(spool)

    def check_for_threshold(self, spool, used_percentage):
        """Check if the used percentage is above a threshold and fire an event if it is."""
        for key, _value in sorted(
            NOTIFICATION_THRESHOLDS.items(), key=lambda x: x[1], reverse=True
        ):
            threshold_name = key
            config_threshold = self.config[f"notification_threshold_{threshold_name}"]

            if threshold_name in self.handled_threshold_events:
                _LOGGER.debug(
                    "SpoolManCoordinator.check_for_threshold: '%s' already handled for spool '%s' in '%s' with '%s'",
                    threshold_name,
                    self._attr_name,
                    self._spool["location"],
                    used_percentage,
                )
                break

            if used_percentage >= config_threshold:
                _LOGGER.debug(
                    "SpoolManCoordinator.check_for_threshold: '%s' reached for spool '%s' in '%s' with '%s'",
                    threshold_name,
                    self._attr_name,
                    self._spool["location"],
                    used_percentage,
                )
                self.hass.bus.fire(
                    EVENT_THRESHOLD_EXCEEDED,
                    {
                        "entity_id": self.entity_id,
                        "spool": spool,
                        "threshold_name": threshold_name,
                        "threshold_value": config_threshold,
                        "used_percentage": used_percentage,
                    },
                )
                self.handled_threshold_events.append(threshold_name)
                break

    def flatten_dict(self, d, parent_key="", sep="_"):
        """Flattens a dictionary."""
        flat_dict = {}
        if isinstance(d, dict):
            for key, value in d.items():
                new_key = f"{parent_key}{sep}{key}" if parent_key else key
                if isinstance(value, dict):
                    # Wenn der Wert ein Dictionary ist, rufen Sie die Funktion rekursiv auf
                    flat_dict.update(self.flatten_dict(value, new_key, sep=sep))
                elif isinstance(value, str):
                    # Wenn der Wert ein String ist, trimmen Sie ihn
                    flat_dict[new_key] = value.strip()
                else:
                    flat_dict[new_key] = value
            return flat_dict

        else:
            return {}

    @property
    def state(self):
        """Return the state of the sensor."""

        return round(self._spool["remaining_weight"], 3)

    @property
    def entity_picture(self):
        """Return the entity picture."""
        filament = self._spool["filament"]
        color_hex = filament["color_hex"]
        return self.generate_entity_picture(color_hex)

    def generate_entity_picture(self, color_hex):
        """Generate an entity picture with the specified color and save it to the www directory."""
        image = Image.new("RGB", (100, 100), f"#{color_hex}")
        image_name = f"spool_{self._spool['id']}.png"

        # Define the directory path
        image_dir = self.hass.config.path(PUBLIC_IMAGE_PATH)

        # Check if the directory exists, and create it if it doesn't
        if not os.path.exists(image_dir):
            os.makedirs(image_dir)

        image_path = os.path.join(image_dir, image_name)
        image.save(image_path)

        # Get the URL for the saved image
        image_url = f"{LOCAL_IMAGE_PATH}/{image_name}"

        return image_url

    async def async_update(self):
        """Fetch the latest data from the coordinator."""
        await self.coordinator.async_request_refresh()

custom_components/spoolman/sensor.py
Disane87/spoolman-homeassistant