118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
"""Switch platform for Olimex ESP32-C6-EVB."""
|
|
import logging
|
|
import aiohttp
|
|
from homeassistant.components.switch import SwitchEntity
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
|
|
from .const import DOMAIN, NUM_RELAYS
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Tight timeout for a local LAN device
|
|
_TIMEOUT = aiohttp.ClientTimeout(total=3)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up switch entities for relays."""
|
|
data = hass.data[DOMAIN][entry.entry_id]
|
|
host = data["host"]
|
|
port = data["port"]
|
|
|
|
switches = [
|
|
OlimexRelaySwitch(entry, host, port, relay_num)
|
|
for relay_num in range(1, NUM_RELAYS + 1)
|
|
]
|
|
async_add_entities(switches, update_before_add=False)
|
|
|
|
|
|
class OlimexRelaySwitch(SwitchEntity):
|
|
"""Switch for Olimex relay.
|
|
|
|
State is set on load via a single GET and on every toggle via the
|
|
state value returned directly in the POST response — no extra round-trip.
|
|
A single persistent aiohttp session is reused for all requests.
|
|
"""
|
|
|
|
_attr_should_poll = False
|
|
|
|
def __init__(self, entry: ConfigEntry, host: str, port: int, relay_num: int):
|
|
"""Initialize the switch."""
|
|
self._entry = entry
|
|
self._host = host
|
|
self._port = port
|
|
self._relay_num = relay_num
|
|
self._attr_name = f"Relay {relay_num}"
|
|
self._attr_unique_id = f"{entry.entry_id}_relay_{relay_num}"
|
|
self._is_on = False
|
|
self._session: aiohttp.ClientSession | None = None
|
|
|
|
async def async_added_to_hass(self):
|
|
"""Open a persistent HTTP session and fetch initial relay state."""
|
|
self._session = aiohttp.ClientSession()
|
|
url = f"http://{self._host}:{self._port}/relay/status?relay={self._relay_num}"
|
|
try:
|
|
async with self._session.get(url, timeout=_TIMEOUT) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
self._is_on = data.get("state", False)
|
|
except Exception as err:
|
|
_LOGGER.debug("Relay %d initial fetch failed: %s", self._relay_num, err)
|
|
self.async_write_ha_state()
|
|
|
|
async def async_will_remove_from_hass(self):
|
|
"""Close the HTTP session when entity is removed."""
|
|
if self._session:
|
|
await self._session.close()
|
|
self._session = None
|
|
|
|
# ------------------------------------------------------------------
|
|
# SwitchEntity interface
|
|
# ------------------------------------------------------------------
|
|
|
|
async def async_turn_on(self, **kwargs):
|
|
"""Turn the relay on."""
|
|
await self._async_set_relay(True)
|
|
|
|
async def async_turn_off(self, **kwargs):
|
|
"""Turn the relay off."""
|
|
await self._async_set_relay(False)
|
|
|
|
async def _async_set_relay(self, on: bool):
|
|
"""POST on/off to the board; read state from the response body directly."""
|
|
action = "on" if on else "off"
|
|
url = f"http://{self._host}:{self._port}/relay/{action}?relay={self._relay_num}"
|
|
try:
|
|
async with self._session.post(url, timeout=_TIMEOUT) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
# Board returns {"status":"ok","state":true/false} — use it directly
|
|
self._is_on = data.get("state", on)
|
|
_LOGGER.debug("Relay %d -> %s (board confirmed: %s)", self._relay_num, action, self._is_on)
|
|
else:
|
|
_LOGGER.error("Relay %d %s failed: HTTP %d", self._relay_num, action, resp.status)
|
|
except Exception as err:
|
|
_LOGGER.error("Relay %d %s error: %s", self._relay_num, action, err)
|
|
self.async_write_ha_state()
|
|
|
|
@property
|
|
def is_on(self):
|
|
"""Return True if relay is on."""
|
|
return self._is_on
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device information."""
|
|
host = self._entry.data.get("host", "unknown")
|
|
return {
|
|
"identifiers": {(DOMAIN, self._entry.entry_id)},
|
|
"name": f"Olimex ESP32-C6 ({host})",
|
|
"manufacturer": "Olimex",
|
|
"model": "ESP32-C6-EVB",
|
|
}
|