## Prerequisites for running AICS BlueZ GATT server ## 1. Check BlueZ version (need 5.48 or newer) bluetoothd –version ## 2. Install Python D-Bus bindings sudo apt-get install python3-dbus python3-gi ## 3. Check your BLE adapter hciconfig -a # Should show hci0 with RUNNING flag ## 4. Enable LE (if not already) sudo hciconfig hci0 up sudo btmgmt le on sudo btmgmt bredr off # For LE-only; skip if you need BR/EDR too sudo btmgmt connectable on sudo btmgmt bondable on # Required for encryption ## 5. Verify bluetoothd is running with experimental mode ## (needed for some GATT features in older BlueZ) ## Add –experimental to /etc/systemd/system/bluetooth.service ## ExecStart=/usr/lib/bluetooth/bluetoothd –experimental ## 6. Reset adapter cleanly sudo systemctl restart bluetooth

#!/usr/bin/env python3 # File: aics_gatt_server.py # Audio Input Control Service — BlueZ D-Bus GATT Server # Run as: sudo python3 aics_gatt_server.py import dbus import dbus.exceptions import dbus.mainloop.glib import dbus.service from gi.repository import GLib import sys import struct # —- D-Bus interface constants —- BLUEZ_SERVICE_NAME = ‘org.bluez’ GATT_MANAGER_IFACE = ‘org.bluez.GattManager1’ DBUS_OM_IFACE = ‘org.freedesktop.DBus.ObjectManager’ DBUS_PROP_IFACE = ‘org.freedesktop.DBus.Properties’ GATT_SERVICE_IFACE = ‘org.bluez.GattService1’ GATT_CHRC_IFACE = ‘org.bluez.GattCharacteristic1’ GATT_DESC_IFACE = ‘org.bluez.GattDescriptor1’ LE_ADVERTISING_MANAGER = ‘org.bluez.LEAdvertisingManager1’ # —- AICS UUIDs (from Bluetooth SIG Assigned Numbers) —- AICS_SVC_UUID = ‘0000184d-0000-1000-8000-00805f9b34fb’ AUDIO_INPUT_STATE_UUID = ‘00002b77-0000-1000-8000-00805f9b34fb’ GAIN_SETTING_PROPS_UUID = ‘00002b78-0000-1000-8000-00805f9b34fb’ AUDIO_INPUT_TYPE_UUID = ‘00002b79-0000-1000-8000-00805f9b34fb’ AUDIO_INPUT_STATUS_UUID = ‘00002b80-0000-1000-8000-00805f9b34fb’ AUDIO_INPUT_CP_UUID = ‘00002b81-0000-1000-8000-00805f9b34fb’ AUDIO_INPUT_DESC_UUID = ‘00002b82-0000-1000-8000-00805f9b34fb’ CCCD_UUID = ‘00002902-0000-1000-8000-00805f9b34fb’ # —- AICS Error codes —- AICS_ERR_INVALID_CC = 0x80 AICS_ERR_OPCODE_NOT_SUP = 0x81 AICS_ERR_MUTE_DISABLED = 0x82 AICS_ERR_OUT_OF_RANGE = 0x83 AICS_ERR_MODE_NOT_ALLOWED = 0x84 # —- AICS Opcodes —- OP_SET_GAIN = 0x01 OP_UNMUTE = 0x02 OP_MUTE = 0x03 OP_SET_MANUAL = 0x04 OP_SET_AUTO = 0x05 # —- AICS State (shared across all characteristics) —- class AICSState: def __init__(self): self.gain_setting = 0 # int8, current gain step self.mute = 0 # 0=NotMuted,1=Muted,2=Disabled self.gain_mode = 2 # 0=ManualOnly,1=AutoOnly,2=Manual,3=Auto self.change_counter = 0x05 # arbitrary init value # Gain Setting Properties (static while connected) self.gs_units = 10 # 10 x 0.1dB = 1.0 dB per step self.gs_minimum = -19 self.gs_maximum = 14 # Audio Input Type (static) self.input_type = 0x02 # 0x02 = Microphone # Audio Input Status self.input_status = 0x01 # Active # Audio Input Description self.description = “Left Mic” def increment_cc(self): self.change_counter = (self.change_counter + 1) & 0xFF def get_state_bytes(self): # Audio Input State: [Gain_Setting][Mute][Gain_Mode][Change_Counter] raw_gain = struct.pack(‘b’, self.gain_setting)[0] # int8 to uint8 return [raw_gain, self.mute, self.gain_mode, self.change_counter] def get_gain_props_bytes(self): raw_min = struct.pack(‘b’, self.gs_minimum)[0] raw_max = struct.pack(‘b’, self.gs_maximum)[0] return [self.gs_units, raw_min, raw_max] # —- Global state instance —- aics = AICSState() # —- Base GATT Service —- class GattService(dbus.service.Object): PATH_BASE = ‘/org/ep/aics’ def __init__(self, bus, index, uuid, primary): self.path = self.PATH_BASE + ‘/service’ + str(index) self.bus = bus self.uuid = uuid self.primary = primary self.characteristics = [] dbus.service.Object.__init__(self, bus, self.path) def get_properties(self): return { GATT_SERVICE_IFACE: { ‘UUID’: self.uuid, ‘Primary’: self.primary, ‘Characteristics’: dbus.Array( self.get_characteristic_paths(), signature=’o’) } } def get_path(self): return dbus.ObjectPath(self.path) def add_characteristic(self, chrc): self.characteristics.append(chrc) def get_characteristic_paths(self): return [c.get_path() for c in self.characteristics] def get_characteristics(self): return self.characteristics @dbus.service.method(DBUS_PROP_IFACE, in_signature=’s’, out_signature=’a{sv}’) def GetAll(self, interface): if interface != GATT_SERVICE_IFACE: raise dbus.exceptions.DBusException( ‘Not supported’, name=’org.freedesktop.DBus.Error.NotSupported’) return self.get_properties()[GATT_SERVICE_IFACE] # —- Base GATT Characteristic —- class GattCharacteristic(dbus.service.Object): def __init__(self, bus, index, uuid, flags, service): self.path = service.path + ‘/char’ + str(index) self.bus = bus self.uuid = uuid self.service = service self.flags = flags self.descriptors = [] self.notifying = False dbus.service.Object.__init__(self, bus, self.path) def get_properties(self): return { GATT_CHRC_IFACE: { ‘Service’: self.service.get_path(), ‘UUID’: self.uuid, ‘Flags’: self.flags, ‘Descriptors’: dbus.Array( self.get_descriptor_paths(), signature=’o’) } } def get_path(self): return dbus.ObjectPath(self.path) def add_descriptor(self, desc): self.descriptors.append(desc) def get_descriptor_paths(self): return [d.get_path() for d in self.descriptors] def get_descriptors(self): return self.descriptors @dbus.service.method(DBUS_PROP_IFACE, in_signature=’s’, out_signature=’a{sv}’) def GetAll(self, interface): if interface != GATT_CHRC_IFACE: raise dbus.exceptions.DBusException( ‘Not supported’, name=’org.freedesktop.DBus.Error.NotSupported’) return self.get_properties()[GATT_CHRC_IFACE] @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): print(f’Default ReadValue on {self.uuid}’) return [] @dbus.service.method(GATT_CHRC_IFACE, in_signature=’aya{sv}’) def WriteValue(self, value, options): print(f’Default WriteValue on {self.uuid}’) @dbus.service.method(GATT_CHRC_IFACE) def StartNotify(self): self.notifying = True print(f’StartNotify on {self.uuid}’) @dbus.service.method(GATT_CHRC_IFACE) def StopNotify(self): self.notifying = False print(f’StopNotify on {self.uuid}’) @dbus.service.signal(DBUS_PROP_IFACE, signature=’sa{sv}as’) def PropertiesChanged(self, interface, changed, invalidated): pass def notify_value(self, value): if self.notifying: self.PropertiesChanged( GATT_CHRC_IFACE, {‘Value’: dbus.Array(value, signature=’y’)}, []) # —- 1. Audio Input State Characteristic —- class AudioInputStateChrc(GattCharacteristic): def __init__(self, bus, index, service): GattCharacteristic.__init__( self, bus, index, AUDIO_INPUT_STATE_UUID, [‘read’, ‘notify’], service) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): value = aics.get_state_bytes() print(f’AudioInputState Read: {[hex(b) for b in value]}’) return dbus.Array(value, signature=’y’) # —- 2. Gain Setting Properties Characteristic —- class GainSettingPropsChrc(GattCharacteristic): def __init__(self, bus, index, service): GattCharacteristic.__init__( self, bus, index, GAIN_SETTING_PROPS_UUID, [‘read’], service) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): value = aics.get_gain_props_bytes() print(f’GainSettingProps Read: units={aics.gs_units}, ‘ f’min={aics.gs_minimum}, max={aics.gs_maximum}’) return dbus.Array(value, signature=’y’) # —- 3. Audio Input Type Characteristic —- class AudioInputTypeChrc(GattCharacteristic): def __init__(self, bus, index, service): GattCharacteristic.__init__( self, bus, index, AUDIO_INPUT_TYPE_UUID, [‘read’], service) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): print(f’AudioInputType Read: {hex(aics.input_type)}’) return dbus.Array([aics.input_type], signature=’y’) # —- 4. Audio Input Status Characteristic —- class AudioInputStatusChrc(GattCharacteristic): def __init__(self, bus, index, service): GattCharacteristic.__init__( self, bus, index, AUDIO_INPUT_STATUS_UUID, [‘read’, ‘notify’], service) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): print(f’AudioInputStatus Read: {hex(aics.input_status)}’) return dbus.Array([aics.input_status], signature=’y’) # —- 5. Audio Input Control Point Characteristic —- class AudioInputControlPointChrc(GattCharacteristic): def __init__(self, bus, index, service, state_chrc): GattCharacteristic.__init__( self, bus, index, AUDIO_INPUT_CP_UUID, [‘write’], service) self.state_chrc = state_chrc # Reference to notify state changes @dbus.service.method(GATT_CHRC_IFACE, in_signature=’aya{sv}’) def WriteValue(self, value, options): value = list(value) print(f’ControlPoint Write: {[hex(b) for b in value]}’) if len(value) < 2: raise dbus.exceptions.DBusException( ‘Invalid length’, name=’org.bluez.Error.Failed’) opcode = value[0] client_cc = value[1] # Validate opcode if opcode not in [OP_SET_GAIN, OP_UNMUTE, OP_MUTE, OP_SET_MANUAL, OP_SET_AUTO]: raise dbus.exceptions.DBusException( f’Opcode {hex(opcode)} not supported’, name=’org.bluez.Error.Failed’) # 0x81 # Validate Change_Counter if client_cc != aics.change_counter: print(f’CC mismatch: client={client_cc}, ‘ f’server={aics.change_counter}’) raise dbus.exceptions.DBusException( ‘Invalid Change Counter’, name=’org.bluez.Error.Failed’) # 0x80 # — Execute opcode — changed = False if opcode == OP_SET_GAIN: if len(value) < 3: raise dbus.exceptions.DBusException( ‘Missing gain operand’, name=’org.bluez.Error.Failed’) new_gain = struct.unpack(‘b’, bytes([value[2]]))[0] # uint8 to int8 if new_gain < aics.gs_minimum or new_gain > aics.gs_maximum: raise dbus.exceptions.DBusException( f’Gain {new_gain} out of range ‘ f'[{aics.gs_minimum}, {aics.gs_maximum}]’, name=’org.bluez.Error.Failed’) # 0x83 # Only apply in manual mode if aics.gain_mode in [0, 2]: # Manual Only or Manual if aics.gain_setting != new_gain: aics.gain_setting = new_gain changed = True # In auto mode: write succeeds but gain not applied elif opcode == OP_UNMUTE: if aics.mute == 2: # Disabled raise dbus.exceptions.DBusException( ‘Mute is disabled’, name=’org.bluez.Error.Failed’) # 0x82 if aics.mute != 0: aics.mute = 0 changed = True elif opcode == OP_MUTE: if aics.mute == 2: # Disabled raise dbus.exceptions.DBusException( ‘Mute is disabled’, name=’org.bluez.Error.Failed’) # 0x82 if aics.mute != 1: aics.mute = 1 changed = True elif opcode == OP_SET_MANUAL: if aics.gain_mode in [0, 1]: # Manual Only or Automatic Only raise dbus.exceptions.DBusException( ‘Gain mode change not allowed’, name=’org.bluez.Error.Failed’) # 0x84 if aics.gain_mode == 3: # Currently Automatic aics.gain_mode = 2 # Switch to Manual changed = True elif opcode == OP_SET_AUTO: if aics.gain_mode in [0, 1]: # Fixed modes raise dbus.exceptions.DBusException( ‘Gain mode change not allowed’, name=’org.bluez.Error.Failed’) # 0x84 if aics.gain_mode == 2: # Currently Manual aics.gain_mode = 3 # Switch to Automatic changed = True # If state changed, increment CC and notify if changed: aics.increment_cc() new_state = aics.get_state_bytes() print(f’State changed! New state: {[hex(b) for b in new_state]}’) self.state_chrc.notify_value(new_state) # —- 6. Audio Input Description Characteristic —- class AudioInputDescriptionChrc(GattCharacteristic): def __init__(self, bus, index, service): GattCharacteristic.__init__( self, bus, index, AUDIO_INPUT_DESC_UUID, [‘read’, ‘write-without-response’, ‘notify’], service) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’a{sv}’, out_signature=’ay’) def ReadValue(self, options): encoded = [ord(c) for c in aics.description] print(f’AudioInputDescription Read: “{aics.description}”‘) return dbus.Array(encoded, signature=’y’) @dbus.service.method(GATT_CHRC_IFACE, in_signature=’aya{sv}’) def WriteValue(self, value, options): new_desc = ”.join([chr(b) for b in value]) old_desc = aics.description aics.description = new_desc print(f’AudioInputDescription changed: “{old_desc}” → “{new_desc}”‘) if old_desc != new_desc: self.notify_value(list(value)) # —- GATT Application (ObjectManager) —- class GattApplication(dbus.service.Object): def __init__(self, bus): self.path = ‘/’ self.services = [] dbus.service.Object.__init__(self, bus, self.path) self.add_aics_service(bus) def get_path(self): return dbus.ObjectPath(self.path) def add_service(self, service): self.services.append(service) def add_aics_service(self, bus): # Create the AICS secondary service svc = GattService(bus, 0, AICS_SVC_UUID, False) # False = Secondary # Create all 6 characteristics state_chrc = AudioInputStateChrc(bus, 0, svc) props_chrc = GainSettingPropsChrc(bus, 1, svc) type_chrc = AudioInputTypeChrc(bus, 2, svc) status_chrc = AudioInputStatusChrc(bus, 3, svc) cp_chrc = AudioInputControlPointChrc(bus, 4, svc, state_chrc) desc_chrc = AudioInputDescriptionChrc(bus, 5, svc) svc.add_characteristic(state_chrc) svc.add_characteristic(props_chrc) svc.add_characteristic(type_chrc) svc.add_characteristic(status_chrc) svc.add_characteristic(cp_chrc) svc.add_characteristic(desc_chrc) self.add_service(svc) self.services.append(svc) # Also keep characteristic objects accessible for c in svc.get_characteristics(): self.services.append(c) # reuse list for managed objects @dbus.service.method(DBUS_OM_IFACE, out_signature=’a{oa{sa{sv}}}’) def GetManagedObjects(self): response = {} for svc in self.services: response[svc.get_path()] = svc.get_properties() if hasattr(svc, ‘get_characteristics’): for chrc in svc.get_characteristics(): response[chrc.get_path()] = chrc.get_properties() return response # —- Main entry point —- def main(): dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() # Find the first Bluetooth adapter adapter_path = None om = dbus.Interface( bus.get_object(BLUEZ_SERVICE_NAME, ‘/’), DBUS_OM_IFACE) objects = om.GetManagedObjects() for path, ifaces in objects.items(): if GATT_MANAGER_IFACE in ifaces: adapter_path = path break if not adapter_path: print(‘ERROR: No Bluetooth adapter with GattManager1 found.’) print(‘Make sure bluetoothd is running and hci0 is up.’) sys.exit(1) print(f’Using adapter: {adapter_path}’) gatt_mgr = dbus.Interface( bus.get_object(BLUEZ_SERVICE_NAME, adapter_path), GATT_MANAGER_IFACE) app = GattApplication(bus) print(‘Registering AICS GATT Application…’) gatt_mgr.RegisterApplication( app.get_path(), {}, reply_handler=lambda: print(‘AICS GATT server registered! Waiting for connections…’), error_handler=lambda e: print(f’Registration failed: {e}’)) mainloop = GLib.MainLoop() try: mainloop.run() except KeyboardInterrupt: print(‘\nShutting down AICS server…’) gatt_mgr.UnregisterApplication(app.get_path()) if __name__ == ‘__main__’: main()

## Testing commands using bluetoothctl (interactive) $ bluetoothctl # Step 1: scan, connect, pair [bluetooth]# scan on [bluetooth]# pair AA:BB:CC:DD:EE:FF [bluetooth]# connect AA:BB:CC:DD:EE:FF [bluetooth]# trust AA:BB:CC:DD:EE:FF # Step 2: Discover GATT services [bluetooth]# menu gatt [bluetooth]# list-attributes AA:BB:CC:DD:EE:FF # Find the Audio Input Control Service attributes # Output shows UUIDs for each characteristic # Step 3: Select and read Audio Input State [bluetooth]# select-attribute /org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF/service0006/char0007 [bluetooth]# read # Value: 00 00 02 05 # Step 4: Enable notifications [bluetooth]# notify on # Step 5: Select Control Point and write Mute command [bluetooth]# select-attribute /org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF/service0006/char000f [bluetooth]# write “0x03 0x05” # Watch terminal — you should see notification arrive in state char terminal # Step 6: Test error — stale CC [bluetooth]# write “0x03 0x05” # Error: ATT error: 0x80 # Step 7: Read new state to get updated CC [bluetooth]# select-attribute …char0007 [bluetooth]# read # Value: 00 01 02 06 (Gain=0, Muted, Manual, CC=6)

## Add this function to aics_gatt_server.py for testing ## Call it via a separate thread or timer to simulate hardware events def simulate_hardware_mute_disable(state_chrc): “””Simulate a physical privacy switch that disables mute control””” import time time.sleep(5) # Wait 5 seconds after server starts print(“\n[HARDWARE EVENT] Privacy switch pressed — Mute → Disabled”) aics.mute = 2 # Set to Disabled aics.increment_cc() # CC must increment new_state = aics.get_state_bytes() print(f”Notifying clients: {[hex(b) for b in new_state]}”) state_chrc.notify_value(new_state) # Now if client tries to Mute or Unmute: # Server returns ATT Error 0x82 (Mute Disabled) # Only a hardware event (like this function) can re-enable it

## HCI-level dump to see ATT PDUs (run as root in separate terminal) sudo hcidump -i hci0 -t -a -X 2>/dev/null | grep -A5 ‘ATT’ ## Or use btmon for cleaner output: sudo btmon ## Expected ATT traffic when client reads Audio Input State: ## > ACL Data RX: handle 0x0040 flags 0x02 dlen 9 ## ATT: Read Request (0x0a) len 2 ## Handle: 0x0022 ## ## < ACL Data TX: handle 0x0040 flags 0x00 dlen 10 ## ATT: Read Response (0x0b) len 4 ## Value: 00 00 02 05 ← [Gain][Mute][Mode][CC] ## When client writes Set Gain and server responds: ## > ACL Data RX: handle 0x0040 flags 0x02 dlen 9 ## ATT: Write Request (0x12) len 3 ## Handle: 0x0025 ← AICP handle ## Value: 01 05 03 ← SetGain, CC=5, gain=3 ## ## < ACL Data TX: handle 0x0040 flags 0x00 dlen 5 ## ATT: Write Response (0x13) ← success ## ## < ACL Data TX: handle 0x0040 flags 0x00 dlen 9 ## ATT: Handle Value Notification (0x1b) len 5 ## Handle: 0x0022 ← Audio Input State notifying new value ## Value: 03 00 02 06 ← Gain=3, NotMuted, Manual, CC=6