bluetooth-speaker-manager/main.py

import os, sys
import logging
DEBUG = os.environ.get('DEBUG')
logging.basicConfig(stream=sys.stdout,
    format='[%(asctime)s] %(levelname)s %(module)s/%(funcName)s - %(message)s',
    level=logging.DEBUG if DEBUG else logging.INFO)

import time
import json

import asyncio
from aiomqtt import Client
from dbus_next.aio import MessageBus
from dbus_next.service import ServiceInterface, method
from dbus_next.constants import BusType
from dbus_next.errors import DBusError
from dbus_next import Variant, Message, MessageType


bus = None
agent_instance = None
pairing_task = None

# --- Bluetooth constants and agent ---
BLUEZ_SERVICE = 'org.bluez'
ADAPTER_IFACE = 'org.bluez.Adapter1'
DEVICE_IFACE = 'org.bluez.Device1'
MEDIA_PLAYER_IFACE = 'org.bluez.MediaPlayer1'
MEDIA_TRANSPORT_IFACE = 'org.bluez.MediaTransport1'
AGENT_IFACE = 'org.bluez.Agent1'
AGENT_MANAGER_IFACE = 'org.bluez.AgentManager1'
AGENT_PATH = '/io/bluetooth_speaker/agent'
CAPABILITY = 'DisplayYesNo'
CALLS_SERVICE_UUID = '0000111e-0000-1000-8000-00805f9b34fb'
AUDIO_SERVICE_UUID = '0000110d-0000-1000-8000-00805f9b34fb'

class Agent(ServiceInterface):
    def __init__(self, interface_name):
        super().__init__(interface_name)
        logging.info('Agent instance created')

    @method()
    def Release(self):
        logging.info('Agent Released')

    @method()
    def RequestPinCode(self, device: 'o') -> 's':
        logging.info(f"RequestPinCode for {device}, returning static PIN")
        return "0000"

    @method()
    def RequestPasskey(self, device: 'o') -> 'u':
        logging.info(f"RequestPasskey for {device}")
        return 0

    @method()
    def DisplayPasskey(self, device: 'o', passkey: 'u', entered: 'q'):
        logging.info(f"DisplayPasskey for {device}: {passkey}")

    @method()
    def DisplayPinCode(self, device: 'o', pincode: 's'):
        logging.info(f"DisplayPinCode for {device}: {pincode}")

    @method()
    async def RequestConfirmation(self, device: 'o', passkey: 'u'):
        logging.info(f"RequestConfirmation for {device} with passkey {passkey}")
        # Automatically confirm and trust
        asyncio.create_task(trust_device(device))

    @method()
    async def RequestAuthorization(self, device: 'o'):
        logging.info(f"RequestAuthorization for {device}")
        # Automatically authorize and trust
        asyncio.create_task(trust_device(device))

    @method()
    async def AuthorizeService(self, device: 'o', uuid: 's'):
        logging.info(f"AuthorizeService request for device {device} with UUID {uuid}")
        if uuid.lower() == CALLS_SERVICE_UUID:
            logging.warning("Rejecting Hands-Free Profile (HFP) connection.")
            raise DBusError('org.bluez.Error.Rejected', 'HFP profile not supported')
        logging.info(f"Authorizing service UUID {uuid}")

    @method()
    def Cancel(self):
        logging.info('Pairing Cancelled')

async def trust_device(device_path):
    logging.info(f'Trusting device {device_path}')
    try:
        introspection = await bus.introspect(BLUEZ_SERVICE, device_path)
        device_obj = bus.get_proxy_object(BLUEZ_SERVICE, device_path, introspection)

        device_props = device_obj.get_interface('org.freedesktop.DBus.Properties')
        await device_props.call_set(DEVICE_IFACE, 'Trusted', Variant('b', True))
        logging.info(f'Trusted device {device_path}')
    except Exception as e:
        logging.error(f'Failed to trust device {device_path}: {e}')

async def get_adapter():
    introspection = await bus.introspect(BLUEZ_SERVICE, '/')
    manager_obj = bus.get_proxy_object(BLUEZ_SERVICE, '/', introspection)
    manager_iface = manager_obj.get_interface('org.freedesktop.DBus.ObjectManager')
    managed_objects = await manager_iface.call_get_managed_objects()

    for path, ifaces in managed_objects.items():
        if ADAPTER_IFACE in ifaces:
            adapter_introspection = await bus.introspect(BLUEZ_SERVICE, path)
            return bus.get_proxy_object(BLUEZ_SERVICE, path, adapter_introspection)
    return None

async def register_agent():
    global agent_instance
    agent_instance = Agent(AGENT_IFACE)
    bus.export(AGENT_PATH, agent_instance)

    introspection = await bus.introspect(BLUEZ_SERVICE, '/org/bluez')
    manager_obj = bus.get_proxy_object(BLUEZ_SERVICE, '/org/bluez', introspection)
    agent_manager = manager_obj.get_interface(AGENT_MANAGER_IFACE)

    try:
        await agent_manager.call_register_agent(AGENT_PATH, CAPABILITY)
        logging.info(f"Agent registered at {AGENT_PATH} with capability {CAPABILITY}")
        await agent_manager.call_request_default_agent(AGENT_PATH)
        logging.info("Agent set as default")
    except Exception as e:
        logging.error(f'Failed to register agent: {e}')
        logging.info('Trying to unregister and register again')
        try:
            await agent_manager.call_unregister_agent(AGENT_PATH)
            await agent_manager.call_register_agent(AGENT_PATH, CAPABILITY)
            await agent_manager.call_request_default_agent(AGENT_PATH)
            logging.info("Agent registered after unregistering")
        except Exception as e2:
            logging.error(f'Failed to register agent again: {e2}')

async def set_adapter_alias(alias):
    logging.info(f"Setting Bluetooth adapter alias to '{alias}'")
    adapter_obj = await get_adapter()
    if not adapter_obj:
        logging.error('Bluetooth adapter not found, cannot set alias.')
        return
    
    adapter_props = adapter_obj.get_interface('org.freedesktop.DBus.Properties')
    try:
        await adapter_props.call_set(ADAPTER_IFACE, 'Alias', Variant('s', alias))
        logging.info(f"Successfully set adapter alias to '{alias}'")
    except Exception as e:
        logging.error(f"Failed to set adapter alias: {e}")


# --- End Bluetooth ---

async def manage_bluetooth():
    await register_agent()
    await set_adapter_alias("Home Audio")
    # The agent will handle things, this task can just sleep
    while True:
        await asyncio.sleep(3600)


async def enable_pairing():
    """Enable pairing for 120 seconds. This task can be cancelled and restarted."""
    adapter_obj = await get_adapter()
    if not adapter_obj:
        logging.error('Bluetooth adapter not found')
        return

    adapter_props = adapter_obj.get_interface('org.freedesktop.DBus.Properties')
    try:
        await adapter_props.call_set(ADAPTER_IFACE, 'Discoverable', Variant('b', True))
        await adapter_props.call_set(ADAPTER_IFACE, 'Pairable', Variant('b', True))
        logging.info('Adapter is discoverable and pairable for 120 seconds')

        await asyncio.sleep(120)

        logging.info('Pairing timeout reached. Making adapter non-discoverable.')
        await adapter_props.call_set(ADAPTER_IFACE, 'Discoverable', Variant('b', False))
    except asyncio.CancelledError:
        logging.info('Pairing timer cancelled, likely by a new pair request.')
        raise
    except Exception as e:
        logging.error(f"Failed to manage pairing state: {e}")


async def disconnect_connected_device():
    logging.info("Attempting to disconnect any connected device.")
    introspection = await bus.introspect(BLUEZ_SERVICE, '/')
    manager_obj = bus.get_proxy_object(BLUEZ_SERVICE, '/', introspection)
    manager_iface = manager_obj.get_interface('org.freedesktop.DBus.ObjectManager')
    managed_objects = await manager_iface.call_get_managed_objects()

    for path, ifaces in managed_objects.items():
        if DEVICE_IFACE in ifaces:
            device_props = ifaces[DEVICE_IFACE]
            if device_props.get('Connected', Variant('b', False)).value:
                logging.info(f"Found connected device: {path}. Disconnecting...")
                try:
                    device_introspection = await bus.introspect(BLUEZ_SERVICE, path)
                    device_obj = bus.get_proxy_object(BLUEZ_SERVICE, path, device_introspection)
                    device_iface = device_obj.get_interface(DEVICE_IFACE)
                    await device_iface.call_disconnect()
                    logging.info(f"Successfully disconnected {path}")
                    return # Assume only one device is connected
                except Exception as e:
                    logging.error(f"Failed to disconnect {path}: {e}")
    logging.info("No connected device found to disconnect.")


async def send_media_command(command):
    """Finds a media player and sends a command to it."""
    logging.info(f"Attempting to send media command: {command}")
    introspection = await bus.introspect(BLUEZ_SERVICE, '/')
    manager_obj = bus.get_proxy_object(BLUEZ_SERVICE, '/', introspection)
    manager_iface = manager_obj.get_interface('org.freedesktop.DBus.ObjectManager')
    managed_objects = await manager_iface.call_get_managed_objects()

    for path, ifaces in managed_objects.items():
        if MEDIA_PLAYER_IFACE in ifaces:
            logging.info(f"Found media player: {path}. Sending command '{command}'...")
            try:
                player_introspection = await bus.introspect(BLUEZ_SERVICE, path)
                player_obj = bus.get_proxy_object(BLUEZ_SERVICE, path, player_introspection)
                player_iface = player_obj.get_interface(MEDIA_PLAYER_IFACE)

                method_name = f'call_{command}'
                dbus_method = getattr(player_iface, method_name)
                await dbus_method()

                logging.info(f"Successfully sent command '{command}' to {path}")
                return # Assume only one media player is active
            except Exception as e:
                logging.error(f"Failed to send media command to {path}: {e}")

    logging.warning("No active media player found to send command to.")


async def adjust_volume(direction, amount):
    """Adjusts the volume of the media transport."""
    logging.info(f"Attempting to adjust volume {direction} by {amount}")
    introspection = await bus.introspect(BLUEZ_SERVICE, '/')
    manager_obj = bus.get_proxy_object(BLUEZ_SERVICE, '/', introspection)
    manager_iface = manager_obj.get_interface('org.freedesktop.DBus.ObjectManager')
    managed_objects = await manager_iface.call_get_managed_objects()

    for path, ifaces in managed_objects.items():
        if MEDIA_TRANSPORT_IFACE in ifaces:
            logging.info(f"Found media transport: {path}. Adjusting volume...")
            try:
                transport_introspection = await bus.introspect(BLUEZ_SERVICE, path)
                transport_obj = bus.get_proxy_object(BLUEZ_SERVICE, path, transport_introspection)
                transport_props = transport_obj.get_interface('org.freedesktop.DBus.Properties')

                current_volume_variant = await transport_props.call_get(MEDIA_TRANSPORT_IFACE, 'Volume')
                current_volume = current_volume_variant.value
                logging.info(f"Current volume is {current_volume}")

                if direction == 'up':
                    new_volume = current_volume + amount
                else: # direction == 'down'
                    new_volume = current_volume - amount

                # The Volume on MediaTransport1 is a uint16, but AVRCP uses 0-127.
                # We'll clamp to this range and hope BlueZ handles scaling.
                new_volume = max(0, min(127, new_volume))

                logging.info(f"Setting new volume to {new_volume}")
                await transport_props.call_set(MEDIA_TRANSPORT_IFACE, 'Volume', Variant('q', new_volume))

                logging.info(f"Successfully adjusted volume on {path}")
                return
            except Exception as e:
                logging.error(f"Failed to adjust volume on {path}: {e}")

    logging.warning("No active media transport found to adjust volume.")


async def process_bluetooth_command(topic, text):
    global pairing_task
    logging.info('Bluetooth command: %s', text)
    if text == "pair":
        if pairing_task and not pairing_task.done():
            logging.info('A pairing process is already active. Cancelling it to restart the timer.')
            pairing_task.cancel()
        pairing_task = asyncio.create_task(enable_pairing())
    elif text == "kick":
        await disconnect_connected_device()
    elif text in ["play", "pause", "next", "prev"]:
        command = "previous" if text == "prev" else text
        await send_media_command(command)
    elif text.startswith("up ") or text.startswith("down "):
        parts = text.split()
        if len(parts) == 2 and parts[1].isdigit():
            direction = parts[0]
            amount = int(parts[1])
            await adjust_volume(direction, amount)
        else:
            logging.warning(f"Invalid volume command format: {text}")

async def process_mqtt(message):
    text = message.payload.decode()
    topic = message.topic.value
    logging.debug('MQTT topic: %s, message: %s', topic, text)

    if topic.startswith('iot/12ser/bluetooth'):
        await process_bluetooth_command(topic, text)
    else:
        logging.debug('Invalid topic, returning')
        return

async def fetch_mqtt():
    await asyncio.sleep(3)

    async with Client(
        hostname='10.55.0.106',
        port=1883,
    ) as client:
        await client.subscribe('iot/12ser/#')
        async for message in client.messages:
            loop = asyncio.get_event_loop()
            loop.create_task(process_mqtt(message))


def suppress_hfp_rejection_error(loop, context):
    exception = context.get('exception')
    if isinstance(exception, DBusError) and 'HFP profile not supported' in str(exception):
        # This is the expected error from AuthorizeService, so we can suppress the traceback.
        logging.info('Suppressed expected DBusError for HFP rejection exception.')
        return
    
    # For all other exceptions, fall back to the default handler.
    loop.default_exception_handler(context)


async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(suppress_hfp_rejection_error)

    global bus
    bus = await MessageBus(bus_type=BusType.SYSTEM).connect()

    logging.info('')
    logging.info('==========================')
    logging.info('Booting up...')

    manage_task = asyncio.create_task(manage_bluetooth())
    mqtt_task = asyncio.create_task(fetch_mqtt())
    await asyncio.gather(manage_task, mqtt_task)


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