Add native Linux controls and tray menu for BTD 700

This commit is contained in:
Justin
2026-09-05 20:43:35 +02:00
commit 3579667c6e
25 changed files with 2047 additions and 0 deletions
+12
View File
@@ -0,0 +1,12 @@
__pycache__/
*.py[cod]
.venv/
build/
dist/
*.egg-info/
.agents/
.codex/
analysis/
*.exe
*.dll
*.dfu
+148
View File
@@ -0,0 +1,148 @@
# BTD 700 Control für Linux
Native GTK-4-App zur Steuerung des **Sennheiser BTD 700**, mit einem Menü im
Infobereich von GNOME/KDE. Eigenständige Implementierung des HID-Steuerprotokolls,
analysiert anhand der offiziellen Windows-Anwendung **Dongle Control 1.0.5**.
**Keine Firmware-Updates, Firmware-Downloads oder Update-Schnittstellen.** Die App
benötigt weder Windows/Wine noch eine Netzwerkverbindung. Sennheiser-Programmcode,
Logos und Firmware werden nicht mitgeliefert. Dies ist ein unabhängiges Projekt.
## Starten
```bash
./run.sh
```
Das Fenster zeigt den echten Dongle-Status. Beim Schließen bleibt die App im
Infobereich aktiv. „Beenden“ im Symbolmenü oder das Beenden-Symbol im Fenster
beendet die App vollständig. Wenn der Desktop keinen Infobereich anbietet,
beendet das Schließen des Fensters die App.
```bash
python3 install.py # Anwendungsmenü-Eintrag für diesen Projektordner
./run.sh --background # Nur im Infobereich starten
./run.sh --demo # Simulierter Dongle, keinerlei USB-Zugriff
```
Den Projektordner nach der Installation behalten. „Beim Anmelden starten“ in der
App aktiviert bei Bedarf den Autostart im Infobereich. Er ist standardmäßig aus.
## Steuerung
- **Standard / Gaming / Auracast** direkt im Fenster oder Symbolmenü wählen.
- Aktiven **Codec, Auflösung, Abtastrate und Verbindungsstatus** ablesen.
- **Codecs** auswählen, die der Dongle für die aktuelle Verbindung meldet.
Die Auswahl steht im Standard-Modus zur Verfügung; Gaming setzt seinen Codec selbst.
- **Bluetooth Classic / LE Audio / Automatisch** auswählen, entsprechend den
vom angeschlossenen Kopfhörer gemeldeten Möglichkeiten.
- Bereits gekoppelte Kopfhörer **verbinden / trennen**.
- **Auracast:** Name, Passwortschutz, öffentliche Auffindbarkeit und Qualität.
Das Symbolmenü bietet Schalter und Untermenüs; für Name/Passwort öffnet es die Eingabe im Fenster.
- **Werkseinstellungen** nur nach ausdrücklicher Bestätigung im Dialog. Dabei
werden die Kopplungen gelöscht. Die App führt niemals automatisch einen Reset aus.
Das Passwort wird verdeckt eingegeben und weder gespeichert noch protokolliert.
Ein leeres Passwortfeld behält das bisherige Passwort; „Passwortschutz“ ausschalten
sendet ohne Passwortschutz. Der Name erlaubt 416 ASCII-Buchstaben/Ziffern mit
inneren Leerzeichen; ein leerer Name stellt den Gerätenamen wieder her. Passwörter
verwenden 416 ASCII-Buchstaben/Ziffern, entsprechend den Original-Eingabegrenzen.
**„Öffentlich auffindbar“ ist die Auracast-Ankündigung, kein Ein/Aus-Schalter für
Audio.** Zum Starten/Beenden der Übertragung den Audiomodus umstellen.
Die angezeigten Bit-/kHz-Werte stammen vom Dongle. Die USB-Abtastrate wird vom
Audiosystem (z. B. PipeWire) bestimmt und ist kein hier nachgewiesener Stellbefehl.
Koppeln erfolgt über die Taste am Dongle; „Verbinden“ versucht die bestehende Kopplung.
## Voraussetzungen
Python ≥ 3.10, PyGObject, GTK ≥ 4.10, libadwaita ≥ 1.5 und libdbusmenu mit
GObject-Introspection. Auf dem hier verwendeten Bazzite sind die benötigten
Bibliotheken bereits vorhanden; keine zusätzlichen Pakete waren erforderlich.
Für andere Systeme typischerweise:
```bash
# Fedora (klassisch, veränderliches System)
sudo dnf install python3-gobject gtk4 libadwaita libdbusmenu
# Debian / Ubuntu
sudo apt install python3-gi gir1.2-gtk-4.0 gir1.2-adw-1 gir1.2-dbusmenu-glib-0.4
```
Auf GNOME muss der Desktop StatusNotifier/AppIndicator-Symbole unterstützen;
beispielsweise mit „AppIndicator and KStatusNotifierItem Support“. Auf dem
getesteten GNOME ist derselbe Dienst bereits für JetBrains Toolbox und andere Apps aktiv.
KDE stellt diese Schnittstelle über seinen Infobereich bereit.
Fehlt die Menü-Bibliothek, bleibt das Fenster nutzbar. Das Terminalinterface
benötigt ausschließlich die Python-Standardbibliothek.
## USB-Zugriff
Es wird ausschließlich die passende HID-Steuerschnittstelle des Geräts
`3542:3001` geöffnet, identifiziert über ihren Report-Deskriptor. Die zweite
Schnittstelle für Updates bleibt geschlossen. Audiotreiber werden nicht getrennt.
Falls die App fehlende Berechtigungen meldet:
```bash
sudo install -m 0644 packaging/70-btd700-control.rules /etc/udev/rules.d/70-btd700-control.rules
sudo udevadm control --reload-rules
```
Anschließend den Dongle aus- und wieder einstecken. Die Regel beschränkt den Zugriff
auf die Steuerungsschnittstelle und den aktiven lokalen Benutzer. Die App nicht
als root starten. Auf dem getesteten System ist bereits Zugriff vorhanden.
## Terminal
```bash
./run.sh devices
./run.sh status
./run.sh mode standard
./run.sh mode gaming
./run.sh mode standard --transport auto
./run.sh codec adaptive
./run.sh disconnect
./run.sh connect
./run.sh auracast --name 'Wohnzimmer' --quality high --public on
./run.sh auracast --password --encryption on # Passwort verdeckt abfragen
./run.sh mode auracast
```
`status` gibt JSON aus und liest kein Passwort. Nur eine Instanz darf den Dongle
halten: Vor CLI-Befehlen eine laufende GUI über „Beenden“ schließen. Mehrere Dongles
lassen sich mit `--device /dev/hidrawN` vor dem Unterbefehl auswählen.
## Verifikation
```bash
python3 -m unittest discover -s tests -v
python3 tools/check_gui.py
```
`check_gui.py` verwendet **ausschließlich einen simulierten Dongle**, öffnet eine
Demo-Oberfläche in der bestehenden Sitzung und prüft die Menüaktionen über den
echten D-Bus-Menüexport. Es speichert Fensterbilder unter `/tmp/btd700-gui-test*.png`.
Nachgewiesen am angeschlossenen BTD 700 mit Firmware **3.11.0**:
- Erkennung und Öffnen der korrekten HID-Schnittstelle ohne Treiberwechsel.
- Echte Antworten auf Status-, Modus-, Codec-, Audioqualitäts-, LE-Zustands-,
Transport-, Auracast-Konfigurations-, Namens- und Firmwareversionsabfragen.
- Anzeige im GTK-Fenster und Registrierung im GNOME-Infobereich.
Schreibbefehle wurden aus der Originalsoftware rekonstruiert und in Simulation
geprüft. Der Nutzer hat die Funktion der App am eigenen Dongle bestätigt. Eine
Aufschlüsselung aller dabei getesteten Funktionen liegt nicht vor; der automatisierte
Hardware-Umschalttest wurde bisher nicht ausgeführt.
Das vorbereitete `tools/check_hardware.py --run` schaltet Modi und Codec um,
prüft Auracast-Werte und versucht anschließend die gesicherten Originalwerte
wiederherzustellen. Es führt weder Firmware-Updates noch einen Werksreset aus.
Den Hardwaretest nur bewusst starten: Er unterbricht kurz die Audiowiedergabe
und verändert vorübergehend gespeicherte Einstellungen einschließlich des Passworts. Bei USB-Trennung während
einer Änderung ist eine Wiederherstellung nicht garantiert.
Protokoll, Herkunft und bekannte Grenzen: [docs/PROTOCOL.md](docs/PROTOCOL.md).
+3
View File
@@ -0,0 +1,3 @@
"""Independent Linux controls for the Sennheiser BTD 700."""
__version__ = "0.1.0"
+85
View File
@@ -0,0 +1,85 @@
import argparse
import getpass
import json
import sys
def main():
parser = argparse.ArgumentParser(description='Native Linux-Steuerung für den BTD 700; ohne Firmware-Updates.')
parser.add_argument('--device', help='Steuergerät aus dem Befehl devices, z. B. /dev/hidraw6')
parser.add_argument('--background', action='store_true', help='Nur im Infobereich starten')
parser.add_argument('--demo', action='store_true', help='Vorführmodus ohne USB-Zugriff')
parser.add_argument('--smoke-seconds', type=int, default=0, help=argparse.SUPPRESS)
subs = parser.add_subparsers(dest='command')
subs.add_parser('devices', help='BTD-700-Steuerschnittstellen auflisten')
subs.add_parser('status', help='Aktuellen Status als JSON ausgeben (ohne Passwort)')
mode = subs.add_parser('mode', help='Audiomodus wählen')
mode.add_argument('value', choices=['standard', 'gaming', 'auracast'])
mode.add_argument('--transport', choices=['auto', 'classic', 'le'])
codec = subs.add_parser('codec', help='Codec im Standard-Modus wählen')
codec.add_argument('value', choices=['sbc', 'aptx', 'adaptive', 'lossless', 'qmap', 'lc3'])
subs.add_parser('connect', help='Gekoppelte Kopfhörer verbinden')
subs.add_parser('disconnect', help='Kopfhörer trennen')
broadcast = subs.add_parser('auracast', help='Auracast-Einstellungen speichern')
broadcast.add_argument('--name')
broadcast.add_argument('--password', action='store_true', help='Neues Passwort verdeckt abfragen')
broadcast.add_argument('--public', choices=['on', 'off'])
broadcast.add_argument('--encryption', choices=['on', 'off'])
broadcast.add_argument('--quality', choices=['16k', '24k', 'high'])
reset = subs.add_parser('reset', help='Alle Kopplungen und Einstellungen löschen')
reset.add_argument('--confirm', action='store_true')
args = parser.parse_args()
if not args.command:
try:
from .app import run
return run(background=args.background, demo=args.demo, device=args.device, smoke_seconds=args.smoke_seconds)
except (ImportError, ValueError) as exc:
print(f'GTK-Oberfläche nicht verfügbar: {exc}\nAbhängigkeiten: siehe README.md. CLI: ./run.sh status', file=sys.stderr)
return 1
from .transport import DeviceError, Hidraw, discover
from .controller import Controller
from .protocol import ProtocolError
if args.command == 'devices':
from dataclasses import asdict
print(json.dumps([asdict(d) for d in discover()], indent=2))
return 0
transport = None
try:
if args.demo:
from .demo import DemoTransport
transport = DemoTransport()
else:
devices = [d for d in discover() if not args.device or d.path == args.device]
if len(devices) != 1:
raise DeviceError('Kein eindeutiger BTD 700 gefunden. devices ausführen; bei mehreren Dongles --device angeben.')
transport = Hidraw(devices[0])
controller = Controller(transport)
if args.command == 'mode':
controller.set_mode(['standard', 'gaming', 'auracast'].index(args.value),
{'auto': 3, 'classic': 1, 'le': 2}.get(args.transport))
elif args.command == 'codec':
controller.set_codec(1 << ['sbc', 'aptx', 'adaptive', 'lossless', 'qmap', 'lc3'].index(args.value))
elif args.command in ('connect', 'disconnect'):
controller.set_connection(args.command == 'connect')
elif args.command == 'auracast':
controller.set_broadcast(name=args.name,
password=getpass.getpass('Neues Auracast-Passwort: ') if args.password else None,
public=None if args.public is None else args.public == 'on',
encrypted=None if args.encryption is None else args.encryption == 'on',
quality={'16k': 0, '24k': 1, 'high': 2}.get(args.quality))
elif args.command == 'reset':
controller.factory_reset(confirmed=args.confirm)
print('Zurücksetzen angefordert. Kopfhörer danach neu koppeln.')
return 0
print(json.dumps(controller.snapshot().public_dict(), indent=2, ensure_ascii=False))
return 0
except (OSError, DeviceError, ProtocolError, ValueError) as exc:
print(f'Fehler: {exc}', file=sys.stderr)
return 1
finally:
if transport:
transport.close()
if __name__ == '__main__':
raise SystemExit(main())
+382
View File
@@ -0,0 +1,382 @@
"""Native GTK 4 / libadwaita controls and a persistent notification-area menu."""
import sys
import gi
gi.require_version('Gtk', '4.0')
gi.require_version('Adw', '1')
from gi.repository import Adw, Gio, GLib, Gtk
from .integration import APP_ID, autostart_path, set_autostart
from .protocol import CODECS, MODES, STATES
from .worker import Worker
class Application(Adw.Application):
def __init__(self, *, background=False, demo=False, device=None, smoke_seconds=0):
super().__init__(application_id=APP_ID + ('.Demo' if demo else ''), flags=Gio.ApplicationFlags.DEFAULT_FLAGS)
self.background = background
self.demo = demo
self.device = device
self.smoke_seconds = smoke_seconds
self.status = None
self.window = None
self.tray = None
self.worker = None
self.busy = False
self.updating = False
self.dirty = False
self.form_submitted = False
self.last_error = ''
self.codec_values = []
self.transport_values = []
self.received_status = False
self.smoke_ok = False
def do_startup(self):
Adw.Application.do_startup(self)
self.hold()
self._build_window()
try:
from .tray import Tray
self.tray = Tray(self)
except (ImportError, ValueError, GLib.Error) as exc:
self.last_error = f'Infobereich nicht verfügbar: {exc}'
self.worker = Worker(self._on_status, demo=self.demo, device_path=self.device)
self.worker.start()
self.worker.commands.put(None)
if self.smoke_seconds:
GLib.timeout_add_seconds(self.smoke_seconds, self._finish_smoke)
def do_activate(self):
if not self.background or not self.window:
self.show_window()
self.background = False # A second launch opens the existing window.
GLib.timeout_add_seconds(4, self._ensure_accessible)
def do_shutdown(self):
if self.worker:
self.worker.stop()
self.worker.join(timeout=12)
if self.tray:
self.tray.close()
Adw.Application.do_shutdown(self)
def _ensure_accessible(self):
if not self.tray or not self.tray.available:
self.show_window()
return False
def _finish_smoke(self):
self.smoke_ok = self.received_status and bool(self.tray and self.tray.available)
print(f'GUI_SMOKE status={self.received_status} tray={bool(self.tray and self.tray.available)}', flush=True)
self.quit()
return False
def _build_window(self):
self.window = Adw.ApplicationWindow(application=self, title='BTD 700 Control',
default_width=620, default_height=800)
self.window.set_size_request(420, 420)
self.window.connect('close-request', self._close_window)
toolbar = Adw.ToolbarView()
header = Adw.HeaderBar()
header.set_title_widget(Adw.WindowTitle(title='BTD 700', subtitle='Linux Control' + (' · DEMO' if self.demo else '')))
refresh = Gtk.Button(icon_name='view-refresh-symbolic', tooltip_text='Status aktualisieren')
refresh.connect('clicked', lambda *_: self.worker.commands.put(None))
header.pack_start(refresh)
self.refresh_button = refresh
quit_button = Gtk.Button(icon_name='application-exit-symbolic', tooltip_text='App vollständig beenden')
quit_button.connect('clicked', lambda *_: self.quit())
header.pack_end(quit_button)
toolbar.add_top_bar(header)
self.toast = Adw.ToastOverlay()
scroll = Gtk.ScrolledWindow(hscrollbar_policy=Gtk.PolicyType.NEVER)
clamp = Adw.Clamp(maximum_size=600, tightening_threshold=500)
content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=22,
margin_top=24, margin_bottom=28, margin_start=20, margin_end=20)
clamp.set_child(content)
scroll.set_child(clamp)
self.toast.set_child(scroll)
toolbar.set_content(self.toast)
self.window.set_content(toolbar)
self.scroll = scroll
heading = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
icon = Gtk.Image(icon_name='audio-headphones-symbolic', pixel_size=48)
heading.append(icon)
self.codec_label = Gtk.Label(label='BTD 700 verbinden')
self.codec_label.add_css_class('title-1')
heading.append(self.codec_label)
self.state_label = Gtk.Label(label='Suche nach deinem USB-Dongle …', wrap=True)
self.state_label.add_css_class('dim-label')
heading.append(self.state_label)
self.quality_label = Gtk.Label(label='')
self.quality_label.add_css_class('caption')
heading.append(self.quality_label)
content.append(heading)
self.error_label = Gtk.Label(wrap=True, xalign=0)
self.error_label.add_css_class('error')
self.error_label.set_visible(False)
content.append(self.error_label)
self.device_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=22, sensitive=False)
content.append(self.device_box)
mode_group = Adw.PreferencesGroup(title='Audiomodus', description='Der Modus wird direkt am Dongle umgeschaltet.')
mode_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0, homogeneous=True)
mode_box.add_css_class('linked')
self.mode_buttons = []
for mode, label in MODES.items():
button = Gtk.ToggleButton(label=label)
if self.mode_buttons:
button.set_group(self.mode_buttons[0])
button.connect('clicked', lambda b, v=mode: self._mode_clicked(b, v))
mode_box.append(button)
self.mode_buttons.append(button)
mode_group.add(mode_box)
self.device_box.append(mode_group)
audio = Adw.PreferencesGroup(title='Kopfhörer')
self.codec_row = Adw.ComboRow(title='Codec')
self.codec_row.connect('notify::selected', self._codec_selected)
audio.add(self.codec_row)
self.transport_row = Adw.ComboRow(title='Bluetooth-Transport')
self.transport_row.connect('notify::selected', self._transport_selected)
audio.add(self.transport_row)
self.connection_row = Adw.ActionRow(title='Verbindung')
self.connection_button = Gtk.Button(label='Verbinden', valign=Gtk.Align.CENTER)
self.connection_button.connect('clicked', lambda *_: self.perform('set_connection', self.status.state < 2) if self.status else None)
self.connection_row.add_suffix(self.connection_button)
audio.add(self.connection_row)
self.device_box.append(audio)
self.auracast_group = Adw.PreferencesGroup(title='Auracast',
description='Der Audiomodus „Auracast“ startet die Übertragung an mehrere Empfänger.')
self.name_entry = Adw.EntryRow(title='Name der Übertragung')
self.name_entry.connect('changed', self._form_changed)
self.auracast_group.add(self.name_entry)
self.public_row = Adw.SwitchRow(title='Öffentlich auffindbar',
subtitle='Übertragung in der Auracast-Suche anzeigen')
self.public_row.connect('notify::active', self._form_changed)
self.auracast_group.add(self.public_row)
self.encryption_row = Adw.SwitchRow(title='Passwortschutz')
self.encryption_row.connect('notify::active', self._form_changed)
self.auracast_group.add(self.encryption_row)
self.password_entry = Adw.PasswordEntryRow(title='Neues Passwort')
self.password_entry.set_tooltip_text('416 Buchstaben oder Ziffern. Leer lassen behält das bisherige Passwort.')
self.password_entry.connect('changed', self._password_changed)
self.auracast_group.add(self.password_entry)
self.broadcast_quality = Adw.ComboRow(title='Übertragungsqualität',
model=Gtk.StringList.new(['Standard · 16 kHz', 'Standard · 24 kHz', 'Hohe Qualität']))
self.broadcast_quality.connect('notify::selected', self._form_changed)
self.auracast_group.add(self.broadcast_quality)
actions = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8, halign=Gtk.Align.END, margin_top=12)
self.discard_button = Gtk.Button(label='Verwerfen', sensitive=False)
self.discard_button.connect('clicked', self._discard)
self.save_button = Gtk.Button(label='Speichern', sensitive=False)
self.save_button.add_css_class('suggested-action')
self.save_button.connect('clicked', self._save_broadcast)
actions.append(self.discard_button)
actions.append(self.save_button)
self.auracast_group.add(actions)
hint = Gtk.Label(label='Passwort leer lassen, um das bisherige zu behalten.',
wrap=True, xalign=0, margin_top=8)
hint.add_css_class('caption')
hint.add_css_class('dim-label')
self.auracast_group.add(hint)
self.device_box.append(self.auracast_group)
integration = Adw.PreferencesGroup(title='App')
self.tray_row = Adw.ActionRow(title='Im Infobereich weiterlaufen', subtitle='Infobereich wird verbunden …')
integration.add(self.tray_row)
autostart = Adw.SwitchRow(title='Beim Anmelden starten', subtitle='Startet unauffällig im Infobereich',
active=autostart_path().exists())
autostart.set_sensitive(not self.demo)
autostart.connect('notify::active', self._autostart_changed)
integration.add(autostart)
content.append(integration)
self.firmware_label = Gtk.Label(label='Unabhängige Steuerungs-App · Keine Firmware-Updates', wrap=True)
self.firmware_label.add_css_class('caption')
self.firmware_label.add_css_class('dim-label')
content.append(self.firmware_label)
reset = Gtk.Button(label='Dongle auf Werkseinstellungen zurücksetzen …', halign=Gtk.Align.CENTER)
reset.add_css_class('flat')
reset.connect('clicked', lambda *_: self.confirm_reset())
self.reset_button = reset
content.append(reset)
def show_window(self, section=None):
self.window.present()
if section == 'auracast':
self.name_entry.grab_focus()
elif section == 'password':
self.password_entry.grab_focus()
def _close_window(self, *_):
if self.tray and self.tray.available:
self.window.set_focus(None)
self.window.set_visible(False)
else:
self.quit()
return True
def tray_changed(self):
available = bool(self.tray and self.tray.available)
self.tray_row.set_subtitle('Beim Schließen bleibt das Symbol in der oberen Leiste.' if available else
'Kein Infobereich verfügbar; Schließen beendet die App.')
def perform(self, method, *args, **kwargs):
if self.busy or not self.status:
return
self.refresh_button.grab_focus()
self.busy = True
self.last_error = ''
self.worker.submit(method, args, kwargs)
self._render()
def _on_status(self, status, error, completed):
self.status = status
self.received_status |= status is not None
if completed:
self.busy = False
self.last_error = error
if self.form_submitted:
if not error:
self.dirty = False
self.updating = True
self.password_entry.set_text('')
self.updating = False
self.form_submitted = False
self.toast.add_toast(Adw.Toast.new(error or 'Vom Dongle bestätigt'))
elif not status:
self.last_error = error
elif self.last_error.startswith(('Kein BTD 700', 'USB-Zugriff fehlt', '[Errno')):
self.last_error = ''
self._render()
return False
def _render(self):
self.updating = True
s = self.status
self.device_box.set_sensitive(s is not None and not self.busy)
self.reset_button.set_sensitive(s is not None and not self.busy)
self.error_label.set_label(self.last_error)
self.error_label.set_visible(bool(self.last_error))
if s:
self.codec_label.set_label('Auracast' if s.mode == 2 else s.codec_name)
self.state_label.set_label('Einstellung wird übernommen …' if self.busy else STATES.get(s.state, f'Status {s.state}'))
self.quality_label.set_label(s.quality)
for mode, button in enumerate(self.mode_buttons):
button.set_active(mode == s.mode)
button.set_sensitive(mode != 1 or s.gaming_allowed)
values = [v for v in CODECS if v & s.codecs]
if values != self.codec_values:
self.codec_values = values
self.codec_row.set_model(Gtk.StringList.new([CODECS[v] for v in values]))
self.codec_row.set_selected(next((i for i, v in enumerate(values) if v & s.codec), Gtk.INVALID_LIST_POSITION))
self.codec_row.set_sensitive(s.mode == 0 and s.state >= 2 and len(values) > 1)
self.codec_row.set_subtitle('Auswahl im Standard-Modus' if s.mode != 0 else 'Vom Dongle angebotene Codecs')
values = [3] + [v for v in (1, 2) if s.transports & v]
if s.transport in (1, 2) and s.transport not in values:
values.append(s.transport)
if values != self.transport_values:
self.transport_values = values
names = {3: 'Automatisch', 1: 'Bluetooth Classic', 2: 'LE Audio'}
self.transport_row.set_model(Gtk.StringList.new([names[v] for v in values]))
self.transport_row.set_selected(values.index(s.transport) if s.transport in values else Gtk.INVALID_LIST_POSITION)
self.transport_row.set_sensitive(s.mode == 0)
self.connection_row.set_subtitle(STATES.get(s.state, 'Unbekannt'))
self.connection_button.set_label('Trennen' if s.state >= 2 else 'Verbinden')
self.connection_button.set_sensitive(s.mode != 2)
if not self.dirty:
self.name_entry.set_text(s.broadcast_name)
self.public_row.set_active(bool(s.broadcast_public))
self.encryption_row.set_active(bool(s.broadcast_encrypted))
self.broadcast_quality.set_selected(s.broadcast_quality)
self.firmware_label.set_label(f'Firmware {s.firmware} · Unabhängige App · Keine Firmware-Updates')
else:
self.codec_label.set_label('BTD 700 verbinden')
self.state_label.set_label('Warte auf den USB-Dongle …')
self.quality_label.set_label('')
self.save_button.set_sensitive(self.dirty and not self.busy)
self.discard_button.set_sensitive(self.dirty and not self.busy)
self.updating = False
if self.tray:
self.tray.update(s, self.busy, self.last_error)
def _mode_clicked(self, button, mode):
if not self.updating and button.get_active() and self.status and mode != self.status.mode:
self.perform('set_mode', mode)
def _codec_selected(self, row, _):
i = row.get_selected()
if not self.updating and i < len(self.codec_values):
self.perform('set_codec', self.codec_values[i])
def _transport_selected(self, row, _):
i = row.get_selected()
if not self.updating and self.status and i < len(self.transport_values):
self.perform('set_mode', self.status.mode, transport=self.transport_values[i])
def _form_changed(self, *_):
if not self.updating:
self.dirty = True
self.save_button.set_sensitive(not self.busy)
self.discard_button.set_sensitive(not self.busy)
def _password_changed(self, *_):
if not self.updating and self.password_entry.get_text():
self.encryption_row.set_active(True)
self._form_changed()
def _discard(self, *_):
self.dirty = False
self.updating = True
self.password_entry.set_text('')
self.updating = False
self._render()
def _save_broadcast(self, *_):
if not self.status:
return
name = self.name_entry.get_text()
self.form_submitted = True
self.perform('set_broadcast',
name=name if name != self.status.broadcast_name else None,
password=self.password_entry.get_text() or None,
public=self.public_row.get_active(), quality=self.broadcast_quality.get_selected(),
encrypted=self.encryption_row.get_active())
def toggle_encryption(self):
if self.status and self.status.broadcast_encrypted:
self.perform('set_broadcast', encrypted=False)
else:
self.show_window('password')
self.encryption_row.set_active(True)
def confirm_reset(self):
self.show_window()
if not self.status or self.busy:
return
dialog = Adw.AlertDialog(heading='BTD 700 zurücksetzen?',
body='Alle Kopplungen und gespeicherten Einstellungen werden gelöscht. Danach musst du die Kopfhörer erneut koppeln.')
dialog.add_response('cancel', 'Abbrechen')
dialog.add_response('reset', 'Zurücksetzen')
dialog.set_response_appearance('reset', Adw.ResponseAppearance.DESTRUCTIVE)
dialog.set_default_response('cancel')
dialog.set_close_response('cancel')
dialog.connect('response', lambda _, response: self.perform('factory_reset', confirmed=True) if response == 'reset' else None)
dialog.present(self.window)
def _autostart_changed(self, row, _):
try:
set_autostart(row.get_active())
except OSError as exc:
self.last_error = f'Autostart konnte nicht gespeichert werden: {exc}'
self._render()
def run(**kwargs):
app = Application(**kwargs)
result = app.run([sys.argv[0]])
if kwargs.get('smoke_seconds') and not app.smoke_ok:
return 1
return result
+177
View File
@@ -0,0 +1,177 @@
"""Device operations with readback and capability checks."""
from dataclasses import asdict, dataclass
import time
from .protocol import CODECS, MODES, STATES, TRANSPORTS, Command as C, validate_text
from .transport import DeviceError
@dataclass
class Status:
state: int
mode: int
transport: int
connected_transport: int
codecs: int
codec: int
resolution: int
frequency: int
le_state: int
transports: int
broadcast_public: int
broadcast_quality: int
broadcast_encrypted: int
broadcast_name: str
firmware: str
gaming: int = -1
@property
def codec_name(self):
return ' + '.join(name for bit, name in CODECS.items() if self.codec & bit) or ''
@property
def quality(self):
bits = {1: '16 Bit', 2: '24 Bit'}.get(self.resolution, '')
frequency = {1: '44,1 kHz', 2: '48 kHz', 3: '96 kHz'}.get(self.frequency, '')
return f'{bits} / {frequency}'
@property
def gaming_allowed(self):
if self.gaming >= 0:
return self.gaming == 1
return self.mode == 1 or (self.state >= 2 and (self.connected_transport in (2, 3) or bool(self.codecs & 4)))
def public_dict(self):
result = asdict(self)
result.update(state_name=STATES.get(self.state, f'Unbekannt ({self.state})'),
mode_name=MODES.get(self.mode, f'Unbekannt ({self.mode})'),
codec_name=self.codec_name, audio_quality=self.quality,
transport_name=TRANSPORTS.get(self.transport, f'Unbekannt ({self.transport})'))
return result
class Controller:
def __init__(self, transport):
self.transport = transport
self.firmware = None
def snapshot(self) -> Status:
q = self.transport.request
state = q(C.GET_STATE)[0]
mode = q(C.GET_MODE)
quality = q(C.GET_QUALITY)
codecs = q(C.GET_CODECS)[0]
codec = q(C.GET_CODEC)[0]
le = q(C.GET_LE_STATE)[0]
transports = q(C.GET_TRANSPORTS)[0] if state >= 2 else 0
broadcast = q(C.GET_BROADCAST)
name = q(C.GET_NAME).split(b'\0', 1)[0].decode('utf-8', errors='replace')
if self.firmware is None:
try:
self.firmware = '.'.join(map(str, q(C.GET_VERSION)[:3]))
except TimeoutError:
self.firmware = 'Unbekannt'
# Firmware 3.11 does not respond to GET_GAMING; use events/fallback,
# matching the original application's capability logic.
event = self.transport.events.get(23, b'')
return Status(state, mode[0], mode[1], mode[2] if len(mode) > 2 else mode[1],
codecs, codec, *quality[:2], le, transports, *broadcast[:3],
name, self.firmware, event[0] if event else -1)
def _verify(self, getter, expected, *, timeout=5.0, exact=True, null_padded=False):
deadline = time.monotonic() + timeout
while True:
result = self.transport.request(getter)
comparison = result.rstrip(b'\0') if null_padded else result
matches = comparison == expected if exact else comparison.startswith(expected)
if matches:
return
if time.monotonic() >= deadline:
raise DeviceError('Der Dongle hat die angeforderte Einstellung nicht bestätigt. Bitte den aktuellen Status prüfen.')
time.sleep(0.15)
def set_mode(self, mode: int, transport: int | None = None):
status = self.snapshot()
if mode not in MODES:
raise ValueError('Unbekannter Audiomodus.')
if mode == 1 and not status.gaming_allowed:
raise ValueError('Gaming ist für die aktuelle Verbindung nicht verfügbar.')
selected = status.transport if transport is None else transport
if selected not in (1, 2, 3):
raise ValueError('Unbekannter Bluetooth-Transport.')
if transport is not None and selected in (1, 2) and status.transports and not status.transports & selected:
raise ValueError('Die Kopfhörer unterstützen diesen Transport nicht.')
payload = bytes((mode, selected))
self.transport.request(C.SET_MODE, payload)
self._verify(C.GET_MODE, payload, exact=False)
def set_codec(self, codec: int):
status = self.snapshot()
if status.mode != 0 or status.state < 2:
raise ValueError('Codec-Auswahl benötigt verbundene Kopfhörer im Standard-Modus.')
if codec not in CODECS or not status.codecs & codec:
raise ValueError('Dieser Codec wird für die aktuelle Verbindung nicht angeboten.')
payload = bytes((codec,))
self.transport.request(C.SET_CODEC, payload)
self._verify(C.GET_CODEC, payload)
def set_connection(self, connected: bool):
self.transport.request(C.SET_CONNECTION, bytes((int(connected),)))
deadline = time.monotonic() + 8
while True:
state = self.transport.request(C.GET_STATE)[0]
if (state >= 2) == connected:
return
if time.monotonic() >= deadline:
raise DeviceError('Verbindung wurde angefordert, aber noch nicht bestätigt. Sind die gekoppelten Kopfhörer eingeschaltet?')
time.sleep(0.25)
def set_broadcast(self, *, name=None, password=None, public=None, quality=None, encrypted=None):
# Validate the entire change before issuing any setter.
name_data = validate_text(name) if name is not None else None
key_data = validate_text(password, key=True) if password is not None else None
if any(value is not None and not isinstance(value, bool) for value in (public, encrypted)):
raise ValueError('Auracast-Schalter benötigen einen booleschen Wert.')
if quality is not None and quality not in (0, 1, 2):
raise ValueError('Unbekannte Auracast-Qualität.')
before = self.snapshot()
info = [before.broadcast_public, before.broadcast_quality, before.broadcast_encrypted]
if public is not None:
info[0] = int(public)
if quality is not None:
info[1] = quality
if password is not None:
info[2] = int(bool(password))
if encrypted is not None:
info[2] = int(encrypted)
if info[2] and key_data == b'':
raise ValueError('Verschlüsselung benötigt ein Passwort mit 416 Zeichen.')
if info[2] and key_data is None:
existing = self.transport.request(C.GET_KEY).rstrip(b'\0')
if len(existing) < 4:
raise ValueError('Bitte zuerst ein Passwort mit 416 Zeichen eingeben.')
# Preserve the original settings while changing a key, then write
# key, name, and final settings in the order used by the original app.
try:
if key_data is not None and before.broadcast_encrypted:
self.transport.request(C.SET_BROADCAST, bytes((before.broadcast_public, before.broadcast_quality, before.broadcast_encrypted)))
if key_data is not None:
self.transport.request(C.SET_KEY, key_data)
self._verify(C.GET_KEY, key_data, null_padded=True)
if name_data is not None:
self.transport.request(C.SET_NAME, name_data)
if name_data:
self._verify(C.GET_NAME, name_data, null_padded=True)
else:
self.transport.request(C.GET_NAME)
self.transport.request(C.SET_BROADCAST, bytes(info))
self._verify(C.GET_BROADCAST, bytes(info))
except Exception as exc:
# Do not silently repeat writes or claim atomicity: settings persist
# independently on the dongle.
raise DeviceError('Auracast wurde nur teilweise oder nicht übernommen. Aktuellen Namen und Passwortschutz prüfen.') from exc
def factory_reset(self, *, confirmed=False):
if not confirmed:
raise ValueError('Zurücksetzen muss ausdrücklich bestätigt werden; alle Kopplungen werden gelöscht.')
self.transport.request(C.FACTORY_RESET)
+37
View File
@@ -0,0 +1,37 @@
"""Explicit, isolated demonstration device; never opens USB."""
from .protocol import Command as C, encode
class DemoTransport:
def __init__(self):
self.events = {}
self.values = {C.GET_STATE: b'\x03', C.GET_MODE: b'\0\3\1',
C.GET_CODECS: b'\x0f', C.GET_CODEC: b'\4', C.GET_LE_STATE: b'\1',
C.GET_QUALITY: b'\2\2', C.GET_TRANSPORTS: b'\1',
C.GET_BROADCAST: b'\1\2\0', C.GET_NAME: b'BTD700 Demo'+bytes(21),
C.GET_KEY: bytes(16), C.GET_VERSION: b'\3\x0b\0'}
def request(self, command, payload=b''):
encode(command, payload)
if command in self.values:
return self.values[command]
if command == C.SET_MODE:
self.values[C.GET_MODE] = payload + b'\1'
elif command == C.SET_CODEC:
self.values[C.GET_CODEC] = payload
elif command == C.SET_CONNECTION:
self.values[C.GET_STATE] = b'\2' if payload[0] else b'\1'
elif command == C.SET_BROADCAST:
self.values[C.GET_BROADCAST] = payload
elif command == C.SET_NAME:
self.values[C.GET_NAME] = (payload or b'BTD700 Demo').ljust(32, b'\0')
elif command == C.SET_KEY:
self.values[C.GET_KEY] = payload.ljust(16, b'\0')
elif command == C.FACTORY_RESET:
self.__init__()
else:
raise TimeoutError('Nicht unterstützte Demo-Abfrage')
return b''
def close(self):
pass
+37
View File
@@ -0,0 +1,37 @@
"""User-scoped desktop integration."""
from pathlib import Path
import os
import sys
ROOT = Path(__file__).resolve().parent.parent
APP_ID = 'io.github.btd700linux.Control'
def desktop_entry(*, background=False):
checkout = (ROOT / 'run.sh').is_file()
path = str(ROOT / 'run.sh') if checkout else sys.executable
module_args = '' if checkout else ' -m btd700'
icon = str(ROOT / 'packaging/btd700-control.svg') if checkout else 'audio-headphones'
# Desktop Entry Exec quoting is not shell quoting.
escaped = path.replace('\\', '\\\\').replace('"', '\\"').replace('`', '\\`').replace('$', '\\$').replace('%', '%%')
return ('[Desktop Entry]\nType=Application\nName=BTD 700 Control\n'
'Comment=Linux-Steuerung für den Sennheiser BTD 700\n'
f'Exec="{escaped}"{module_args}' + (' --background' if background else '') + '\n'
f'Icon={icon}\nTerminal=false\n'
'Categories=AudioVideo;Audio;\n'
'Keywords=Sennheiser;Bluetooth;aptX;Auracast;Dongle;\n'
f'StartupWMClass={APP_ID}\n')
def autostart_path():
base = Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config'))
return base / 'autostart' / f'{APP_ID}.desktop'
def set_autostart(enabled):
path = autostart_path()
if enabled:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(desktop_entry(background=True))
else:
path.unlink(missing_ok=True)
+104
View File
@@ -0,0 +1,104 @@
"""BTD 700 control protocol. No updater transport or arbitrary commands."""
from dataclasses import dataclass
from enum import IntEnum
import re
REPORT_ID = 0x34
REPORT_SIZE = 64
class ProtocolError(Exception):
pass
class Command(IntEnum):
GET_MODE = 0x01
SET_MODE = 0x02
GET_CODECS = 0x03
SET_CODEC = 0x04
GET_CODEC = 0x05
GET_STATE = 0x06
GET_LE_STATE = 0x07
GET_QUALITY = 0x08
GET_BROADCAST = 0x09
SET_BROADCAST = 0x0A
GET_KEY = 0x0B
SET_KEY = 0x0C
GET_NAME = 0x0D
SET_NAME = 0x0E
GET_VERSION = 0x12
FACTORY_RESET = 0x13
SET_CONNECTION = 0x14
GET_TRANSPORTS = 0x15
GET_GAMING = 0x17
READ_COMMANDS = frozenset(c for c in Command if c.name.startswith('GET_'))
EVENTS = frozenset((2, 3, 4, 15, 16, 17, 22, 23))
CODECS = {1: 'SBC', 2: 'aptX Classic', 4: 'aptX Adaptive', 8: 'aptX Lossless',
16: 'aptX Lite (QMAP)', 32: 'LC3'}
MODES = {0: 'Standard', 1: 'Gaming', 2: 'Auracast'}
TRANSPORTS = {0: 'Nicht verbunden', 1: 'Bluetooth Classic', 2: 'LE Audio', 3: 'Automatisch'}
STATES = {0: 'Bereit', 1: 'Kopfhörer getrennt', 2: 'Kopfhörer verbunden',
3: 'Musikwiedergabe', 4: 'Sprachanruf'}
MIN_LENGTHS = {Command.GET_MODE: 2, Command.GET_CODECS: 1, Command.GET_CODEC: 1,
Command.GET_STATE: 1, Command.GET_LE_STATE: 1, Command.GET_QUALITY: 2,
Command.GET_BROADCAST: 3, Command.GET_VERSION: 3,
Command.GET_TRANSPORTS: 1, Command.GET_GAMING: 1}
def validate_text(value: str, *, key: bool = False) -> bytes:
"""Match the original app's 416 ASCII character input bounds; empty resets."""
pattern = r'[A-Za-z0-9]{4,16}' if key else r'[A-Za-z0-9 ]{4,16}'
if value and (not re.fullmatch(pattern, value) or value != value.strip()):
what = 'Passwort: 416 Buchstaben oder Ziffern.' if key else (
'Name: 416 Buchstaben, Ziffern oder innere Leerzeichen.')
raise ValueError(what)
return value.encode('ascii')
def encode(command: Command, payload: bytes = b'') -> bytes:
try:
command = Command(command)
except ValueError as exc:
raise ValueError('Unbekannter Steuerbefehl; kein Zugriff auf Update-Befehle.') from exc
if command in READ_COMMANDS or command == Command.FACTORY_RESET:
valid = not payload
elif command == Command.SET_MODE:
valid = len(payload) == 2 and payload[0] in MODES and payload[1] in (1, 2, 3)
elif command == Command.SET_CODEC:
valid = len(payload) == 1 and payload[0] in CODECS
elif command == Command.SET_CONNECTION:
valid = payload in (b'\0', b'\1')
elif command == Command.SET_BROADCAST:
valid = len(payload) == 3 and payload[0] in (0, 1) and payload[1] in (0, 1, 2) and payload[2] in (0, 1)
elif command in (Command.SET_NAME, Command.SET_KEY):
# Existing device names can contain an underscore. New user input is
# validated more narrowly by validate_text, preserving originals on undo.
valid = len(payload) <= 16 and all(32 <= b <= 126 for b in payload)
else:
valid = False
if not valid:
raise ValueError(f'Ungültige Parameter für {command.name}.')
return bytes((REPORT_ID, 0xFE, command, len(payload))) + payload + bytes(60 - len(payload))
def acknowledge(event: int) -> bytes:
if event not in EVENTS:
raise ProtocolError(f'Unbekannte Benachrichtigung: {event:#x}')
return bytes((REPORT_ID, 0xFD, event, 0)) + bytes(60)
@dataclass(frozen=True)
class Message:
kind: int
command: int
payload: bytes
def decode(report: bytes) -> Message | None:
if not report or report[0] != REPORT_ID:
return None # Media keys share the same interface.
if len(report) < 4 or report[1] not in (0xFC, 0xFF) or report[3] > 60 or len(report) < 4 + report[3]:
raise ProtocolError('Ungültige Antwort des Dongles.')
return Message(report[1], report[2], report[4:4 + report[3]])
+115
View File
@@ -0,0 +1,115 @@
"""Linux hidraw access, limited to the BTD 700 control interface."""
from dataclasses import dataclass
from pathlib import Path
import fcntl
import os
import select
import struct
import time
from .protocol import Command, EVENTS, MIN_LENGTHS, ProtocolError, acknowledge, decode, encode
class DeviceError(Exception):
pass
@dataclass(frozen=True)
class Device:
path: str
serial: str
name: str = 'Sennheiser BTD 700'
def discover(sysfs: Path = Path('/sys/class/hidraw')) -> list[Device]:
devices = []
for node in sorted(sysfs.glob('hidraw*')):
try:
info = dict(line.split('=', 1) for line in (node / 'device/uevent').read_text().splitlines() if '=' in line)
if info.get('HID_ID', '').upper() != '0003:00003542:00003001':
continue
descriptor = (node / 'device/report_descriptor').read_bytes()
if b'\x06\xa2\xff' not in descriptor or b'\x85\x34' not in descriptor:
continue
devices.append(Device('/dev/' + node.name, info.get('HID_UNIQ', '')))
except (OSError, ValueError):
continue
return devices
class Hidraw:
def __init__(self, device: Device, timeout: float = 1.0):
self.device = device
self.timeout = timeout
self.fd = None
self.events = {}
try:
self.fd = os.open(device.path, os.O_RDWR | os.O_NONBLOCK | os.O_CLOEXEC | os.O_NOFOLLOW)
info = bytearray(8)
fcntl.ioctl(self.fd, 0x80084803, info, True) # HIDIOCGRAWINFO
bus, vendor, product = struct.unpack('IHH', info)
if (bus, vendor, product) != (3, 0x3542, 0x3001):
raise DeviceError('Gerätepfad gehört nicht mehr zum BTD 700.')
try:
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError as exc:
raise DeviceError('Der Dongle ist bereits in einer anderen Instanz geöffnet.') from exc
except PermissionError as exc:
self.close()
raise DeviceError('USB-Zugriff fehlt. Die mitgelieferte udev-Regel installieren und den Dongle neu einstecken.') from exc
except BaseException:
self.close()
raise
def close(self):
if self.fd is not None:
os.close(self.fd)
self.fd = None
def __enter__(self):
return self
def __exit__(self, *_):
self.close()
def _write(self, report):
if self.fd is None:
raise DeviceError('Dongle ist nicht geöffnet.')
if not select.select([], [self.fd], [], self.timeout)[1]:
raise TimeoutError('USB-Schreibzugriff hat nicht geantwortet.')
if os.write(self.fd, report) != len(report):
raise DeviceError('Unvollständiger USB-Schreibvorgang.')
def _receive(self):
report = os.read(self.fd, 1024)
if not report:
raise DeviceError('Dongle wurde entfernt.')
message = decode(report)
if message and message.kind == 0xFC and message.command in EVENTS:
self.events[message.command] = message.payload
self._write(acknowledge(message.command))
return message
def request(self, command: Command, payload: bytes = b'') -> bytes:
report = encode(command, payload) # Validate before any I/O.
try:
# Discard stale replies, acknowledging events. The cap prevents an
# unrelated consumer-control input flood from blocking indefinitely.
for _ in range(128):
if not select.select([self.fd], [], [], 0)[0]:
break
self._receive()
self._write(report)
deadline = time.monotonic() + self.timeout
while (remaining := deadline - time.monotonic()) > 0:
if not select.select([self.fd], [], [], remaining)[0]:
break
message = self._receive()
if message and message.kind == 0xFF and message.command == command:
if len(message.payload) < MIN_LENGTHS.get(command, 0):
raise ProtocolError(f'Unvollständige Antwort auf {command.name}.')
return message.payload
raise TimeoutError(f'Der Dongle antwortet nicht auf {command.name}.')
except (OSError, ProtocolError):
self.close()
raise
+176
View File
@@ -0,0 +1,176 @@
"""StatusNotifierItem + libdbusmenu, compatible with GNOME AppIndicator and KDE."""
import gi
from gi.repository import Gio, GLib
from .protocol import CODECS, MODES, STATES
gi.require_version('Dbusmenu', '0.4')
from gi.repository import Dbusmenu
INTERFACE = 'org.kde.StatusNotifierItem'
XML = '''<node><interface name="org.kde.StatusNotifierItem">
<property name="Category" type="s" access="read"/>
<property name="Id" type="s" access="read"/>
<property name="Title" type="s" access="read"/>
<property name="Status" type="s" access="read"/>
<property name="WindowId" type="u" access="read"/>
<property name="IconName" type="s" access="read"/>
<property name="IconThemePath" type="s" access="read"/>
<property name="IconPixmap" type="a(iiay)" access="read"/>
<property name="OverlayIconName" type="s" access="read"/>
<property name="OverlayIconPixmap" type="a(iiay)" access="read"/>
<property name="AttentionIconName" type="s" access="read"/>
<property name="AttentionIconPixmap" type="a(iiay)" access="read"/>
<property name="AttentionMovieName" type="s" access="read"/>
<property name="ToolTip" type="(sa(iiay)ss)" access="read"/>
<property name="ItemIsMenu" type="b" access="read"/>
<property name="Menu" type="o" access="read"/>
<method name="Activate"><arg type="i" direction="in"/><arg type="i" direction="in"/></method>
<method name="SecondaryActivate"><arg type="i" direction="in"/><arg type="i" direction="in"/></method>
<method name="ContextMenu"><arg type="i" direction="in"/><arg type="i" direction="in"/></method>
<method name="Scroll"><arg type="i" direction="in"/><arg type="s" direction="in"/></method>
<signal name="NewTitle"/><signal name="NewIcon"/><signal name="NewToolTip"/>
<signal name="NewStatus"><arg type="s"/></signal>
</interface></node>'''
class Tray:
def __init__(self, app):
self.app = app
self.status = None
self.available = False
self._signature = None
self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
self.registration = self.bus.register_object('/StatusNotifierItem',
Gio.DBusNodeInfo.new_for_xml(XML).interfaces[0], self._method, self._property, None)
self.server = Dbusmenu.Server.new('/Menu')
self.watch = Gio.bus_watch_name_on_connection(self.bus, 'org.kde.StatusNotifierWatcher',
Gio.BusNameWatcherFlags.NONE, self._watcher_appeared, self._watcher_lost)
self.update(None, False)
def _watcher_appeared(self, connection, _name, _owner):
def done(bus, result):
try:
bus.call_finish(result)
self.available = True
except GLib.Error:
self.available = False
self.app.tray_changed()
connection.call('org.kde.StatusNotifierWatcher', '/StatusNotifierWatcher',
'org.kde.StatusNotifierWatcher', 'RegisterStatusNotifierItem',
GLib.Variant('(s)', (connection.get_unique_name(),)), None,
Gio.DBusCallFlags.NONE, 3000, None, done)
def _watcher_lost(self, *_):
self.available = False
self.app.tray_changed()
def _property(self, _bus, _sender, _path, _iface, name):
tooltip = 'BTD 700 anschließen'
if self.status:
tooltip = f'{MODES.get(self.status.mode, "")} · {self.status.codec_name} · {self.status.quality}'
props = {
'Category': ('s', 'Hardware'), 'Id': ('s', 'btd700-control-demo' if self.app.demo else 'btd700-control'),
'Title': ('s', 'BTD 700 Control · DEMO' if self.app.demo else 'BTD 700 Control'), 'Status': ('s', 'Active'),
'WindowId': ('u', 0), 'IconName': ('s', 'audio-headphones-symbolic'),
'IconThemePath': ('s', ''), 'IconPixmap': ('a(iiay)', []),
'OverlayIconName': ('s', ''), 'OverlayIconPixmap': ('a(iiay)', []),
'AttentionIconName': ('s', ''), 'AttentionIconPixmap': ('a(iiay)', []),
'AttentionMovieName': ('s', ''),
'ToolTip': ('(sa(iiay)ss)', ('audio-headphones-symbolic', [], 'BTD 700', tooltip)),
'ItemIsMenu': ('b', True), 'Menu': ('o', '/Menu')}
if name in props:
return GLib.Variant(*props[name])
return None
def _method(self, _bus, _sender, _path, _iface, method, _params, invocation):
if method in ('Activate', 'SecondaryActivate', 'ContextMenu'):
self.app.show_window()
invocation.return_value(None)
@staticmethod
def _item(label, action=None, *, enabled=True, checked=None, radio=False):
item = Dbusmenu.Menuitem.new()
item.property_set('label', label)
item.property_set_bool('visible', True)
item.property_set_bool('enabled', enabled)
if checked is not None:
item.property_set('toggle-type', 'radio' if radio else 'checkmark')
item.property_set_int('toggle-state', int(checked))
if action:
item.connect('item-activated', lambda *_: action())
return item
def _submenu(self, root, label):
item = self._item(label)
item.property_set('children-display', 'submenu')
root.child_append(item)
return item
def update(self, status, busy, error=''):
signature = (repr(status), busy, error)
if signature == self._signature:
return
self._signature = signature
self.status = status
root = Dbusmenu.Menuitem.new()
add = root.child_append
add(self._item('BTD 700 Control · DEMO' if self.app.demo else 'BTD 700 Control', self.app.show_window))
if status:
add(self._item(f'{status.codec_name} · {status.quality}', enabled=False))
add(self._item(STATES.get(status.state, 'Unbekannter Status'), enabled=False))
modes = self._submenu(root, 'Audiomodus')
for value, label in MODES.items():
modes.child_append(self._item(label,
lambda v=value: self.app.perform('set_mode', v),
enabled=not busy and (value != 1 or status.gaming_allowed),
checked=status.mode == value, radio=True))
codecs = self._submenu(root, 'Codec')
for value, label in CODECS.items():
if status.codecs & value:
codecs.child_append(self._item(label,
lambda v=value: self.app.perform('set_codec', v),
enabled=not busy and status.mode == 0 and status.state >= 2,
checked=bool(status.codec & value), radio=True))
if not codecs.get_children():
codecs.child_append(self._item('Keine verbundenen Kopfhörer', enabled=False))
transports = self._submenu(root, 'Bluetooth-Transport')
for value, label in ((3, 'Automatisch'), (1, 'Bluetooth Classic'), (2, 'LE Audio')):
transports.child_append(self._item(label,
lambda v=value: self.app.perform('set_mode', status.mode, transport=v),
enabled=not busy and status.mode == 0 and (value == 3 or bool(status.transports & value)),
checked=status.transport == value, radio=True))
auracast = self._submenu(root, 'Auracast')
auracast.child_append(self._item('Name und Passwort …', lambda: self.app.show_window('auracast')))
auracast.child_append(self._item('Öffentlich auffindbar',
lambda: self.app.perform('set_broadcast', public=not status.broadcast_public),
enabled=not busy, checked=bool(status.broadcast_public)))
auracast.child_append(self._item('Passwortschutz',
lambda: self.app.toggle_encryption(), enabled=not busy,
checked=bool(status.broadcast_encrypted)))
quality = self._submenu(auracast, 'Übertragungsqualität')
for value, label in enumerate(('Standard · 16 kHz', 'Standard · 24 kHz', 'Hohe Qualität')):
quality.child_append(self._item(label,
lambda v=value: self.app.perform('set_broadcast', quality=v),
enabled=not busy, checked=status.broadcast_quality == value, radio=True))
add(self._item('Kopfhörer trennen' if status.state >= 2 else 'Kopfhörer verbinden',
lambda: self.app.perform('set_connection', status.state < 2), enabled=not busy and status.mode != 2))
add(self._item('Werkseinstellungen …', self.app.confirm_reset, enabled=not busy))
else:
add(self._item('Dongle nicht verbunden' if not error else 'USB-Zugriff prüfen Fenster öffnen',
self.app.show_window))
if busy:
add(self._item('Einstellung wird übernommen …', enabled=False))
if error:
add(self._item('Fehler anzeigen …', self.app.show_window))
separator = Dbusmenu.Menuitem.new()
separator.property_set('type', 'separator')
add(separator)
add(self._item('Fenster öffnen', self.app.show_window))
add(self._item('Beenden', self.app.quit))
self.root = root # Keep Python signal closures alive.
self.server.set_root(root)
self.bus.emit_signal(None, '/StatusNotifierItem', INTERFACE, 'NewToolTip', None)
def close(self):
Gio.bus_unwatch_name(self.watch)
self.bus.unregister_object(self.registration)
+67
View File
@@ -0,0 +1,67 @@
"""One thread owns the USB handle; the GTK main loop never waits for USB."""
import queue
import threading
from gi.repository import GLib
from .controller import Controller
from .transport import DeviceError, Hidraw, discover
class Worker(threading.Thread):
def __init__(self, callback, *, device_path=None, demo=False):
super().__init__(name='btd700-usb', daemon=True)
self.callback = callback
self.device_path = device_path
self.demo = demo
self.commands = queue.Queue()
self.stopping = threading.Event()
self.controller = None
self.transport = None
def submit(self, method, args, kwargs):
self.commands.put((method, args, kwargs))
def stop(self):
self.stopping.set()
self.commands.put(None)
def run(self):
try:
while not self.stopping.is_set():
try:
action = self.commands.get(timeout=2)
except queue.Empty:
action = None
if self.stopping.is_set():
break
error = ''
try:
if self.controller is None:
if self.demo:
from .demo import DemoTransport
self.transport = DemoTransport()
else:
devices = discover()
if self.device_path:
devices = [d for d in devices if d.path == self.device_path]
if not devices:
raise DeviceError('Kein BTD 700 gefunden. Bitte den Dongle einstecken.')
if len(devices) > 1:
raise DeviceError('Mehrere BTD 700 gefunden. Mit --device /dev/hidrawN auswählen.')
self.transport = Hidraw(devices[0])
self.controller = Controller(self.transport)
if action:
method, args, kwargs = action
try:
getattr(self.controller, method)(*args, **kwargs)
except (ValueError, DeviceError, TimeoutError) as exc:
error = str(exc)
status = self.controller.snapshot()
GLib.idle_add(self.callback, status, error, bool(action))
except Exception as exc:
if self.transport:
self.transport.close()
self.transport = self.controller = None
GLib.idle_add(self.callback, None, error or str(exc), bool(action))
finally:
if self.transport:
self.transport.close()
+142
View File
@@ -0,0 +1,142 @@
# Rekonstruiertes BTD-700-Steuerprotokoll
Stand der Untersuchung: 5. September 2026. Zweck: unabhängige Linux-Steuerung des
eigenen USB-Dongles. Die Windows-Anwendung wurde statisch untersucht, nicht ausgeführt.
## Herkunft und Reproduzierbarkeit
- [Offizielle Produktseite](https://uk.sennheiser-hearing.com/products/btd-700)
- [Offizieller Dongle-Control-Download](https://uk.sennheiser-hearing.com/pages/sennheiser-dongle-control)
- [Dort verlinktes Windows-ZIP](https://eu-central-1-akqa.graphassets.com/AGz66yvUcQ42Ggm7CrXdgz/cmgrvi8excrci07uu3ivz166x)
- Archiv: `windows-signed-v1.0.5/Sennheiser Dongle Control.exe`, Version 1.0.5.0,
ProductVersion `1.0.5+eac62d73c43c8572999e7e68cbaec6e4536a1318`.
- SHA256 ZIP: `1d1057b7eb64ab08e41d76723c343c691196f0affe1d5fc9168f01f9908a8cc7`
- SHA256 EXE: `e176f1ab7d4aae40308c152b0bd95227ac5a16fedb99efef7e9229345d8014c0`
- SHA256 eingebettete App-Assembly:
`2e8c89ea0333b0a9dd5bb0e851cd685810edaa6e32caee7645511d646ee970b4`
Die EXE ist ein .NET-Single-File-Bundle, Manifestversion 6 mit 451 Einträgen.
Die benötigte Assembly kann mit `tools/extract_control_assembly.py` aus einer
lokalen Original-EXE extrahiert werden. Das Bundleformat wurde mit den
[Manifest](https://github.com/dotnet/runtime/blob/main/src/installer/managed/Microsoft.NET.HostModel/Bundle/Manifest.cs)-
und [FileEntry](https://github.com/dotnet/runtime/blob/main/src/installer/managed/Microsoft.NET.HostModel/Bundle/FileEntry.cs)-
Definitionen des .NET-Runtimes abgeglichen.
Analysewerkzeug: ILSpy CLI 11.0.0.9375. Relevante Typen:
`BTDTool.BTD700Tool`, `_BTD700_HOSTCMD`, `_BTD700_DONGLECMD`, `_BTD700_*`-Enums,
`BTD700Context`, `HidDeviceExt.sendGenericCommand`,
`ViewModels.MainAppWindowViewModel`, `Views.AppBtd700Features`.
Die neue Implementierung enthält nur rekonstruierte Protokollfakten und eigenen
Code. Original-Binaries, dekompilierte Originalquellen und Ressourcen gehören
nicht zum Projekt. Die Herstellerlizenz der Originalsoftware bleibt davon getrennt.
## USB und Framing
VID `0x3542`, PID `0x3001`, USB-HID-Interface 0. Die Steuersammlung verwendet
Vendor Usage Page `0xFFA2`, Report ID **0x34**. Auf dem Testsystem liegt sie unter
`/dev/hidraw6`; die Nummer wird dynamisch ermittelt. Interface 1 ist für diese
Steuer-App nicht erforderlich und wird nicht geöffnet.
64-Byte-Output-Report, ungenutzte Bytes mit Null gefüllt:
| Byte | Bedeutung |
|---|---|
| 0 | Report-ID `34` |
| 1 | `FE` Host-Befehl; `FF` Dongle-Antwort; `FC` Dongle-Ereignis; `FD` Ereignisbestätigung |
| 2 | Befehls- oder Ereignisnummer |
| 3 | Nutzdatenlänge, maximal 60 |
| 4… | Nutzdaten |
Host-Abfrage Beispiel: `34 FE 06 00` + 60 Nullbytes.
Antwort Beispiel: `34 FF 06 01 03` (Audio läuft).
Bestätigung des Ereignisses 15: `34 FD 0F 00` + 60 Nullbytes.
Ereignisse 2/3/4/15/16/17/22/23 werden bestätigt. Andere Reports können von
Medientasten auf derselben Schnittstelle stammen und werden ignoriert. Antworten
werden nach Richtung und Befehlsnummer zugeordnet; Längen werden geprüft.
Nur eine Anfrage ist gleichzeitig aktiv. Schreibbefehle werden nicht blind
wiederholt; für Einstellungen wird anschließend der Wert erneut abgefragt.
Es gibt keine Transaktions-ID und keine garantierte Atomizität mehrerer Einstellungen.
## Befehle
Alle Nummern hexadezimal. Bei Lesezugriffen ist die Anfrage-Nutzlast leer.
Die nachfolgende Nutzlast beschreibt bei `GET` die Antwort, bei `SET` die Anfrage.
| ID | Operation | Nutzlast |
|---|---|---|
| 01 | GET Modus/Transport | Modus, konfigurierter Transport, optional aktuell verbundener Transport |
| 02 | SET Modus/Transport | Modus, Transport |
| 03 | GET verfügbare Codecs | Codec-Bitmaske |
| 04 | SET Codec | Einzelnes Codec-Bit, nicht Bitindex |
| 05 | GET aktiver Codec | Codec-Bitmaske |
| 06 | GET Dongle-Zustand | Zustand |
| 07 | GET LE-Audio-Zustand | LE-Zustand |
| 08 | GET Audioqualität | Auflösung, Frequenz |
| 09 | GET Auracast-Konfiguration | öffentlich, Qualität, Verschlüsselung |
| 0A | SET Auracast-Konfiguration | öffentlich, Qualität, Verschlüsselung |
| 0B | GET Auracast-Schlüssel | Zeichenbytes; nur bei expliziter Passwort-Operation gelesen |
| 0C | SET Auracast-Schlüssel | 016 Bytes; UI beschränkt auf leer oder 416 ASCII-Zeichen |
| 0D | GET Auracast-Name | Zeichenbytes; am Gerät 32 Bytes, mit Null aufgefüllt |
| 0E | SET Auracast-Name | bis 16 Zeichenbytes; leer setzt Gerätenamen zurück |
| 12 | GET Firmwareversion | drei Versionsbytes, nur Anzeige |
| 13 | Werksreset | leer; nur nach Nutzerbestätigung |
| 14 | Bluetooth verbinden/trennen | 1 / 0 |
| 15 | GET Kopfhörer-Transportmöglichkeiten | Bitmaske |
| 17 | GET Gaming-Verfügbarkeit | im Original definiert; Firmware 3.11 antwortet nicht, daher nicht regelmäßig abgefragt |
Modus: 0 Standard, 1 Gaming, 2 Auracast.
Transport: 0 getrennt, 1 BR/EDR, 2 LE Audio, 3 Dual/automatisch. Setzen von 0 wird
in der neuen App nicht angeboten; Trennen hat einen eigenen Befehl.
Codec-Bits: `01` SBC, `02` aptX Classic, `04` aptX Adaptive/Low Latency,
`08` aptX Lossless, `10` aptX Lite/QMAP, `20` LC3. Nur die vom Dongle angebotenen
Bits werden im Menü angezeigt. Die Firmware kann die Liste abhängig vom Modus ändern.
Dongle-Zustand: 0 keiner/bereit, 1 getrennt, 2 verbunden, 3 Audio, 4 Sprache.
LE-Zustand: 0 keiner, 1 getrennt, 2 verbunden, 3 Unicast, 4 Broadcast.
Auflösung: 1 = 16 Bit, 2 = 24 Bit.
Frequenz: 1 = 44,1 kHz, 2 = 48 kHz, 3 = 96 kHz.
Auracast: öffentlich 0/1; Qualität 0 = SQ 16 kHz, 1 = SQ 24 kHz, 2 = HQ;
Verschlüsselung 0/1. „Öffentlich“ bezeichnet die Ankündigung/Auffindbarkeit,
nicht das Aktivieren des Audiomodus.
Gaming-Verfügbarkeit fällt wie in der Original-App auf den Verbindungsstatus,
aptX-Adaptive-Bit und LE-Transport zurück, solange kein Ereignis 23 eingetroffen ist.
## Tatsächlich gelesene Antworten (Firmware 3.11.0)
| Anfrage | Antwort ohne Null-Padding | Interpretation |
|---|---|---|
| 06 | `34 FF 06 01 03` | Audio läuft |
| 01 | `34 FF 01 03 01 03 01` | Gaming, automatisch, verbunden per Classic |
| 03 | `34 FF 03 01 04` | aktuell aptX Adaptive angeboten |
| 05 | `34 FF 05 01 04` | aptX Adaptive aktiv |
| 07 | `34 FF 07 01 01` | LE getrennt |
| 08 | `34 FF 08 02 02 02` | 24 Bit / 48 kHz |
| 09 | `34 FF 09 03 01 02 00` | öffentlich, HQ, unverschlüsselt konfiguriert |
| 12 | `34 FF 12 03 03 0B 00` | Firmware 3.11.0 |
| 15 | `34 FF 15 01 01` | Kopfhörer unterstützt Classic |
Auch der Auracast-Name wurde erfolgreich gelesen. Private Kennungen/Schlüssel
werden hier nicht dokumentiert. Die verfügbare Audiokonfiguration belegt nicht,
dass Audio mit diesem Profil bitgenau übertragen wird.
## Getrennte Update-Funktion und Grenzen
Die App implementiert ausschließlich Report `0x34` auf der Kontrollschnittstelle.
Sie enthält keine DFU-/Upgrade-Kommandos, Firmwaredateien, Firmwareparser,
Firmware-Download-URLs oder Umschaltung in einen Update-Modus.
Die Firmwareversion wird nur über den Kontrollbefehl `0x12` gelesen.
Der Nutzer hat die Funktion der App am eigenen Dongle bestätigt. Der automatisierte
Hardware-Umschalttest wurde noch nicht ausgeführt; die Rückmeldung belegt keine
vollständige Prüfung jedes Schreibbefehls. Insbesondere Wiederverbindung, echte
Auracast-Empfänger und Firmwareunterschiede benötigen weitere systematische Tests.
Infobereich: `org.kde.StatusNotifierItem` plus `com.canonical.dbusmenu` über
libdbusmenu. Grundlage ist die
[StatusNotifier-Spezifikation](https://specifications.freedesktop.org/status-notifier-item/latest-single/).
GNOME-Registrierung und D-Bus-Menüaktionen wurden lokal geprüft. Name/Passwort
verwenden fokussierte GTK-Eingaben, da D-Bus-Menüs keine Texteingabefelder vorsehen.
+37
View File
@@ -0,0 +1,37 @@
# Validierung am 5. September 2026
Getestetes System: Bazzite, GNOME, Python 3.14, GTK 4.22, libadwaita 1.9.
Angeschlossener BTD 700: USB `3542:3001`, Firmware 3.11.0.
- 21 Tests mit `python3 -m unittest discover -s tests -q`: bestanden.
- Fenster-/Infobereich-Test `python3 tools/check_gui.py`: bestanden, ausschließlich
mit einem simulierten Dongle. Aktionen wurden über `com.canonical.dbusmenu.Event`
ausgelöst, nicht nur direkt gegen Controller-Methoden.
- Geprüfte Demo-Abläufe: Standard/Gaming/Auracast, Codec, Transport, Trennen/Verbinden,
öffentliche Auffindbarkeit, Broadcastqualität, Name/Passwort speichern,
Passwortschutz ausschalten, Fenster schließen/aus Menü wieder öffnen,
ungespeicherte Eingaben bei Statusabfragen erhalten und verwerfen.
- Eigene Fensterbilder bei 620 × 800 und 420 × 600 Pixeln gerendert und visuell
geprüft. Oberfläche scrollbar, Bedienelemente erreichbar. Die zuvor überlange
Passwortbeschriftung wurde gekürzt und um einen sichtbaren Hinweis ergänzt.
- Echte Hardwareabfragen: erfolgreich, siehe PROTOCOL.md. Anzeige:
Gaming / aptX Adaptive / 24 Bit / 48 kHz / laufende Musikwiedergabe.
- Live-Symbol exportiert StatusNotifierItem mit aktuellem Tooltip und einem
D-Bus-Menü. GNOME-StatusNotifierWatcher nimmt die Registrierung an.
- App über persönlichen Desktop-Eintrag installierbar; Desktop-Datei validiert.
Dauerhafter Start in der Benutzersitzung mit transienter Unit
`btd700-control.service`; kein Systemdienst und kein aktivierter Autostart.
- Zweiter USB-Zugriff während laufender App: korrekt mit verständlicher Meldung
abgelehnt, ohne die laufende Instanz zu beeinträchtigen.
- Extraktionsskript gegen die offizielle Windows-EXE geprüft; SHA256 der
extrahierten Assembly entspricht der bei der Analyse verwendeten Datei.
Der Nutzer hat anschließend bestätigt, dass die App am eigenen Dongle funktioniert.
Welche einzelnen Funktionen dabei getestet wurden, wurde nicht näher aufgeschlüsselt.
**Noch offen:** automatisierte Hardware-Schreibtests sowie eine systematische
Prüfung mit Audio-/Auracast-Empfängern. `tools/check_hardware.py --run` wurde bisher
nicht ausgeführt. Dieser Test unterbricht kurz Audio und verändert vorübergehend
Auracast-Einstellungen einschließlich des Passworts; anschließend versucht er,
die ursprünglichen Werte wiederherzustellen. Ein Werksreset wird dabei nicht
ausgeführt; Firmware-Updates sind nicht implementiert.
+13
View File
@@ -0,0 +1,13 @@
#!/usr/bin/env python3
"""Install a launcher for this checkout, without root or system packages."""
from pathlib import Path
import os
from btd700.integration import APP_ID, desktop_entry
base = Path(os.environ.get('XDG_DATA_HOME', Path.home() / '.local/share'))
applications = base / 'applications'
applications.mkdir(parents=True, exist_ok=True)
destination = applications / f'{APP_ID}.desktop'
destination.write_text(desktop_entry())
print(f'Installiert: {destination}')
print('Im Anwendungsmenü „BTD 700 Control“ öffnen. Der Projektordner muss erhalten bleiben.')
+3
View File
@@ -0,0 +1,3 @@
# Grant the active local desktop user access to the BTD 700 control interface.
# The second interface (firmware updater) is deliberately excluded.
SUBSYSTEM=="hidraw", ATTRS{idVendor}=="3542", ATTRS{idProduct}=="3001", ENV{ID_USB_INTERFACE_NUM}=="00", TAG+="uaccess"
+7
View File
@@ -0,0 +1,7 @@
<svg xmlns="http://www.w3.org/2000/svg" width="128" height="128" viewBox="0 0 128 128">
<rect x="4" y="4" width="120" height="120" rx="28" fill="#243347"/>
<path d="M34 71V58a30 30 0 0 1 60 0v13" fill="none" stroke="#edf3fa" stroke-width="8" stroke-linecap="round"/>
<rect x="26" y="64" width="18" height="31" rx="7" fill="#edf3fa"/>
<rect x="84" y="64" width="18" height="31" rx="7" fill="#edf3fa"/>
<path d="m61 45 13 12-20 17 13 12V45L54 57l20 17-13 12" fill="none" stroke="#82b8f4" stroke-width="3.5" stroke-linejoin="round"/>
</svg>

After

Width:  |  Height:  |  Size: 538 B

+16
View File
@@ -0,0 +1,16 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "btd700-control"
version = "0.1.0"
description = "Independent native Linux controls and tray menu for the Sennheiser BTD 700"
requires-python = ">=3.10"
readme = "README.md"
[project.scripts]
btd700-control = "btd700.__main__:main"
[tool.setuptools.packages.find]
include = ["btd700*"]
Executable
+4
View File
@@ -0,0 +1,4 @@
#!/usr/bin/env bash
set -euo pipefail
cd -- "$(dirname -- "${BASH_SOURCE[0]}")"
exec /usr/bin/python3 -m btd700 "$@"
+86
View File
@@ -0,0 +1,86 @@
import unittest
from btd700.controller import Controller
from btd700.demo import DemoTransport
from btd700.protocol import Command as C
from btd700.transport import DeviceError
class RecordingDevice(DemoTransport):
def __init__(self):
super().__init__()
self.writes = []
def request(self, command, payload=b''):
if not command.name.startswith('GET_'):
self.writes.append((command, payload))
return super().request(command, payload)
class ControllerTests(unittest.TestCase):
def setUp(self):
self.device = RecordingDevice()
self.controller = Controller(self.device)
def test_snapshot_does_not_read_password(self):
self.device.values.pop(C.GET_KEY)
status = self.controller.snapshot()
self.assertEqual(status.codec_name, 'aptX Adaptive')
self.assertEqual(status.quality, '24 Bit / 48 kHz')
self.assertEqual(status.firmware, '3.11.0')
self.assertEqual(self.device.writes, [])
self.assertNotIn('password', status.public_dict())
def test_mode_preserves_auto_transport(self):
self.controller.set_mode(1)
self.assertEqual(self.device.writes, [(C.SET_MODE, b'\1\3')])
self.assertEqual(self.controller.snapshot().mode, 1)
def test_codec_capabilities_and_mode(self):
self.controller.set_codec(2)
self.assertEqual(self.controller.snapshot().codec, 2)
with self.assertRaises(ValueError):
self.controller.set_codec(32)
self.controller.set_mode(1)
with self.assertRaises(ValueError):
self.controller.set_codec(4)
def test_validation_happens_before_writes(self):
with self.assertRaises(ValueError):
self.controller.set_broadcast(name='Valid Name', password='bad')
self.assertEqual(self.device.writes, [])
with self.assertRaises(ValueError):
self.controller.set_broadcast(name='Valid Name', encrypted=True)
self.assertEqual(self.device.writes, [])
def test_broadcast_name_key_info_order(self):
self.controller.set_broadcast(name='Studio Audio', password='Secret123', public=False, quality=1)
self.assertEqual(self.device.writes, [(C.SET_KEY, b'Secret123'),
(C.SET_NAME, b'Studio Audio'), (C.SET_BROADCAST, b'\0\1\1')])
status = self.controller.snapshot()
self.assertEqual(status.broadcast_name, 'Studio Audio')
self.assertEqual(status.broadcast_encrypted, 1)
self.assertNotIn('Secret123', str(status.public_dict()))
def test_existing_password_not_cleared_on_other_change(self):
self.controller.set_broadcast(password='Secret123')
self.device.writes.clear()
self.controller.set_broadcast(quality=0)
self.assertEqual(self.device.writes, [(C.SET_BROADCAST, b'\1\0\1')])
self.assertEqual(self.device.values[C.GET_KEY].rstrip(b'\0'), b'Secret123')
def test_reset_requires_confirmation(self):
with self.assertRaises(ValueError):
self.controller.factory_reset()
self.assertEqual(self.device.writes, [])
def test_reconnect_and_disconnect(self):
self.controller.set_connection(False)
self.assertEqual(self.controller.snapshot().state, 1)
self.controller.set_connection(True)
self.assertEqual(self.controller.snapshot().state, 2)
def test_readback_mismatch_is_error(self):
original = self.device.request
self.device.request = lambda command, payload=b'': b'\0' if command == C.GET_CODEC else original(command, payload)
with self.assertRaises(DeviceError):
self.controller._verify(C.GET_CODEC, b'\4', timeout=0)
+63
View File
@@ -0,0 +1,63 @@
import tempfile
import unittest
from pathlib import Path
from btd700.protocol import Command as C, ProtocolError, acknowledge, decode, encode, validate_text
from btd700.transport import discover
class ProtocolTests(unittest.TestCase):
def test_captured_status_response(self):
message = decode(bytes.fromhex('34ff0103010301'))
self.assertEqual(message.payload, b'\1\3\1')
self.assertEqual(message.command, C.GET_MODE)
def test_captured_get_state_request(self):
self.assertEqual(encode(C.GET_STATE), bytes.fromhex('34fe0600') + bytes(60))
def test_all_non_control_command_numbers_rejected(self):
for value in range(256):
if value not in set(C):
with self.assertRaises(ValueError):
encode(value)
def test_setter_validation(self):
for command, args in ((C.SET_MODE, b'\3\3'), (C.SET_MODE, b'\0\0'),
(C.SET_CODEC, b'\3'), (C.SET_CONNECTION, b'\2'),
(C.SET_BROADCAST, b'\1\3\0'), (C.GET_STATE, b'\1'),
(C.SET_KEY, b'x' * 17), (C.FACTORY_RESET, b'\1')):
with self.assertRaises(ValueError):
encode(command, args)
def test_malformed_and_unrelated_reports(self):
for report in (b'\x34', b'\x34\xff\1', bytes.fromhex('34ff013d'), bytes.fromhex('34ff0602ff'), bytes.fromhex('34000600')):
with self.assertRaises(ProtocolError):
decode(report)
self.assertIsNone(decode(b'\1\0'))
self.assertIsNone(decode(b''))
def test_event_ack(self):
self.assertEqual(acknowledge(15), bytes.fromhex('34fd0f00') + bytes(60))
with self.assertRaises(ProtocolError):
acknowledge(255)
def test_name_and_key_bounds(self):
self.assertEqual(validate_text('Studio Audio'), b'Studio Audio')
self.assertEqual(validate_text(''), b'')
self.assertEqual(validate_text('A' * 16, key=True), b'A' * 16)
for name in ('Hi', 'x' * 17, ' Musik', 'Musik ', 'Müsik', 'abc\0def'):
with self.assertRaises(ValueError):
validate_text(name)
with self.assertRaises(ValueError):
validate_text('abc def', key=True)
def test_discovery_excludes_updater_and_other_devices(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
for number, vendor, descriptor in ((0, '3542', b'\x06\xa2\xff\x85\x34'),
(1, '3542', b'\x06\x00\xff\x85\x05'),
(2, '046D', b'\x06\xa2\xff\x85\x34')):
node = root / f'hidraw{number}' / 'device'
node.mkdir(parents=True)
(node / 'uevent').write_text(f'HID_ID=0003:0000{vendor}:00003001\nHID_UNIQ=test\n')
(node / 'report_descriptor').write_bytes(descriptor)
self.assertEqual([d.path for d in discover(root)], ['/dev/hidraw0'])
+56
View File
@@ -0,0 +1,56 @@
import os
import socket
import threading
import unittest
from btd700.protocol import Command as C, ProtocolError
from btd700.transport import Hidraw
class TransportTests(unittest.TestCase):
def setUp(self):
self.host, self.device = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM)
self.transport = Hidraw.__new__(Hidraw)
self.transport.fd = self.host.detach()
self.transport.timeout = 0.1
self.transport.events = {}
def tearDown(self):
self.transport.close()
self.device.close()
def test_event_and_unrelated_response_do_not_complete_request(self):
ack = []
def device():
self.device.recv(64)
os.write(self.device.fileno(), bytes.fromhex('0100'))
os.write(self.device.fileno(), bytes.fromhex('34ff01020003'))
os.write(self.device.fileno(), bytes.fromhex('34fc0f0103'))
ack.append(self.device.recv(64))
os.write(self.device.fileno(), bytes.fromhex('34ff060103'))
thread = threading.Thread(target=device)
thread.start()
self.assertEqual(self.transport.request(C.GET_STATE), b'\3')
thread.join(timeout=1)
self.assertEqual(self.transport.events[15], b'\3')
self.assertEqual(ack[0][:4], bytes.fromhex('34fd0f00'))
def test_timeout_is_bounded(self):
with self.assertRaises(TimeoutError):
self.transport.request(C.GET_STATE)
def test_short_payload_is_rejected(self):
def device():
self.device.recv(64)
os.write(self.device.fileno(), bytes.fromhex('34ff080102'))
thread = threading.Thread(target=device)
thread.start()
with self.assertRaises(ProtocolError):
self.transport.request(C.GET_QUALITY)
thread.join(timeout=1)
def test_invalid_command_does_not_touch_device(self):
with self.assertRaises(ValueError):
self.transport.request(255)
self.device.settimeout(0.01)
with self.assertRaises(TimeoutError):
self.device.recv(64)
+143
View File
@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""Exercise the GTK UI and real D-Bus menu against an isolated demo device."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from btd700.app import Application
from gi.repository import Gio, GLib, Gtk
app = Application(demo=True)
step = 0
wait_ticks = 0
passed = False
bus_pending = False
def menu_item(labels):
root = app.tray.root
for label in labels:
root = next(item for item in root.get_children() if item.property_get('label') == label)
return root
def click_menu(*labels):
global bus_pending
item = menu_item(labels)
assert item.property_get_bool('enabled'), labels
bus_pending = True
def complete(bus, result):
global bus_pending
try:
bus.call_finish(result)
except Exception:
app.quit()
raise
bus_pending = False
app.tray.bus.call(app.tray.bus.get_unique_name(), '/Menu', 'com.canonical.dbusmenu', 'Event',
GLib.Variant('(isvu)', (item.get_id(), 'clicked', GLib.Variant('i', 0), 0)),
None, Gio.DBusCallFlags.NONE, 2000, None, complete)
def screenshot(path):
window = app.window
paintable = Gtk.WidgetPaintable.new(window)
snapshot = Gtk.Snapshot.new()
paintable.snapshot(snapshot, window.get_width(), window.get_height())
node = snapshot.to_node()
if node is None:
raise RuntimeError('Window snapshot is not ready')
texture = window.get_renderer().render_texture(node, None)
texture.save_to_png(path)
def tick():
global step, passed, wait_ticks
try:
wait_ticks += 1
if wait_ticks > 200:
raise TimeoutError(f'GUI step {step} timed out')
if not app.status or not app.tray or not app.tray.available or app.busy or bus_pending:
return True
assert not app.last_error, app.last_error
if step == 0:
assert app.status.mode == 0
click_menu('Audiomodus', 'Gaming')
elif step == 1:
assert app.status.mode == 1
assert not menu_item(('Codec', 'SBC')).property_get_bool('enabled')
click_menu('Audiomodus', 'Standard')
elif step == 2:
assert app.status.mode == 0
click_menu('Codec', 'SBC')
elif step == 3:
assert app.status.codec == 1
click_menu('Bluetooth-Transport', 'Bluetooth Classic')
elif step == 4:
assert app.status.transport == 1
click_menu('Kopfhörer trennen')
elif step == 5:
assert app.status.state == 1
click_menu('Kopfhörer verbinden')
elif step == 6:
assert app.status.state == 2
click_menu('Audiomodus', 'Auracast')
elif step == 7:
assert app.status.mode == 2
click_menu('Auracast', 'Öffentlich auffindbar')
elif step == 8:
assert app.status.broadcast_public == 0
click_menu('Auracast', 'Übertragungsqualität', 'Standard · 24 kHz')
elif step == 9:
assert app.status.broadcast_quality == 1
click_menu('Auracast', 'Name und Passwort …')
elif step == 10:
assert app.window.get_visible()
app.name_entry.set_text('Studio Linux')
app.password_entry.set_text('DemoSecret123')
app.save_button.emit('clicked')
elif step == 11:
assert app.status.broadcast_name == 'Studio Linux'
assert app.status.broadcast_encrypted == 1
assert not app.password_entry.get_text()
assert not app.dirty
click_menu('Auracast', 'Passwortschutz')
elif step == 12:
assert app.status.broadcast_encrypted == 0
app.window.close()
assert not app.window.get_visible()
assert app.tray.available
click_menu('Fenster öffnen')
elif step == 13:
assert app.window.get_visible()
app.name_entry.set_text('Unsaved edit')
app._on_status(app.status, '', False)
assert app.name_entry.get_text() == 'Unsaved edit'
app.discard_button.emit('clicked')
assert app.name_entry.get_text() == 'Studio Linux'
app.mode_buttons[0].emit('clicked') if app.mode_buttons[0].get_active() else app.mode_buttons[0].set_active(True)
app.perform('set_mode', 0)
elif step == 14:
screenshot('/tmp/btd700-gui-test.png')
adjustment = app.scroll.get_vadjustment()
adjustment.set_value(adjustment.get_upper() - adjustment.get_page_size())
elif step == 15:
screenshot('/tmp/btd700-gui-test-bottom.png')
app.window.set_default_size(420, 600)
elif step == 16:
screenshot('/tmp/btd700-gui-test-narrow.png')
print('PASS: GTK form, all D-Bus tray control groups, encryption, close/reopen, dirty edits, screenshots', flush=True)
passed = True
app.quit()
return False
print('GUI step', step, 'passed', flush=True)
step += 1
wait_ticks = 0
except Exception as exc:
print('FAIL: GUI step', step, type(exc).__name__, str(exc), flush=True)
app.quit()
return False
return True
GLib.timeout_add(250, tick)
app.run(['btd700-gui-test'])
raise SystemExit(0 if passed else 1)
+65
View File
@@ -0,0 +1,65 @@
#!/usr/bin/env python3
"""Opt-in reversible control test. Never resets or updates the dongle."""
import argparse
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from btd700.controller import Controller
from btd700.protocol import Command as C, encode
from btd700.transport import Hidraw, discover
parser = argparse.ArgumentParser(description='Steuerfunktionen testen und Originalwerte wiederherstellen; unterbricht kurz Audio.')
parser.add_argument('--run', action='store_true', required=True)
args = parser.parse_args()
devices = discover()
assert len(devices) == 1, 'Exactly one BTD 700 required'
with Hidraw(devices[0]) as transport:
c = Controller(transport)
original = c.snapshot()
original_name = transport.request(C.GET_NAME).rstrip(b'\0')
original_key = transport.request(C.GET_KEY).rstrip(b'\0')
original_info = transport.request(C.GET_BROADCAST)
# Refuse to mutate unless all saved values can be restored by this driver.
encode(C.SET_KEY, original_key)
encode(C.SET_NAME, original_name)
encode(C.SET_BROADCAST, original_info)
encode(C.SET_MODE, bytes((original.mode, original.transport)))
print('Original:', original.mode, original.transport, original.codec, original.quality, flush=True)
try:
c.set_mode(0)
print('PASS: Standard mode readback', flush=True)
s = c.snapshot()
chosen = next((value for value in (1, 2, 4, 8, 16, 32) if s.codecs & value), None)
if chosen:
c.set_codec(chosen)
print('PASS: codec readback', chosen, flush=True)
c.set_mode(2)
print('PASS: Auracast mode readback', flush=True)
c.set_broadcast(name='BTD700 Linux', public=not bool(original_info[0]), quality=(original_info[1]+1)%3)
print('PASS: broadcast name, visibility and quality readback', flush=True)
if len(original_key) <= 16 and all(32 <= b <= 126 for b in original_key):
c.set_broadcast(password='LinuxTest2026', encrypted=True)
assert transport.request(C.GET_KEY).rstrip(b'\0') == b'LinuxTest2026'
print('PASS: password and encryption readback (secret omitted)', flush=True)
finally:
transport.request(C.SET_KEY, original_key)
transport.request(C.SET_NAME, original_name)
transport.request(C.SET_BROADCAST, original_info)
transport.request(C.SET_MODE, bytes((0, original.transport)))
deadline = time.monotonic()+10
while time.monotonic() < deadline:
s = c.snapshot()
if s.state >= 2:
break
time.sleep(.25)
if s.state >= 2 and s.codecs & original.codec:
c.set_codec(original.codec)
transport.request(C.SET_MODE, bytes((original.mode, original.transport)))
c._verify(C.GET_MODE, bytes((original.mode, original.transport)), exact=False)
assert transport.request(C.GET_NAME).rstrip(b'\0') == original_name
assert transport.request(C.GET_KEY).rstrip(b'\0') == original_key
assert transport.request(C.GET_BROADCAST) == original_info
final = c.snapshot()
print('RESTORED: mode, transport, Auracast name, key, visibility, quality, encryption', flush=True)
print('Final:', final.mode, final.transport, final.codec, final.quality, flush=True)
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""Extract only the application assembly from a locally supplied .NET bundle.
No downloading or execution of the original application. Output remains private
analysis material; it is not needed at runtime and must not be distributed here.
"""
import argparse
import io
from pathlib import Path
import struct
import zlib
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('exe', type=Path)
parser.add_argument('output', type=Path, help='Output DLL path, outside this project')
args = parser.parse_args()
data = args.exe.read_bytes()
signature = bytes.fromhex('8b1202b96a612038727b930214d7a03213f5b9e6efae3318ee3b2dce24b36aae')
position = data.find(signature)
if position < 8:
parser.error('No .NET bundle signature found')
offset = struct.unpack_from('<q', data, position - 8)[0]
if not 0 <= offset < len(data):
parser.error('Invalid manifest offset')
stream = io.BytesIO(data)
stream.seek(offset)
def unpack(fmt):
return struct.unpack(fmt, stream.read(struct.calcsize(fmt)))
def read_string():
length = 0
for shift in range(0, 35, 7):
value = unpack('<B')[0]
length |= (value & 127) << shift
if not value & 128:
if length > 4096:
raise ValueError('Unexpected string length')
return stream.read(length).decode('utf-8')
raise ValueError('Invalid string prefix')
major, minor, count = unpack('<IIi')
if major != 6 or not 1 <= count <= 10000:
parser.error('Expected a version-6 .NET bundle')
read_string()
stream.read(40)
for _ in range(count):
offset, size, compressed, kind = unpack('<qqqB')
name = read_string()
if name != 'Sennheiser Dongle Control.dll':
continue
length = compressed or size
if not 0 <= offset <= offset + length <= len(data) or not 0 < size < 32 * 1024 * 1024:
parser.error('Invalid assembly bounds')
blob = data[offset:offset + length]
if compressed:
decoder = zlib.decompressobj(-15)
blob = decoder.decompress(blob, size + 1)
if len(blob) != size or not blob.startswith(b'MZ'):
parser.error('Assembly size or signature mismatch')
with args.output.open('xb') as output:
output.write(blob)
print(f'Extracted {name}: {len(blob)} bytes to {args.output}')
break
else:
parser.error('Application assembly not found')