86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
"""Relay status sensor updater for Olimex ESP32-C6-EVB."""
|
|
import asyncio
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
import aiohttp
|
|
from homeassistant.helpers.update_coordinator import (
|
|
CoordinatorEntity,
|
|
DataUpdateCoordinator,
|
|
UpdateFailed,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.components.sensor import SensorEntity
|
|
|
|
from .const import DOMAIN, DEFAULT_SCAN_INTERVAL
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class OlimexDataUpdateCoordinator(DataUpdateCoordinator):
|
|
"""Coordinator to fetch relay statuses from the board."""
|
|
|
|
def __init__(self, hass: HomeAssistant, host: str, port: int, scan_interval: int) -> None:
|
|
"""Initialize the coordinator."""
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
name=DOMAIN,
|
|
update_interval=timedelta(seconds=scan_interval),
|
|
)
|
|
self.host = host
|
|
self.port = port
|
|
|
|
async def _fetch_single_relay(self, session: aiohttp.ClientSession, relay_num: int):
|
|
"""Fetch one relay's status, returning (relay_num, state)."""
|
|
try:
|
|
url = f"http://{self.host}:{self.port}/relay/status?relay={relay_num}"
|
|
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
return relay_num, data.get("state", False)
|
|
return relay_num, False
|
|
except Exception as err:
|
|
_LOGGER.debug("Error fetching relay %d status: %s", relay_num, err)
|
|
return relay_num, False
|
|
|
|
async def _async_update_data(self):
|
|
"""Fetch all relay statuses from the device in parallel."""
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
results = await asyncio.gather(
|
|
*[self._fetch_single_relay(session, n) for n in range(1, 5)]
|
|
)
|
|
return {f"relay_{num}": state for num, state in results}
|
|
except Exception as err:
|
|
raise UpdateFailed(f"Error communicating with device: {err}")
|
|
|
|
|
|
class RelayStatusSensor(CoordinatorEntity, SensorEntity):
|
|
"""Sensor for relay status."""
|
|
|
|
def __init__(self, coordinator, entry, relay_num):
|
|
"""Initialize the sensor."""
|
|
super().__init__(coordinator)
|
|
self._entry = entry
|
|
self._relay_num = relay_num
|
|
self._attr_name = f"Relay {relay_num} Status"
|
|
self._attr_unique_id = f"{entry.entry_id}_relay_{relay_num}_status"
|
|
|
|
@property
|
|
def native_value(self):
|
|
"""Return the relay status."""
|
|
if self.coordinator.data:
|
|
return self.coordinator.data.get(f"relay_{self._relay_num}", False)
|
|
return False
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device information."""
|
|
return {
|
|
"identifiers": {(DOMAIN, self._entry.entry_id)},
|
|
"name": f"Olimex ESP32-C6 ({self._entry.data['host']})",
|
|
"manufacturer": "Olimex",
|
|
"model": "ESP32-C6-EVB",
|
|
}
|