Writing An Integration For Home Assistant
In an earlier post, I mentioned that I wished there was a bass and treble adjustment in the CamillaDSP integration for home assistant. There was another problem I wished it could have solved. You see, for the integration to work, both CamillaDSP and CamillaDSP GUI servers need to be running. I did not want the CamillaDSP GUI server to be running all the time just to be able to change the configs. The CamillaDSP server itself is capable of switching configs. Just that the home assistant integration did not use those APIs.
So like any good engineer, I decided to write the integration myself :). Let me tell you something upfront. Understanding the lifecycle of a Home Assistant integration is not easy. I had to read random blogs and forums to get a basic understanding. Then I read a lot of code to come up with my own integration. It is not clean, but it lets me do everything I want without needing an extra CamillaDSP server!
For the development of the integration, I ran an instance of home assistant on my local dev machine so I can iterate quickly. Here is the docker compose file I used for development.
version: '3.0'
services:
home-assistant:
container_name: home-assistant
image: "ghcr.io/home-assistant/home-assistant:stable"
restart: unless-stopped
network_mode: host
privileged: true
init: false
volumes:
- /home/chandanp/workspace/home_assistant:/config
- /etc/localtime:/etc/localtime:ro
I ran the container with podman compose up. Next I created home_assistant/custom_components/my_camilladsp and copied over the code from https://github.com/HEnquist/pycamilladsp/tree/master/camilladsp to home_assistant/custom_components/my_camilladsp/camilladsp. That is the API created by the author of CamillaDSP to be able to talk to CamillaDSP server.
Rest of the files that I am going to mention below are all going under home_assistant/custom_components/my_camilladsp/. The first file I added was manifest.json with the following content.
{
"domain": "mycamilladsp",
"name": "My Camilla DSP",
"codeowners": ["@re-ynd"],
"iot_class": "local_polling",
"config_flow": true,
"dependencies": [],
"requirements": ["websocket-client", "PyYAML"],
"version": "1.0.0"
}
The manifest gives basic information like the name, dependencies, version number etc of my integration to Home Assistant. I added websocket-client and PyYAML as dependencies because they are required by the CamillaDSP API. Next I added a few constants in const.py.
DOMAIN = "mycamilladsp"
NAME = "Camilla DSP"
Then I added strings.json with the following content
{
"config": {
"title": "Camilla DSP",
"error": {
"cannot_connect": "Failed to connect!",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"host": "Host",
"port": "Port"
}
}
}
}
}
To connect to the CamillaDSP server, we need to know the host and port of the server. I could have hardcoded it, but anyway went with a config.py file.
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.const import (
CONF_HOST,
CONF_PORT,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from .camilla_dsp_api import CamillaDspApi, APIConnectionError
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST, default="192.168.0.77", description={"suggested_value": "192.168.0.77"}): str,
vol.Required(CONF_PORT, default="1234", description={"suggested_value": "1234"}): str,
}
)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
api = CamillaDspApi(data[CONF_HOST], data[CONF_PORT])
try:
api.connect()
except APIConnectionError as err:
raise CannotConnect from err
return {"title": f"Camilla DSP - {data[CONF_HOST]}:{data[CONF_PORT]}"}
class CamillaDspConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
_input_data: dict[str, Any]
@staticmethod
@callback
def async_get_options_flow(config_entry):
return OptionsFlowHandler(config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
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 Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
if "base" not in errors:
await self.async_set_unique_id(info.get("title"))
self._abort_if_unique_id_configured()
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 OptionsFlowHandler(OptionsFlow):
def __init__(self, config_entry: ConfigEntry) -> None:
self.config_entry = config_entry
self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None):
if user_input is not None:
options = self.config_entry.options | user_input
return self.async_create_entry(title="", data=options)
return self.async_show_form(step_id="init", data_schema=STEP_USER_DATA_SCHEMA)
class CannotConnect(HomeAssistantError):
I copied most of the code from https://github.com/msp1974/HAIntegrationExamples/blob/main/msp_integration_101_template/README.md. But if you write a basic __init.py__ file, you can now install the integration in Home Assistant and configure it. For debugging, I was monitoring the Home Assistant logs which you get to see in the terminal where you are running podman compose up.
Another way to see only the logs from your integration is to add the following lines to home_assistant/configuration.yaml and restarting Home Assistant.
logger:
default: info
logs:
custom_components.my_camilladsp: debug
Remember to change my_camilladsp with your package directory. Then you can tail -F home-assistant.log | grep -i "camilla" to see your logs. Anyway, this is how the UI looks like when you add the integration in Home Assistant.
Next we need to add a proper __init__.py with the following content.
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .coordinator import CamillaDspCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.MEDIA_PLAYER,
Platform.NUMBER,
]
type MyConfigEntry = ConfigEntry[RuntimeData]
@dataclass
class RuntimeData:
coordinator: DataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, config_entry: MyConfigEntry) -> bool:
coordinator = CamillaDspCoordinator(hass, config_entry)
await coordinator.async_config_entry_first_refresh()
if not coordinator.api.connected:
raise ConfigEntryNotReady
config_entry.async_on_unload(
config_entry.add_update_listener(_async_update_listener)
)
config_entry.runtime_data = RuntimeData(coordinator)
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
return True
async def _async_update_listener(hass: HomeAssistant, config_entry):
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_remove_config_entry_device(
hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry
) -> bool:
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: MyConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)
I am basically letting Home Assistant know that I am integrating a MEDIA_PLAYER and NUMBER entities. The media player entity is the one that will let me switch the config files from music to movie etc. The number entities will let me change the bass and treble configuration in CamillaDSP. Now comes the meat of the integration which is the class that makes the API calls to CamillaDSP server camilla_dsp_api.py.
import logging
from enum import StrEnum
from homeassistant.components.media_player import MediaPlayerState
from .camilladsp.camilladsp import CamillaClient
from .camilladsp.general import ProcessingState
_LOGGER = logging.getLogger(__name__)
class CamillaDspConfig(StrEnum):
DEFAULT = "default"
MUSIC = "music"
MOVIE = "movie"
class CamillaDspEq(StrEnum):
BASS = "Bass"
TREBLE = "Treble"
class CamillaDspApi:
def __init__(self, host: str, port: str) -> None:
self.host = host
self.port = port
self.connected: bool = False
self.client = CamillaClient(self.host, self.port)
@property
def controller_name(self) -> str:
return self.host.replace(".", "_")
def connect(self) -> bool:
camilla_dsp_info = "CamillaDSP (" + self.host + ":" + self.port + ")"
_LOGGER.debug("Connecting to " + camilla_dsp_info)
try:
self.client.connect()
_LOGGER.info("Connected to " + camilla_dsp_info)
self.connected = True
return True
except Exception as e:
_LOGGER.error("Failed to connect to " + camilla_dsp_info)
raise APIConnectionError("Unable to connect to " + camilla_dsp_info)
def disconnect(self) -> bool:
camilla_dsp_info = "CamillaDSP (" + self.host + ":" + self.port + ")"
_LOGGER.debug("Disconnecting from " + camilla_dsp_info)
self.client.disconnect()
self.connected = False
return True
def get_state(self) -> str:
state = MediaPlayerState.OFF
match self.client.general.state():
case ProcessingState.INACTIVE:
state = MediaPlayerState.STANDBY
case ProcessingState.PAUSED:
state = MediaPlayerState.PAUSED
case ProcessingState.RUNNING:
state = MediaPlayerState.PLAYING
case ProcessingState.STALLED:
state = MediaPlayerState.IDLE
case ProcessingState.STARTING:
state = MediaPlayerState.ON
return state
def get_volume(self) -> float:
return self.client.volume.main_volume()
def set_volume(self, volume: float):
self.client.volume.set_main_volume(volume)
self._volume = volume
def get_mute(self) -> bool:
return self.client.volume.main_mute()
def set_mute(self, mute: bool):
self.client.volume.set_main_mute(mute)
self._mute = mute
def get_source(self) -> str:
source = self.client.config.file_path()
if source.endswith("music.yml"):
return CamillaDspConfig.MUSIC
elif source.endswith("movie.yml"):
return CamillaDspConfig.MOVIE
else:
return CamillaDspConfig.DEFAULT
def set_source(self, source: str) -> None:
base = "/storage/workspace/camilladsp/configs/"
match source:
case CamillaDspConfig.MUSIC:
self.client.config.set_file_path(base + "51-rew-80hz-music.yml")
case CamillaDspConfig.MOVIE:
self.client.config.set_file_path(base + "51-rew-80hz-movie.yml")
case _:
self.client.config.set_file_path(base + "51-rew-80hz.yml")
self.client.general.reload()
def get_sources(self) -> list[str]:
return [CamillaDspConfig.DEFAULT, CamillaDspConfig.MUSIC, CamillaDspConfig.MOVIE]
def get_eq(self) -> dict[CamillaDspEq, float]:
config = self.client.config.active()
return {
CamillaDspEq.BASS: config["filters"][CamillaDspEq.BASS]["parameters"]["gain"],
CamillaDspEq.TREBLE: config["filters"][CamillaDspEq.TREBLE]["parameters"]["gain"]
}
def set_eq(self, filter, db_value):
config = self.client.config.active()
config["filters"][filter]["parameters"]["gain"] = db_value
self.client.config.set_active(config)
class APIConnectionError(Exception):
For details about the API, see the pyCamillaDSP documentation. Next we have coordinator.py which updates the local state with data from CamillaDSP.
from dataclasses import dataclass
from datetime import timedelta
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_PORT,
)
from homeassistant.core import DOMAIN, HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .camilla_dsp_api import CamillaDspApi, APIConnectionError, CamillaDspEq
_LOGGER = logging.getLogger(__name__)
@dataclass
class CamillaDspData:
state: str
volume: float
mute: bool
source: str
source_list: list[str]
eq: dict[CamillaDspEq, float]
class CamillaDspCoordinator(DataUpdateCoordinator):
data: CamillaDspData
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
self.host = config_entry.data[CONF_HOST]
self.port = config_entry.data[CONF_PORT]
self.poll_interval = 10
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN} ({config_entry.unique_id})",
update_method=self.async_update_data,
update_interval=timedelta(seconds=self.poll_interval),
)
# Initialise your api here
self.api = CamillaDspApi(host=self.host, port=self.port)
async def async_update_data(self):
try:
if not self.api.connected:
self.api.connect()
state = self.api.get_state()
volume = self.api.get_volume()
mute = self.api.get_mute()
source = self.api.get_source()
sources = self.api.get_sources()
eq = self.api.get_eq()
except APIConnectionError as err:
_LOGGER.error(err)
raise UpdateFailed(err) from err
except Exception as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
return CamillaDspData(state, volume, mute, source, sources, eq)
Home Assistant will be calling async_update_data every 10 seconds (poll_interval) and we connect to camillaDSP and fetch the current state and store that information locally. Next I created a base class entity.py for all the entities that belong to this integration.
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.device_registry import DeviceInfo
from .coordinator import CamillaDspCoordinator, CamillaDspData
from .const import (
DOMAIN,
NAME,
)
class CamillaDspEntity(CoordinatorEntity):
def __init__(
self,
unique_id: str,
coordinator: CamillaDspCoordinator,
) -> None:
super().__init__(coordinator)
self._attr_has_entity_name = True
self._attr_unique_id = unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, str(self.unique_id))},
name=NAME,
manufacturer="HEnquist",
model="",
)
self._data: CamillaDspData = self.coordinator.data
Now for the UI side of Home Assistant. I added a media_player.py entity that allows me to change the master volume of CamillaDSP and the configs.
import logging
import voluptuous as vol
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityDescription,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import MyConfigEntry
from .coordinator import CamillaDspCoordinator
from .entity import CamillaDspEntity
_LOGGER = logging.getLogger(__name__)
ENTITY_DESC = MediaPlayerEntityDescription(
key="mediaplayer",
translation_key="mediaplayer",
)
async def async_setup_entry(hass: HomeAssistant,
config_entry: MyConfigEntry,
async_add_entities: AddEntitiesCallback) -> None:
# This gets the data update coordinator from the config entry runtime data as specified in your __init__.py
coordinator: CamillaDspCoordinator = config_entry.runtime_data.coordinator
entities = []
entities.append(CamillaDspMediaPlayer(config_entry.entry_id, coordinator, ENTITY_DESC))
async_add_entities(entities, update_before_add=False)
class CamillaDspMediaPlayer(CamillaDspEntity, MediaPlayerEntity): # type: ignore[misc]
_attr_media_content_type = MediaType.MUSIC
_attr_device_class = MediaPlayerDeviceClass.RECEIVER
_attr_supported_features = (
MediaPlayerEntityFeature.VOLUME_MUTE |
MediaPlayerEntityFeature.VOLUME_SET |
MediaPlayerEntityFeature.SELECT_SOURCE
)
_attr_source_list = []
entity_description = MediaPlayerEntityDescription
def __init__(
self,
unique_id: str,
coordinator: CamillaDspCoordinator,
description: MediaPlayerEntityDescription,
) -> None:
super().__init__(unique_id, coordinator)
self._attr_name = None
self.entity_description = description
self._set_attrs_from_data()
@callback
def _handle_coordinator_update(self) -> None:
self._data = self.coordinator.data
self._set_attrs_from_data()
super()._handle_coordinator_update()
def _set_attrs_from_data(self):
if self._data is not None:
self._attr_available = self._data.state
self._attr_state = self._data.state
self._attr_volume_level = self._convertFromDb(self._data.volume)
self._attr_is_volume_muted = self._data.mute
self._attr_source = self._data.source
self._attr_source_list = self._data.source_list
else:
self._attr_available = MediaPlayerState.OFF
def _convertFromDb(self, volume: float) -> float:
return round((50 + volume) / 50, 2)
@callback
def _update_attrs_write_ha_state(self) -> None:
self._set_attrs_from_data()
self.async_write_ha_state()
def available(self) -> bool:
return self._attr_available
async def async_set_volume_level(self, volume: float) -> None:
volume_db = self._convertToDb(volume)
await self.async_set_volume_level_db(volume_db)
def _convertToDb(self, volume: float) -> float:
return round((volume * 50 - 50), 2)
async def async_set_volume_level_db(self, volume_db: float) -> None:
self.coordinator.api.set_volume(volume_db)
self._data.volume = volume_db
self._update_attrs_write_ha_state()
async def async_mute_volume(self, mute: bool) -> None:
self.coordinator.api.set_mute(mute)
self._data.mute = mute
self._update_attrs_write_ha_state()
async def async_select_source(self, source: str) -> None:
self.coordinator.api.set_source(source)
self._data.source = source
self._update_attrs_write_ha_state()
What I am letting Home Assistant (HA) know is that this media player is capable of changing volume (MediaPlayerEntityFeature.VOLUME_SET), can be muted or un-muted (MediaPlayerEntityFeature.VOLUME_MUTE) and it can change source (MediaPlayerEntityFeature.SELECT_SOURCE). Basically, changing the source is what causes the CamillaDSP to change the config files (music or movie). When HA calls the relevant methods, I simply call the API to CamillaDSP to do the appropriate thing. I do the same thing in number.py which controls the bass and treble settings.
import logging
from homeassistant.components.number import NumberEntity, NumberMode, NumberDeviceClass
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import MyConfigEntry
from .camilla_dsp_api import CamillaDspEq
from .coordinator import CamillaDspCoordinator
from .entity import CamillaDspEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant,
config_entry: MyConfigEntry,
async_add_entities: AddEntitiesCallback) -> None:
coordinator: CamillaDspCoordinator = config_entry.runtime_data.coordinator
entities = []
entities.append(CamillaDspEqEntity(config_entry.entry_id, coordinator, CamillaDspEq.BASS))
entities.append(CamillaDspEqEntity(config_entry.entry_id, coordinator, CamillaDspEq.TREBLE))
async_add_entities(entities, update_before_add=False)
class CamillaDspEqEntity(CamillaDspEntity, NumberEntity):
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
unique_id: str,
coordinator: CamillaDspCoordinator,
type: CamillaDspEq,
) -> None:
super().__init__(unique_id, coordinator)
self._attr_unique_id = f"{unique_id}-{type}"
self._attr_name = type
self._attr_native_min_value, self._attr_native_max_value = [-12, 12]
self._attr_native_step = 0.5
self._attr_mode = NumberMode.SLIDER
self._attr_device_class = NumberDeviceClass.SOUND_PRESSURE
self. _attr_native_unit_of_measurement = "db"
self.type = type;
self._set_attrs_from_data()
def _set_attrs_from_data(self):
self._attr_native_value = self._data.eq[self.type]
@callback
def _handle_coordinator_update(self) -> None:
self._data = self.coordinator.data
self._set_attrs_from_data()
super()._handle_coordinator_update()
async def async_set_native_value(self, value: float) -> None:
self.coordinator.api.set_eq(self.type, value)
self._attr_native_value = value
self.async_write_ha_state()
Here I am creating 2 entities (CamillaDspEq.BASS, CamillaDspEq.TREBLE) of the type configuration (EntityCategory.CONFIG). Whenever the values are changed in HA UI, the async_set_native_value will be called and I simply call the relevant CamillaDSP API to apply the filters. That is it. After this we have a fully functioning integration.
This how it looks like in my Home Assistant UI with mini media player card.
References
- https://github.com/HEnquist/camilladsp
- https://github.com/HEnquist/camillagui-backend
- https://github.com/mdsimon2/RPi-CamillaDSP
- https://www.roomeqwizard.com/
- https://www.youtube.com/watch?v=tLUClxjUTC4
- https://www.youtube.com/watch?v=O8YpPSXKzuE
- https://github.com/kwerner72/homeassistant-camilladsp