projects-jenz/torchlight_changes_unloze/torchlight3/Torchlight/FFmpegAudioPlayer.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import logging
import traceback
import asyncio
import datetime
import time
import socket
import struct
import sys

SAMPLEBYTES = 2

class FFmpegAudioPlayerFactory():
    VALID_CALLBACKS = ["Play", "Stop", "Update"]

    def __init__(self, master):
        self.Logger = logging.getLogger(__class__.__name__)
        self.Master = master
        self.Torchlight = self.Master.Torchlight

    def __del__(self):
        self.Master.Logger.info("~FFmpegAudioPlayerFactory()")
        self.Quit()

    def NewPlayer(self):
        self.Logger.debug(sys._getframe().f_code.co_name)
        Player = FFmpegAudioPlayer(self)
        return Player

    def Quit(self):
        self.Master.Logger.info("FFmpegAudioPlayerFactory->Quit()")


class FFmpegAudioPlayer():
    def __init__(self, master):
        self.Master = master
        self.Torchlight = self.Master.Torchlight
        self.Playing = False

        self.Host = (
            self.Torchlight().Config["VoiceServer"]["Host"],
            self.Torchlight().Config["VoiceServer"]["Port"]
        )
        self.SampleRate = float(self.Torchlight().Config["VoiceServer"]["SampleRate"])

        self.StartedPlaying = None
        self.StoppedPlaying = None
        self.Seconds = 0.0

        self.Writer = None
        self.Process = None

        self.Callbacks = []

    def __del__(self):
        self.Master.Logger.debug("~FFmpegAudioPlayer()")
        self.Stop()

    def PlayURI(self, uri, position, rubberband = None, dec_params = None, bitrate = None, 
            backwards = None, *args):
        if position:
            PosStr = str(datetime.timedelta(seconds = position))
            Command = ["/usr/bin/ffmpeg", "-ss", PosStr, "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]
        else:
            Command = ["/usr/bin/ffmpeg", "-i", uri, "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(int(self.SampleRate)), "-f", "s16le", "-vn", *args]

        self.Playing = True
        if dec_params:
            Command += dec_params

        if rubberband and backwards:
            Command += ["-filter:a"]
            rubberCommand = ""
            for rubber in rubberband:
                rubberCommand += rubber + ", "
            rubberCommand = rubberCommand[:-2]
            Command += [rubberCommand + "[reversed];[reversed]areverse"] #[reversed] is intermediate stream label so reverse knows what stream label to reverse
        else:
            if rubberband:
                Command += ["-filter:a"]
                rubberCommand = ""
                for rubber in rubberband:
                    rubberCommand += rubber + ", "
                rubberCommand = rubberCommand[:-2]
                Command += [rubberCommand]
            if backwards:
                Command += ["-af"]
                Command += ["areverse"]
        if bitrate:
            Command += ["-ab ", str(bitrate), "k"]
            self.Master.Logger.debug(f"command: {Command}")
        Command += ["-"]
        #self.Master.Logger.debug(f"command: {Command}")
        asyncio.ensure_future(self._stream_subprocess(Command))
        return True

    def Stop(self, force = True):
        if not self.Playing:
            return False

        if self.Process:
            try:
                self.Process.terminate()
                self.Process.kill()
                self.Process = None
            except ProcessLookupError:
                pass

        if self.Writer:
            if force:
                Socket = self.Writer.transport.get_extra_info("socket")
                if Socket:
                    Socket.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
                        struct.pack("ii", 1, 0))

                self.Writer.transport.abort()

            self.Writer.close()

        self.Playing = False

        self.Callback("Stop")
        del self.Callbacks

        return True

    def AddCallback(self, cbtype, cbfunc):
        if not cbtype in FFmpegAudioPlayerFactory.VALID_CALLBACKS:
            return False

        self.Callbacks.append((cbtype, cbfunc))
        return True

    def Callback(self, cbtype, *args, **kwargs):
        for Callback in self.Callbacks:
            if Callback[0] == cbtype:
                try:
                    Callback[1](*args, **kwargs)
                except Exception as e:
                    self.Master.Logger.error(traceback.format_exc())

    async def _updater(self):
        LastSecondsElapsed = 0.0

        while self.Playing:
            SecondsElapsed = time.time() - self.StartedPlaying

            if SecondsElapsed > self.Seconds:
                SecondsElapsed = self.Seconds

            self.Callback("Update", LastSecondsElapsed, SecondsElapsed)

            if SecondsElapsed >= self.Seconds:
                if not self.StoppedPlaying:
                    print("BUFFER UNDERRUN!")
                self.Stop(False)
                return

            LastSecondsElapsed = SecondsElapsed

            await asyncio.sleep(0.1)

    async def _read_stream(self, stream, writer):
        Started = False

        while stream and self.Playing:
            Data = await stream.read(65536)

            if Data:
                writer.write(Data)
                await writer.drain()

                Bytes = len(Data)
                Samples = Bytes / SAMPLEBYTES
                Seconds = Samples / self.SampleRate

                self.Seconds += Seconds

                if not Started:
                    Started = True
                    self.Callback("Play")
                    self.StartedPlaying = time.time()
                    asyncio.ensure_future(self._updater())
            else:
                self.Process = None
                break

        self.StoppedPlaying = time.time()

    async def _stream_subprocess(self, cmd):
        if not self.Playing:
            return

        _, self.Writer = await asyncio.open_connection(self.Host[0], self.Host[1])

        Process = await asyncio.create_subprocess_exec(*cmd,
                stdout = asyncio.subprocess.PIPE, stderr = asyncio.subprocess.DEVNULL)
        self.Process = Process

        await self._read_stream(Process.stdout, self.Writer)
        await Process.wait()

        if self.Seconds == 0.0:
            self.Stop()