medialink: hands-free listening ignores the microphone's opening click

docs/guides/medialink.md

@@ -155,7 +155,10 @@ The gate can't tell a voice from a door slamming.
 When both sides speak at once, one gets the floor and the other's words
 are lost until the turn ends, as with two buttons. While the peer talks,
 this side doesn't listen, and it waits for the speaker to finish before it
-listens again, so it doesn't send the peer's own voice back.
+listens again, so it doesn't send the peer's own voice back. It also
+ignores the microphone's first 60 ms each time it opens (`settle_ms`),
+since a microphone can click as it starts, and the click would otherwise
+take a turn of nothing back after every turn the peer took.
 [`medialink_hands_free.py`](https://github.com/PyDevices/pydevices/blob/main/examples/medialink_hands_free.py)
 is the whole thing.
 

lib/medialink/intercom.py

@@ -43,6 +43,13 @@
 it. ``press()`` and ``release()`` still work, for a button beside it. Set
 ``intercom.voice`` to None to stop listening, and back to start again.
 
+A microphone's first moment after it opens can hold a click: on a P4 board
+the first 20 ms frame peaked 4 to 80 times over the room, after a turn
+played or not. The voice activity took that for speech, so each time the
+peer finished, this side took a turn of nothing back. Listening throws away
+the first ``settle_ms`` after the microphone opens, before the voice
+activity hears it.
+
 Either side may have no speaker (a baby monitor) or no microphone (its
 listener): pass None.
 """
@@ -56,7 +63,7 @@
 class Intercom:
     def __init__(
         self, link, mic, speaker, *, frame_ms=20, codec=None, prebuffer=5, voice=None, preroll_ms=120,
-        target_ms=None, catch_up_ms=2000,
+        target_ms=None, catch_up_ms=2000, settle_ms=60,
     ):
         self.link = link
         self.mic = mic
@@ -90,6 +97,8 @@ def __init__(
         self.voice = voice
         self._preroll = []
         self._preroll_frames = preroll_ms // frame_ms
+        self._settle_frames = -(-settle_ms // frame_ms)
+        self._settle = 0
         self._mic_open = False
         self._auto = False
         self._refused = None
@@ -203,11 +212,16 @@ def _listen_step(self):
             self.mic.open()
             self._mic_open = True
             self._filled = 0
+            self._settle = self._settle_frames
         view = memoryview(self._capture)
         self._filled += self.mic.readinto(view[self._filled:])
         if self._filled < self.frame_bytes:
             return
         self._filled = 0
+        if self._settle:
+            # The microphone's opening click, not the room.
+            self._settle -= 1
+            return
         frame = bytes(view)
         self._preroll.append(frame)
         if len(self._preroll) > self._preroll_frames + 1:

tests/test_medialink_voice.py

@@ -121,6 +121,34 @@ def listen():
         # The tone arrived, from a little before it was first heard.
         self.assertIn(tone(8000)[:FRAME * 4], bytes(heard.got))
 
+    def listen_after_a_click(self, **kwargs):
+        """A side that has heard the room for a while opens its microphone
+        again, as it does after each of the peer's turns, and the first
+        frame is a click."""
+        fmt = Format(16000, 1)
+        click = struct.pack("<h", 12000) * (FRAME // 2)
+        sound = click + hiss(16000, seed=5)
+        network, a, b = pair()
+        voice = medialink.VoiceActivity(detector="level", hold_ms=40)
+        for frame in frames(hiss(16000, seed=4)):
+            voice.feed(frame)
+        mic = FakeMic(fmt, sound)
+        intercom = medialink.Intercom(a, mic, FakeSpeaker(fmt), voice=voice, codec=codec.PCM16, **kwargs)
+        serve = Serve(b)
+        a.connect("b")
+        try:
+            while mic.at < len(sound):
+                intercom.service()
+        finally:
+            serve.stop()
+        intercom.close()
+        return intercom.stats["voice_turns"]
+
+    def test_the_microphones_opening_click_is_not_speech(self):
+        self.assertEqual(self.listen_after_a_click(), 0)
+        # Without the settling time, the click takes a turn.
+        self.assertEqual(self.listen_after_a_click(settle_ms=0), 1)
+
     def test_no_voice_means_no_microphone(self):
         fmt = Format(16000, 1)
         network, a, b = pair()