Add Home Assistant custom integration for Systant
- Create basic integration structure with manifest.json - Add config flow for easy setup via UI - Implement MQTT-based host discovery and sensor creation - Auto-discover Systant hosts via systant/+/stats topic - Create device entities with last_seen sensor for each host - Add Python tooling to flake.nix for HA development Integration features: - Automatic host discovery via MQTT - Device representation for each monitored host - Extensible sensor architecture - Proper Home Assistant integration patterns
This commit is contained in:
parent
36e590becb
commit
c7949285d1
25
flake.nix
25
flake.nix
@ -23,22 +23,41 @@
|
||||
{
|
||||
devShells.default = pkgs.mkShell {
|
||||
buildInputs = with pkgs; [
|
||||
# Elixir/Erlang for server
|
||||
elixir
|
||||
erlang
|
||||
|
||||
# AI/Development tools
|
||||
claude-code
|
||||
aider-chat
|
||||
|
||||
# Node.js tools
|
||||
nodejs_20
|
||||
yarn
|
||||
deno
|
||||
|
||||
# Python for Home Assistant integration
|
||||
python311
|
||||
python311Packages.pip
|
||||
python311Packages.ipython
|
||||
|
||||
python311Packages.pytest
|
||||
python311Packages.homeassistant
|
||||
python311Packages.pytest-homeassistant-custom-component
|
||||
python311Packages.black
|
||||
python311Packages.isort
|
||||
python311Packages.flake8
|
||||
python311Packages.mypy
|
||||
];
|
||||
|
||||
shellHook = ''
|
||||
echo "Elixir development environment loaded"
|
||||
elixir --version
|
||||
echo "Systant development environment loaded"
|
||||
echo "Elixir: $(elixir --version | head -1)"
|
||||
echo "Python: $(python --version)"
|
||||
echo "Available tools: pytest, black, isort, flake8, mypy"
|
||||
echo ""
|
||||
echo "Directories:"
|
||||
echo " server/ - Elixir systant daemon"
|
||||
echo " home-assistant-integration/ - Home Assistant custom integration"
|
||||
'';
|
||||
};
|
||||
packages = {
|
||||
|
||||
@ -0,0 +1,42 @@
|
||||
"""The Systant System Monitor integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS = ["sensor", "binary_sensor"]
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Systant integration."""
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Systant from a config entry."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
||||
# Store the config entry data
|
||||
hass.data[DOMAIN][entry.entry_id] = entry.data
|
||||
|
||||
# Forward the setup to the sensor platform
|
||||
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."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
@ -0,0 +1,63 @@
|
||||
"""Config flow for Systant integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
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
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required("mqtt_prefix", default="systant"): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Systant."""
|
||||
|
||||
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:
|
||||
await self.async_set_unique_id(user_input["mqtt_prefix"])
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=f"Systant ({user_input['mqtt_prefix']})",
|
||||
data=user_input,
|
||||
)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except InvalidConfig:
|
||||
errors["base"] = "invalid_config"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
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 InvalidConfig(HomeAssistantError):
|
||||
"""Error to indicate there is invalid configuration."""
|
||||
@ -0,0 +1,11 @@
|
||||
"""Constants for the Systant integration."""
|
||||
|
||||
DOMAIN = "systant"
|
||||
|
||||
# MQTT topics
|
||||
MQTT_STATS_TOPIC = "systant/+/stats"
|
||||
MQTT_COMMANDS_TOPIC = "systant/+/commands"
|
||||
|
||||
# Default configuration
|
||||
DEFAULT_NAME = "Systant"
|
||||
DEFAULT_MQTT_PREFIX = "systant"
|
||||
@ -0,0 +1,13 @@
|
||||
{
|
||||
"domain": "systant",
|
||||
"name": "Systant System Monitor",
|
||||
"codeowners": ["@ryanpandya"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["mqtt"],
|
||||
"documentation": "https://github.com/ryanpandya/systant",
|
||||
"integration_type": "service",
|
||||
"iot_class": "local_push",
|
||||
"mqtt": ["systant/+/stats", "systant/+/commands"],
|
||||
"requirements": [],
|
||||
"version": "0.1.0"
|
||||
}
|
||||
161
home-assistant-integration/custom_components/systant/sensor.py
Normal file
161
home-assistant-integration/custom_components/systant/sensor.py
Normal file
@ -0,0 +1,161 @@
|
||||
"""Support for Systant sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components import mqtt
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from .const import DOMAIN, MQTT_STATS_TOPIC
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Systant sensors from config entry."""
|
||||
mqtt_prefix = config_entry.data.get("mqtt_prefix", "systant")
|
||||
|
||||
# Create a coordinator that will manage discovered hosts
|
||||
coordinator = SystantCoordinator(hass, mqtt_prefix)
|
||||
await coordinator.async_setup()
|
||||
|
||||
# Store coordinator in hass data
|
||||
hass.data[DOMAIN][config_entry.entry_id] = coordinator
|
||||
|
||||
|
||||
class SystantCoordinator:
|
||||
"""Coordinate discovery and management of Systant hosts."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, mqtt_prefix: str) -> None:
|
||||
"""Initialize the coordinator."""
|
||||
self.hass = hass
|
||||
self.mqtt_prefix = mqtt_prefix
|
||||
self.hosts: dict[str, SystantHost] = {}
|
||||
self._unsubscribe_mqtt = None
|
||||
|
||||
async def async_setup(self) -> None:
|
||||
"""Set up MQTT subscription for host discovery."""
|
||||
topic = f"{self.mqtt_prefix}/+/stats"
|
||||
|
||||
self._unsubscribe_mqtt = await mqtt.async_subscribe(
|
||||
self.hass, topic, self._message_received, 0
|
||||
)
|
||||
|
||||
_LOGGER.info("Subscribed to %s for Systant host discovery", topic)
|
||||
|
||||
@callback
|
||||
def _message_received(self, message: mqtt.ReceiveMessage) -> None:
|
||||
"""Handle received MQTT message."""
|
||||
try:
|
||||
# Extract hostname from topic: systant/hostname/stats
|
||||
topic_parts = message.topic.split("/")
|
||||
if len(topic_parts) != 3:
|
||||
return
|
||||
|
||||
hostname = topic_parts[1]
|
||||
|
||||
# Parse message payload
|
||||
payload = json.loads(message.payload)
|
||||
|
||||
# Create or update host
|
||||
if hostname not in self.hosts:
|
||||
_LOGGER.info("Discovered new Systant host: %s", hostname)
|
||||
self.hosts[hostname] = SystantHost(self.hass, hostname, self.mqtt_prefix)
|
||||
|
||||
# Update host with new data
|
||||
self.hosts[hostname].update_data(payload)
|
||||
|
||||
except json.JSONDecodeError:
|
||||
_LOGGER.warning("Invalid JSON received from %s", message.topic)
|
||||
except Exception as err:
|
||||
_LOGGER.error("Error processing message from %s: %s", message.topic, err)
|
||||
|
||||
async def async_unload(self) -> None:
|
||||
"""Unload the coordinator."""
|
||||
if self._unsubscribe_mqtt:
|
||||
self._unsubscribe_mqtt()
|
||||
|
||||
|
||||
class SystantHost:
|
||||
"""Represent a Systant host with its sensors."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, hostname: str, mqtt_prefix: str) -> None:
|
||||
"""Initialize the host."""
|
||||
self.hass = hass
|
||||
self.hostname = hostname
|
||||
self.mqtt_prefix = mqtt_prefix
|
||||
self.sensors: list[SystantSensor] = []
|
||||
self.last_seen = None
|
||||
|
||||
# Create sensors for this host
|
||||
self._create_sensors()
|
||||
|
||||
def _create_sensors(self) -> None:
|
||||
"""Create sensor entities for this host."""
|
||||
# Last seen sensor
|
||||
last_seen_sensor = SystantLastSeenSensor(self.hostname, self.mqtt_prefix)
|
||||
self.sensors.append(last_seen_sensor)
|
||||
|
||||
# Add sensors to Home Assistant
|
||||
self.hass.async_create_task(
|
||||
self.hass.helpers.entity_platform.async_add_entities(
|
||||
self.sensors, update_before_add=True
|
||||
)
|
||||
)
|
||||
|
||||
def update_data(self, data: dict[str, Any]) -> None:
|
||||
"""Update sensors with new data."""
|
||||
self.last_seen = dt_util.utcnow()
|
||||
|
||||
# Update all sensors
|
||||
for sensor in self.sensors:
|
||||
sensor.update_from_data(data)
|
||||
|
||||
|
||||
class SystantSensor(SensorEntity):
|
||||
"""Base class for Systant sensors."""
|
||||
|
||||
def __init__(self, hostname: str, mqtt_prefix: str, sensor_type: str) -> None:
|
||||
"""Initialize the sensor."""
|
||||
self._hostname = hostname
|
||||
self._mqtt_prefix = mqtt_prefix
|
||||
self._sensor_type = sensor_type
|
||||
self._attr_unique_id = f"{mqtt_prefix}_{hostname}_{sensor_type}"
|
||||
self._attr_name = f"Systant {hostname} {sensor_type.replace('_', ' ').title()}"
|
||||
self._attr_device_info = {
|
||||
"identifiers": {(DOMAIN, f"{mqtt_prefix}_{hostname}")},
|
||||
"name": f"Systant {hostname}",
|
||||
"manufacturer": "Systant",
|
||||
"model": "System Monitor",
|
||||
"via_device": (DOMAIN, mqtt_prefix),
|
||||
}
|
||||
|
||||
def update_from_data(self, data: dict[str, Any]) -> None:
|
||||
"""Update sensor from MQTT data."""
|
||||
# To be implemented by subclasses
|
||||
pass
|
||||
|
||||
|
||||
class SystantLastSeenSensor(SystantSensor):
|
||||
"""Sensor for last seen timestamp."""
|
||||
|
||||
def __init__(self, hostname: str, mqtt_prefix: str) -> None:
|
||||
"""Initialize the last seen sensor."""
|
||||
super().__init__(hostname, mqtt_prefix, "last_seen")
|
||||
self._attr_device_class = "timestamp"
|
||||
self._attr_icon = "mdi:clock-outline"
|
||||
|
||||
def update_from_data(self, data: dict[str, Any]) -> None:
|
||||
"""Update with current timestamp."""
|
||||
self._attr_native_value = dt_util.utcnow()
|
||||
self.async_write_ha_state()
|
||||
Loading…
Reference in New Issue
Block a user