From 7275dfd272a2402abd4d6358a13ebd0c2993f395 Mon Sep 17 00:00:00 2001 From: Dolly132 <109222243+Dolly132@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:38:44 +0300 Subject: [PATCH 1/3] refactor(stream): Use aiohttp instead of curl **Summary** Well, YouTube is so strict sometimes, so it blocks buffer from actually being sent to the ffmpeg process in time, and we can't control that, however, a 15 seconds timeout was set, youtube usually takes more than 5 seconds to send buffers in time. **Why aiohttp?** I had to get rid of curl because handling it is harder, aiohttp is way easier and more reliable for requests like YouTube and MyInstants. - Solves https://github.com/srcdslab/torchlight/issues/119 - Now local files don't need a curl process to be made, it can be done inside ffmpeg itself - Set the duration of the youtube video before playing it instead of calculating it during the stream. --- src/torchlight/AudioClip.py | 3 + src/torchlight/Commands.py | 3 + src/torchlight/FFmpegAudioPlayer.py | 273 +++++++++++++++------------- 3 files changed, 150 insertions(+), 129 deletions(-) diff --git a/src/torchlight/AudioClip.py b/src/torchlight/AudioClip.py index 1de48de..d3aa204 100644 --- a/src/torchlight/AudioClip.py +++ b/src/torchlight/AudioClip.py @@ -33,6 +33,9 @@ def __init__( def __del__(self) -> None: self.logger.debug("~AudioClip()") + def SetDuration(self, duration: float) -> None: + self.audio_player.SetDuration(duration) + def Play( self, seconds: int | None = None, diff --git a/src/torchlight/Commands.py b/src/torchlight/Commands.py index 520cdf2..832a69c 100644 --- a/src/torchlight/Commands.py +++ b/src/torchlight/Commands.py @@ -966,6 +966,9 @@ async def _func(self, message: list[str], player: Player) -> int: self.torchlight.last_url = url + # Set the duration of the audio clip because youtube already provides it in the metadata + # Helps fixing youtube not sending buffer in time. + audio_clip.SetDuration(float(info["duration"])) return audio_clip.Play(real_time) diff --git a/src/torchlight/FFmpegAudioPlayer.py b/src/torchlight/FFmpegAudioPlayer.py index 0ac9c8f..527b848 100644 --- a/src/torchlight/FFmpegAudioPlayer.py +++ b/src/torchlight/FFmpegAudioPlayer.py @@ -1,14 +1,18 @@ import asyncio import datetime import logging +import os import socket import struct import time import traceback -from asyncio import StreamReader, StreamWriter +from asyncio import StreamReader, StreamWriter, Task from asyncio.subprocess import Process from collections.abc import Callable from typing import Any +from urllib.parse import urlparse + +import aiohttp from torchlight.Torchlight import Torchlight @@ -39,11 +43,13 @@ def __init__(self, torchlight: Torchlight) -> None: self.started_playing: float | None = None self.stopped_playing: float | None = None - self.seconds = 0.0 + self.seconds: float = 0.0 + self.duration_set: bool = False self.writer: StreamWriter | None = None self.ffmpeg_process: Process | None = None - self.curl_process: Process | None = None + self.stream_task: Task | None = None + self.session: aiohttp.ClientSession | None = None self.callbacks: list[tuple[str, Callable]] = [] @@ -51,7 +57,110 @@ def __del__(self) -> None: self.logger.debug("~FFmpegAudioPlayer()") self.Stop() - # @profile + async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> None: + parsed = urlparse(uri) + is_local_file = parsed.scheme in ("file", "") or os.path.exists(uri) + + if is_local_file: + file_path = parsed.path if parsed.scheme == "file" else uri + + if "-i" in ffmpeg_command: + idx = ffmpeg_command.index("-i") + ffmpeg_command[idx + 1] = file_path + else: + ffmpeg_command.extend(["-i", file_path]) + + try: + _, self.writer = await asyncio.open_connection(self.host, self.port) + except Exception as e: + self.logger.error("Failed to connect to voice server at %s:%s - %s", self.host, self.port, e) + self.Stop(False) + return + + stdin_mode = asyncio.subprocess.DEVNULL if is_local_file else asyncio.subprocess.PIPE + + try: + self.ffmpeg_process = await asyncio.create_subprocess_exec( + *ffmpeg_command, + stdin=stdin_mode, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL + ) + except Exception as e: + self.logger.error("Failed to spawn FFmpeg process: %s", e) + self.Stop(False) + return + + if self.ffmpeg_process.stdout: + asyncio.ensure_future(self._read_stream(self.ffmpeg_process.stdout, self.writer)) + + if is_local_file: + return + + headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0.0.0 Safari/537.36" + ) + } + timeout = aiohttp.ClientTimeout(total=None, connect=10.0, sock_read=15.0) + bytes_downloaded = 0 + max_network_retries = 5 + + try: + if self.session is None or self.session.closed: + self.session = aiohttp.ClientSession() + + proxy_url = self.proxy if self.proxy else None + + for attempt in range(1, max_network_retries + 1): + if not self.playing or self.ffmpeg_process.returncode is not None: + break + + req_headers = headers.copy() + if bytes_downloaded > 0: + req_headers["Range"] = f"bytes={bytes_downloaded}-" + + try: + async with self.session.get( + uri, headers=req_headers, timeout=timeout, proxy=proxy_url + ) as resp: + if resp.status not in (200, 206): + self.logger.error("HTTP stream failed with status %d", resp.status) + break + + async for chunk in resp.content.iter_chunked(32 * 1024): + if not self.playing or self.ffmpeg_process.returncode is not None: + break + + bytes_downloaded += len(chunk) + + if self.ffmpeg_process.stdin: + self.ffmpeg_process.stdin.write(chunk) + await self.ffmpeg_process.stdin.drain() + + break # Success + + except (asyncio.TimeoutError, aiohttp.ClientError) as err: + self.logger.warning("Stream network drop/timeout (%s). Retrying (%d/%d)...", err, attempt, max_network_retries) + await asyncio.sleep(0.5) + + except Exception as e: + self.logger.error("Unexpected streaming error: %s", e, exc_info=True) + finally: + if self.ffmpeg_process and self.ffmpeg_process.stdin: + try: + self.ffmpeg_process.stdin.close() + await self.ffmpeg_process.stdin.wait_closed() + except Exception: + pass + + def SetDuration(self, duration: float) -> None: + self.seconds = duration + if self.seconds > 0: + self.duration_set = True + def PlayURI( self, uri: str, @@ -71,28 +180,10 @@ def PlayURI( if pitch is None: pitch = self.pitch - curl_command = [ - "/usr/bin/curl", - "--silent", - "--show-error", - "--connect-timeout", - "1", - "--retry", - "2", - "--retry-delay", - "1", - "--output", - "-", - "-L", - uri, - ] - if self.proxy: - curl_command.extend( - [ - "-x", - self.proxy, - ] - ) + self.seconds = 0.0 + self.started_playing = None + self.stopped_playing = None + ffmpeg_command = [ "/usr/bin/ffmpeg", "-i", @@ -113,25 +204,14 @@ def PlayURI( if position is not None: pos_str = str(datetime.timedelta(seconds=position)) - ffmpeg_command.extend( - [ - "-ss", - pos_str, - ] - ) + ffmpeg_command.extend(["-ss", pos_str]) self.position = position if duration is not None: - ffmpeg_command.extend( - [ - "-t", - str(duration), - ] - ) + ffmpeg_command.extend(["-t", str(duration)]) ffmpeg_command.append("-") - self.logger.debug(curl_command) self.logger.debug(ffmpeg_command) self.playing = True @@ -139,16 +219,21 @@ def PlayURI( self.logger.info("Playing %s", self.uri) - asyncio.ensure_future(self._stream_subprocess(curl_command, ffmpeg_command)) + self.stream_task = asyncio.ensure_future( + self._stream_url_to_ffmpeg(uri, ffmpeg_command) + ) return True - # @profile def Stop(self, force: bool = True) -> bool: if not self.playing: return False self.playing = False + if self.stream_task and not self.stream_task.done(): + self.stream_task.cancel() + self.stream_task = None + if self.ffmpeg_process: try: self.ffmpeg_process.terminate() @@ -157,14 +242,6 @@ def Stop(self, force: bool = True) -> bool: self.logger.debug(exc) self.ffmpeg_process = None - if self.curl_process: - try: - self.curl_process.terminate() - self.curl_process.kill() - except ProcessLookupError as exc: - self.logger.debug(exc) - self.curl_process = None - if self.writer: if force: writer_socket = self.writer.transport.get_extra_info("socket") @@ -176,7 +253,6 @@ def Stop(self, force: bool = True) -> bool: struct.pack("ii", 1, 0), ) except OSError as exc: - # Errno 9: Bad file descriptor if exc.errno == 9: self.logger.error("Unable to setsockopt: %s", exc) @@ -195,15 +271,12 @@ def Stop(self, force: bool = True) -> bool: self.writer = None self.logger.info("Stopped %s", self.uri) - self.uri = "" self.Callback("Stop") - del self.callbacks return True - # @profile def AddCallback(self, cbtype: str, cbfunc: Callable) -> bool: if cbtype not in self.VALID_CALLBACKS: return False @@ -211,7 +284,6 @@ def AddCallback(self, cbtype: str, cbfunc: Callable) -> bool: self.callbacks.append((cbtype, cbfunc)) return True - # @profile def Callback(self, cbtype: str, *args: Any, **kwargs: Any) -> None: for callback in self.callbacks: if callback[0] == cbtype: @@ -221,7 +293,6 @@ def Callback(self, cbtype: str, *args: Any, **kwargs: Any) -> None: except Exception: self.logger.error(traceback.format_exc()) - # @profile async def _updater(self) -> None: try: last_seconds_elapsed = 0.0 @@ -232,27 +303,36 @@ async def _updater(self) -> None: if self.started_playing: seconds_elapsed = time.time() - self.started_playing - if seconds_elapsed > self.seconds: + if self.duration_set and seconds_elapsed > self.seconds: seconds_elapsed = self.seconds self.Callback("Update", last_seconds_elapsed, seconds_elapsed) - if seconds_elapsed >= self.seconds: - if not self.stopped_playing: - self.logger.debug("BUFFER UNDERRUN!") + is_ffmpeg_done = ( + self.ffmpeg_process is None + or self.ffmpeg_process.returncode is not None + ) + + if self.duration_set and self.seconds > 0 and seconds_elapsed >= self.seconds: + if is_ffmpeg_done: + self.logger.debug("Playback naturally finished (time reached).") + self.Stop(False) + return + + elif not self.duration_set and is_ffmpeg_done: + self.logger.debug("Playback naturally finished (FFmpeg EOF).") self.Stop(False) return last_seconds_elapsed = seconds_elapsed - await asyncio.sleep(0.1) + except Exception as exc: self.Stop() self.torchlight.SayChat(f"Error: {str(exc)}") raise exc - # @profile - async def _read_stream(self, stream: StreamReader | None, writer: StreamWriter) -> None: + async def _read_stream(self, stream: StreamReader, writer: StreamWriter) -> None: try: started = False @@ -269,7 +349,8 @@ async def _read_stream(self, stream: StreamReader | None, writer: StreamWriter) samples = bytes_len / SAMPLEBYTES seconds = samples / self.sample_rate - self.seconds += seconds + if not self.duration_set: + self.seconds += seconds if not started: self.logger.info("Streaming %s", self.uri) @@ -283,69 +364,3 @@ async def _read_stream(self, stream: StreamReader | None, writer: StreamWriter) self.Stop() self.torchlight.SayChat(f"Error: {str(exc)}") raise exc - - # @profile - async def _stream_subprocess(self, curl_command: list[str], ffmpeg_command: list[str]) -> None: - if not self.playing: - return - - try: - _, self.writer = await asyncio.open_connection(self.host, self.port) - - self.curl_process = await asyncio.create_subprocess_exec( - *curl_command, - stdout=asyncio.subprocess.PIPE, - ) - - self.ffmpeg_process = await asyncio.create_subprocess_exec( - *ffmpeg_command, - stdin=asyncio.subprocess.PIPE, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, - ) - - asyncio.create_task(self._wait_for_process_exit(self.curl_process)) - - asyncio.create_task(self._write_stream(self.curl_process.stdout, self.ffmpeg_process.stdin)) - - asyncio.create_task(self._read_stream(self.ffmpeg_process.stdout, self.writer)) - - if self.ffmpeg_process is not None: - await self.ffmpeg_process.wait() - - if self.seconds == 0.0: - self.Stop() - - except Exception as exc: - self.Stop() - self.torchlight.SayChat(f"Error: {str(exc)}") - raise exc - - async def _write_stream(self, stream: StreamReader | None, writer: StreamWriter | None) -> None: - try: - while True: - if not stream: - break - chunk = await stream.read(65536) - if not chunk: - break - - if writer: - writer.write(chunk) - await writer.drain() - if writer: - writer.close() - except Exception as exc: - self.Stop() - self.torchlight.SayChat(f"Error: {str(exc)}") - raise exc - - async def _wait_for_process_exit(self, curl_process: Process) -> None: - try: - await curl_process.wait() - if curl_process.returncode != 0 and curl_process.returncode != -15: - raise Exception(f"Curl process exited with error code {curl_process.returncode}") - except Exception as exc: - self.Stop() - self.torchlight.SayChat(f"Error: {str(exc)}") - raise exc From de98ad0a5d8674fad04497d75bf5b06fa6e1920f Mon Sep 17 00:00:00 2001 From: Dolly132 <109222243+Dolly132@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:43:03 +0300 Subject: [PATCH 2/3] fix ruff warnings --- src/torchlight/FFmpegAudioPlayer.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/torchlight/FFmpegAudioPlayer.py b/src/torchlight/FFmpegAudioPlayer.py index 527b848..08b2968 100644 --- a/src/torchlight/FFmpegAudioPlayer.py +++ b/src/torchlight/FFmpegAudioPlayer.py @@ -135,15 +135,18 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> No break bytes_downloaded += len(chunk) - + if self.ffmpeg_process.stdin: self.ffmpeg_process.stdin.write(chunk) await self.ffmpeg_process.stdin.drain() - break # Success + break except (asyncio.TimeoutError, aiohttp.ClientError) as err: - self.logger.warning("Stream network drop/timeout (%s). Retrying (%d/%d)...", err, attempt, max_network_retries) + self.logger.warning( + "Stream network drop/timeout (%s). Retrying (%d/%d)...", + err, attempt, max_network_retries + ) await asyncio.sleep(0.5) except Exception as e: @@ -153,8 +156,8 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> No try: self.ffmpeg_process.stdin.close() await self.ffmpeg_process.stdin.wait_closed() - except Exception: - pass + except Exception as e: + self.logger.debug("Failed to cleanly close FFmpeg stdin: %s", e) def SetDuration(self, duration: float) -> None: self.seconds = duration @@ -309,7 +312,7 @@ async def _updater(self) -> None: self.Callback("Update", last_seconds_elapsed, seconds_elapsed) is_ffmpeg_done = ( - self.ffmpeg_process is None + self.ffmpeg_process is None or self.ffmpeg_process.returncode is not None ) From c1b6d7673413eb9b4e79b85974cb225532375864 Mon Sep 17 00:00:00 2001 From: Dolly132 <109222243+Dolly132@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:47:40 +0300 Subject: [PATCH 3/3] fix ruff --- src/torchlight/FFmpegAudioPlayer.py | 21 +++++---------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/src/torchlight/FFmpegAudioPlayer.py b/src/torchlight/FFmpegAudioPlayer.py index 08b2968..a714a2a 100644 --- a/src/torchlight/FFmpegAudioPlayer.py +++ b/src/torchlight/FFmpegAudioPlayer.py @@ -81,10 +81,7 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> No try: self.ffmpeg_process = await asyncio.create_subprocess_exec( - *ffmpeg_command, - stdin=stdin_mode, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL + *ffmpeg_command, stdin=stdin_mode, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL ) except Exception as e: self.logger.error("Failed to spawn FFmpeg process: %s", e) @@ -123,9 +120,7 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> No req_headers["Range"] = f"bytes={bytes_downloaded}-" try: - async with self.session.get( - uri, headers=req_headers, timeout=timeout, proxy=proxy_url - ) as resp: + async with self.session.get(uri, headers=req_headers, timeout=timeout, proxy=proxy_url) as resp: if resp.status not in (200, 206): self.logger.error("HTTP stream failed with status %d", resp.status) break @@ -144,8 +139,7 @@ async def _stream_url_to_ffmpeg(self, uri: str, ffmpeg_command: list[str]) -> No except (asyncio.TimeoutError, aiohttp.ClientError) as err: self.logger.warning( - "Stream network drop/timeout (%s). Retrying (%d/%d)...", - err, attempt, max_network_retries + "Stream network drop/timeout (%s). Retrying (%d/%d)...", err, attempt, max_network_retries ) await asyncio.sleep(0.5) @@ -222,9 +216,7 @@ def PlayURI( self.logger.info("Playing %s", self.uri) - self.stream_task = asyncio.ensure_future( - self._stream_url_to_ffmpeg(uri, ffmpeg_command) - ) + self.stream_task = asyncio.ensure_future(self._stream_url_to_ffmpeg(uri, ffmpeg_command)) return True def Stop(self, force: bool = True) -> bool: @@ -311,10 +303,7 @@ async def _updater(self) -> None: self.Callback("Update", last_seconds_elapsed, seconds_elapsed) - is_ffmpeg_done = ( - self.ffmpeg_process is None - or self.ffmpeg_process.returncode is not None - ) + is_ffmpeg_done = self.ffmpeg_process is None or self.ffmpeg_process.returncode is not None if self.duration_set and self.seconds > 0 and seconds_elapsed >= self.seconds: if is_ffmpeg_done: