118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
"""Olimex ESP32-C6-EVB Integration for Home Assistant."""
|
|
import logging
|
|
import aiohttp
|
|
from homeassistant.config_entries import ConfigEntry, SOURCE_IMPORT
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
|
|
from homeassistant.components.webhook import (
|
|
async_register as webhook_register,
|
|
async_unregister as webhook_unregister,
|
|
)
|
|
|
|
from .const import DOMAIN, CONF_CALLBACK_IP, DEFAULT_CALLBACK_IP
|
|
from .webhook import handle_input_event
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SWITCH]
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
|
|
"""Set up the Olimex ESP32-C6-EVB component."""
|
|
hass.data.setdefault(DOMAIN, {})
|
|
|
|
# Handle YAML configuration
|
|
if DOMAIN in config:
|
|
for device_config in config[DOMAIN]:
|
|
hass.async_create_task(
|
|
hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_IMPORT},
|
|
data=device_config,
|
|
)
|
|
)
|
|
|
|
_LOGGER.debug("Olimex ESP32-C6-EVB integration initialized")
|
|
return True
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Set up Olimex ESP32-C6-EVB from a config entry."""
|
|
_LOGGER.info("Setting up Olimex ESP32-C6-EVB entry: %s", entry.entry_id)
|
|
|
|
host = entry.data.get(CONF_HOST, "192.168.0.181")
|
|
port = entry.data.get(CONF_PORT, 80)
|
|
callback_ip = entry.data.get(CONF_CALLBACK_IP, DEFAULT_CALLBACK_IP)
|
|
|
|
hass.data.setdefault(DOMAIN, {})
|
|
hass.data[DOMAIN][entry.entry_id] = {
|
|
"host": host,
|
|
"port": port,
|
|
}
|
|
|
|
# Tell the board where to POST input events
|
|
try:
|
|
callback_url = f"http://{callback_ip}:8123/api/webhook/{entry.entry_id}"
|
|
register_url = f"http://{host}:{port}/register?callback_url={callback_url}"
|
|
_LOGGER.info("Registering webhook with board at %s:%d (callback %s)", host, port, callback_url)
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(register_url, timeout=aiohttp.ClientTimeout(total=5)) as response:
|
|
if response.status == 200:
|
|
_LOGGER.info("Board webhook registered successfully")
|
|
else:
|
|
_LOGGER.warning("Board registration returned status %d", response.status)
|
|
except Exception as err:
|
|
_LOGGER.warning("Failed to register webhook with board: %s", err)
|
|
|
|
# Register HA webhook handler to receive input events from the board
|
|
# Unregister first in case a previous failed setup left it registered
|
|
try:
|
|
webhook_unregister(hass, entry.entry_id)
|
|
except Exception:
|
|
pass
|
|
webhook_register(
|
|
hass,
|
|
DOMAIN,
|
|
"Olimex Input Event",
|
|
entry.entry_id,
|
|
handle_input_event,
|
|
)
|
|
_LOGGER.info("HA webhook handler registered for entry %s", entry.entry_id)
|
|
|
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
|
_LOGGER.info("Olimex ESP32-C6-EVB configured for %s:%d", host, port)
|
|
return True
|
|
|
|
|
|
|
|
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
|
"""Clean up when integration is fully removed (called after unload)."""
|
|
_LOGGER.info("Removing Olimex ESP32-C6-EVB entry: %s", entry.entry_id)
|
|
# Remove any leftover domain data bucket if it's now empty
|
|
if DOMAIN in hass.data and not hass.data[DOMAIN]:
|
|
hass.data.pop(DOMAIN)
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
"""Unload a config entry."""
|
|
try:
|
|
_LOGGER.info("Unloading Olimex ESP32-C6-EVB entry: %s", entry.entry_id)
|
|
|
|
# Unregister webhook handler
|
|
webhook_unregister(hass, entry.entry_id)
|
|
|
|
# Unload all platforms
|
|
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
|
|
|
if unload_ok:
|
|
# Clean up data
|
|
if entry.entry_id in hass.data.get(DOMAIN, {}):
|
|
hass.data[DOMAIN].pop(entry.entry_id)
|
|
_LOGGER.debug("Cleaned up data for entry %s", entry.entry_id)
|
|
|
|
_LOGGER.info("Successfully unloaded Olimex ESP32-C6-EVB entry: %s", entry.entry_id)
|
|
return unload_ok
|
|
except Exception as err:
|
|
_LOGGER.error("Error unloading Olimex ESP32-C6-EVB entry: %s", err, exc_info=True)
|
|
return False
|