"""Home Assistant integration for invertergui MQTT topics.""" from __future__ import annotations from typing import Any import voluptuous as vol from homeassistant.const import CONF_NAME from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import config_validation as cv, discovery from homeassistant.helpers.typing import ConfigType from .const import ( ATTR_ESS_MAX_CHARGE_POWER, ATTR_ESS_MAX_DISCHARGE_POWER, ATTR_ESS_MODE, ATTR_ESS_SETPOINT, ATTR_CURRENT_LIMIT, ATTR_MODE, CONF_COMMAND_TOPIC, CONF_STATE_TOPIC, CONF_STATUS_TOPIC, CONF_TOPIC_ROOT, CONF_VENUS_GUIDE_COMPAT, CONF_VENUS_PORTAL_ID, CONF_VENUS_TOPIC_PREFIX, DATA_BRIDGE, DEFAULT_COMMAND_TOPIC, DEFAULT_NAME, DEFAULT_STATE_TOPIC, DEFAULT_STATUS_TOPIC, DEFAULT_TOPIC_ROOT, DEFAULT_VENUS_GUIDE_COMPAT, DEFAULT_VENUS_PORTAL_ID, DEFAULT_VENUS_TOPIC_PREFIX, DOMAIN, PANEL_MODES, PLATFORMS, SERVICE_SET_ESS_CONTROL, SERVICE_SET_REMOTE_PANEL_STATE, ) from .coordinator import VictronMqttBridge CONFIG_SCHEMA = vol.Schema( { DOMAIN: vol.Schema( { vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, vol.Optional(CONF_STATE_TOPIC, default=DEFAULT_STATE_TOPIC): cv.string, vol.Optional( CONF_COMMAND_TOPIC, default=DEFAULT_COMMAND_TOPIC ): cv.string, vol.Optional(CONF_STATUS_TOPIC, default=DEFAULT_STATUS_TOPIC): cv.string, vol.Optional(CONF_TOPIC_ROOT): cv.string, vol.Optional(CONF_VENUS_PORTAL_ID, default=DEFAULT_VENUS_PORTAL_ID): cv.string, vol.Optional(CONF_VENUS_TOPIC_PREFIX, default=DEFAULT_VENUS_TOPIC_PREFIX): cv.string, vol.Optional(CONF_VENUS_GUIDE_COMPAT, default=DEFAULT_VENUS_GUIDE_COMPAT): cv.boolean, } ) }, extra=vol.ALLOW_EXTRA, ) SERVICE_SET_REMOTE_PANEL_STATE_SCHEMA = vol.Schema( { vol.Optional(ATTR_MODE): vol.In(PANEL_MODES), vol.Optional(ATTR_CURRENT_LIMIT): vol.Coerce(float), }, extra=vol.PREVENT_EXTRA, ) SERVICE_SET_ESS_CONTROL_SCHEMA = vol.Schema( { vol.Optional(ATTR_ESS_SETPOINT): vol.Coerce(float), vol.Optional(ATTR_ESS_MAX_CHARGE_POWER): vol.Coerce(float), vol.Optional(ATTR_ESS_MAX_DISCHARGE_POWER): vol.Coerce(float), vol.Optional(ATTR_ESS_MODE): vol.Coerce(int), }, extra=vol.PREVENT_EXTRA, ) def mqtt_topic_root(topic: str) -> str: """Match invertergui MQTT root behavior.""" cleaned = topic.strip().strip("/") if not cleaned: return DEFAULT_TOPIC_ROOT if cleaned.endswith("/updates"): root = cleaned[: -len("/updates")] if root: return root return cleaned async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up Victron MK2 MQTT integration from YAML.""" conf = config.get(DOMAIN) if conf is None: return True setup_conf: dict[str, Any] = dict(conf) if not setup_conf.get(CONF_TOPIC_ROOT): setup_conf[CONF_TOPIC_ROOT] = mqtt_topic_root(setup_conf[CONF_STATE_TOPIC]) bridge = VictronMqttBridge(hass, setup_conf) await bridge.async_setup() hass.data.setdefault(DOMAIN, {}) hass.data[DOMAIN][DATA_BRIDGE] = bridge await _register_services(hass, bridge) for platform in PLATFORMS: hass.async_create_task( discovery.async_load_platform(hass, platform, DOMAIN, {}, config) ) return True async def _register_services(hass: HomeAssistant, bridge: VictronMqttBridge) -> None: """Register integration services.""" if hass.services.has_service(DOMAIN, SERVICE_SET_REMOTE_PANEL_STATE): return async def handle_set_remote_panel_state(call: ServiceCall) -> None: mode = call.data.get(ATTR_MODE) current_limit = call.data.get(ATTR_CURRENT_LIMIT) if mode is None and current_limit is None: raise HomeAssistantError("Provide at least one of mode or current_limit") if current_limit is not None and current_limit < 0: raise HomeAssistantError("current_limit must be >= 0") payload: dict[str, Any] = {"kind": "panel_state"} if mode is not None: payload["switch"] = mode if current_limit is not None: payload["current_limit"] = float(current_limit) await bridge.async_publish_command(payload) hass.services.async_register( DOMAIN, SERVICE_SET_REMOTE_PANEL_STATE, handle_set_remote_panel_state, schema=SERVICE_SET_REMOTE_PANEL_STATE_SCHEMA, ) async def handle_set_ess_control(call: ServiceCall) -> None: setpoint = call.data.get(ATTR_ESS_SETPOINT) max_charge = call.data.get(ATTR_ESS_MAX_CHARGE_POWER) max_discharge = call.data.get(ATTR_ESS_MAX_DISCHARGE_POWER) ess_mode = call.data.get(ATTR_ESS_MODE) if all(value is None for value in (setpoint, max_charge, max_discharge, ess_mode)): raise HomeAssistantError( "Provide at least one of ess_setpoint, ess_max_charge_power, ess_max_discharge_power, or ess_mode" ) if max_charge is not None and max_charge < 0: raise HomeAssistantError("ess_max_charge_power must be >= 0") if max_discharge is not None and max_discharge < 0: raise HomeAssistantError("ess_max_discharge_power must be >= 0") if ess_mode is not None and ess_mode not in (9, 10): raise HomeAssistantError("ess_mode must be 9 or 10") commands: list[dict[str, Any]] = [] if setpoint is not None: commands.append({"kind": "ess_setpoint", "value": float(setpoint)}) if max_charge is not None: commands.append({"kind": "ess_max_charge_power", "value": float(max_charge)}) if max_discharge is not None: commands.append({"kind": "ess_max_discharge_power", "value": float(max_discharge)}) if ess_mode is not None: commands.append({"kind": "ess_mode", "value": int(ess_mode)}) for payload in commands: await bridge.async_publish_command(payload) hass.services.async_register( DOMAIN, SERVICE_SET_ESS_CONTROL, handle_set_ess_control, schema=SERVICE_SET_ESS_CONTROL_SCHEMA, )