diff --git a/.coveragerc b/.coveragerc index 792e0c3785c..30df1694a25 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1297,6 +1297,7 @@ omit = homeassistant/components/touchline/climate.py homeassistant/components/tplink_lte/* homeassistant/components/tplink_omada/__init__.py + homeassistant/components/tplink_omada/binary_sensor.py homeassistant/components/tplink_omada/controller.py homeassistant/components/tplink_omada/coordinator.py homeassistant/components/tplink_omada/entity.py diff --git a/homeassistant/components/tplink_omada/__init__.py b/homeassistant/components/tplink_omada/__init__.py index 824ea8df423..1367f8757af 100644 --- a/homeassistant/components/tplink_omada/__init__.py +++ b/homeassistant/components/tplink_omada/__init__.py @@ -18,7 +18,7 @@ from .config_flow import CONF_SITE, create_omada_client from .const import DOMAIN from .controller import OmadaSiteController -PLATFORMS: list[Platform] = [Platform.SWITCH, Platform.UPDATE] +PLATFORMS: list[Platform] = [Platform.SWITCH, Platform.UPDATE, Platform.BINARY_SENSOR] async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: diff --git a/homeassistant/components/tplink_omada/binary_sensor.py b/homeassistant/components/tplink_omada/binary_sensor.py new file mode 100644 index 00000000000..caaae3465b7 --- /dev/null +++ b/homeassistant/components/tplink_omada/binary_sensor.py @@ -0,0 +1,120 @@ +"""Support for TPLink Omada binary sensors.""" +from __future__ import annotations + +from collections.abc import Callable, Generator + +from attr import dataclass +from tplink_omada_client.definitions import GatewayPortMode, LinkStatus +from tplink_omada_client.devices import OmadaDevice, OmadaGateway, OmadaGatewayPort + +from homeassistant.components.binary_sensor import ( + BinarySensorDeviceClass, + BinarySensorEntity, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import DOMAIN +from .controller import OmadaGatewayCoordinator, OmadaSiteController +from .entity import OmadaDeviceEntity + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up binary sensors.""" + controller: OmadaSiteController = hass.data[DOMAIN][config_entry.entry_id] + omada_client = controller.omada_client + + gateway_coordinator = await controller.get_gateway_coordinator() + if not gateway_coordinator: + return + + gateway = await omada_client.get_gateway(gateway_coordinator.mac) + + async_add_entities( + get_gateway_port_status_sensors(gateway, hass, gateway_coordinator) + ) + + await gateway_coordinator.async_request_refresh() + + +def get_gateway_port_status_sensors( + gateway: OmadaGateway, hass: HomeAssistant, coordinator: OmadaGatewayCoordinator +) -> Generator[BinarySensorEntity, None, None]: + """Generate binary sensors for gateway ports.""" + for port in gateway.port_status: + if port.mode == GatewayPortMode.WAN: + yield OmadaGatewayPortBinarySensor( + coordinator, + gateway, + GatewayPortBinarySensorConfig( + port_number=port.port_number, + id_suffix="wan_link", + name_suffix="Internet Link", + device_class=BinarySensorDeviceClass.CONNECTIVITY, + update_func=lambda p: p.wan_connected, + ), + ) + if port.mode == GatewayPortMode.LAN: + yield OmadaGatewayPortBinarySensor( + coordinator, + gateway, + GatewayPortBinarySensorConfig( + port_number=port.port_number, + id_suffix="lan_status", + name_suffix="LAN Status", + device_class=BinarySensorDeviceClass.CONNECTIVITY, + update_func=lambda p: p.link_status == LinkStatus.LINK_UP, + ), + ) + + +@dataclass +class GatewayPortBinarySensorConfig: + """Config for a binary status derived from a gateway port.""" + + port_number: int + id_suffix: str + name_suffix: str + device_class: BinarySensorDeviceClass + update_func: Callable[[OmadaGatewayPort], bool] + + +class OmadaGatewayPortBinarySensor(OmadaDeviceEntity[OmadaGateway], BinarySensorEntity): + """Binary status of a property on an internet gateway.""" + + _attr_has_entity_name = True + + def __init__( + self, + coordinator: OmadaGatewayCoordinator, + device: OmadaDevice, + config: GatewayPortBinarySensorConfig, + ) -> None: + """Initialize the gateway port binary sensor.""" + super().__init__(coordinator, device) + self._config = config + self._attr_unique_id = f"{device.mac}_{config.port_number}_{config.id_suffix}" + self._attr_device_class = config.device_class + + self._attr_name = f"Port {config.port_number} {config.name_suffix}" + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + gateway = self.coordinator.data[self.device.mac] + + port = next( + p for p in gateway.port_status if p.port_number == self._config.port_number + ) + if port: + self._attr_is_on = self._config.update_func(port) + self._attr_available = True + else: + self._attr_available = False + + self.async_write_ha_state() diff --git a/homeassistant/components/tplink_omada/controller.py b/homeassistant/components/tplink_omada/controller.py index 508a8b914da..194f18ae9bf 100644 --- a/homeassistant/components/tplink_omada/controller.py +++ b/homeassistant/components/tplink_omada/controller.py @@ -1,6 +1,10 @@ """Controller for sharing Omada API coordinators between platforms.""" -from tplink_omada_client.devices import OmadaSwitch, OmadaSwitchPortDetails +from tplink_omada_client.devices import ( + OmadaGateway, + OmadaSwitch, + OmadaSwitchPortDetails, +) from tplink_omada_client.omadasiteclient import OmadaSiteClient from homeassistant.core import HomeAssistant @@ -8,6 +12,7 @@ from homeassistant.core import HomeAssistant from .coordinator import OmadaCoordinator POLL_SWITCH_PORT = 300 +POLL_GATEWAY = 300 class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]): @@ -31,9 +36,31 @@ class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]): return {p.port_id: p for p in ports} +class OmadaGatewayCoordinator(OmadaCoordinator[OmadaGateway]): + """Coordinator for getting details about the site's gateway.""" + + def __init__( + self, + hass: HomeAssistant, + omada_client: OmadaSiteClient, + mac: str, + ) -> None: + """Initialize my coordinator.""" + super().__init__(hass, omada_client, "Gateway", POLL_GATEWAY) + self.mac = mac + + async def poll_update(self) -> dict[str, OmadaGateway]: + """Poll a the gateway's current state.""" + gateway = await self.omada_client.get_gateway(self.mac) + return {self.mac: gateway} + + class OmadaSiteController: """Controller for the Omada SDN site.""" + _gateway_coordinator: OmadaGatewayCoordinator | None = None + _initialized_gateway_coordinator = False + def __init__(self, hass: HomeAssistant, omada_client: OmadaSiteClient) -> None: """Create the controller.""" self._hass = hass @@ -56,3 +83,18 @@ class OmadaSiteController: ) return self._switch_port_coordinators[switch.mac] + + async def get_gateway_coordinator(self) -> OmadaGatewayCoordinator | None: + """Get coordinator for site's gateway, or None if there is no gateway.""" + if not self._initialized_gateway_coordinator: + self._initialized_gateway_coordinator = True + devices = await self._omada_client.get_devices() + gateway = next((d for d in devices if d.type == "gateway"), None) + if not gateway: + return None + + self._gateway_coordinator = OmadaGatewayCoordinator( + self._hass, self._omada_client, gateway.mac + ) + + return self._gateway_coordinator