"""The Netio switch component.""" from collections import namedtuple from datetime import timedelta import logging from typing import Any, override from pynetio import Netio import voluptuous as vol from homeassistant import util from homeassistant.components.http import HomeAssistantView from homeassistant.components.switch import ( PLATFORM_SCHEMA as SWITCH_PLATFORM_SCHEMA, SwitchEntity, ) from homeassistant.const import ( CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME, EVENT_HOMEASSISTANT_STOP, STATE_ON, ) from homeassistant.core import HomeAssistant, callback from homeassistant.helpers import config_validation as cv, issue_registry as ir from homeassistant.helpers.entity_platform import ( AddConfigEntryEntitiesCallback, AddEntitiesCallback, ) from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN from .coordinator import NetioConfigEntry, NetioDataUpdateCoordinator _LOGGER = logging.getLogger(__name__) PARALLEL_UPDATES = 1 ATTR_START_DATE = "start_date" ATTR_TOTAL_CONSUMPTION_KWH = "total_energy_kwh" CONF_OUTLETS = "outlets" DEFAULT_PORT = 1234 DEFAULT_USERNAME = "admin" Device = namedtuple("Device", ["netio", "entities"]) # noqa: PYI024 DEVICES: dict[str, Device] = {} MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10) REQ_CONF = [CONF_HOST, CONF_OUTLETS] URL_API_NETIO_EP = "/api/netio/{host}" PLATFORM_SCHEMA = SWITCH_PLATFORM_SCHEMA.extend( { vol.Required(CONF_HOST): cv.string, vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port, vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string, vol.Required(CONF_PASSWORD): cv.string, vol.Optional(CONF_OUTLETS): {cv.string: cv.string}, } ) async def async_setup_entry( hass: HomeAssistant, entry: NetioConfigEntry, async_add_entities: AddConfigEntryEntitiesCallback, ) -> None: """Set up Netio switches from a config entry.""" coordinator = entry.runtime_data async_add_entities( NetioOutputSwitch(coordinator, output_id) for output_id in coordinator.data ) def setup_platform( hass: HomeAssistant, config: ConfigType, add_entities: AddEntitiesCallback, discovery_info: DiscoveryInfoType | None = None, ) -> None: """Set up the Netio platform.""" ir.create_issue( hass, DOMAIN, "deprecated_yaml", breaks_in_ha_version="2027.3.0", is_fixable=False, issue_domain=DOMAIN, severity=ir.IssueSeverity.WARNING, translation_key="deprecated_yaml", translation_placeholders={ "url": f"/config/integrations/dashboard/add?domain={DOMAIN}", }, ) host = config[CONF_HOST] username = config[CONF_USERNAME] password = config[CONF_PASSWORD] port = config[CONF_PORT] if not DEVICES: hass.http.register_view(NetioApiView) dev = Netio(host, port, username, password) DEVICES[host] = Device(dev, []) # Throttle the update for all Netio switches of one Netio dev.update = util.Throttle(MIN_TIME_BETWEEN_SCANS)(dev.update) for key in config[CONF_OUTLETS]: switch = NetioSwitch(DEVICES[host].netio, key, config[CONF_OUTLETS][key]) DEVICES[host].entities.append(switch) add_entities(DEVICES[host].entities) hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, dispose) def dispose(event): """Close connections to Netio Devices.""" for value in DEVICES.values(): value.netio.stop() class NetioApiView(HomeAssistantView): """WSGI handler class.""" url = URL_API_NETIO_EP name = "api:netio" @callback def get(self, request, host): """Request handler.""" data = request.query states, consumptions, cumulated_consumptions, start_dates = [], [], [], [] for i in range(1, 5): out = f"output{i}" states.append(data.get(f"{out}_state") == STATE_ON) consumptions.append(float(data.get(f"{out}_consumption", 0))) cumulated_consumptions.append( float(data.get(f"{out}_cumulatedConsumption", 0)) / 1000 ) start_dates.append(data.get(f"{out}_consumptionStart", "")) _LOGGER.debug( "%s: %s, %s, %s since %s", host, states, consumptions, cumulated_consumptions, start_dates, ) ndev = DEVICES[host].netio ndev.consumptions = consumptions ndev.cumulated_consumptions = cumulated_consumptions ndev.states = states ndev.start_dates = start_dates for dev in DEVICES[host].entities: dev.async_write_ha_state() return self.json(True) class NetioOutputSwitch(CoordinatorEntity[NetioDataUpdateCoordinator], SwitchEntity): """Switch for a single output of a Netio device.""" _attr_has_entity_name = True def __init__(self, coordinator: NetioDataUpdateCoordinator, output_id: int) -> None: """Initialize the switch.""" super().__init__(coordinator) self._output_id = output_id self._attr_unique_id = f"{coordinator.device.SerialNumber}_{output_id}" self._attr_device_info = coordinator.device_info @property @override def available(self) -> bool: """Return True if the output is present in coordinator data.""" return super().available and self._output_id in self.coordinator.data @property @override def name(self) -> str: """Return the output name configured on the device.""" output = self.coordinator.data.get(self._output_id) if output is None or not output.Name: return f"Output {self._output_id}" return output.Name @property @override def is_on(self) -> bool | None: """Return true if the output is on.""" if (output := self.coordinator.data.get(self._output_id)) is None: return None return output.State == 1 @override async def async_turn_on(self, **kwargs: Any) -> None: """Turn the output on.""" await self.coordinator.async_set_output(self._output_id, True) @override async def async_turn_off(self, **kwargs: Any) -> None: """Turn the output off.""" await self.coordinator.async_set_output(self._output_id, False) class NetioSwitch(SwitchEntity): """Provide a Netio linked switch.""" def __init__(self, netio, outlet, name) -> None: """Initialize the Netio switch.""" self._name = name self.outlet = outlet self.netio = netio @property @override def name(self) -> str: """Return the device's name.""" return self._name @property @override def available(self) -> bool: """Return true if entity is available.""" return not hasattr(self, "telnet") @override def turn_on(self, **kwargs: Any) -> None: """Turn switch on.""" self._set(True) @override def turn_off(self, **kwargs: Any) -> None: """Turn switch off.""" self._set(False) def _set(self, value): val = list("uuuu") val[int(self.outlet) - 1] = "1" if value else "0" val = "".join(val) self.netio.get(f"port list {val}") self.netio.states[int(self.outlet) - 1] = value self.schedule_update_ha_state() @property @override def is_on(self) -> bool: """Return the switch's status.""" return self.netio.states[int(self.outlet) - 1] def update(self) -> None: """Update the state.""" self.netio.update()