controller.py

Go to the documentation of this file.

"""Controller for sharing Omada API coordinators between platforms."""
 
from tplink_omada_client import OmadaSiteClient
from tplink_omada_client.devices import OmadaSwitch
 
from homeassistant.core import HomeAssistant
 
from .coordinator import (
    OmadaClientsCoordinator,
    OmadaDevicesCoordinator,
    OmadaGatewayCoordinator,
    OmadaSwitchPortCoordinator,
)
 
 
class OmadaSiteController:
    """Controller for the Omada SDN site."""
 
    _gateway_coordinator: OmadaGatewayCoordinator | None = None
 
    def __init__(
        self,
        hass: HomeAssistant,
        omada_client: OmadaSiteClient,
    ) -> None:
        """Create the controller."""
        self._hass_hass = hass
        self._omada_client_omada_client = omada_client
 
        self._switch_port_coordinators: dict[str, OmadaSwitchPortCoordinator] = {}
        self._devices_coordinator_devices_coordinator = OmadaDevicesCoordinator(hass, omada_client)
        self._clients_coordinator_clients_coordinator = OmadaClientsCoordinator(hass, omada_client)
 
    async def initialize_first_refresh(self) -> None:
        """Initialize the all coordinators, and perform first refresh."""
        await self._devices_coordinator_devices_coordinator.async_config_entry_first_refresh()
 
        devices = self._devices_coordinator_devices_coordinator.data.values()
        gateway = next((d for d in devices if d.type == "gateway"), None)
        if gateway:
            self._gateway_coordinator_gateway_coordinator = OmadaGatewayCoordinator(
                self._hass_hass, self._omada_client_omada_client, gateway.mac
            )
            await self._gateway_coordinator_gateway_coordinator.async_config_entry_first_refresh()
 
        await self.clients_coordinatorclients_coordinator.async_config_entry_first_refresh()
 
    @property
    def omada_client(self) -> OmadaSiteClient:
        """Get the connected client API for the site to manage."""
        return self._omada_client_omada_client
 
    def get_switch_port_coordinator(
        self, switch: OmadaSwitch
    ) -> OmadaSwitchPortCoordinator:
        """Get coordinator for network port information of a given switch."""
        if switch.mac not in self._switch_port_coordinators:
            self._switch_port_coordinators[switch.mac] = OmadaSwitchPortCoordinator(
                self._hass_hass, self._omada_client_omada_client, switch
            )
 
        return self._switch_port_coordinators[switch.mac]
 
    @property
    def gateway_coordinator(self) -> OmadaGatewayCoordinator | None:
        """Gets the coordinator for site's gateway, or None if there is no gateway."""
        return self._gateway_coordinator_gateway_coordinator
 
    @property
    def devices_coordinator(self) -> OmadaDevicesCoordinator:
        """Gets the coordinator for site's devices."""
        return self._devices_coordinator_devices_coordinator
 
    @property
    def clients_coordinator(self) -> OmadaClientsCoordinator:
        """Gets the coordinator for site's clients."""
        return self._clients_coordinator_clients_coordinator