From 5bcf669915fa761591b73939f5e2b361d79c0ab1 Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Sat, 26 Sep 2026 17:54:16 +0300 Subject: [PATCH 1/2] Voice: speak on the USB speakerphone when the car is away; owner's trailing @handle addresses the car Co-Authored-By: Claude Opus 5.5 --- carwatch/agent.py | 9 ++++- carwatch/listen.py | 11 ++++-- scripts/car-speak.sh | 12 +++++++ tests/test_speak_route.py | 72 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 101 insertions(+), 3 deletions(-) create mode 100644 tests/test_speak_route.py diff --git a/carwatch/agent.py b/carwatch/agent.py index 552b702..411f33c 100644 --- a/carwatch/agent.py +++ b/carwatch/agent.py @@ -221,7 +221,14 @@ def _mentions_me(msg: dict, handle: str, owner: str = "") -> bool: # #32. A message is addressed to the car when the handle LEADS it, or # when it is a reply to one of the car's own posts. A mention buried in # a sentence is somebody talking about the car; stay quiet. - if not _addressed_to(msg, handle): + # The owner typing the literal @handle anywhere is talking TO the car: + # "can you speak using the jabra @eclass" (petrus, 26 Sep) went + # unanswered because the handle trailed. #32's third-person false + # triggers came from agents, which the owner gate below still stops. + owner_tagged = (handle.lower() in body.lower() + and bool((owner or "").strip()) + and _owner_ok(sender, owner)) + if not owner_tagged and not _addressed_to(msg, handle): return False # ONLY the owner addresses the car through the room. Fellow agents # DISCUSSING the car ("eclass", "E Class" in ordinary sentences) kept diff --git a/carwatch/listen.py b/carwatch/listen.py index c2a15ba..8351af0 100644 --- a/carwatch/listen.py +++ b/carwatch/listen.py @@ -433,8 +433,15 @@ def _speak(text: str) -> bool: subprocess.run(["bluetoothctl", "connect", car], capture_output=True, timeout=10) time.sleep(2) - target = f"bluealsa:DEV={car},PROFILE=a2dp" - bt = True + # Only a LIVE link counts. The saved MAC outlives the car: on the + # kitchen table (26 Sep, Jabra) every answer went to the absent + # car's A2DP and the speakerphone on the desk stayed silent. + if _bt_pcm_mac("a2dpsrc/sink") == car: + target = f"bluealsa:DEV={car},PROFILE=a2dp" + bt = True + else: + print(f"speak: car {car} not connected, trying USB/headset", + flush=True) if not target: target = _usb_audio_device("playback") if not target: diff --git a/scripts/car-speak.sh b/scripts/car-speak.sh index f3f5ada..824e7e4 100644 --- a/scripts/car-speak.sh +++ b/scripts/car-speak.sh @@ -40,6 +40,18 @@ speak() { # Bond can exist while A2DP is down (measured: Paired yes, Connected no). bluetoothctl connect "$mac" >/dev/null 2>&1 || true sleep 2 + # The saved MAC outlives the car: away from it (26 Sep, Jabra on the + # kitchen table) speak to a USB speakerphone instead of into silence. + if ! bluetoothctl info "$mac" 2>/dev/null | grep -q "Connected: yes"; then + local usb + usb=$(aplay -l 2>/dev/null | awk '/^card [0-9]+:.*USB Audio/{sub(":","",$2); print $2; exit}') + if [ -n "$usb" ]; then + timeout 60 aplay -q -D "plughw:$usb,0" "$wav" 2>/dev/null || \ + echo "USB playback failed - is the voice listener holding the speakerphone?" + rm -f "$wav" + return + fi + fi # aplay = send direction; bluealsa-aplay was the receive tool and # blocked forever (27 Aug). Timeout guards the wedge. timeout 30 aplay --buffer-time=1000000 -D "bluealsa:DEV=$mac,PROFILE=a2dp" "$wav" 2>/dev/null || \ diff --git a/tests/test_speak_route.py b/tests/test_speak_route.py new file mode 100644 index 0000000..23e5ed8 --- /dev/null +++ b/tests/test_speak_route.py @@ -0,0 +1,72 @@ +"""Voice answers were silent on the Jabra (petrus, 26 Sep 2026, vadelma). + +~/.carwatch/car-bt-mac held the car's MAC while the car was away, so +_speak aimed every answer at bluealsa A2DP for a head unit that was not +connected and never tried the USB speakerphone. The room gate also dropped +"can you speak using the jabra @eclass" because the handle trailed. These +tests pin both: the car is used only when its A2DP link is live, and the +owner's literal @handle anywhere addresses the car while agents' mentions +still do not. subprocess and piper are mocked: nothing touches real audio. +""" +import os +import sys +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +CAR = "88:5A:85:66:67:C2" + + +class TestSpeakRoute(unittest.TestCase): + def setUp(self): + from carwatch import listen + self.L = listen + wav = os.path.join(os.path.dirname(__file__), "_speak_route.wav") + open(wav, "wb").close() + self.wav = wav + self.addCleanup(lambda: os.path.exists(wav) and os.unlink(wav)) + + def _route(self, a2dp_mac): + played = [] + + def run(cmd, **kw): + if cmd[0] == "aplay": + played.append(cmd[cmd.index("-D") + 1]) + return mock.Mock(returncode=0) + + from carwatch import voiceroom + with mock.patch.object(voiceroom, "tts_wav", return_value=self.wav), \ + mock.patch.object(self.L, "_car_a2dp_mac", return_value=CAR), \ + mock.patch.object(self.L, "_bt_pcm_mac", + side_effect=lambda s: a2dp_mac if s == "a2dpsrc/sink" else None), \ + mock.patch.object(self.L, "_usb_audio_device", return_value="plughw:0,0"), \ + mock.patch.object(self.L, "_echo_tail_sec", return_value=0.0), \ + mock.patch.object(self.L.subprocess, "run", side_effect=run), \ + mock.patch.object(self.L.time, "sleep"): + self.assertTrue(self.L._speak("hello")) + return played + + def test_absent_car_falls_back_to_usb(self): + self.assertEqual(self._route(None), ["plughw:0,0"]) + + def test_connected_car_still_wins(self): + self.assertEqual(self._route(CAR), [f"bluealsa:DEV={CAR},PROFILE=a2dp"]) + + +class TestOwnerTrailingHandle(unittest.TestCase): + def setUp(self): + from carwatch import agent + self.agent = agent + + def test_owner_trailing_handle_is_addressed(self): + msg = {"from": "petrus", "body": "can you speak using the jabra @eclass"} + self.assertTrue(self.agent._mentions_me(msg, "@eclass", owner="petrus")) + + def test_agent_trailing_handle_stays_quiet(self): + msg = {"from": "@claudeMB", "body": "the line @eclass just posted"} + self.assertFalse(self.agent._mentions_me(msg, "@eclass", owner="petrus")) + + def test_owner_spoken_name_mid_sentence_stays_quiet(self): + msg = {"from": "petrus", "body": "I parked the eclass outside"} + self.assertFalse(self.agent._mentions_me(msg, "@eclass", owner="petrus")) From 0557c65e564e4015e616fc795be695bec3c5d4dc Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Sat, 26 Sep 2026 18:01:56 +0300 Subject: [PATCH 2/2] Grounding: tell the car what it speaks and listens through (live aplay -l + car BT state) Co-Authored-By: Claude Opus 5.5 --- carwatch/selfstate.py | 38 ++++++++++++++++++++++++++++++++++++++ tests/test_speak_route.py | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/carwatch/selfstate.py b/carwatch/selfstate.py index 7f4d938..1579021 100644 --- a/carwatch/selfstate.py +++ b/carwatch/selfstate.py @@ -15,6 +15,7 @@ import glob import os +import re import shutil import subprocess @@ -170,6 +171,42 @@ def manual_status() -> str | None: return None +def _usb_audio_names(aplay_l: str) -> list[str]: + """Card names of USB audio devices in `aplay -l` output, e.g. + 'card 0: MS [Jabra Speak2 40 MS], device 0: USB Audio [USB Audio]'.""" + names = [] + for line in (aplay_l or "").splitlines(): + m = re.match(r"card \d+: \S+ \[([^\]]+)\], device \d+: USB Audio", line) + if m and m.group(1) not in names: + names.append(m.group(1)) + return names + + +def voice() -> str | None: + """What the car speaks and listens through right now. Without this the + car told petrus it had no Jabra while one sat plugged into it (26 Sep).""" + usb = _usb_audio_names(_run(["aplay", "-l"]) or "") + car = None + try: + mac = open(os.path.expanduser("~/.carwatch/car-bt-mac")).read().strip() + if mac: + info = _run(["bluetoothctl", "info", mac]) or "" + car = "Connected: yes" in info + except Exception: + pass + parts = [] + if car: + parts.append("you speak through the car's own speakers over Bluetooth") + if usb: + dev = " and ".join(usb) + parts.append(f"a USB speakerphone is plugged into your onboard computer ({dev}); " + "you hear through its microphone" + + ("" if car else " and speak your answers aloud through its speaker")) + if car is False: + parts.append("the car's Bluetooth audio is not connected right now") + return "; ".join(parts) or None + + def live_facts() -> dict[str, str]: """Sensor readings the car may legitimately assert about itself.""" facts: dict[str, str] = {} @@ -193,6 +230,7 @@ def live_facts() -> dict[str, str]: ("brain", serving_model()), ("network", network()), ("your manual", manual_status()), + ("your voice", voice()), ): if val: facts[key] = val diff --git a/tests/test_speak_route.py b/tests/test_speak_route.py index 23e5ed8..d85c21f 100644 --- a/tests/test_speak_route.py +++ b/tests/test_speak_route.py @@ -70,3 +70,42 @@ def test_agent_trailing_handle_stays_quiet(self): def test_owner_spoken_name_mid_sentence_stays_quiet(self): msg = {"from": "petrus", "body": "I parked the eclass outside"} self.assertFalse(self.agent._mentions_me(msg, "@eclass", owner="petrus")) + + +class TestVoiceFact(unittest.TestCase): + """The car said "I don't have a Jabra speaker" with one plugged in.""" + + APLAY = ("**** List of PLAYBACK Hardware Devices ****\n" + "card 0: MS [Jabra Speak2 40 MS], device 0: USB Audio [USB Audio]\n" + " Subdevices: 1/1\n" + "card 1: vc4hdmi0 [vc4-hdmi-0], device 0: MAI PCM i2s-hifi-0 [MAI PCM i2s-hifi-0]\n") + + def setUp(self): + from carwatch import selfstate + self.S = selfstate + + def test_usb_names_skip_hdmi(self): + self.assertEqual(self.S._usb_audio_names(self.APLAY), ["Jabra Speak2 40 MS"]) + + def _voice(self, bt_info): + def run(cmd, timeout=5): + return self.APLAY if cmd[0] == "aplay" else bt_info + with mock.patch.object(self.S, "_run", side_effect=run), \ + mock.patch("builtins.open", mock.mock_open(read_data=CAR)): + return self.S.voice() + + def test_car_away_names_the_jabra_as_the_speaker(self): + v = self._voice("Device x\n\tConnected: no\n") + self.assertIn("Jabra Speak2 40 MS", v) + self.assertIn("speak your answers aloud through its speaker", v) + self.assertIn("Bluetooth audio is not connected", v) + + def test_car_connected_speaks_through_the_car(self): + v = self._voice("Device x\n\tConnected: yes\n") + self.assertIn("car's own speakers", v) + self.assertNotIn("aloud through its speaker", v) + + def test_nothing_plugged_in_says_nothing(self): + with mock.patch.object(self.S, "_run", return_value=""), \ + mock.patch("builtins.open", side_effect=OSError): + self.assertIsNone(self.S.voice())