Add FFmpegOpusAudio and other voice improvements

Rework FFmpeg player and add FFmpegOpusAudio

I have extracted some of the base FFmpeg source code into its own
base class and reimplemented the PCM and the new Opus variants.

Support avconv probing

Also fix a few things

Update `__all__`

Fix the bugs

Rework probe functions and add factory function

Probing involves subprocess so it has been reworked into an async
factory function.

Add docs + a few tweaks

* Removed unnecessary read() and is_opus() functions from FFmpegAudio
* Clear self._stdout in cleanup()
* Add 20 second process communication timeout to probe functions
* Capped probe function bitrate values at 512

Change AudioPlayer to use more accurate, monotonic time.perf_counter()

Add lazy opus loading

The library now no longer loads libopus on import, only on
opus.Encoder creation or manually.

Fix review nits

discord/oggparse.py

@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+
+"""
+The MIT License (MIT)
+
+Copyright (c) 2015-2019 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 struct
+
+from .errors import DiscordException
+
+class OggError(DiscordException):
+    """An exception that is thrown for Ogg stream parsing errors."""
+    pass
+
+# https://tools.ietf.org/html/rfc3533
+# https://tools.ietf.org/html/rfc7845
+
+class OggPage:
+    _header = struct.Struct('<xBQIIIB')
+
+    def __init__(self, stream):
+        try:
+            header = stream.read(struct.calcsize(self._header.format))
+
+            self.flag, self.gran_pos, self.serial, \
+            self.pagenum, self.crc, self.segnum = self._header.unpack(header)
+
+            self.segtable = stream.read(self.segnum)
+            bodylen = sum(struct.unpack('B'*self.segnum, self.segtable))
+            self.data = stream.read(bodylen)
+        except Exception:
+            raise OggError('bad data stream') from None
+
+    def iter_packets(self):
+        packetlen = offset = 0
+        partial = True
+
+        for seg in self.segtable:
+            if seg == 255:
+                packetlen += 255
+                partial = True
+            else:
+                packetlen += seg
+                yield self.data[offset:offset+packetlen], True
+                offset += packetlen
+                packetlen = 0
+                partial = False
+
+        if partial:
+            yield self.data[offset:], False
+
+class OggStream:
+    def __init__(self, stream):
+        self.stream = stream
+
+    def _next_page(self):
+        head = self.stream.read(4)
+        if head == b'OggS':
+            return OggPage(self.stream)
+        else:
+            raise OggError('invalid header magic')
+
+    def _iter_pages(self):
+        page = self._next_page()
+        while page:
+            yield page
+            page = self._next_page()
+
+    def iter_packets(self):
+        partial = b''
+        for page in self._iter_pages():
+            for data, complete in page.iter_packets():
+                partial += data
+                if complete:
+                    yield partial
+                    partial = b''

discord/opus.py

@@ -38,6 +38,8 @@ c_int_ptr = ctypes.POINTER(ctypes.c_int)
 c_int16_ptr = ctypes.POINTER(ctypes.c_int16)
 c_float_ptr = ctypes.POINTER(ctypes.c_float)
 
+_lib = None
+
 class EncoderStruct(ctypes.Structure):
     pass
 
@@ -100,25 +102,29 @@ def libopus_loader(name):
 
     return lib
 
-try:
-    if sys.platform == 'win32':
-        _basedir = os.path.dirname(os.path.abspath(__file__))
-        _bitness = 'x64' if sys.maxsize > 2**32 else 'x86'
-        _filename = os.path.join(_basedir, 'bin', 'libopus-0.{}.dll'.format(_bitness))
-        _lib = libopus_loader(_filename)
-    else:
-        _lib = libopus_loader(ctypes.util.find_library('opus'))
-except Exception:
-    _lib = None
+def _load_default():
+    global _lib
+    try:
+        if sys.platform == 'win32':
+            _basedir = os.path.dirname(os.path.abspath(__file__))
+            _bitness = 'x64' if sys.maxsize > 2**32 else 'x86'
+            _filename = os.path.join(_basedir, 'bin', 'libopus-0.{}.dll'.format(_bitness))
+            _lib = libopus_loader(_filename)
+        else:
+            _lib = libopus_loader(ctypes.util.find_library('opus'))
+    except Exception:
+        _lib = None
+
+    return _lib is not 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
-    :func:`ctypes.util.find_library` and then loads that one
-    if available.
+    :func:`ctypes.util.find_library` and then loads that one if available.
 
-    Not loading a library leads to voice not working.
+    Not loading a library and attempting to use PCM based AudioSources will
+    lead to voice not working.
 
     This function propagates the exceptions thrown.
 
@@ -221,7 +227,8 @@ class Encoder:
         self.application = application
 
         if not is_loaded():
-            raise OpusNotLoaded()
+            if not _load_default():
+                raise OpusNotLoaded()
 
         self._state = self._create_state()
         self.set_bitrate(128)

discord/player.py

@@ -31,16 +31,21 @@ import asyncio
 import logging
 import shlex
 import time
+import json
+import re
 
 from .errors import ClientException
 from .opus import Encoder as OpusEncoder
+from .oggparse import OggStream
 
 log = logging.getLogger(__name__)
 
 __all__ = (
     'AudioSource',
     'PCMAudio',
+    'FFmpegAudio',
     'FFmpegPCMAudio',
+    'FFmpegOpusAudio',
     'PCMVolumeTransformer',
 )
 
@@ -107,7 +112,55 @@ class PCMAudio(AudioSource):
             return b''
         return ret
 
-class FFmpegPCMAudio(AudioSource):
+class FFmpegAudio(AudioSource):
+    """Represents an FFmpeg (or AVConv) based AudioSource.
+
+    User created AudioSources using FFmpeg differently from how :class:`FFmpegPCMAudio` and
+    :class:`FFmpegOpusAudio` work should subclass this.
+    """
+
+    def __init__(self, source, *, executable='ffmpeg', args, **subprocess_kwargs):
+        args = [executable, *args]
+        kwargs = {'stdout': subprocess.PIPE}
+        kwargs.update(subprocess_kwargs)
+
+        self._process = self._spawn_process(args, **kwargs)
+        self._stdout = self._process.stdout
+
+    def _spawn_process(self, args, **subprocess_kwargs):
+        process = None
+        try:
+            process = subprocess.Popen(args, **subprocess_kwargs)
+        except FileNotFoundError:
+            executable = args.partition(' ')[0] if isinstance(args, str) else args[0]
+            raise ClientException(executable + ' was not found.') from None
+        except subprocess.SubprocessError as exc:
+            raise ClientException('Popen failed: {0.__class__.__name__}: {0}'.format(exc)) from exc
+        else:
+            return process
+
+    def cleanup(self):
+        proc = self._process
+        if proc is None:
+            return
+
+        log.info('Preparing to terminate ffmpeg process %s.', proc.pid)
+
+        try:
+            proc.kill()
+        except Exception:
+            log.exception("Ignoring error attempting to kill ffmpeg process %s", proc.pid)
+
+        if proc.poll() is None:
+            log.info('ffmpeg process %s has not terminated. Waiting to terminate...', proc.pid)
+            proc.communicate()
+            log.info('ffmpeg process %s should have terminated with a return code of %s.', proc.pid, proc.returncode)
+        else:
+            log.info('ffmpeg process %s successfully terminated with return code of %s.', proc.pid, proc.returncode)
+
+        self._process = self._stdout = None
+
+class FFmpegPCMAudio(FFmpegAudio):
     """An audio source from FFmpeg (or AVConv).
 
     This launches a sub-process to a specific input file given.
@@ -131,10 +184,10 @@ class FFmpegPCMAudio(AudioSource):
     stderr: Optional[:term:`py:file object`]
         A file-like object to pass to the Popen constructor.
         Could also be an instance of ``subprocess.PIPE``.
-    options: Optional[:class:`str`]
-        Extra command line arguments to pass to ffmpeg after the ``-i`` flag.
     before_options: Optional[:class:`str`]
         Extra command line arguments to pass to ffmpeg before the ``-i`` flag.
+    options: Optional[:class:`str`]
+        Extra command line arguments to pass to ffmpeg after the ``-i`` flag.
 
     Raises
     --------
@@ -143,9 +196,8 @@ class FFmpegPCMAudio(AudioSource):
     """
 
     def __init__(self, source, *, executable='ffmpeg', pipe=False, stderr=None, before_options=None, options=None):
-        stdin = None if not pipe else source
-
-        args = [executable]
+        args = []
+        subprocess_kwargs = {'stdin': source if pipe else None, 'stderr': stderr}
 
         if isinstance(before_options, str):
             args.extend(shlex.split(before_options))
@@ -159,14 +211,7 @@ class FFmpegPCMAudio(AudioSource):
 
         args.append('pipe:1')
 
-        self._process = None
-        try:
-            self._process = subprocess.Popen(args, stdin=stdin, stdout=subprocess.PIPE, stderr=stderr)
-            self._stdout = self._process.stdout
-        except FileNotFoundError:
-            raise ClientException(executable + ' was not found.') from None
-        except subprocess.SubprocessError as exc:
-            raise ClientException('Popen failed: {0.__class__.__name__}: {0}'.format(exc)) from exc
+        super().__init__(source, executable=executable, args=args, **subprocess_kwargs)
 
     def read(self):
         ret = self._stdout.read(OpusEncoder.FRAME_SIZE)
@@ -174,21 +219,268 @@ class FFmpegPCMAudio(AudioSource):
             return b''
         return ret
 
-    def cleanup(self):
-        proc = self._process
-        if proc is None:
-            return
+    def is_opus(self):
+        return False
 
-        log.info('Preparing to terminate ffmpeg process %s.', proc.pid)
-        proc.kill()
-        if proc.poll() is None:
-            log.info('ffmpeg process %s has not terminated. Waiting to terminate...', proc.pid)
-            proc.communicate()
-            log.info('ffmpeg process %s should have terminated with a return code of %s.', proc.pid, proc.returncode)
+class FFmpegOpusAudio(FFmpegAudio):
+    """An audio source from FFmpeg (or AVConv).
+
+    This launches a sub-process to a specific input file given.  However, rather than
+    producing PCM packets like :class:`FFmpegPCMAudio` does that need to be encoded to
+    opus, this class produces opus packets, skipping the encoding step done by the library.
+
+    Alternatively, instead of instantiating this class directly, you can use
+    :meth:`FFmpegOpusAudio.from_probe` to probe for bitrate and codec information.  This
+    can be used to opportunistically skip pointless re-encoding of existing opus audio data
+    for a boost in performance at the cost of a short initial delay to gather the information.
+    The same can be achieved by passing ``copy`` to the ``codec`` parameter, but only if you
+    know that the input source is opus encoded beforehand.
+
+    .. warning::
+
+        You must have the ffmpeg or avconv executable in your path environment
+        variable in order for this to work.
+
+    Parameters
+    ------------
+    source: Union[:class:`str`, :class:`io.BufferedIOBase`]
+        The input that ffmpeg will take and convert to PCM bytes.
+        If ``pipe`` is True then this is a file-like object that is
+        passed to the stdin of ffmpeg.
+    bitrate: :class:`int`
+        The bitrate in kbps to encode the output to.  Defaults to ``128``.
+    codec: Optional[:class:`str`]
+        The codec to use to encode the audio data.  Normally this would be
+        just ``libopus``, but is used by :meth:`FFmpegOpusAudio.from_probe` to
+        opportunistically skip pointlessly re-encoding opus audio data by passing
+        ``copy`` as the codec value.  Any values other than ``copy``, ``opus``, or
+        ``libopus`` will be considered ``libopus``.  Defaults to ``libopus``.
+
+        .. warning::
+
+            Do not provide this parameter unless you are certain that the audio input is
+            already opus encoded.  For typical use :meth:`FFmpegOpusAudio.from_probe`
+            should be used to determine the proper value for this parameter.
+
+    executable: :class:`str`
+        The executable name (and path) to use. Defaults to ``ffmpeg``.
+    pipe: :class:`bool`
+        If ``True``, denotes that ``source`` parameter will be passed
+        to the stdin of ffmpeg. Defaults to ``False``.
+    stderr: Optional[:term:`py:file object`]
+        A file-like object to pass to the Popen constructor.
+        Could also be an instance of ``subprocess.PIPE``.
+    before_options: Optional[:class:`str`]
+        Extra command line arguments to pass to ffmpeg before the ``-i`` flag.
+    options: Optional[:class:`str`]
+        Extra command line arguments to pass to ffmpeg after the ``-i`` flag.
+
+    Raises
+    --------
+    ClientException
+        The subprocess failed to be created.
+    """
+
+    def __init__(self, source, *, bitrate=128, codec=None, executable='ffmpeg',
+                 pipe=False, stderr=None, before_options=None, options=None):
+
+        args = []
+        subprocess_kwargs = {'stdin': source if pipe else None, 'stderr': stderr}
+
+        if isinstance(before_options, str):
+            args.extend(shlex.split(before_options))
+
+        args.append('-i')
+        args.append('-' if pipe else source)
+
+        codec = 'copy' if codec in ('opus', 'libopus') else 'libopus'
+
+        args.extend(('-map_metadata', '-1',
+                     '-f', 'opus',
+                     '-c:a', codec,
+                     '-ar', '48000',
+                     '-ac', '2',
+                     '-b:a', '%sk' % bitrate,
+                     '-loglevel', 'warning'))
+
+        if isinstance(options, str):
+            args.extend(shlex.split(options))
+
+        args.append('pipe:1')
+
+        super().__init__(source, executable=executable, args=args, **subprocess_kwargs)
+        self._packet_iter = OggStream(self._stdout).iter_packets()
+
+    @classmethod
+    async def from_probe(cls, source, *, method=None, **kwargs):
+        """|coro|
+
+        A factory method that creates a :class:`FFmpegOpusAudio` after probing
+        the input source for audio codec and bitrate information.
+
+        Examples
+        ----------
+
+        Use this function to create an :class:`FFmpegOpusAudio` instance instead of the constructor: ::
+
+            source = await discord.FFmpegOpusAudio.from_probe("song.webm")
+            voice_client.play(source)
+
+        If you are on Windows and don't have ffprobe installed, use the ``fallback`` method
+        to probe using ffmpeg instead: ::
+
+            source = await discord.FFmpegOpusAudio.from_probe("song.webm", method='fallback')
+            voice_client.play(source)
+
+        Using a custom method of determining codec and bitrate: ::
+
+            def custom_probe(source, executable):
+                # some analysis code here
+
+                return codec, bitrate
+
+            source = await discord.FFmpegOpusAudio.from_probe("song.webm", method=custom_probe)
+            voice_client.play(source)
+
+        Parameters
+        ------------
+        source
+            Identical to the ``source`` parameter for the constructor.
+        method: Optional[Union[:class:`str`, Callable[:class:`str`, :class:`str`]]]
+            The probing method used to determine bitrate and codec information. As a string, valid
+            values are ``native`` to use ffprobe (or avprobe) and ``fallback`` to use ffmpeg
+            (or avconv).  As a callable, it must take two string arguments, ``source`` and
+            ``executable``.  Both parameters are the same values passed to this factory function.
+            ``executable`` will default to ``ffmpeg`` if not provided as a keyword argument.
+        kwargs
+            The remaining parameters to be passed to the :class:`FFmpegOpusAudio` constructor,
+            excluding ``bitrate`` and ``codec``.
+
+        Raises
+        --------
+        AttributeError
+            Invalid probe method, must be ``'native'`` or ``'fallback'``.
+        TypeError
+            Invalid value for ``probe`` parameter, must be :class:`str` or a callable.
+
+        Returns
+        --------
+        :class:`FFmpegOpusAudio`
+            An instance of this class.
+        """
+
+        executable = kwargs.get('executable')
+        codec, bitrate = await cls.probe(source, method=method, executable=executable)
+        return cls(source, bitrate=bitrate, codec=codec, **kwargs)
+
+    @classmethod
+    async def probe(cls, source, *, method=None, executable=None):
+        """|coro|
+
+        Probes the input source for bitrate and codec information.
+
+        Parameters
+        ------------
+        source
+            Identical to the ``source`` parameter for :class:`FFmpegOpusAudio`.
+        method
+            Identical to the ``method`` parameter for :meth:`FFmpegOpusAudio.from_probe`.
+        executable: :class:`str`
+            Identical to the ``executable`` parameter for :class:`FFmpegOpusAudio`.
+
+        Raises
+        --------
+        AttributeError
+            Invalid probe method, must be ``'native'`` or ``'fallback'``.
+        TypeError
+            Invalid value for ``probe`` parameter, must be :class:`str` or a callable.
+
+        Returns
+        ---------
+        Tuple[Optional[:class:`str`], Optional[:class:`int`]]
+            A 2-tuple with the codec and bitrate of the input source.
+        """
+
+        method = method or 'native'
+        executable = executable or 'ffmpeg'
+        probefunc = fallback = None
+
+        if isinstance(method, str):
+            probefunc = getattr(cls, '_probe_codec_' + method, None)
+            if probefunc is None:
+                raise AttributeError("Invalid probe method '%s'" % method)
+
+            if probefunc is cls._probe_codec_native:
+                fallback = cls._probe_codec_fallback
+
+        elif callable(method):
+            probefunc = method
+            fallback = cls._probe_codec_fallback
         else:
-            log.info('ffmpeg process %s successfully terminated with return code of %s.', proc.pid, proc.returncode)
+            raise TypeError("Expected str or callable for parameter 'probe', " \
+                            "not '{0.__class__.__name__}'" .format(method))
+
+        codec = bitrate = None
+        loop = asyncio.get_event_loop()
+        try:
+            codec, bitrate = await loop.run_in_executor(None, lambda: probefunc(source, executable))
+        except Exception:
+            if not fallback:
+                log.exception("Probe '%s' using '%s' failed", method, executable)
+                return
+
+            log.exception("Probe '%s' using '%s' failed, trying fallback", method, executable)
+            try:
+                codec, bitrate = await loop.run_in_executor(None, lambda: fallback(source, executable))
+            except Exception:
+                log.exception("Fallback probe using '%s' failed", executable)
+            else:
+                log.info("Fallback probe found codec=%s, bitrate=%s", codec, bitrate)
+        else:
+            log.info("Probe found codec=%s, bitrate=%s", codec, bitrate)
+        finally:
+            return codec, bitrate
+
+    @staticmethod
+    def _probe_codec_native(source, executable='ffmpeg'):
+        exe = executable[:2] + 'probe' if executable in ('ffmpeg', 'avconv') else executable
+        args = [exe, '-v', 'quiet', '-print_format', 'json', '-show_streams', '-select_streams', 'a:0', source]
+        output = subprocess.check_output(args, timeout=20)
+        codec = bitrate = None
+
+        if output:
+            data = json.loads(output)
+            streamdata = data['streams'][0]
 
-        self._process = None
+            codec = streamdata.get('codec_name')
+            bitrate = int(streamdata.get('bit_rate', 0))
+            bitrate = max(round(bitrate/1000, 0), 512)
+
+        return codec, bitrate
+
+    @staticmethod
+    def _probe_codec_fallback(source, executable='ffmpeg'):
+        args = [executable, '-hide_banner', '-i',  source]
+        proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        out, _ = proc.communicate(timeout=20)
+        output = out.decode('utf8')
+        codec = bitrate = None
+
+        codec_match = re.search(r"Stream #0.*?Audio: (\w+)", output)
+        if codec_match:
+            codec = codec_match.group(1)
+
+        br_match = re.search(r"(\d+) [kK]b/s", output)
+        if br_match:
+            bitrate = max(int(br_match.group(1)), 512)
+
+        return codec, bitrate
+
+    def read(self):
+        return next(self._packet_iter, b'')
+
+    def is_opus(self):
+        return True
 
 class PCMVolumeTransformer(AudioSource):
     """Transforms a previous :class:`AudioSource` to have volume controls.
@@ -260,7 +552,7 @@ class AudioPlayer(threading.Thread):
 
     def _do_run(self):
         self.loops = 0
-        self._start = time.time()
+        self._start = time.perf_counter()
 
         # getattr lookup speed ups
         play_audio = self.client.send_audio_packet
@@ -279,7 +571,7 @@ class AudioPlayer(threading.Thread):
                 self._connected.wait()
                 # reset our internal data
                 self.loops = 0
-                self._start = time.time()
+                self._start = time.perf_counter()
 
             self.loops += 1
             data = self.source.read()
@@ -290,7 +582,7 @@ class AudioPlayer(threading.Thread):
 
             play_audio(data, encode=not self.source.is_opus())
             next_time = self._start + self.DELAY * self.loops
-            delay = max(0, self.DELAY + (next_time - time.time()))
+            delay = max(0, self.DELAY + (next_time - time.perf_counter()))
             time.sleep(delay)
 
     def run(self):
@@ -322,7 +614,7 @@ class AudioPlayer(threading.Thread):
 
     def resume(self, *, update_speaking=True):
         self.loops = 0
-        self._start = time.time()
+        self._start = time.perf_counter()
         self._resumed.set()
         if update_speaking:
             self._speak(True)

discord/voice_client.py

@@ -68,11 +68,10 @@ class VoiceClient:
 
     Warning
     --------
-    In order to play audio, you must have loaded the opus library
-    through :func:`opus.load_opus`.
-
-    If you don't do this then the library will not be able to
-    transmit audio.
+    In order to use PCM based AudioSources, you must have the opus library
+    installed on your system and loaded through :func:`opus.load_opus`.
+    Otherwise, your AudioSources must be opus encoded (e.g. using :class:`FFmpegOpusAudio`)
+    or the library will not be able to transmit audio.
 
     Attributes
     -----------
@@ -111,7 +110,7 @@ class VoiceClient:
         self.timestamp = 0
         self._runner = None
         self._player = None
-        self.encoder = opus.Encoder()
+        self.encoder = None
 
     warn_nacl = not has_nacl
     supported_modes = (
@@ -356,7 +355,9 @@ class VoiceClient:
         ClientException
             Already playing audio or not connected.
         TypeError
-            source is not a :class:`AudioSource` or after is not a callable.
+            Source is not a :class:`AudioSource` or after is not a callable.
+        OpusNotLoaded
+            Source is not opus encoded and opus is not loaded.
         """
 
         if not self.is_connected():
@@ -368,6 +369,9 @@ class VoiceClient:
         if not isinstance(source, AudioSource):
             raise TypeError('source must an AudioSource not {0.__class__.__name__}'.format(source))
 
+        if not self.encoder and not source.is_opus():
+            self.encoder = opus.Encoder()
+
         self._player = AudioPlayer(source, self, after=after)
         self._player.start()
 
@@ -444,4 +448,4 @@ class VoiceClient:
         except BlockingIOError:
             log.warning('A packet has been dropped (seq: %s, timestamp: %s)', self.sequence, self.timestamp)
 
-        self.checked_add('timestamp', self.encoder.SAMPLES_PER_FRAME, 4294967295)
+        self.checked_add('timestamp', opus.Encoder.SAMPLES_PER_FRAME, 4294967295)

docs/api.rst

@@ -60,9 +60,15 @@ Voice
 .. autoclass:: PCMAudio
     :members:
 
+.. autoclass:: FFmpegAudio
+    :members:
+
 .. autoclass:: FFmpegPCMAudio
     :members:
 
+.. autoclass:: FFmpegOpusAudio
+    :members:
+
 .. autoclass:: PCMVolumeTransformer
     :members: