Voice: raise the USB speakerphone's volume before speaking

carwatch/listen.py

@@ -402,6 +402,22 @@ def _car_a2dp_mac():
         return None
 
 
+def _usb_speaker_volume(target: str, pct: int = 90) -> None:
+    """Bring a USB speakerphone's playback level up before an answer. The
+    Jabra Speak2 40 came up at 53% (-21 dB): aplay exited 0 and petrus heard
+    nothing across the room (26 Sep). Best effort; control names vary."""
+    m = re.match(r"plughw:(\d+),", target or "")
+    if not m:
+        return
+    for control in ("PCM", "Speaker"):
+        try:
+            if subprocess.run(["amixer", "-q", "-c", m.group(1), "sset", control, f"{pct}%", "unmute"],
+                              capture_output=True, timeout=5).returncode == 0:
+                return
+        except Exception:
+            return
+
+
 def _speak(text: str) -> bool:
     """Voice a reply through the car's A2DP sink (the brief's music channel),
     falling back to USB playback, then the headset channels. Call ONLY with
@@ -480,6 +496,8 @@ def _speak(text: str) -> bool:
             pass
         if bt:
             time.sleep(1.5)   # let a shared headset fall back from HFP mode
+        else:
+            _usb_speaker_volume(target)
         rc = subprocess.run(
             ["aplay", "--buffer-time=1000000", "-D", target, wav],
             capture_output=True, timeout=play_timeout).returncode

tests/test_speak_route.py

@@ -29,8 +29,10 @@ def setUp(self):
 
     def _route(self, a2dp_mac):
         played = []
+        self.calls = []
 
         def run(cmd, **kw):
+            self.calls.append(cmd[0])
             if cmd[0] == "aplay":
                 played.append(cmd[cmd.index("-D") + 1])
             return mock.Mock(returncode=0)
@@ -50,6 +52,16 @@ def run(cmd, **kw):
     def test_absent_car_falls_back_to_usb(self):
         self.assertEqual(self._route(None), ["plughw:0,0"])
 
+    def test_usb_volume_is_raised_before_playback(self):
+        # 26 Sep: the Jabra sat at 53% and petrus heard nothing.
+        self._route(None)
+        self.assertIn("amixer", self.calls)
+        self.assertLess(self.calls.index("amixer"), self.calls.index("aplay"))
+
+    def test_car_playback_leaves_the_mixer_alone(self):
+        self._route(CAR)
+        self.assertNotIn("amixer", self.calls)
+
     def test_connect_timeout_still_falls_back_to_usb(self):
         # 26 Sep on vadelma: connect to the absent car hung past 10 s and the
         # TimeoutExpired aborted the whole reply.