commit 3579667c6e2d15a6ca447d2aa75a9a9a074f346f
Author: Justin <39440218+justin-eckenweber@users.noreply.github.com>
Date: Sat Sep 5 20:43:35 2026 +0200
Add native Linux controls and tray menu for BTD 700
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..2a475f9
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+.venv/
+build/
+dist/
+*.egg-info/
+.agents/
+.codex/
+analysis/
+*.exe
+*.dll
+*.dfu
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..11312dd
--- /dev/null
+++ b/README.md
@@ -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 4–16 ASCII-Buchstaben/Ziffern mit
+inneren Leerzeichen; ein leerer Name stellt den Gerätenamen wieder her. Passwörter
+verwenden 4–16 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).
diff --git a/btd700/__init__.py b/btd700/__init__.py
new file mode 100644
index 0000000..c1f1ef6
--- /dev/null
+++ b/btd700/__init__.py
@@ -0,0 +1,3 @@
+"""Independent Linux controls for the Sennheiser BTD 700."""
+
+__version__ = "0.1.0"
diff --git a/btd700/__main__.py b/btd700/__main__.py
new file mode 100644
index 0000000..d877b4d
--- /dev/null
+++ b/btd700/__main__.py
@@ -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())
diff --git a/btd700/app.py b/btd700/app.py
new file mode 100644
index 0000000..b6d7eea
--- /dev/null
+++ b/btd700/app.py
@@ -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('4–16 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
diff --git a/btd700/controller.py b/btd700/controller.py
new file mode 100644
index 0000000..6a16051
--- /dev/null
+++ b/btd700/controller.py
@@ -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 4–16 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 4–16 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)
diff --git a/btd700/demo.py b/btd700/demo.py
new file mode 100644
index 0000000..cc3d3db
--- /dev/null
+++ b/btd700/demo.py
@@ -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
diff --git a/btd700/integration.py b/btd700/integration.py
new file mode 100644
index 0000000..dba10d8
--- /dev/null
+++ b/btd700/integration.py
@@ -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)
diff --git a/btd700/protocol.py b/btd700/protocol.py
new file mode 100644
index 0000000..3f792a6
--- /dev/null
+++ b/btd700/protocol.py
@@ -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 4–16 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: 4–16 Buchstaben oder Ziffern.' if key else (
+ 'Name: 4–16 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]])
diff --git a/btd700/transport.py b/btd700/transport.py
new file mode 100644
index 0000000..507e94d
--- /dev/null
+++ b/btd700/transport.py
@@ -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
diff --git a/btd700/tray.py b/btd700/tray.py
new file mode 100644
index 0000000..daabbf1
--- /dev/null
+++ b/btd700/tray.py
@@ -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 = '''
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+'''
+
+
+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)
diff --git a/btd700/worker.py b/btd700/worker.py
new file mode 100644
index 0000000..ebc8f34
--- /dev/null
+++ b/btd700/worker.py
@@ -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()
diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md
new file mode 100644
index 0000000..65a30f4
--- /dev/null
+++ b/docs/PROTOCOL.md
@@ -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 | 0–16 Bytes; UI beschränkt auf leer oder 4–16 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.
diff --git a/docs/VALIDATION.md b/docs/VALIDATION.md
new file mode 100644
index 0000000..cf0ec0a
--- /dev/null
+++ b/docs/VALIDATION.md
@@ -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.
diff --git a/install.py b/install.py
new file mode 100644
index 0000000..6482540
--- /dev/null
+++ b/install.py
@@ -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.')
diff --git a/packaging/70-btd700-control.rules b/packaging/70-btd700-control.rules
new file mode 100644
index 0000000..7bcef07
--- /dev/null
+++ b/packaging/70-btd700-control.rules
@@ -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"
diff --git a/packaging/btd700-control.svg b/packaging/btd700-control.svg
new file mode 100644
index 0000000..44c453b
--- /dev/null
+++ b/packaging/btd700-control.svg
@@ -0,0 +1,7 @@
+
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..98450fb
--- /dev/null
+++ b/pyproject.toml
@@ -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*"]
diff --git a/run.sh b/run.sh
new file mode 100755
index 0000000..cfcd7ec
--- /dev/null
+++ b/run.sh
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd -- "$(dirname -- "${BASH_SOURCE[0]}")"
+exec /usr/bin/python3 -m btd700 "$@"
diff --git a/tests/test_controller.py b/tests/test_controller.py
new file mode 100644
index 0000000..70ed30b
--- /dev/null
+++ b/tests/test_controller.py
@@ -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)
diff --git a/tests/test_protocol.py b/tests/test_protocol.py
new file mode 100644
index 0000000..1d10d5d
--- /dev/null
+++ b/tests/test_protocol.py
@@ -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'])
diff --git a/tests/test_transport.py b/tests/test_transport.py
new file mode 100644
index 0000000..5573ea4
--- /dev/null
+++ b/tests/test_transport.py
@@ -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)
diff --git a/tools/check_gui.py b/tools/check_gui.py
new file mode 100644
index 0000000..b3523f7
--- /dev/null
+++ b/tools/check_gui.py
@@ -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)
diff --git a/tools/check_hardware.py b/tools/check_hardware.py
new file mode 100644
index 0000000..327d0a0
--- /dev/null
+++ b/tools/check_hardware.py
@@ -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)
diff --git a/tools/extract_control_assembly.py b/tools/extract_control_assembly.py
new file mode 100644
index 0000000..9fcbdc0
--- /dev/null
+++ b/tools/extract_control_assembly.py
@@ -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(' 4096:
+ raise ValueError('Unexpected string length')
+ return stream.read(length).decode('utf-8')
+ raise ValueError('Invalid string prefix')
+
+
+major, minor, count = unpack('