diff --git a/linux/Main.qml b/linux/Main.qml index 9e95b34..983a2a5 100644 --- a/linux/Main.qml +++ b/linux/Main.qml @@ -156,6 +156,13 @@ ApplicationWindow { checked: airPodsTrayApp.deviceInfo.conversationalAwareness onCheckedChanged: airPodsTrayApp.setConversationalAwareness(checked) } + + Switch { + visible: airPodsTrayApp.airpodsConnected + text: "Hearing Aid" + checked: airPodsTrayApp.deviceInfo.hearingAidEnabled + onCheckedChanged: airPodsTrayApp.setHearingAidEnabled(checked) + } } RoundButton { diff --git a/linux/README.md b/linux/README.md index 6fcbf56..742d991 100644 --- a/linux/README.md +++ b/linux/README.md @@ -6,7 +6,10 @@ A native Linux application to control your AirPods, with support for: - Conversational Awareness - Battery monitoring - Auto play/pause on ear detection -- Seamless handoff between phone and PC +- Hearing Aid features + - Supports adjusting hearing aid- amplification, balance, tone, ambient noise reduction, own voice amplification, and conversation boost + - Supports setting the values for left and right hearing aids (this is not a hearing test! you need to have an audiogram to set the values) +- Seamless handoff between Android and Linux ## Prerequisites @@ -97,3 +100,35 @@ systemctl --user enable --now mpris-proxy - Switch between noise control modes - View battery levels - Control playback + +## Hearing Aid + +To use hearing aid features, you need to have an audiogram. To enable/disable hearing aid, you can use the toggle in the main app. But, to adjust the settings and set the audiogram, you need to use a different script which is located in this folder as `hearing_aid.py`. You can run it with: + +```bash +python3 hearing_aid.py +``` + +The script will load the current settings from the AirPods and allow you to adjust them. You can set the audiogram by providing the values for 8 frequencies (250Hz, 500Hz, 1kHz, 2kHz, 3kHz, 4kHz, 6kHz, 8kHz) for both left and right ears. There are also options to adjust amplification, balance, tone, ambient noise reduction, own voice amplification, and conversation boost. + +AirPods check for the DeviceID characteristic to see if the connected device is an Apple device and only then allow hearing aid features. To set the DeviceID characteristic, you need to add this line to your bluetooth configuration file (usually located at `/etc/bluetooth/main.conf`): + +``` +DeviceID = bluetooth:004C:0000:0000 +``` + +Then, restart the bluetooth service: + +```bash +sudo systemctl restart bluetooth +``` + +Here, you might need to re-pair your AirPods because they seem to cache this info. + +### Troubleshooting + +It is possible that the AirPods disconnect after a short period of time and play the disconnect sound. This is likely due to the AirPods expecting some information from an Apple device. Since I have not implemented everything that an Apple device does, the AirPods may disconnect. You don't need to reconnect them manually; the script will handle reconnection automatically for hearing aid features. So, once you are done setting the hearing aid features, change back the `DeviceID` to whatever it was before. + +### Why a separate script? + +Because I discovered that QBluetooth doesn't support connecting to a socket with its PSM, only a UUID can be used. I could add a dependency on BlueZ, but then having two bluetooth interfaces seems unnecessary. So, I decided to use a separate script for hearing aid features. In the future, QBluetooth will be replaced with BlueZ native calls, and then everything will be in one application. \ No newline at end of file diff --git a/linux/airpods_packets.h b/linux/airpods_packets.h index 2593345..3316f8d 100644 --- a/linux/airpods_packets.h +++ b/linux/airpods_packets.h @@ -107,6 +107,16 @@ namespace AirPodsPackets inline std::optional parseState(const QByteArray &data) { return Type::parseState(data); } } + // Hearing Aid + namespace HearingAid + { + using Type = BasicControlCommand<0x2C>; + static const QByteArray ENABLED = Type::ENABLED; + static const QByteArray DISABLED = Type::DISABLED; + static const QByteArray HEADER = Type::HEADER; + inline std::optional parseState(const QByteArray &data) { return Type::parseState(data); } + } + // Allow Off Option namespace AllowOffOption { diff --git a/linux/deviceinfo.hpp b/linux/deviceinfo.hpp index 6e5e17f..7a4c7a0 100644 --- a/linux/deviceinfo.hpp +++ b/linux/deviceinfo.hpp @@ -15,6 +15,7 @@ class DeviceInfo : public QObject Q_PROPERTY(QString batteryStatus READ batteryStatus WRITE setBatteryStatus NOTIFY batteryStatusChanged) Q_PROPERTY(int noiseControlMode READ noiseControlModeInt WRITE setNoiseControlModeInt NOTIFY noiseControlModeChangedInt) Q_PROPERTY(bool conversationalAwareness READ conversationalAwareness WRITE setConversationalAwareness NOTIFY conversationalAwarenessChanged) + Q_PROPERTY(bool hearingAidEnabled READ hearingAidEnabled WRITE setHearingAidEnabled NOTIFY hearingAidEnabledChanged) Q_PROPERTY(int adaptiveNoiseLevel READ adaptiveNoiseLevel WRITE setAdaptiveNoiseLevel NOTIFY adaptiveNoiseLevelChanged) Q_PROPERTY(QString deviceName READ deviceName WRITE setDeviceName NOTIFY deviceNameChanged) Q_PROPERTY(Battery *battery READ getBattery CONSTANT) @@ -67,6 +68,16 @@ public: } } + bool hearingAidEnabled() const { return m_hearingAidEnabled; } + void setHearingAidEnabled(bool enabled) + { + if (m_hearingAidEnabled != enabled) + { + m_hearingAidEnabled = enabled; + emit hearingAidEnabledChanged(enabled); + } + } + int adaptiveNoiseLevel() const { return m_adaptiveNoiseLevel; } void setAdaptiveNoiseLevel(int level) { @@ -159,6 +170,7 @@ public: setNoiseControlMode(NoiseControlMode::Off); setBluetoothAddress(""); getEarDetection()->reset(); + setHearingAidEnabled(false); } void saveToSettings(QSettings &settings) @@ -168,6 +180,7 @@ public: settings.setValue("model", static_cast(model())); settings.setValue("magicAccIRK", magicAccIRK()); settings.setValue("magicAccEncKey", magicAccEncKey()); + settings.setValue("hearingAidEnabled", hearingAidEnabled()); settings.endGroup(); } void loadFromSettings(const QSettings &settings) @@ -176,6 +189,7 @@ public: setModel(static_cast(settings.value("DeviceInfo/model", (int)(AirPodsModel::Unknown)).toInt())); setMagicAccIRK(settings.value("DeviceInfo/magicAccIRK", QByteArray()).toByteArray()); setMagicAccEncKey(settings.value("DeviceInfo/magicAccEncKey", QByteArray()).toByteArray()); + setHearingAidEnabled(settings.value("DeviceInfo/hearingAidEnabled", false).toBool()); } void updateBatteryStatus() @@ -191,6 +205,7 @@ signals: void noiseControlModeChanged(NoiseControlMode mode); void noiseControlModeChangedInt(int mode); void conversationalAwarenessChanged(bool enabled); + void hearingAidEnabledChanged(bool enabled); void adaptiveNoiseLevelChanged(int level); void deviceNameChanged(const QString &name); void primaryChanged(); @@ -202,6 +217,7 @@ private: QString m_batteryStatus; NoiseControlMode m_noiseControlMode = NoiseControlMode::Transparency; bool m_conversationalAwareness = false; + bool m_hearingAidEnabled = false; int m_adaptiveNoiseLevel = 50; QString m_deviceName; Battery *m_battery; diff --git a/linux/hearing-aid-adjustments.py b/linux/hearing-aid-adjustments.py new file mode 100644 index 0000000..2312b8e --- /dev/null +++ b/linux/hearing-aid-adjustments.py @@ -0,0 +1,480 @@ +import sys +import socket +import struct +import threading +from queue import Queue +import logging +import signal + +# Configure logging +logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') + +from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QSlider, QCheckBox, QPushButton, QLineEdit, QFormLayout, QGridLayout +from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject + +OPCODE_READ_REQUEST = 0x0A +OPCODE_WRITE_REQUEST = 0x12 +OPCODE_HANDLE_VALUE_NTF = 0x1B + +ATT_HANDLES = { + 'TRANSPARENCY': 0x18, + 'LOUD_SOUND_REDUCTION': 0x1B, + 'HEARING_AID': 0x2A, +} + +ATT_CCCD_HANDLES = { + 'TRANSPARENCY': ATT_HANDLES['TRANSPARENCY'] + 1, + 'LOUD_SOUND_REDUCTION': ATT_HANDLES['LOUD_SOUND_REDUCTION'] + 1, + 'HEARING_AID': ATT_HANDLES['HEARING_AID'] + 1, +} + +PSM_ATT = 31 + +class ATTManager: + def __init__(self, mac_address): + self.mac_address = mac_address + self.sock = None + self.responses = Queue() + self.listeners = {} + self.notification_thread = None + self.running = False + # Avoid logging full MAC address to prevent sensitive data exposure + mac_tail = ':'.join(mac_address.split(':')[-2:]) if isinstance(mac_address, str) and ':' in mac_address else '[redacted]' + logging.info(f"ATTManager initialized") + + def connect(self): + logging.info("Attempting to connect to ATT socket") + self.sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_L2CAP) + self.sock.connect((self.mac_address, PSM_ATT)) + self.sock.settimeout(0.1) + self.running = True + self.notification_thread = threading.Thread(target=self._listen_notifications) + self.notification_thread.start() + logging.info("Connected to ATT socket") + + def disconnect(self): + logging.info("Disconnecting from ATT socket") + self.running = False + if self.sock: + logging.info("Closing socket") + self.sock.close() + if self.notification_thread: + logging.info("Stopping notification thread") + self.notification_thread.join(timeout=1.0) + logging.info("Disconnected from ATT socket") + + def register_listener(self, handle, listener): + if handle not in self.listeners: + self.listeners[handle] = [] + self.listeners[handle].append(listener) + logging.debug(f"Registered listener for handle {handle}") + + def unregister_listener(self, handle, listener): + if handle in self.listeners: + self.listeners[handle].remove(listener) + logging.debug(f"Unregistered listener for handle {handle}") + + def enable_notifications(self, handle): + self.write_cccd(handle, b'\x01\x00') + logging.info(f"Enabled notifications for handle {handle.name}") + + def read(self, handle): + handle_value = ATT_HANDLES[handle.name] + lsb = handle_value & 0xFF + msb = (handle_value >> 8) & 0xFF + pdu = bytes([OPCODE_READ_REQUEST, lsb, msb]) + logging.debug(f"Sending read request for handle {handle.name}: {pdu.hex()}") + self._write_raw(pdu) + response = self._read_response() + logging.debug(f"Read response for handle {handle.name}: {response.hex()}") + return response + + def write(self, handle, value): + handle_value = ATT_HANDLES[handle.name] + lsb = handle_value & 0xFF + msb = (handle_value >> 8) & 0xFF + pdu = bytes([OPCODE_WRITE_REQUEST, lsb, msb]) + value + logging.debug(f"Sending write request for handle {handle.name}: {pdu.hex()}") + self._write_raw(pdu) + try: + self._read_response() + logging.debug(f"Write response received for handle {handle.name}") + except: + logging.warning(f"No write response received for handle {handle.name}") + + def write_cccd(self, handle, value): + handle_value = ATT_CCCD_HANDLES[handle.name] + lsb = handle_value & 0xFF + msb = (handle_value >> 8) & 0xFF + pdu = bytes([OPCODE_WRITE_REQUEST, lsb, msb]) + value + logging.debug(f"Sending CCCD write request for handle {handle.name}: {pdu.hex()}") + self._write_raw(pdu) + try: + self._read_response() + logging.debug(f"CCCD write response received for handle {handle.name}") + except: + logging.warning(f"No CCCD write response received for handle {handle.name}") + + def _write_raw(self, pdu): + self.sock.send(pdu) + logging.debug(f"Sent PDU: {pdu.hex()}") + + def _read_pdu(self): + try: + data = self.sock.recv(512) + logging.debug(f"Received PDU: {data.hex()}") + return data + except socket.timeout: + return None + except: + raise + + def _read_response(self, timeout=2.0): + try: + response = self.responses.get(timeout=timeout)[1:] # Skip opcode + logging.debug(f"Response received: {response.hex()}") + return response + except: + logging.error("No response received within timeout") + raise Exception("No response received") + + def _listen_notifications(self): + logging.info("Starting notification listener thread") + while self.running: + try: + pdu = self._read_pdu() + except: + break + if pdu is None: + continue + if len(pdu) > 0 and pdu[0] == OPCODE_HANDLE_VALUE_NTF: + logging.debug(f"Notification PDU received: {pdu.hex()}") + handle = pdu[1] | (pdu[2] << 8) + value = pdu[3:] + logging.debug(f"Notification for handle {handle}: {value.hex()}") + if handle in self.listeners: + for listener in self.listeners[handle]: + listener(value) + else: + self.responses.put(pdu) + logging.info("Notification listener thread stopped, trying to reconnect") + if self.running: + try: + self.connect() + except Exception as e: + logging.error(f"Reconnection failed: {e}") + +class HearingAidSettings: + def __init__(self, left_eq, right_eq, left_amp, right_amp, left_tone, right_tone, + left_conv, right_conv, left_anr, right_anr, net_amp, balance, own_voice): + self.left_eq = left_eq + self.right_eq = right_eq + self.left_amplification = left_amp + self.right_amplification = right_amp + self.left_tone = left_tone + self.right_tone = right_tone + self.left_conversation_boost = left_conv + self.right_conversation_boost = right_conv + self.left_ambient_noise_reduction = left_anr + self.right_ambient_noise_reduction = right_anr + self.net_amplification = net_amp + self.balance = balance + self.own_voice_amplification = own_voice + logging.debug(f"HearingAidSettings created: amp={net_amp}, balance={balance}, tone={left_tone}, anr={left_anr}, conv={left_conv}") + +def parse_hearing_aid_settings(data): + logging.debug(f"Parsing hearing aid settings from data: {data.hex()}") + if len(data) < 104: + logging.warning("Data too short for parsing") + return None + buffer = data + offset = 0 + + offset += 4 + + logging.info(f"Parsing hearing aid settings, starting read at offset 4, value: {buffer[offset]:02x}") + + left_eq = [] + for i in range(8): + val, = struct.unpack(' 0.5 + offset += 4 + left_anr, = struct.unpack(' 0.5 + offset += 4 + right_anr, = struct.unpack(' 0 else amp + + left_eq = [float(input_box.text() or 0) for input_box in self.left_eq_inputs] + right_eq = [float(input_box.text() or 0) for input_box in self.right_eq_inputs] + + settings = HearingAidSettings( + left_eq, right_eq, left_amp, right_amp, tone, tone, + conv, conv, anr, anr, amp, balance, own_voice + ) + threading.Thread(target=send_hearing_aid_settings, args=(self.att_manager, settings)).start() + + def reset_settings(self): + logging.debug("Resetting settings to defaults") + self.amp_slider.setValue(0) + self.balance_slider.setValue(0) + self.tone_slider.setValue(0) + self.anr_slider.setValue(50) + self.conv_checkbox.setChecked(False) + self.own_voice_slider.setValue(50) + self.on_value_changed() + + def closeEvent(self, event): + logging.info("Closing app") + self.att_manager.disconnect() + event.accept() + +if __name__ == "__main__": + mac = None + if len(sys.argv) != 2: + logging.error("Usage: python hearing-aid-adjustments.py ") + sys.exit(1) + mac = sys.argv[1] + mac_regex = r'^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$' + import re + if not re.match(mac_regex, mac): + logging.error("Invalid MAC address format") + sys.exit(1) + logging.info(f"Starting app") + app = QApplication(sys.argv) + + def quit_app(signum, frame): + app.quit() + + signal.signal(signal.SIGINT, quit_app) + + window = HearingAidApp(mac) + window.show() + sys.exit(app.exec_()) diff --git a/linux/imgs/main-app.png b/linux/imgs/main-app.png new file mode 100644 index 0000000..2f64f20 Binary files /dev/null and b/linux/imgs/main-app.png differ diff --git a/linux/main.cpp b/linux/main.cpp index a9c3b88..9e8d7b9 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -42,6 +42,7 @@ class AirPodsTrayApp : public QObject { Q_PROPERTY(bool hideOnStart READ hideOnStart CONSTANT) Q_PROPERTY(DeviceInfo *deviceInfo READ deviceInfo CONSTANT) Q_PROPERTY(QString phoneMacStatus READ phoneMacStatus NOTIFY phoneMacStatusChanged) + Q_PROPERTY(bool hearingAidEnabled READ hearingAidEnabled WRITE setHearingAidEnabled NOTIFY hearingAidEnabledChanged) public: AirPodsTrayApp(bool debugMode, bool hideOnStart, QQmlApplicationEngine *parent = nullptr) @@ -131,6 +132,7 @@ public: bool hideOnStart() const { return m_hideOnStart; } DeviceInfo *deviceInfo() const { return m_deviceInfo; } QString phoneMacStatus() const { return m_phoneMacStatus; } + bool hearingAidEnabled() const { return m_deviceInfo->hearingAidEnabled(); } private: bool debugMode; @@ -367,6 +369,16 @@ public slots: emit phoneMacStatusChanged(); } + void setHearingAidEnabled(bool enabled) + { + LOG_INFO("Setting hearing aid to: " << (enabled ? "enabled" : "disabled")); + QByteArray packet = enabled ? AirPodsPackets::HearingAid::ENABLED + : AirPodsPackets::HearingAid::DISABLED; + + writePacketToSocket(packet, "Hearing aid packet written: "); + m_deviceInfo->setHearingAidEnabled(enabled); + } + bool writePacketToSocket(const QByteArray &packet, const QString &logMessage) { if (socket && socket->isOpen()) @@ -682,6 +694,14 @@ private slots: LOG_INFO("Conversational awareness state received: " << m_deviceInfo->conversationalAwareness()); } } + // Hearing Aid state + else if (data.startsWith(AirPodsPackets::HearingAid::HEADER)) { + if (auto result = AirPodsPackets::HearingAid::parseState(data)) + { + m_deviceInfo->setHearingAidEnabled(result.value()); + LOG_INFO("Hearing aid state received: " << m_deviceInfo->hearingAidEnabled()); + } + } // Noise Control Mode else if (data.size() == 11 && data.startsWith(AirPodsPackets::NoiseControl::HEADER)) { @@ -944,6 +964,7 @@ signals: void retryAttemptsChanged(int attempts); void oneBudANCModeChanged(bool enabled); void phoneMacStatusChanged(); + void hearingAidEnabledChanged(bool enabled); private: QBluetoothSocket *socket = nullptr;