aiohomematic

import asyncio import logging import os from aiohomematic.central import CentralUnit from aiohomematic.client import Client, CallbackService from aiohomematic.config import HostConfig _LOGGER = logging.getLogger(__name__) async def main(): logging.basicConfig(level=logging.INFO) # Replace with your CCU IP, port, and RPC username/password ccu_host = os.environ.get('HOMEMATIC_CCU_HOST', '192.168.1.100') ccu_port = int(os.environ.get('HOMEMATIC_CCU_PORT', '2001')) # or 2010 for IP, 2000 for wired ccu_username = os.environ.get('HOMEMATIC_CCU_USERNAME', 'Admin') ccu_password = os.environ.get('HOMEMATIC_CCU_PASSWORD', 'password') # The local IP/port aiohomematic's XML-RPC client listens on for CCU callbacks client_host = os.environ.get('HOMEMATIC_CLIENT_HOST', '192.168.1.50') # Must be reachable by CCU client_port = int(os.environ.get('HOMEMATIC_CLIENT_PORT', '8080')) print(f"Connecting to CCU at {ccu_host}:{ccu_port}") print(f"Listening for callbacks on {client_host}:{client_port}") # 1. Configure the Homematic CCU connection host_config = HostConfig( name="my_ccu", host=ccu_host, port=ccu_port, path='/xmlrpc', username=ccu_username, password=ccu_password, tls=False ) # 2. Setup the callback service (the local XML-RPC server) callback_service = CallbackService(client_host, client_port) await callback_service.start() # 3. Create the aiohomematic client that connects to the CCU client = Client(callback_service, client_host, client_port, skip_certificates=True) # skip_certificates if no valid certs # 4. Create the CentralUnit and pass the host_config and client central = CentralUnit(client, host_config) try: # 5. Start the central unit (connect to CCU, fetch devices) await central.start() print("CentralUnit started. Waiting for devices...") # Example: print some device info after a delay await asyncio.sleep(10) # Give time for devices to be discovered for device_address, device in central.devices.items(): print(f"Device: {device.device_type} ({device_address})") for channel_address, channel in device.channels.items(): print(f" Channel: {channel.channel_type} ({channel_address})") # Keep running to receive events print("Running indefinitely. Press Ctrl+C to exit.") while True: await asyncio.sleep(3600) except asyncio.CancelledError: print("Application cancelled.") except Exception as e: _LOGGER.exception("An error occurred: %s", e) finally: # 6. Stop the central unit and callback service await central.stop() await callback_service.stop() print("CentralUnit and CallbackService stopped.") if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print("Exiting.")