xRoute Developer Guide

import asyncio
from bleak import BleakClient, BleakScanner

# --- Configuration ---
DEVICE_NAME = "LabobinxSmart"
UART_SERVICE_UUID = "0000ffe0-0000-1000-8000-00805f9b34fb"
UART_CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb"

# This will hold the device address once found
device_address = None

def notification_handler(sender, data):
    """Handles incoming data from the device."""
    message = data.decode('utf-8').strip()
    print(f"[RECV] {message}")
    # Here you can add logic to parse the key-value pairs
    if message.startswith("M.V1.val="):
        try:
            voltage_scaled = int(message.split('=')[1])
            real_voltage = voltage_scaled / 10.0
            print(f"--> Parsed Voltage: {real_voltage} V")
        except (ValueError, IndexError):
            print("--> Could not parse voltage value.")

async def main():
    global device_address

    print("Scanning for xRoute device...")
    device = await BleakScanner.find_device_by_name(DEVICE_NAME)

    if not device:
        print(f"Could not find device named '{DEVICE_NAME}'. Please ensure it is powered on and advertising.")
        return

    print(f"Found {DEVICE_NAME} at address: {device.address}")

    async with BleakClient(device) as client:
        if not client.is_connected:
            print(f"Failed to connect to {device.address}")
            return

        print("Connected successfully!")
        
        # Subscribe to notifications
        print(f"Subscribing to notifications...")
        await client.start_notify(UART_CHAR_UUID, notification_handler)
        
        print("--- Listening for data and sending commands. Press Ctrl+C to exit. ---")
        try:
            while True:
                # Example: Toggle relay 1 every 10 seconds
                command = "sw1\n"
                print(f"\n[SEND] Toggling relay 1 with command: '{command.strip()}'")
                await client.write_gatt_char(UART_CHAR_UUID, command.encode('utf-8'))
                await asyncio.sleep(10)
        
        except asyncio.CancelledError:
            print("Task cancelled.")
        finally:
            await client.stop_notify(UART_CHAR_UUID)
            print("Unsubscribed and disconnected.")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nProgram stopped by user.")