Refactor playables a bit, general fixes n stuff

disco/voice/__init__.py

@@ -0,0 +1,2 @@
+from disco.voice.client import *
+from disco.voice.player import *

disco/voice/opus.py

@@ -1,13 +1,10 @@
 import sys
 import array
+import struct
 import gevent
 import ctypes
 import ctypes.util
-
-try:
-    from cStringIO import cStringIO as StringIO
-except:
-    from StringIO import StringIO
+import subprocess
 
 from gevent.queue import Queue
 from holster.enum import Enum
@@ -100,10 +97,16 @@ class OpusEncoder(BaseOpus):
         self.samples_per_frame = int(self.sampling_rate / 1000 * self.frame_length)
         self.frame_size = self.samples_per_frame * self.sample_size
 
-        self.inst = self.create()
-        self.set_bitrate(128)
-        self.set_fec(True)
-        self.set_expected_packet_loss_percent(0.15)
+        self._inst = None
+
+    @property
+    def inst(self):
+        if not self._inst:
+            self._inst = self.create()
+            self.set_bitrate(128)
+            self.set_fec(True)
+            self.set_expected_packet_loss_percent(0.15)
+        return self._inst
 
     def set_bitrate(self, kbps):
         kbps = min(128, max(16, int(kbps)))
@@ -156,8 +159,8 @@ class OpusDecoder(BaseOpus):
 
 
 class BufferedOpusEncoder(OpusEncoder):
-    def __init__(self, data, *args, **kwargs):
-        self.data = StringIO(data)
+    def __init__(self, f, *args, **kwargs):
+        self.data = f
         self.frames = Queue(kwargs.pop('queue_size', 4096))
         super(BufferedOpusEncoder, self).__init__(*args, **kwargs)
         gevent.spawn(self._encoder_loop)
@@ -182,10 +185,10 @@ class BufferedOpusEncoder(OpusEncoder):
 class GIPCBufferedOpusEncoder(OpusEncoder):
     FIN = 1
 
-    def __init__(self, data, *args, **kwargs):
+    def __init__(self, f, *args, **kwargs):
         import gipc
 
-        self.data = StringIO(data)
+        self.data = f
         self.parent_pipe, self.child_pipe = gipc.pipe(duplex=True)
         self.frames = Queue(kwargs.pop('queue_size', 4096))
         super(GIPCBufferedOpusEncoder, self).__init__(*args, **kwargs)
@@ -232,3 +235,39 @@ class GIPCBufferedOpusEncoder(OpusEncoder):
                 return
 
             pipe.put(encoder.encode(data, encoder.samples_per_frame))
+
+
+class DCADOpusEncoder(OpusEncoder):
+    def __init__(self, pipe, *args, **kwargs):
+        command = kwargs.pop('command', 'dcad')
+        super(DCADOpusEncoder, self).__init__(*args, **kwargs)
+        self.proc = subprocess.Popen([
+            command,
+            # '--channels', str(self.channels),
+            # '--rate', str(self.sampling_rate),
+            # '--size', str(self.frame_length),
+            '--bitrate', '128',
+            '--fec',
+            '--packet-loss-percent', '30',
+            '--input', 'pipe:0',
+            '--output', 'pipe:1',
+        ], stdin=pipe, stdout=subprocess.PIPE)
+        self.header_size = struct.calcsize('<h')
+
+    def have_frame(self):
+        return bool(self.proc)
+
+    def next_frame(self):
+        header = self.proc.stdout.read(self.header_size)
+        if len(header) < self.header_size:
+            self.proc = None
+            return
+
+        size = struct.unpack('<h', header)[0]
+
+        data = self.proc.stdout.read(size)
+        if len(data) < size:
+            self.proc = None
+            return
+
+        return data

disco/voice/player.py

@@ -4,28 +4,79 @@ import struct
 import subprocess
 
 from six.moves import queue
+from holster.enum import Enum
+from holster.emitter import Emitter
 
 from disco.voice.client import VoiceState
-from disco.voice.opus import BufferedOpusEncoder, GIPCBufferedOpusEncoder
+from disco.voice.opus import OpusEncoder, BufferedOpusEncoder
 
+try:
+    from cStringIO import cStringIO as StringIO
+except:
+    from StringIO import StringIO
 
-class BaseFFmpegPlayable(object):
+
+class FFmpegPlayable(object):
     def __init__(self, source='-', command='avconv', sampling_rate=48000, channels=2, **kwargs):
-        args = [command, '-i', source, '-f', 's16le', '-ar', str(sampling_rate), '-ac', str(channels), '-loglevel', 'warning', 'pipe:1']
-        self.proc = subprocess.Popen(args, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        data, _ = self.proc.communicate()
-        super(BaseFFmpegPlayable, self).__init__(data, sampling_rate, channels, **kwargs)
+        self.source = source
+        self.command = command
+        self.sampling_rate = sampling_rate
+        self.channels = channels
+        self.kwargs = kwargs
+
+        self._proc = None
+        self._child = None
+
+    def pipe(self, other, streaming=True):
+        if issubclass(other, OpusEncoder):
+            if not streaming:
+                stdout, _ = self._proc.communicate()
+                self._child = other(StringIO(stdout), self.sampling_rate, self.channels, **self.kwargs)
+            else:
+                self._child = other(self.out_pipe, self.sampling_rate, self.channels, **self.kwargs)
+
+    @property
+    def samples_per_frame(self):
+        return self._child.samples_per_frame
+
+    @property
+    def proc(self):
+        if not self._proc:
+            args = [
+                self.command,
+                '-i', self.source,
+                '-f', 's16le',
+                '-ar', str(self.sampling_rate),
+                '-ac', str(self.channels),
+                '-loglevel', 'warning',
+                'pipe:1'
+            ]
+            self._proc = subprocess.Popen(args, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        return self._proc
+
+    @property
+    def out_pipe(self):
+        return self.proc.stdout
+
+    @property
+    def in_pipe(self):
+        return self.proc.stdin
 
+    def have_frame(self):
+        return self._child and self._child.have_frame()
 
-class FFmpegPlayable(BaseFFmpegPlayable, BufferedOpusEncoder):
-    pass
+    def next_frame(self):
+        return self._child.next_frame()
 
 
-class GIPCFFmpegPlayable(BaseFFmpegPlayable, GIPCBufferedOpusEncoder):
-    pass
+def create_ffmpeg_playable(*args, **kwargs):
+    cls = kwargs.pop('cls', BufferedOpusEncoder)
+    playable = FFmpegPlayable(*args, **kwargs)
+    playable.pipe(cls)
+    return playable
 
 
-def create_youtube_dl_playable(url, cls=FFmpegPlayable, *args, **kwargs):
+def create_youtube_dl_playable(url, *args, **kwargs):
     import youtube_dl
 
     ydl = youtube_dl.YoutubeDL({'format': 'webm[abr>0]/bestaudio/best'})
@@ -34,7 +85,9 @@ def create_youtube_dl_playable(url, cls=FFmpegPlayable, *args, **kwargs):
     if 'entries' in info:
         info = info['entries'][0]
 
-    return cls(info['url'], *args, **kwargs), info
+    playable = create_ffmpeg_playable(info['url'], *args, **kwargs)
+    playable.info = info
+    return playable
 
 
 class OpusPlayable(object):
@@ -71,25 +124,58 @@ class OpusPlayable(object):
 
 
 class Player(object):
+    Events = Enum(
+        'START_PLAY',
+        'STOP_PLAY',
+        'PAUSE_PLAY',
+        'RESUME_PLAY',
+        'DISCONNECT'
+    )
+
     def __init__(self, client):
         self.client = client
+
+        # Queue contains playable items
         self.queue = queue.Queue()
+
+        # Whether we're playing music (true for lifetime)
         self.playing = True
-        self.run_task = gevent.spawn(self.run)
+
+        # Set to an event when playback is paused
         self.paused = None
+
+        # Current playing item
+        self.now_playing = None
+
+        # Current play task
+        self.play_task = None
+
+        # Core task
+        self.run_task = gevent.spawn(self.run)
+
+        # Event triggered when playback is complete
         self.complete = gevent.event.Event()
 
+        # Event emitter for metadata
+        self.events = Emitter(gevent.spawn)
+
     def disconnect(self):
         self.client.disconnect()
+        self.events.emit(self.Events.DISCONNECT)
+
+    def skip(self):
+        self.play_task.kill()
 
     def pause(self):
         if self.paused:
             return
         self.paused = gevent.event.Event()
+        self.events.emit(self.Events.PAUSE_PLAY)
 
     def resume(self):
         self.paused.set()
         self.paused = None
+        self.events.emit(self.Events.RESUME_PLAY)
 
     def play(self, item):
         start = time.time()
@@ -123,7 +209,12 @@ class Player(object):
         self.client.set_speaking(True)
 
         while self.playing:
-            self.play(self.queue.get())
+            self.now_playing = self.queue.get()
+
+            self.events.emit(self.Events.START_PLAY, self.now_playing)
+            self.play_task = gevent.spawn(self.play, self.now_playing)
+            self.play_task.join()
+            self.events.emit(self.Events.STOP_PLAY, self.now_playing)
 
             if self.client.state == VoiceState.DISCONNECTED:
                 self.playing = False

examples/music.py

@@ -5,7 +5,7 @@ from disco.voice.client import VoiceException
 
 
 def download(url):
-    return create_youtube_dl_playable(url)[0]
+    return create_youtube_dl_playable(url)
 
 
 class MusicPlugin(Plugin):

requirements.txt

@@ -3,4 +3,4 @@ holster==1.0.11
 inflection==0.3.1
 requests==2.11.1
 six==1.10.0
-websocket-client==0.37.0
+websocket-client==0.40.0