buffer.py

@@ -1,55 +0,0 @@
-from stream import ByteSource
-from queue import Queue
-from threading import Lock
-from time import sleep
-from stream_listener import StreamListener
-
-# a simple FIFO buffer for ensuring stream stability
-class Buffer(ByteSource):
-
-  def __init__(self, upstream, size=200000):
-    self.fifo = Queue()
-    self.targetSize = size
-    self.size = 0
-    self.quit = False
-    self.mutex = Lock()
-    self.listener = StreamListener(upstream, self)
-    self.listener.start()
-
-  def stop(self):
-    self.quit = True
-    self.listener.stop()
-
-  def peek(self):
-    self.mutex.acquire()
-    try:
-      if self.size > 0 and not self.quit:
-        return self.fifo.queue[0]
-      else:
-        return None
-    finally:
-      self.mutex.release()
-
-  def read(self):
-    self.mutex.acquire()
-    try:
-      if self.size > 0 and not self.quit:
-        out = self.fifo.get_nowait()
-        self.size -= len(out)
-        return out
-      else:
-        return None
-    finally:
-      self.mutex.release()
-
-  def write(self, chunk):
-    while not self.quit:
-      try:
-        self.mutex.acquire()
-        if self.size < self.targetSize:
-          self.size += len(chunk)
-          self.fifo.put(chunk)
-          break
-      finally:
-        self.mutex.release()
-      sleep(0.1)

default.nix

@@ -4,5 +4,5 @@ pkgs.python3Packages.buildPythonApplication {
   pname = "radio";
   src = self;
   version = "0.1";
-  propagatedBuildInputs = with pkgs.python3Packages; [ pip ffmpeg-python flask requests pkgs.ffmpeg libtorrent-rasterbar ];
+  propagatedBuildInputs = with pkgs.python3Packages; [ pip ffmpeg-python flask requests pkgs.ffmpeg ];
 }

downloader.py

@@ -1,100 +1,60 @@
 #
-# Downloads the video/audio as a stream from a provided link using yt-dlp
+# Downloads the video/audio as a stream from a provided link using youtube-dl
 # does not save the file, only the most recent fragment is held. Thus, this is
 # ideal for devices with little memory
+# TODO gather video metadata before download
 #
 
-import ffmpeg
 import tempfile
 import sys
 import subprocess
 import os
 import pip
 import signal
-import json
-import shutil
-import glob
 from logger import logger
 from threading import Thread, main_thread
 from time import sleep
-from stream import StreamSource
 
-def updateYtdlp():
+def updateYoutubeDL():
-  pip.main(['install', '--target=' + dirpath, '--upgrade', 'yt-dlp'])
+  pip.main(['install', '--target=' + dirpath, '--upgrade', 'youtube_dl'])
 
 def importYoutubeDL():
-  return __import__('yt-dlp')
+  return __import__('youtube_dl')
 
 dirpath = tempfile.mkdtemp()
 sys.path.append(dirpath)
-updateYtdlp()
+updateYoutubeDL()
 
-baseOpts = [
+class Downloader(Thread):
-  "-f", "bestaudio/best", # select for best audio
-  # "--audio-format", "mp3", "-x", # cannot do because it requires a tmp file to re-encode
-  "--prefer-ffmpeg",
-  "--no-mark-watched",
-  "--geo-bypass",
-  "--no-playlist",
-  "--retries", "100",
-  "--extractor-retries", "100",
-#      "--throttled-rate", "100K", # get around youtube throttling; probably not needed anymore
-  "--no-call-home",
-  "--sponsorblock-remove", "sponsor,intro,selfpromo,interaction,preview,music_offtopic",
-]
 
-def runYtdlp(opts):
-  # update yt-dlp
-  # TODO: do this only every once in a while
-  # updateYtdlp()
-
-  env = dict(os.environ)
-  env["PYTHONPATH"] = dirpath
-  cmd = [
-    sys.executable,
-    dirpath + "/bin/yt-dlp"
-  ] + baseOpts + opts
-  popen = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, preexec_fn=os.setsid)
-  logger.add(popen.stderr, "yt-dlp.log")
-  return popen
-
-def getVideoInfo(url):
-  popen = runYtdlp([
-    "-j", # dump all metadata as json
-    "--playlist-items", "1", # don't go through every single item in playlist
-    url
-  ])
-
-  j = popen.communicate()[0]
-
-  # make sure the ytdlp instance is closed
-  if popen.poll() is None: # is alive?
-    os.killpg(os.getpgid(popen.pid), signal.SIGTERM) # kill
-  popen.stdout.close()
-
-  # keys of interest (but not always present)
-  # title, direct, extractor, upload_date
-  # format_id (machine readable)
-  # view_count, uploader, duration, view_count, average_rating
-  # age_limit, like_count, dislike_count, thumbnail,
-
-  try:
-    return json.loads(j)
-  except:
-    return None
-
-# Downloads using yt-dlp
-class YtdlpDownloader(Thread, StreamSource):
   def __init__(self, url, cb):
     Thread.__init__(self)
 
+    # update youtube-dl
+    # TODO: do this only every once in a while
+    # updateYoutubeDL()
+
     self.cb = cb
     self.exit = False
 
-    self.popen = runYtdlp([
+    env = dict(os.environ)
-      "-o", "-", # output stream to stdout
+    env["PYTHONPATH"] = dirpath
+    cmd = [
+      sys.executable,
+      dirpath + "/bin/youtube-dl",
+      "-o", "-",
+      "-f", "bestaudio/best",
+      # "--audio-format", "mp3", "-x", # cannot do because it requires a tmp file to re-encode
+      "--prefer-ffmpeg",
+      "--no-mark-watched",
+      "--geo-bypass",
+      "--no-playlist",
+      "--retries", "100",
+      "--no-call-home",
       url
-    ])
+    ]
+    self.popen = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, preexec_fn=os.setsid)
+    logger.add(self.popen.stderr, "youtube-dl.log")
     self.start()
 
   def isAlive(self):
@@ -104,6 +64,7 @@ class YtdlpDownloader(Thread, StreamSource):
     if self.isAlive():
       os.killpg(os.getpgid(self.popen.pid), signal.SIGTERM)
     self.popen.stdout.close()
+    # self.popen.wait()
     self.exit = True
 
   # checks to see if the current download has finished
@@ -115,40 +76,4 @@ class YtdlpDownloader(Thread, StreamSource):
       sleep(0.1)
 
   def getStream(self):
-    return self.popen.stdout
+    return self.popen.stdout
-
-# Downloads the file and plays it (must be a direct URL, ex: cannot be a yt link)
-class DirectDownloader(Thread, StreamSource):
-  def __init__(self, url, cb):
-    Thread.__init__(self)
-
-    self.cb = cb
-    self.exit = False
-
-    self.process = ( ffmpeg
-      .input(url, re=None)
-      .output('pipe:', format='mp3')
-      .run_async(pipe_stdout=True, pipe_stderr=True)
-    )
-    logger.add(self.process.stderr, "direct-downloader.log")
-
-    self.start()
-
-  def isAlive(self):
-    return self.process.poll() is None
-
-  # checks to see if the current download has finished
-  # if yes, cleans up and fires callback
-  def run(self):
-    while main_thread().is_alive() and not self.exit:
-      if not self.isAlive():
-        self.cb()
-      sleep(0.1)
-
-  def stop(self):
-    self.process.stdout.close()
-    self.process.stderr.close()
-    self.exit = True
-
-  def getStream(self):
-    return self.process.stdout

logger.py

@@ -37,9 +37,9 @@ class Logger(Thread):
             output = non_block_read(stream)
             if output == None or output == b'':
               break
+            print(output.decode('ascii'))
             f.write(output)
             f.flush()
-            print(output.decode('utf-8'))
       finally:
         self.mutex.release()
       sleep(0.1)

nullsrc.py

@@ -1,10 +1,9 @@
 import ffmpeg
 from logger import logger
-from stream import StreamSource
 
 # A null audio source
 
-class NullSrc(StreamSource):
+class NullSrc(object):
 
   def __init__(self):
     self.process = ( ffmpeg
@@ -17,6 +16,7 @@ class NullSrc(StreamSource):
   def stop(self):
     self.process.stdout.close()
     self.process.stderr.close()
+    # self.process.wait()
 
   def getStream(self):
     return self.process.stdout

radio.py

@@ -1,25 +1,21 @@
 import downloader
 import uploader
 import transcoder
-import torrent
-import buffer
 from time import sleep
 from flask import Flask, request
 from queue import Queue
 import json
 import requests
-import sys
 
 app = Flask(__name__)
 
-host = sys.argv[1]
+host = "localhost:8000"
-stream = sys.argv[2]
+stream = "stream.mp3"
 
 class Radio(object):
   def __init__(self):
     self.downloader = None
     self.transcoder = None
-    self.buffer = None
     self.uploader = uploader.Uploader(host, stream)
     self.playingUrl = None
     self.queue = Queue()
@@ -27,34 +23,9 @@ class Radio(object):
   # plays the next song in the queue 
   def play(self):
     self.playingUrl = self.queue.get()
-
+    self.downloader = downloader.Downloader(self.playingUrl, self.downloadFinished)
-    # determine what downloader needs to be used and create the downloader
-    if self.playingUrl.startswith("magnet:?"):
-      # it's a torrent
-      torrent.mountTorrent(self.playingUrl)
-      self.playingUrl = torrent.getTorrentMedia()
-      if self.playingUrl is None:
-        # this torrent is unplayable, skip it
-        torrent.umountTorrent()
-        return self.play()
-      else:
-        self.downloader = downloader.DirectDownloader(self.playingUrl, self.downloadFinished)
-    else:
-      # assume http/https
-      info = downloader.getVideoInfo(self.playingUrl)
-      if info is None:
-        # this url is unplayable, skip it
-        return self.play()
-      elif ("direct" in info and info["direct"] == True) or ("format_id" in info and info["format_id"] == "rtmp"): # stdout for rtmp in ytdl is broken
-        # direct source
-        self.downloader = downloader.DirectDownloader(self.playingUrl, self.downloadFinished)
-      else:
-        # requires youtube-dl
-        self.downloader = downloader.YtdlpDownloader(self.playingUrl, self.downloadFinished)
-
     self.transcoder = transcoder.Transcoder(self.downloader)
-    self.buffer = buffer.Buffer(self.transcoder)
+    self.uploader.setUpstream(self.transcoder)
-    self.uploader.setUpstream(self.buffer)
 
   def isPlaying(self):
     return not self.playingUrl is None
@@ -77,9 +48,7 @@ class Radio(object):
   def stopPlaying(self):
     self.downloader.stop()
     self.transcoder.stop()
-    self.buffer.stop()
     self.playingUrl = None
-    torrent.umountTorrent() # make sure torrent is unmounted
 
   # downloader callback function, called when the downloader is finished
   # but may still have bytes left that need to be read and played
@@ -120,7 +89,7 @@ def listeners():
   return str(r.listenerCount())
 
 def run():
-  app.run(host="0.0.0.0", port=int(sys.argv[3]))
+  app.run(host="0.0.0.0")
 
 if __name__ == "__main__":
   run()

setup.py

@@ -13,10 +13,7 @@ setup(
     'util',
     'stream_listener',
     'transcoder',
-    'nullsrc',
+    'nullsrc'
-    'buffer',
-    'stream',
-    'torrent'
     ],
   entry_points={
     'console_scripts': ['radio = radio:run']

stream.py

@@ -1,46 +0,0 @@
-from util import non_block_read, non_block_peek
-
-class StreamSource(object):
-  def getStream(self):
-    """ Gets the python stream for reading from """
-    pass
-
-class ByteSource(object):
-  def peek(self):
-    """ Gets the bytes at the head of the queue (if any) without incrementing
-    Non-blocking!
-    Returns None if no bytes available
-    """
-    pass
-  def read(self):
-    """ Gets the bytes at the head of the queue (if any)
-    Non-blocking!
-    Returns None if no bytes available
-    """
-    pass
-
-class StreamByteSource(ByteSource):
-  def __init__(self, stream):
-    self.stream = stream
-
-  def peek(self):
-    output = non_block_peek(self.stream)
-    if output == None or output == b'':
-      return None
-    return output
-
-  def read(self):
-    output = non_block_read(self.stream)
-    if output == None or output == b'':
-      return None
-    return output
-
-def toByteSource(src):
-  if isinstance(src, ByteSource):
-    return src
-  elif isinstance(src, StreamSource):
-    return StreamByteSource(src.getStream())
-  elif src is None:
-    return None
-  else:
-    raise Exception("not implemented")

stream_listener.py

@@ -1,22 +1,25 @@
 from threading import Thread, main_thread
 from util import non_block_read, non_block_peek
 from time import sleep, time
-from stream import toByteSource
 
 # time (ms) to wait before switching over to backup source 
-FALLBACK_TIME = 100
+FALLBACK_TIME = 500
 
 def current_milli_time():
   return round(time() * 1000)
 
+def empty(output):
+  return output == None or output == b''
+
 #
 # In a new thread, listens for new data on the stream
 # and passes it to the listener when available
 #
-# Upstreams implement class StreamSource or ByteSource
+# Upstreams implement getStream()
 # Downstreams implement write() for bytes to be given to
 #
 class StreamListener(Thread):
+  
   def __init__(self, upstream, downstream, backupUpstream=None):
     Thread.__init__(self)
     self.setUpstream(upstream)
@@ -26,20 +29,26 @@ class StreamListener(Thread):
     self.lastData = 0
 
   def setUpstream(self, upstream):
-    self.src = toByteSource(upstream)
+    if not upstream is None:
+      self.stream = upstream.getStream()
+    else:
+      self.stream = None
 
   def setBackupUpstream(self, upstream):
-    self.backupSrc = toByteSource(upstream)
+    if not upstream is None:
+      self.backupStream = upstream.getStream()
+    else:
+      self.backupStream = None
 
   def setDownstream(self, downstream):
     self.listener = downstream
 
   # blocks until there is nothing left to read from upstream
   def blockUntilEmpty(self):
-    if not self.src is None:
+    if not self.stream is None:
       while True:
-        output = self.src.peek()
+        output = non_block_peek(self.stream)
-        if output == None:
+        if output == None or output == b'':
           break
         sleep(0.1)
 
@@ -50,13 +59,13 @@ class StreamListener(Thread):
     while main_thread().is_alive() and not self.quit:
       while True:
         output = None
-        if not self.src is None:
+        if not self.stream is None:
-          output = self.src.read()
+          output = non_block_read(self.stream)
-        if output == None and not self.backupSrc is None and current_milli_time() - self.lastData > FALLBACK_TIME:
+        if empty(output) and not self.backupStream is None and current_milli_time() - self.lastData > FALLBACK_TIME:
-          output = self.backupSrc.read()
+          output = non_block_read(self.backupStream)
         else:
           self.lastData = current_milli_time()
-        if output == None:
+        if empty(output):
           break
         self.listener.write(output)
       sleep(0.1)

torrent.py

@@ -1,325 +0,0 @@
-from urllib.parse import quote
-import tempfile
-import os
-import mimetypes
-from random import randint
-from time import sleep
-
-import libtorrent as lt
-
-mimetypes.init()
-STATUSES = [
-  'queued', 'checking', 'downloading_metadata', 'downloading', 'finished',
-  'seeding', 'allocating', 'checking_fastresume'
-]
-
-TRACKERS = (
-  "udp://tracker.openbittorrent.com:80/announce",
-  "udp://tracker.publicbt.com:80/announce"
-)
-
-DHT = (
-  ("router.utorrent.com", 6881),
-  ("router.bittorrent.com", 6881),
-  ("dht.transmissionbt.com", 6881),
-  ("router.bitcomet.com", 6881),
-  ("dht.aelitis.com", 6881)
-)
-
-EXTENSIONS = ('ut_pex', 'ut_metadata', 'smart_ban', 'metadata_transfoer')
-
-PORTS = (randint(1024, 2000), randint(1024, 2000))
-
-def get_indexed(func):
-  """Return currently indedex torrent"""
-  def inner(*args, **kwargs):
-    """Executes a method, and returns result[class_instance.index]"""
-    return list(func(*args, **kwargs)())[args[0].index]
-  return inner
-
-class TorrentSession:
-  """Represent a torrent session. May handle multiple torrents"""
-  def __init__(self, ports=PORTS, extensions=EXTENSIONS, dht_routers=DHT):
-    self.session = lt.session()
-    # self.session.set_severity_level(lt.alert.severity_levels.critical)
-    self.session.listen_on(*ports)
-    for extension in extensions:
-      self.session.add_extension(extension)
-    self.session.start_dht()
-    self.session.start_lsd()
-    self.session.start_upnp()
-    self.session.start_natpmp()
-    for router in dht_routers:
-      self.session.add_dht_router(*router)
-    self.torrents = []
-
-  def remove_torrent(self, *args, **kwargs):
-    """Remove torrent from session."""
-    self.session.remove_torrent(*args, **kwargs)
-
-  def add_torrent(self, **kwargs):
-    """Add a torrent to this session
-
-    For accepted parameters reference, see over `Torrent` definition.
-    """
-    torrent = Torrent(session=self, **kwargs)
-    self.torrents.append(torrent)
-    return torrent
-
-  def __iter__(self):
-    """Iterating trough a session will give you all the currently-downloading torrents"""
-    return iter(self.torrents)
-
-class Torrent:
-  """Wrapper over libtorrent"""
-  def __init__(self, magnet_link: str, session: TorrentSession, trackers: tuple = TRACKERS,
-              remove_after: bool = False, **params):
-    """Set default parameters to a magnet link, and add ourselves to a session
-
-    Arguments:
-      magnet_link: Magnet link. Currently torrent files are not supported
-      session: TorrentSession instance
-      trackers: Tracker list to add to magnet link. Defaults to TRACKERS
-                constant
-      remove_after: Delete download dir upon __exit__. Only if params.save_path has not been specified
-      save_path: Path to save the torrent into. A temporary directory
-                  will be created if not specified
-      storage_mode: Property of lt.storage_mode_t
-    """
-    self.session = session
-    self.temp_dir = None
-    self.remove_after = remove_after
-    self.params = {
-      'save_path': None,
-      'storage_mode': lt.storage_mode_t.storage_mode_sparse,
-      **params
-    }
-
-    #: Force trackers into magnet link. Not the best coding practice.
-    trackers = (quote(t, safe='') for t in trackers)
-    self.magnet_link = f'{magnet_link}&tr={"&tr=".join(trackers)}'
-    self.handle = None
-
-  def __enter__(self):
-    if not self.params.get('save_path'):
-      self.temp_dir = tempfile.TemporaryDirectory()
-      self.params['save_path'] = self.temp_dir.name
-    self.handle = lt.add_magnet_uri(self.session.session, self.magnet_link, self.params)
-    return self
-
-  def __exit__(self, *args, **kwargs):
-    if self.temp_dir and self.remove_after:
-      self.temp_dir.cleanup()
-    self.session.remove_torrent(self.handle)
-
-  def sequential(self, value: bool):
-    """Set sequential download"""
-    self.handle.set_sequential_download(value)
-
-  @property
-  def queue(self):
-    """ Download queue """
-    return self.handle.get_download_queue()
-
-  @property
-  def queue_status(self):
-    """ Returns a represented queue status """
-    state_char = [' ', '-', '=', '#']
-
-    def repr_piece(piece):
-      """ Represents a piece """
-      return {
-        piece['piece_index']:
-        [state_char[block['state']] for block in piece['blocks']]
-      }
-
-    return [repr_piece(piece) for piece in self.queue]
-
-  @property
-  def name(self):
-    """ Torrent name """
-    if not self.handle.has_metadata():
-      return "N/A"
-    return self.torrent_info.name()
-
-  @property
-  def status(self):
-    """
-      Return a status dict.
-    """
-    status = self.handle.status()
-    result = {
-      'name': self.name,
-      'download': status.download_rate,
-      'total_download': status.total_download,
-      'upload': status.upload_rate,
-      'total_upload': status.total_upload
-    }
-
-    if not self.finished:
-      result.update({
-        'state': STATUSES[status.state],
-        'total_downloaded': status.total_done,
-        'peers': status.num_peers,
-        'seeds': status.num_seeds,
-        'progress': '%5.4f%%' % (status.progress * 100),
-      })
-
-    return result
-
-  @property
-  def finished(self):
-    """Checks if torrent is finished."""
-    return self.handle.is_finished()
-
-  @property
-  def started(self):
-    """ Checks if handle has metadata"""
-    return self.handle.has_metadata()
-
-  @property
-  def torrent_info(self):
-    """Return handle.torrent_info"""
-    return self.handle.get_torrent_info()
-
-  @property
-  def files(self):
-    """Returns a `TorrentFile` object for each file"""
-    fnum = range(len(self.torrent_info.files()))
-    return [TorrentFile(self, i) for i in fnum]
-
-  def update_priorities(self):
-    """Update file priorities with self.files."""
-    self.handle.prioritize_files([a.priority for a in self.files])
-
-  def download_only(self, file):
-    """ Filter out priorities for every file except this one"""
-    if file not in self.files:
-      return None
-    for file_ in self.files:
-      file.priority = 7 if file == file_ else 0
-    return file
-
-  def wait_for(self, status):
-    """Wait for a specific status
-
-    Example:
-      >>> # This will wait for a torrent to start, and return the torrent
-      >>> torrent = await Torrent("magnet:...").wait_for('started')
-
-      >>> # This will wait for a torrent to finish, and return the torrent
-      >>> torrent = await Torrent("magnet:...").wait_for('finished')
-    """
-    while not getattr(self, status):
-      sleep(0.5)
-
-  def __iter__(self):
-    """Iterating trough a Torrent instance will return each TorrentFile"""
-    return iter(self.files)
-
-class TorrentFile:
-  """ Wrapper over libtorrent.file """
-  def __init__(self, parent: Torrent, index: int):
-    self.root = parent.params.get('save_path')
-    self.index = index
-    self.handle = parent.handle
-    self.torrent = parent
-
-  def wait_for_completion(self, percent):
-    while self.completed_percent < percent:
-      sleep(1)
-
-  @property
-  def full_path(self):
-    return f'{self.root}/{self.path}'
-
-  @property
-  def mime_type(self):
-    """Return file mimetype"""
-    return mimetypes.guess_type(self.path)[0] or ''
-
-  @property
-  def is_media(self):
-    """Return true if file is a media type"""
-    return any(self.mime_type.startswith(f) for f in ('audio', 'video'))
-
-  @property
-  def path(self):
-    """Return torrent path on filesystem"""
-    return self.hfile.path
-
-  @property
-  def file(self):
-    """Return a file object with this file's path open in rb mode """
-    return open(self.path, 'rb')
-
-  @property
-  def filehash(self):
-    """File hash"""
-    return self.hfile.filehash
-
-  @property
-  def size(self):
-    """File size"""
-    return self.hfile.size
-
-  @property
-  @get_indexed
-  def hfile(self):
-    """ Return file from libtorrent """
-    return self.handle.get_torrent_info().files
-
-  @property
-  @get_indexed
-  def priority(self):
-    """ Readonly file priority from libtorrent """
-    return self.handle.file_priorities
-
-  @priority.setter
-  def priority(self, value):
-    self._priority = value
-    self.parent.update_priorities()
-
-  # @property
-  # @get_indexed
-  # def file_progress(self):
-  #   """ Returns file progress """
-  #   return self.handle.file_progress
-
-  @property
-  def completed_percent(self):
-    """ Returns this file completed percentage """
-    # return (self.file_progress / self.size) * 100
-    return self.handle.status().progress * 100
-
-session = TorrentSession()
-torrent = None
-
-def mountTorrent(magnet):
-  global session
-  global torrent
-  print("mounting torrent...")
-  torrent = session.add_torrent(magnet_link=magnet, remove_after=True)
-  torrent.__enter__()
-  torrent.sequential(True)
-  # Wait for torrent to be started
-  torrent.wait_for('started')
-
-def umountTorrent():
-  global session
-  global torrent
-  print("unmounting torrent...")
-  torrent.__exit__()
-
-def getTorrentMedia():
-  global session
-  global torrent
-  print("Getting torrent media...")
-  # Get first match of a media file
-  try:
-    media = next(a for a in torrent if a.is_media and not 'sample' in a.path.lower())
-  except StopIteration:
-    return None # no playable media
-  # Wait for 5% download
-  media.wait_for_completion(5)
-  return media.full_path

transcoder.py

@@ -1,13 +1,9 @@
 import ffmpeg
 from logger import logger
 from stream_listener import StreamListener
-from stream import StreamSource
-import os
 
 # converts the stream to mp3 before sending to the long-lasting mp3 connection
-# write to a fifo file instead of directly via stdin to fool ffmpeg into thinking it has seekable input
+class Transcoder(object):
-# https://lists.libav.org/pipermail/ffmpeg-user/2007-May/008917.html
-class Transcoder(StreamSource):
 
   def __init__(self, upstream):
     self.process = ( ffmpeg
@@ -24,6 +20,7 @@ class Transcoder(StreamSource):
     self.process.stdin.close()
     self.process.stdout.close()
     self.process.stderr.close()
+    # self.process.wait()
 
   def getStream(self):
     return self.process.stdout

uploader.py

@@ -31,10 +31,9 @@ class Uploader(object):
 
   def stop(self):
     self.listener.stop()
-    if not self.process.stdin.closed:
+    self.process.stdin.close()
-      self.process.stdin.close()
+    self.process.stderr.close()
-    if not self.process.stderr.closed:
+    self.process.wait()
-      self.process.stderr.close()
 
   def write(self, chunk):
     while True: