Hi guys! I’m pretty new to Python and the development of new integrations. Yesterday I started digging into your dev docs and got an example project up and running. I’m a senior developer with overall 15 years of experience, so I’m not that new to software development, but I guess I’m missing something at the moment.
I want to integrate my spool manager for my 3d printer into HA by calling their API (GitHub - Donkie/Spoolman: Keep track of your inventory of 3D-printer filament spools.). That works flawlessly and all sensors are created. But, all sensors aren’t added to the integration. The integration itself is working.
What am I missing at the moment?
My __init__.py
import asyncio
import logging
import aiohttp
import websockets
import json
from homeassistant.const import Platform
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import DOMAIN
from .coordinator import SpoolManCoordinator
from .sensor import SpoolSensor # Annahme: Dies ist die Sensor-Klasse
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.SENSOR]
async def async_setup(hass, config):
"""Set up the Spoolman component."""
_LOGGER.info("__init__.async_setup")
return True
async def async_setup_entry(hass, entry):
"""Set up the Spoolman component from a config entry."""
_LOGGER.info("__init__.async_setup_entry")
session = async_create_clientsession(hass)
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, entry):
_LOGGER.info("__init__.async_unload_entry")
"""Unload a config 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):
_LOGGER.info("__init__.async_get_data")
"""Get the latest data from the Spoolman API."""
api_key = hass.data[DOMAIN]["api_key"]
url = hass.data[DOMAIN]["url"]
async with aiohttp.ClientSession() as session:
try:
async with session.get(
url, headers={"Authorization": f"Bearer {api_key}"}
) as response:
data = await response.json()
return data
except Exception as e:
_LOGGER.error(f"Error fetching data from Spoolman API: {e}")
async def async_setup_websocket(hass, websocket_url, callback):
_LOGGER.info("__init__.async_setup_websocket")
"""Set up a WebSocket connection for real-time updates."""
while True:
try:
async with websockets.connect(websocket_url) as ws:
while True:
data = await ws.recv()
data_json = json.loads(data)
await callback()
except Exception as e:
_LOGGER.error(f"WebSocket error: {e}")
My config_flow.py
"""Config flow for spoolman integration."""
from __future__ import annotations
from typing import Any
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 .const import (
DOMAIN,
CONF_API_KEY,
CONF_UPDATE_INTERVAL,
CONF_URL,
CONF_WEBSOCKET_URL,
)
# TODO adjust the data schema to the data that you need
# STEP_USER_DATA_SCHEMA =
class PlaceholderHub:
"""Placeholder class to make tests pass.
TODO Remove this placeholder class and replace with things from your PyPI package.
"""
def __init__(self, host: str) -> None:
"""Initialize."""
self.host = host
async def authenticate(self, username: str, password: str) -> bool:
"""Test if we can authenticate with the host."""
return True
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.
"""
# TODO validate the data can be used to set up a connection.
# If your PyPI package is not built with async, pass your methods
# to the executor:
# await hass.async_add_executor_job(
# your_validate_func, data["username"], data["password"]
# )
hub = PlaceholderHub(data["host"])
if not await hub.authenticate(data["username"], data["password"]):
raise InvalidAuth
# If you cannot connect:
# throw CannotConnect
# If the authentication is wrong:
# InvalidAuth
# Return info that you want to store in the config entry.
return {"title": "Name of the device"}
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for spoolman."""
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:
api_key = user_input[CONF_API_KEY]
url = user_input[CONF_URL]
update_interval = user_input.get(CONF_UPDATE_INTERVAL)
websocket_url = user_input.get(CONF_WEBSOCKET_URL) or f"wss://{url}"
# Test the API key and URLs here if necessary
# If valid, create an entry
# If invalid, set errors
if not errors:
return self.async_create_entry(
title="Spoolman",
data={
CONF_API_KEY: api_key,
CONF_URL: url,
CONF_UPDATE_INTERVAL: update_interval,
CONF_WEBSOCKET_URL: websocket_url,
},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_API_KEY, "blabla", "sdfsdfdsdsf"): str,
vol.Required(
CONF_URL, "blabla", "https://spoolman.disane.dev"
): str,
vol.Optional(CONF_UPDATE_INTERVAL, default=15): vol.All(
vol.Coerce(int), vol.Range(min=1)
), # In minutes
vol.Optional(
CONF_WEBSOCKET_URL, "Websocket", "wss://spoolman.disane.dev"
): str,
}
),
errors=errors,
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""
My sensor.py
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.const import LENGTH_MILLIMETERS
from PIL import Image
from .coordinator import SpoolManCoordinator
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
ICON = "mdi:printer-3d-nozzle"
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
):
_LOGGER.info("Sensor.async_setup_entry")
"""Set up the Spoolman sensor platform."""
coordinator = hass.data[DOMAIN]["coordinator"]
spool_data = coordinator.data
if spool_data:
sensors = [
SpoolSensor(coordinator, spool, config_entry) for spool in spool_data
]
async_add_entities(sensors)
class SpoolSensor(CoordinatorEntity[SpoolManCoordinator], SensorEntity):
"""Representation of a Spoolman Sensor."""
def __init__(
self, coordinator: SpoolManCoordinator, spool, config_entry: ConfigEntry
):
_LOGGER.info("SpoolSensor.__init__")
super().__init__(coordinator)
self._spool = spool
self._filament = self._spool["filament"]
self._entry = config_entry
@property
def name(self):
"""Return the name of the sensor."""
return f"{self._filament['name']}"
@property
def state(self):
"""Return the state of the sensor."""
return round(self._spool["remaining_weight"], 2)
@property
def device_class(self):
"""Return the device class of the sensor."""
return SensorDeviceClass.DISTANCE
@property
def state_class(self):
"""Return the state class of the sensor."""
return SensorStateClass.MEASUREMENT
@property
def unit_of_measurement(self):
"""Return the unit of measurement of the sensor."""
return LENGTH_MILLIMETERS
@property
def icon(self):
"""Return the icon for the sensor."""
return ICON
@property
def device_info(self):
"""Return device information for the sensor."""
return DeviceInfo(
identifiers={(DOMAIN, self._entry.entry_id)},
manufacturer="Spoolman",
name=self.name,
)
@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"
image_path = self.hass.config.path(f"www/spoolman_images/{image_name}")
image.save(image_path)
# Get the URL for the saved image
image_url = f"/local/spoolman_images/{image_name}"
return image_url
@property
def extra_state_attributes(self):
"""Return the attributes of the sensor."""
spool = self._spool
attributes = {}
for key, value in spool.items():
if isinstance(value, dict):
# If the value is a dictionary, iterate over the keys and values
for sub_key, sub_value in value.items():
attributes[f"{key}_{sub_key}"] = sub_value
elif isinstance(value, str):
# If the value is a string, trim it
attributes[key] = value.strip()
else:
attributes[key] = value
return attributes
async def async_update(self):
"""Fetch the latest data from the coordinator."""
await self.hass.data[DOMAIN]["coordinator"].async_request_refresh()
coordinator.py
from .const import (
DOMAIN,
CONF_API_KEY,
CONF_UPDATE_INTERVAL,
CONF_URL,
CONF_WEBSOCKET_URL,
)
import aiohttp
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
import logging
from datetime import timedelta
_LOGGER = logging.getLogger(__name__)
class SpoolManCoordinator(DataUpdateCoordinator):
"""My custom coordinator."""
def __init__(self, hass, entry):
_LOGGER.info("SpoolManCoordinator.__init__")
api_key = entry.data[CONF_API_KEY]
url = f"{entry.data[CONF_URL]}/api/v1/spool"
update_interval = entry.data[CONF_UPDATE_INTERVAL]
websocket_url = entry.data.get(CONF_WEBSOCKET_URL) or f"wss://{url}"
"""Initialize my coordinator."""
super().__init__(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name="Spoolman data",
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(seconds=update_interval),
)
self.my_api = url
self.hass = hass
hass.data[DOMAIN] = {
"api_key": api_key,
"url": url,
"coordinator": self, # ,
}
async def _async_update_data(self):
_LOGGER.info("SpoolManCoordinator._async_update_data")
api_key = self.hass.data[DOMAIN]["api_key"]
url = self.hass.data[DOMAIN]["url"]
async with aiohttp.ClientSession() as session:
try:
async with session.get(
url, headers={"Authorization": f"Bearer {api_key}"}
) as response:
data = await response.json()
return data
except Exception as e:
raise UpdateFailed(f"Error fetching data from Spoolman API: {e}")
manifest.json
{
"domain": "spoolman",
"name": "spoolman",
"codeowners": [
"@mfranke87"
],
"config_flow": true,
"dependencies": [],
"documentation": "https://www.home-assistant.io/integrations/spoolman",
"homekit": {},
"iot_class": "local_polling",
"requirements": [],
"ssdp": [],
"zeroconf": []
}
But the integration is empty:
But the sensors are created:
Please don’t be mad at me, I’m actually learning python and the ecosystem of HA 
I guess this is a little thing I’ve forgot to set but I can’t find anything. Even a simple almost boilerplate sensor integration isn’t working.
Kind regards!