Add gateway WAN/LAN port connectivity entities to TP-Link Omada (#91905)

* Add gateway WAN/LAN port status entities

* Fix copy pasta comments

* Add omada binary_sensor to coveragerc
This commit is contained in:
MarkGodwin 2023-05-24 14:12:51 +01:00 committed by GitHub
parent 872cd47e87
commit e97b331942
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 165 additions and 2 deletions

View file

@ -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

View file

@ -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:

View file

@ -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()

View file

@ -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