shhhh

boof.py

#!/usr/bin/env python3
"""
boof - Bluetooth speaker management utility
Connects to and stops music on multiple speakers
"""

import json
import subprocess
import time
import re
import logging
from typing import List, Dict, Optional

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class BluetoothSpeakerManager:
    def __init__(self, config_file: str = 'speakers.json'):
        """Initialize the speaker manager with configuration file."""
        self.config_file = config_file
        self.speakers = self.load_speakers()
        
    def load_speakers(self) -> List[Dict]:
        """Load speaker configurations from JSON file."""
        try:
            with open(self.config_file, 'r') as f:
                data = json.load(f)
                return data.get('bluetoothSpeakers', [])
        except FileNotFoundError:
            logger.error(f"Configuration file {self.config_file} not found")
            return []
        except json.JSONDecodeError:
            logger.error(f"Invalid JSON in {self.config_file}")
            return []
    
    def run_bluetoothctl_command(self, command: str) -> str:
        """Execute a bluetoothctl command and return output."""
        try:
            result = subprocess.run(
                ['bluetoothctl'] + command.split(),
                capture_output=True,
                text=True,
                timeout=30
            )
            return result.stdout.strip()
        except subprocess.TimeoutExpired:
            logger.warning(f"Command '{command}' timed out")
            return ""
        except Exception as e:
            logger.error(f"Error running command '{command}': {e}")
            return ""
    
    def scan_for_devices(self, duration: int = 10) -> List[str]:
        """Scan for nearby Bluetooth devices."""
        logger.info(f"Scanning for devices for {duration} seconds...")
        
        # Start scanning
        subprocess.run(['bluetoothctl', 'scan', 'on'], capture_output=True)
        time.sleep(duration)
        subprocess.run(['bluetoothctl', 'scan', 'off'], capture_output=True)
        
        # Get list of discovered devices
        output = self.run_bluetoothctl_command("devices")
        devices = []
        
        for line in output.split('\n'):
            if line.startswith('Device'):
                parts = line.split(' ', 2)
                if len(parts) >= 3:
                    mac_address = parts[1]
                    device_name = parts[2]
                    devices.append(f"{mac_address}:{device_name}")
        
        logger.info(f"Found {len(devices)} devices")
        return devices
    
    def match_speaker(self, device_name: str) -> Optional[Dict]:
        """Match a device name against speaker configurations."""
        for speaker in self.speakers:
            pattern = speaker.get('deviceNamePattern', '')
            if re.search(pattern, device_name, re.IGNORECASE):
                return speaker
        return None
    
    def connect_to_device(self, mac_address: str, timeout: int = 15) -> bool:
        """Attempt to connect to a Bluetooth device."""
        logger.info(f"Attempting to connect to {mac_address}")
        
        # Trust and pair the device first
        subprocess.run(['bluetoothctl', 'trust', mac_address], capture_output=True)
        subprocess.run(['bluetoothctl', 'pair', mac_address], capture_output=True)
        
        # Attempt connection
        result = subprocess.run(
            ['bluetoothctl', 'connect', mac_address],
            capture_output=True,
            text=True,
            timeout=timeout
        )
        
        if result.returncode == 0:
            logger.info(f"Successfully connected to {mac_address}")
            return True
        else:
            logger.warning(f"Failed to connect to {mac_address}: {result.stderr}")
            return False
    
    def stop_audio(self, mac_address: str) -> bool:
        """Send audio control commands to stop/pause music."""
        commands = [
            ['playerctl', 'pause'],  # Generic media control
            ['dbus-send', '--system', '--dest=org.bluez', f'/org/bluez/hci0/dev_{mac_address.replace(":", "_")}', 
             'org.bluez.MediaControl1.Pause'],  # BlueZ media control
        ]
        
        for cmd in commands:
            try:
                result = subprocess.run(cmd, capture_output=True, timeout=5)
                if result.returncode == 0:
                    logger.info(f"Audio stopped on {mac_address}")
                    return True
            except (subprocess.TimeoutExpired, FileNotFoundError):
                continue
        
        logger.warning(f"Could not stop audio on {mac_address}")
        return False
    
    def disconnect_device(self, mac_address: str) -> bool:
        """Disconnect from a Bluetooth device."""
        result = subprocess.run(
            ['bluetoothctl', 'disconnect', mac_address],
            capture_output=True,
            text=True
        )
        
        if result.returncode == 0:
            logger.info(f"Disconnected from {mac_address}")
            return True
        else:
            logger.warning(f"Failed to disconnect from {mac_address}")
            return False
    
    def process_speaker(self, mac_address: str, device_name: str, speaker_config: Dict) -> bool:
        """Process a single speaker: connect, stop audio, disconnect."""
        timeout = speaker_config.get('connectionTimeout', 15)
        
        try:
            # Connect to speaker
            if not self.connect_to_device(mac_address, timeout):
                return False
            
            # Wait a moment for connection to stabilize
            time.sleep(2)
            
            # Stop audio
            audio_stopped = self.stop_audio(mac_address)
            
            # Disconnect
            self.disconnect_device(mac_address)
            
            return audio_stopped
            
        except Exception as e:
            logger.error(f"Error processing speaker {device_name}: {e}")
            return False
    
    def run(self):
        """Main execution loop."""
        logger.info("Starting boof - Bluetooth speaker manager")
        
        if not self.speakers:
            logger.error("No speaker configurations loaded. Exiting.")
            return
        
        # Enable Bluetooth controller
        subprocess.run(['bluetoothctl', 'power', 'on'], capture_output=True)
        subprocess.run(['bluetoothctl', 'agent', 'on'], capture_output=True)
        subprocess.run(['bluetoothctl', 'default-agent'], capture_output=True)
        
        # Scan for devices
        devices = self.scan_for_devices()
        
        if not devices:
            logger.info("No devices found. Exiting.")
            return
        
        matched_speakers = []
        
        # Match found devices against speaker configurations
        for device in devices:
            mac_address, device_name = device.split(':', 1)
            speaker_config = self.match_speaker(device_name)
            
            if speaker_config:
                matched_speakers.append((mac_address, device_name, speaker_config))
                logger.info(f"Matched speaker: {device_name} ({speaker_config['brand']})")
        
        if not matched_speakers:
            logger.info("No matching speakers found. Exiting.")
            return
        
        # Process each matched speaker
        logger.info(f"Processing {len(matched_speakers)} speakers...")
        successful = 0
        
        for mac_address, device_name, speaker_config in matched_speakers:
            logger.info(f"Processing {device_name}...")
            
            if self.process_speaker(mac_address, device_name, speaker_config):
                successful += 1
            
            # Brief pause between speakers
            time.sleep(1)
        
        logger.info(f"Completed. Successfully processed {successful}/{len(matched_speakers)} speakers.")

def main():
    """Main entry point."""
    try:
        manager = BluetoothSpeakerManager()
        manager.run()
    except KeyboardInterrupt:
        logger.info("Interrupted by user. Exiting.")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")

if __name__ == "__main__":
    main()

speakers.json

{
  "bluetoothSpeakers": [
    {
      "brand": "JBL",
      "deviceNamePattern": "JBL.*(Flip|Charge|Boombox|Pulse|Xtreme|PartyBox|Authentics|Go|Clip).*",
      "bluetoothVersion": "5.3",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 15
    },
    {
      "brand": "Bose",
      "deviceNamePattern": "Bose.*(SoundLink|Portable|Home|Revolve|Flex|Max).*",
      "bluetoothVersion": "5.3",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 10
    },
    {
      "brand": "Sony",
      "deviceNamePattern": "Sony.*(SRS|ULT|XB|XG|XE).*",
      "bluetoothVersion": "5.2",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 15
    },
    {
      "brand": "Ultimate Ears",
      "deviceNamePattern": "UE.*(Boom|Megaboom|Wonderboom|Hyperboom|Epicboom).*",
      "bluetoothVersion": "5.0",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 12
    },
    {
      "brand": "Sonos",
      "deviceNamePattern": "Sonos.*(Roam|Move|Era).*",
      "bluetoothVersion": "5.0",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 20
    },
    {
      "brand": "Marshall",
      "deviceNamePattern": "Marshall.*(Emberton|Kilburn|Tufton|Acton|Stanmore|Woburn|Middleton).*",
      "bluetoothVersion": "5.0",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 15
    },
    {
      "brand": "Anker Soundcore",
      "deviceNamePattern": "Soundcore.*(Motion|Flare|Select|X).*",
      "bluetoothVersion": "5.0",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 10
    },
    {
      "brand": "Tribit",
      "deviceNamePattern": "Tribit.*(StormBox|XSound|MaxSound).*",
      "bluetoothVersion": "5.0",
      "pairingMethod": "SSP",
      "defaultPin": null,
      "connectionTimeout": 10
    },
    {
      "brand": "Harman/Kardon",
      "deviceNamePattern": "Harman.*(Onyx|Aura|Go|Play).*",
      "bluetoothVersion": "4.2",
      "pairingMethod": "SSP",
      "defaultPin": "0000",
      "connectionTimeout": 15
    }
  ]
}