home_assistant_pattern.py

"""Example showing Home Assistant integration pattern.

This demonstrates how to use the global KebaUdpManager for Home Assistant
where multiple entities (sensors, switches, etc.) need to share the same
UDP socket on port 7090.
"""

import asyncio
import logging
from keba_kecontact import KebaClient, KebaUdpManager

logging.basicConfig(level=logging.INFO)
_LOGGER = logging.getLogger(__name__)


class KebaSensor:
    """Simulates a Home Assistant sensor entity."""

    def __init__(self, name: str, ip_address: str):
        self.name = name
        self._client = KebaClient(ip_address, use_global_handler=True)

    async def async_setup(self):
        """Setup the sensor (called during HA integration setup)."""
        await self._client.connect()
        _LOGGER.info(f"[{self.name}] Sensor setup complete")

    async def async_update(self):
        """Update sensor data (called by HA coordinator)."""
        try:
            report3 = await self._client.get_report_3()
            _LOGGER.info(
                f"[{self.name}] Power: {report3.power_kw} kW, "
                f"Energy: {report3.energy_present_kwh} kWh"
            )
            return {
                "power_kw": report3.power_kw,
                "energy_kwh": report3.energy_present_kwh,
            }
        except Exception as e:
            _LOGGER.error(f"[{self.name}] Update failed: {e}")
            return None

    async def async_remove(self):
        """Remove the sensor (called during HA shutdown)."""
        await self._client.disconnect()
        _LOGGER.info(f"[{self.name}] Sensor removed")


class KebaSwitch:
    """Simulates a Home Assistant switch entity."""

    def __init__(self, name: str, ip_address: str):
        self.name = name
        self._client = KebaClient(ip_address, use_global_handler=True)
        self._is_on = False

    async def async_setup(self):
        """Setup the switch (called during HA integration setup)."""
        await self._client.connect()
        _LOGGER.info(f"[{self.name}] Switch setup complete")

    async def async_turn_on(self):
        """Turn on the charging station."""
        await self._client.enable()
        await self._client.set_current(16000)
        self._is_on = True
        _LOGGER.info(f"[{self.name}] Turned ON")

    async def async_turn_off(self):
        """Turn off the charging station."""
        await self._client.disable()
        self._is_on = False
        _LOGGER.info(f"[{self.name}] Turned OFF")

    async def async_remove(self):
        """Remove the switch (called during HA shutdown)."""
        await self._client.disconnect()
        _LOGGER.info(f"[{self.name}] Switch removed")


async def simulate_homeassistant_integration():
    """Simulate Home Assistant integration lifecycle."""

    _LOGGER.info("=== Starting Home Assistant Integration ===")

    manager = KebaUdpManager.get_instance()
    await manager.start()
    _LOGGER.info(f"Manager started, clients: {manager.client_count}")

    charger1_ip = "192.168.1.100"
    charger2_ip = "192.168.1.101"

    sensor1 = KebaSensor("Charger 1 Power Sensor", charger1_ip)
    sensor2 = KebaSensor("Charger 2 Power Sensor", charger2_ip)
    switch1 = KebaSwitch("Charger 1 Switch", charger1_ip)
    switch2 = KebaSwitch("Charger 2 Switch", charger2_ip)

    await sensor1.async_setup()
    await sensor2.async_setup()
    await switch1.async_setup()
    await switch2.async_setup()

    _LOGGER.info(f"All entities setup, clients: {manager.client_count}")

    _LOGGER.info("\n=== Simulating normal operation ===")

    await switch1.async_turn_on()
    await switch2.async_turn_on()

    for i in range(3):
        _LOGGER.info(f"\n--- Update cycle {i+1} ---")
        await sensor1.async_update()
        await sensor2.async_update()
        await asyncio.sleep(2)

    await switch1.async_turn_off()
    await switch2.async_turn_off()

    _LOGGER.info("\n=== Shutting down integration ===")

    await sensor1.async_remove()
    await sensor2.async_remove()
    await switch1.async_remove()
    await switch2.async_remove()

    _LOGGER.info(f"All entities removed, clients: {manager.client_count}")

    await manager.stop()
    _LOGGER.info("Manager stopped")


async def main():
    """Main entry point."""
    try:
        await simulate_homeassistant_integration()
    except KeyboardInterrupt:
        _LOGGER.info("Interrupted by user")
    except Exception as e:
        _LOGGER.error(f"Error: {e}", exc_info=True)


if __name__ == "__main__":
    asyncio.run(main())