Working voice sending implementation.

Currently you can only send from a stream that implements
``read`` and a ``ffmpeg`` or ``avconv``.

4 files changed, 391 insertions, 3 deletions

diff --git a/discord/__init__.py b/discord/__init__.py
index e4c0a674..476f2419 100644
--- a/discord/__init__.py
+++ b/discord/__init__.py
@@ -31,6 +31,7 @@ from .colour import Color, Colour
 from .invite import Invite
 from .object import Object
 from . import utils
+from . import opus
 
 import logging
 
diff --git a/discord/client.py b/discord/client.py
index 5d1729fb..284585a2 100644
--- a/discord/client.py
+++ b/discord/client.py
@@ -1903,7 +1903,8 @@ class Client:
             'channel': self.voice_channel,
             'data': self._voice_data_found.data,
             'loop': self.loop,
-            'session_id': self.session_id
+            'session_id': self.session_id,
+            'main_ws': self.ws
         }
 
         result = VoiceClient(**kwargs)
diff --git a/discord/opus.py b/discord/opus.py
new file mode 100644
index 00000000..a49d9341
--- /dev/null
+++ b/discord/opus.py
@@ -0,0 +1,159 @@
+# -*- coding: utf-8 -*-
+
+"""
+The MIT License (MIT)
+
+Copyright (c) 2015 Rapptz
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the "Software"),
+to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense,
+and/or sell copies of the Software, and to permit persons to whom the
+Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
+"""
+
+import ctypes
+import ctypes.util
+import array
+from .errors import DiscordException
+import logging
+
+log = logging.getLogger(__name__)
+c_int_ptr = ctypes.POINTER(ctypes.c_int)
+c_int16_ptr = ctypes.POINTER(ctypes.c_int16)
+c_float_ptr = ctypes.POINTER(ctypes.c_float)
+
+class EncoderStruct(ctypes.Structure):
+    pass
+
+EncoderStructPtr = ctypes.POINTER(EncoderStruct)
+
+# A list of exported functions.
+# The first argument is obviously the name.
+# The second one are the types of arguments it takes.
+# The third is the result type.
+exported_functions = [
+    ('opus_strerror', [ctypes.c_int], ctypes.c_char_p),
+    ('opus_encoder_get_size', [ctypes.c_int], ctypes.c_int),
+    ('opus_encoder_create', [ctypes.c_int, ctypes.c_int, ctypes.c_int, c_int_ptr], EncoderStructPtr),
+    ('opus_encode', [EncoderStructPtr, c_int16_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int32),
+    ('opus_encoder_destroy', [EncoderStructPtr], None)
+]
+
+def libopus_loader(name):
+    # create the library...
+    lib = ctypes.cdll.LoadLibrary(name)
+
+    # register the functions...
+    for item in exported_functions:
+        try:
+            func = getattr(lib, item[0])
+        except Exception as e:
+            raise e
+
+        try:
+            func.argtypes = item[1]
+            func.restype = item[2]
+        except KeyError:
+            pass
+
+    return lib
+
+try:
+    _lib = libopus_loader(ctypes.util.find_library('opus'))
+except:
+    _lib = None
+
+def load_opus(name):
+    """Loads the libopus shared library for use with voice.
+
+    If this function is not called then the library uses the function
+    ``ctypes.util.find_library`` and then loads that one if available.
+
+    Not loading a library leads to voice not working.
+
+    This function propagates the exceptions thrown.
+
+    .. warning::
+
+        The bitness of the library must match the bitness of your python
+        interpreter. If the library is 64-bit then your python interpreter
+        must be 64-bit as well. Usually if there's a mismatch in bitness then
+        the load will throw an exception.
+
+    .. note::
+
+        On Windows, the .dll extension is not necessary. However, on Linux
+        the full extension is required to load the library, e.g. ``libopus.so.1``.
+
+    :param name: The filename of the shared library.
+    """
+    global _lib
+    _lib = libopus_loader(name)
+
+class OpusError(DiscordException):
+    """An exception that is thrown for libopus related errors."""
+    def __init__(self, code):
+        self.code = code
+        msg = _lib.opus_strerror(self.code).decode('utf-8')
+        log.info('"{}" has happened'.format(msg))
+        super(DiscordException, self).__init__(msg)
+
+
+# Some constants...
+OK = 0
+APPLICATION_AUDIO    = 2049
+APPLICATION_VOIP     = 2048
+APPLICATION_LOWDELAY = 2051
+
+class Encoder:
+    def __init__(self, sampling, channels, application=APPLICATION_AUDIO):
+        self.sampling_rate = sampling
+        self.channels = channels
+        self.application = application
+
+        self.frame_length = 20
+        self.sample_size = 2 * self.channels # (bit_rate / 8) but bit_rate == 16
+        self.samples_per_frame = int(self.sampling_rate / 1000 * self.frame_length)
+        self.frame_size = self.samples_per_frame * self.sample_size
+
+        self._state = self._create_state()
+
+    def __del__(self):
+        if hasattr(self, '_state'):
+            _lib.opus_encoder_destroy(self._state)
+            self._state = None
+
+    def _create_state(self):
+        ret = ctypes.c_int()
+        result = _lib.opus_encoder_create(self.sampling_rate, self.channels, self.application, ctypes.byref(ret))
+
+        if ret.value != 0:
+            log.info('error has happened in state creation')
+            raise OpusError(ret.value)
+
+        return result
+
+    def encode(self, pcm, frame_size):
+        max_data_bytes = len(pcm)
+        pcm = ctypes.cast(pcm, c_int16_ptr)
+        data = (ctypes.c_char * max_data_bytes)()
+
+        ret = _lib.opus_encode(self._state, pcm, frame_size, data, max_data_bytes)
+        if ret < 0:
+            log.info('error has happened in encode')
+            raise OpusError(ret)
+
+        return array.array('b', data[:ret]).tobytes()
diff --git a/discord/voice_client.py b/discord/voice_client.py
index 9f0ea3ef..a5da879f 100644
--- a/discord/voice_client.py
+++ b/discord/voice_client.py
@@ -45,11 +45,53 @@ import socket
 import json, time
 import logging
 import struct
+import threading
+import subprocess
+import shlex
 
 log = logging.getLogger(__name__)
 
 from . import utils
-from .errors import ClientException
+from .errors import ClientException, InvalidArgument
+from .opus import Encoder as OpusEncoder
+
+class StreamPlayer(threading.Thread):
+    def __init__(self, stream, encoder, connected, player, after, **kwargs):
+        threading.Thread.__init__(self, **kwargs)
+        self.buff = stream
+        self.encoder = encoder
+        self.player = player
+        self._event = threading.Event()
+        self._connected = connected
+        self.after = after
+        self.delay = self.encoder.frame_length / 1000.0
+
+    def run(self):
+        self.loops = 0
+        start = time.time()
+        while not self.is_done():
+            self.loops += 1
+            data = self.buff.read(self.encoder.frame_size)
+            log.info('received {} bytes (out of {})'.format(len(data), self.encoder.frame_size))
+            if len(data) != self.encoder.frame_size:
+                self.stop()
+                break
+
+            self.player(data)
+            next_time = start + self.delay * self.loops
+            delay = max(0, self.delay + (next_time - time.time()))
+            time.sleep(delay)
+
+    def stop(self):
+        self._event.set()
+        if callable(self.after):
+            try:
+                self.after()
+            except:
+                pass
+
+    def is_done(self):
+        return not self._connected.is_set() or self._event.is_set()
 
 class VoiceClient:
     """Represents a Discord voice connection.
@@ -70,15 +112,27 @@ class VoiceClient:
     channel : :class:`Channel`
         The voice channel connected to.
     """
-    def __init__(self, user, connected, session_id, channel, data, loop):
+    def __init__(self, user, connected, main_ws, session_id, channel, data, loop):
         self.user = user
         self._connected = connected
+        self.main_ws = main_ws
         self.channel = channel
         self.session_id = session_id
         self.loop = loop
         self.token = data.get('token')
         self.guild_id = data.get('guild_id')
         self.endpoint = data.get('endpoint')
+        self.sequence = 0
+        self.timestamp = 0
+        self.encoder = OpusEncoder(48000, 2)
+        log.info('created opus encoder with {0.__dict__}'.format(self.encoder))
+
+    def checked_add(self, attr, value, limit):
+        val = getattr(self, attr)
+        if val + value > limit:
+            setattr(self, attr, 0)
+        else:
+            setattr(self, attr, val + value)
 
     @asyncio.coroutine
     def keep_alive_handler(self, delay):
@@ -155,6 +209,8 @@ class VoiceClient:
         yield from self.ws.send(utils.to_json(speaking))
         self._connected.set()
 
+    # connection related
+
     @asyncio.coroutine
     def connect(self):
         log.info('voice connection is connecting...')
@@ -204,3 +260,174 @@ class VoiceClient:
         self.socket.close()
         self._connected.clear()
         yield from self.ws.close()
+
+        payload = {
+            'op': 4,
+            'd': {
+                'guild_id': None,
+                'channel_id': None,
+                'self_mute': True,
+                'self_deaf': False
+            }
+        }
+
+        yield from self.main_ws.send(utils.to_json(payload))
+
+    # audio related
+
+    def _get_voice_packet(self, data):
+        log.info('creating a voice packet')
+        buff = bytearray(len(data) + 12)
+        buff[0] = 0x80
+        buff[1] = 0x78
+
+        for i in range(0, len(data)):
+            buff[i + 12] = data[i]
+
+        struct.pack_into('>H', buff, 2, self.sequence)
+        struct.pack_into('>I', buff, 4, self.timestamp)
+        struct.pack_into('>I', buff, 8, self.ssrc)
+        return buff
+
+    def create_ffmpeg_player(self, filename, *, use_avconv=False, after=None):
+        """Creates a stream player for ffmpeg that launches in a separate thread to play
+        audio.
+
+        The ffmpeg player launches a subprocess of ``ffmpeg`` to a specific
+        filename and then plays that file.
+
+        You must have the ffmpeg or avconv executable in your path environment variable
+        in order for this to work.
+
+        The operations that can be done on the player are the same as those in
+        :meth:`create_stream_player`.
+
+        Examples
+        ----------
+
+        Basic usage: ::
+
+            voice = yield from client.join_voice_channel(channel)
+            player = voice.create_ffmpeg_player('cool.mp3')
+            player.start()
+
+        Parameters
+        -----------
+        filename : str
+            The filename that ffmpeg will take and convert to PCM bytes.
+            This is passed to the ``-i`` flag that ffmpeg takes.
+        use_avconv: bool
+            Use ``avconv`` instead of ``ffmpeg``.
+        after : callable
+            The finalizer that is called after the stream is done being
+            played. All exceptions the finalizer throws are silently discarded.
+
+        Raises
+        -------
+        ClientException
+            Popen failed to due to an error in ``ffmpeg`` or ``avconv``.
+
+        Returns
+        --------
+        StreamPlayer
+            A stream player with specific operations.
+            See :meth:`create_stream_player`.
+        """
+        command = 'ffmpeg' if not use_avconv else 'avconv'
+        cmd = '{} -i "{}" -f s16le -ar {} -ac {} -loglevel warning pipe:1'
+        cmd = cmd.format(command, filename, self.encoder.sampling_rate, self.encoder.channels)
+        try:
+            process = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE)
+        except:
+            raise ClientException('Popen failed: {}'.format(str(e)))
+
+        return StreamPlayer(process.stdout, self.encoder, self._connected, self.play_audio, after)
+
+    def encoder_options(self, *, sample_rate, channels=2):
+        """Sets the encoder options for the OpusEncoder.
+
+        Calling this after you create a stream player
+        via :meth:`create_ffmpeg_player` or :meth:`create_stream_player`
+        has no effect.
+
+        Parameters
+        ----------
+        sample_rate : int
+            Sets the sample rate of the OpusEncoder.
+        channels : int
+            Sets the number of channels for the OpusEncoder.
+            2 for stereo, 1 for mono.
+
+        Raises
+        -------
+        InvalidArgument
+            The values provided are invalid.
+        """
+        if sample_rate not in (8000, 12000, 16000, 24000, 48000):
+            raise InvalidArgument('Sample rate out of range. Valid: [8000, 12000, 16000, 24000, 48000]')
+        if channels not in (1, 2):
+            raise InvalidArgument('Channels must be either 1 or 2.')
+
+        self.encoder = OpusEncoder(sample_rate, channels)
+        log.info('created opus encoder with {0.__dict__}'.format(self.encoder))
+
+    def create_stream_player(self, stream, after=None):
+        """Creates a stream player that launches in a separate thread to
+        play audio.
+
+        The stream player assumes that ``stream.read`` is a valid function
+        that returns a *bytes-like* object.
+
+        The finalizer, ``after`` is called after the stream has been exhausted.
+
+        The following operations are valid on the ``StreamPlayer`` object:
+
+        +------------------+--------------------------------------------------+
+        |    Operation     |                   Description                    |
+        +==================+==================================================+
+        | player.start()   | Starts the audio stream.                         |
+        +------------------+--------------------------------------------------+
+        | player.stop()    | Stops the audio stream.                          |
+        +------------------+--------------------------------------------------+
+        | player.is_done() | Returns a bool indicating if the stream is done. |
+        +------------------+--------------------------------------------------+
+
+        Parameters
+        -----------
+        stream
+            The stream object to read from.
+        after:
+            The finalizer that is called after the stream is exhausted.
+            All exceptions it throws are silently discarded. It is called
+            without parameters.
+
+        Returns
+        --------
+        StreamPlayer
+            A stream player with the operations noted above.
+        """
+
+    def play_audio(self, data):
+        """Sends an audio packet composed of the data.
+
+        You must be connected to play audio.
+
+        Parameters
+        ----------
+        data
+            The *bytes-like object* denoting PCM voice data.
+
+        Raises
+        -------
+        ClientException
+            You are not connected.
+        OpusError
+            Encoding the data failed.
+        """
+
+        self.checked_add('sequence', 1, 65535)
+        encoded_data = self.encoder.encode(data, self.encoder.samples_per_frame)
+        packet = self._get_voice_packet(encoded_data)
+        sent = self.socket.sendto(packet, (self.endpoint_ip, self.voice_port))
+        self.checked_add('timestamp', self.encoder.samples_per_frame, 4294967295)
+