Add voice playback control (#111)

disagreement/voice_client.py

@@ -77,11 +77,14 @@ class VoiceClient:
         self.secret_key: Optional[Sequence[int]] = None
         self._server_ip: Optional[str] = None
         self._server_port: Optional[int] = None
-        self._current_source: Optional[AudioSource] = None
+        self._current_source: Optional[AudioSource] = None
-        self._play_task: Optional[asyncio.Task] = None
+        self._play_task: Optional[asyncio.Task] = None
-        self._sink: Optional[AudioSink] = None
+        self._pause_event = asyncio.Event()
-        self._ssrc_map: dict[int, int] = {}
+        self._pause_event.set()
-        self._ssrc_lock = threading.Lock()
+        self._is_playing = False
+        self._sink: Optional[AudioSink] = None
+        self._ssrc_map: dict[int, int] = {}
+        self._ssrc_lock = threading.Lock()
 
     async def connect(self) -> None:
         if self._ws is None:
@@ -189,31 +192,37 @@ class VoiceClient:
             raise RuntimeError("UDP socket not initialised")
         self._udp.send(frame)
 
-    async def _play_loop(self) -> None:
+    async def _play_loop(self) -> None:
-        assert self._current_source is not None
+        assert self._current_source is not None
-        try:
+        self._is_playing = True
-            while True:
+        try:
-                data = await self._current_source.read()
+            while True:
-                if not data:
+                await self._pause_event.wait()
-                    break
+                data = await self._current_source.read()
-                volume = getattr(self._current_source, "volume", 1.0)
+                if not data:
-                if volume != 1.0:
+                    break
-                    data = _apply_volume(data, volume)
+                volume = getattr(self._current_source, "volume", 1.0)
-                await self.send_audio_frame(data)
+                if volume != 1.0:
-        finally:
+                    data = _apply_volume(data, volume)
-            await self._current_source.close()
+                await self.send_audio_frame(data)
-            self._current_source = None
+        finally:
-            self._play_task = None
+            await self._current_source.close()
+            self._current_source = None
+            self._play_task = None
+            self._is_playing = False
+            self._pause_event.set()
 
-    async def stop(self) -> None:
+    async def stop(self) -> None:
-        if self._play_task:
+        if self._play_task:
-            self._play_task.cancel()
+            self._play_task.cancel()
-            with contextlib.suppress(asyncio.CancelledError):
+            self._pause_event.set()
-                await self._play_task
+            with contextlib.suppress(asyncio.CancelledError):
-            self._play_task = None
+                await self._play_task
-        if self._current_source:
+            self._play_task = None
-            await self._current_source.close()
+            self._is_playing = False
-            self._current_source = None
+        if self._current_source:
+            await self._current_source.close()
+            self._current_source = None
 
     async def play(self, source: AudioSource, *, wait: bool = True) -> None:
         """|coro| Play an :class:`AudioSource` on the voice connection."""
@@ -224,10 +233,31 @@ class VoiceClient:
         if wait:
             await self._play_task
 
-    async def play_file(self, filename: str, *, wait: bool = True) -> None:
+    async def play_file(self, filename: str, *, wait: bool = True) -> None:
-        """|coro| Stream an audio file or URL using FFmpeg."""
+        """|coro| Stream an audio file or URL using FFmpeg."""
-
+
-        await self.play(FFmpegAudioSource(filename), wait=wait)
+        await self.play(FFmpegAudioSource(filename), wait=wait)
+
+    def pause(self) -> None:
+        """Pause the current audio source."""
+
+        if self._play_task and not self._play_task.done():
+            self._pause_event.clear()
+
+    def resume(self) -> None:
+        """Resume playback of a paused source."""
+
+        if self._play_task and not self._play_task.done():
+            self._pause_event.set()
+
+    def is_paused(self) -> bool:
+        """Return ``True`` if playback is currently paused."""
+
+        return bool(self._play_task and not self._pause_event.is_set())
+
+    def is_playing(self) -> bool:
+        """Return ``True`` if audio is actively being played."""
+        return self._is_playing and self._pause_event.is_set()
 
     def listen(self, sink: AudioSink) -> None:
         """Start listening to voice and routing to a sink."""

docs/voice_features.md

@@ -6,6 +6,10 @@ Disagreement includes experimental support for connecting to voice channels. You
 voice = await client.join_voice(guild_id, channel_id)
 await voice.play_file("welcome.mp3")
 await voice.play_file("another.mp3")  # switch sources while connected
+voice.pause()
+voice.resume()
+if voice.is_playing():
+    print("audio is playing")
 await voice.close()
 ```
 

tests/test_voice_client.py

@@ -59,6 +59,17 @@ class DummySource(AudioSource):
         return b""
 
 
+class SlowSource(AudioSource):
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+
+    async def read(self) -> bytes:
+        await asyncio.sleep(0)
+        if self.chunks:
+            return self.chunks.pop(0)
+        return b""
+
+
 @pytest.mark.asyncio
 async def test_voice_client_handshake():
     hello = {"d": {"heartbeat_interval": 50}}
@@ -205,3 +216,49 @@ async def test_voice_client_volume_scaling(monkeypatch):
     samples[1] = int(samples[1] * 0.5)
     expected = samples.tobytes()
     assert udp.sent == [expected]
+
+
+@pytest.mark.asyncio
+async def test_pause_resume_and_status():
+    ws = DummyWebSocket(
+        [
+            {"d": {"heartbeat_interval": 50}},
+            {"d": {"ssrc": 1, "ip": "127.0.0.1", "port": 4000}},
+            {"d": {"secret_key": []}},
+        ]
+    )
+    udp = DummyUDP()
+    vc = VoiceClient(
+        client=DummyVoiceClient(),
+        endpoint="ws://localhost",
+        session_id="sess",
+        token="tok",
+        guild_id=1,
+        user_id=2,
+        ws=ws,
+        udp=udp,
+    )
+    await vc.connect()
+    vc._heartbeat_task.cancel()
+
+    src = SlowSource([b"a", b"b", b"c"])
+    await vc.play(src, wait=False)
+
+    while not udp.sent:
+        await asyncio.sleep(0)
+
+    assert vc.is_playing()
+    vc.pause()
+    assert vc.is_paused()
+    await asyncio.sleep(0)
+    sent = len(udp.sent)
+    await asyncio.sleep(0.01)
+    assert len(udp.sent) == sent
+    assert not vc.is_playing()
+
+    vc.resume()
+    assert not vc.is_paused()
+
+    await vc._play_task
+    assert udp.sent == [b"a", b"b", b"c"]
+    assert not vc.is_playing()