Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-22 09:37:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge branch 'ffmpeg-stats' of https://github.com/Dispatcharr/Dispatcharr into dev

Browse source
This commit is contained in:
SergeantPanda 2025-06-11 17:33:22 -05:00
commit 34ceffb86a
5 changed files with 145 additions and 76 deletions
Download patch file Download diff file Expand all files Collapse all files

2
apps/proxy/config.py
View file

@ -11,6 +11,8 @@ class BaseConfig:
BUFFER_CHUNK_SIZE = 188 * 1361 # ~256KB
# Redis settings
REDIS_CHUNK_TTL = 60 # Number in seconds - Chunks expire after 1 minute
BUFFERING_TIMEOUT = 15 # Seconds to wait for buffering before switching streams
BUFFER_SPEED = 1 # What speed to condsider the stream buffering, 1x is normal speed, 2x is double speed, etc.
class HLSConfig(BaseConfig):
MIN_SEGMENTS = 12

8
apps/proxy/ts_proxy/config_helper.py
View file

@ -85,3 +85,11 @@ class ConfigHelper:
def failover_grace_period():
"""Get extra time (in seconds) to allow for stream switching before disconnecting clients"""
return ConfigHelper.get('FAILOVER_GRACE_PERIOD', 20) # Default to 20 seconds
@staticmethod
def buffering_timeout():
"""Get buffering timeout in seconds"""
return ConfigHelper.get('BUFFERING_TIMEOUT', 15) # Default to 15 seconds
@staticmethod
def buffering_speed():
"""Get buffering speed in bytes per second"""
return ConfigHelper.get('BUFFERING_SPEED',1) # Default to 1x

1
apps/proxy/ts_proxy/constants.py
View file

@ -18,6 +18,7 @@ class ChannelState:
ERROR = "error"
STOPPING = "stopping"
STOPPED = "stopped"
BUFFERING = "buffering"
# Event types
class EventType:

208
apps/proxy/ts_proxy/stream_manager.py
View file

@ -40,6 +40,10 @@ class StreamManager:
self.url_switching = False
self.url_switch_start_time = 0
self.url_switch_timeout = ConfigHelper.url_switch_timeout()
self.buffering = False
self.buffering_timeout = ConfigHelper.buffering_timeout()
self.buffering_speed = ConfigHelper.buffering_speed()
self.buffering_start_time = None
# Store worker_id for ownership checks
self.worker_id = worker_id
@ -382,90 +386,99 @@ class StreamManager:
buffer = b""
last_stats_line = b""
# Read in small chunks
# Read byte by byte for immediate detection
while self.transcode_process and self.transcode_process.stderr:
try:
chunk = self.transcode_process.stderr.read(256) # Smaller chunks for real-time processing
if not chunk:
# Read one byte at a time for immediate processing
byte = self.transcode_process.stderr.read(1)
if not byte:
break
buffer += chunk
buffer += byte
# Look for stats updates (overwrite previous stats with \r)
if b'\r' in buffer and b"frame=" in buffer:
# Split on \r to handle overwriting stats
parts = buffer.split(b'\r')
# Check for frame= at the start of buffer (new stats line)
if buffer == b"frame=":
# We detected the start of a stats line, read until we get a complete line
# or hit a carriage return (which overwrites the previous stats)
while True:
next_byte = self.transcode_process.stderr.read(1)
if not next_byte:
break
# Process all parts except the last (which might be incomplete)
for i, part in enumerate(parts[:-1]):
if part.strip():
if part.startswith(b"frame=") or b"frame=" in part:
# This is a stats line - keep it intact
try:
stats_text = part.decode('utf-8', errors='ignore').strip()
if stats_text and "frame=" in stats_text:
# Extract just the stats portion if there's other content
if "frame=" in stats_text:
frame_start = stats_text.find("frame=")
stats_text = stats_text[frame_start:]
buffer += next_byte
self._parse_ffmpeg_stats(stats_text)
self._log_stderr_content(stats_text)
last_stats_line = part
except Exception as e:
logger.debug(f"Error parsing stats line: {e}")
else:
# Regular content - process line by line
line_content = part
while b'\n' in line_content:
line, line_content = line_content.split(b'\n', 1)
if line.strip():
self._log_stderr_content(line.decode('utf-8', errors='ignore'))
# Break on carriage return (stats overwrite) or newline
if next_byte in (b'\r', b'\n'):
break
# Handle remaining content without newline
if line_content.strip():
self._log_stderr_content(line_content.decode('utf-8', errors='ignore'))
# Also break if we have enough data for a typical stats line
if len(buffer) > 200: # Typical stats line length
break
# Keep the last part as it might be incomplete
buffer = parts[-1]
# Process the stats line immediately
if buffer.strip():
try:
stats_text = buffer.decode('utf-8', errors='ignore').strip()
if stats_text and "frame=" in stats_text:
self._parse_ffmpeg_stats(stats_text)
self._log_stderr_content(stats_text)
except Exception as e:
logger.debug(f"Error parsing immediate stats line: {e}")
# Clear buffer after processing
buffer = b""
continue
# Handle regular line breaks for non-stats content
elif b'\n' in buffer:
while b'\n' in buffer:
line, buffer = buffer.split(b'\n', 1)
if line.strip():
line_text = line.decode('utf-8', errors='ignore').strip()
if line_text and not line_text.startswith("frame="):
self._log_stderr_content(line_text)
elif byte == b'\n':
if buffer.strip():
line_text = buffer.decode('utf-8', errors='ignore').strip()
if line_text and not line_text.startswith("frame="):
self._log_stderr_content(line_text)
buffer = b""
# If we have a potential stats line in buffer without line breaks
elif b"frame=" in buffer and (b"speed=" in buffer or len(buffer) > 200):
# We likely have a complete or substantial stats line
try:
stats_text = buffer.decode('utf-8', errors='ignore').strip()
if "frame=" in stats_text:
# Extract just the stats portion
frame_start = stats_text.find("frame=")
stats_text = stats_text[frame_start:]
# Handle carriage returns (potential stats overwrite)
elif byte == b'\r':
# Check if this might be a stats line
if b"frame=" in buffer:
try:
stats_text = buffer.decode('utf-8', errors='ignore').strip()
if stats_text and "frame=" in stats_text:
self._parse_ffmpeg_stats(stats_text)
self._log_stderr_content(stats_text)
except Exception as e:
logger.debug(f"Error parsing stats on carriage return: {e}")
elif buffer.strip():
# Regular content with carriage return
line_text = buffer.decode('utf-8', errors='ignore').strip()
if line_text:
self._log_stderr_content(line_text)
buffer = b""
self._parse_ffmpeg_stats(stats_text)
self._log_stderr_content(stats_text)
buffer = b"" # Clear buffer after processing
except Exception as e:
logger.debug(f"Error parsing buffered stats: {e}")
# Prevent buffer from growing too large
if len(buffer) > 4096:
# Try to preserve any potential stats line at the end
if b"frame=" in buffer[-1024:]:
buffer = buffer[-1024:]
else:
buffer = buffer[-512:]
# Prevent buffer from growing too large for non-stats content
elif len(buffer) > 1024 and b"frame=" not in buffer:
# Process whatever we have if it's not a stats line
if buffer.strip():
line_text = buffer.decode('utf-8', errors='ignore').strip()
if line_text:
self._log_stderr_content(line_text)
buffer = b""
except Exception as e:
logger.error(f"Error reading stderr: {e}")
logger.error(f"Error reading stderr byte: {e}")
break
# Process any remaining buffer content
if buffer.strip():
try:
remaining_text = buffer.decode('utf-8', errors='ignore').strip()
if remaining_text:
if "frame=" in remaining_text:
self._parse_ffmpeg_stats(remaining_text)
self._log_stderr_content(remaining_text)
except Exception as e:
logger.debug(f"Error processing remaining buffer: {e}")
except Exception as e:
# Catch any other exceptions in the thread to prevent crashes
try:
@ -484,13 +497,18 @@ class StreamManager:
content_lower = content.lower()
# Check for stream info lines first and delegate to ChannelService
# Only parse INPUT streams (which have hex identifiers like [0x100]) not output streams
if "stream #" in content_lower and ("video:" in content_lower or "audio:" in content_lower):
from .services.channel_service import ChannelService
if "video:" in content_lower:
ChannelService.parse_and_store_stream_info(self.channel_id, content, "video")
elif "audio:" in content_lower:
ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio")
# Check if this is an input stream by looking for the hex identifier pattern [0x...]
if "stream #0:" in content_lower and "[0x" in content_lower:
from .services.channel_service import ChannelService
if "video:" in content_lower:
ChannelService.parse_and_store_stream_info(self.channel_id, content, "video")
elif "audio:" in content_lower:
ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio")
else:
# This is likely an output stream (no hex identifier), don't parse it
logger.debug(f"Skipping output stream info: {content}")
# Determine log level based on content
if any(keyword in content_lower for keyword in ['error', 'failed', 'cannot', 'invalid', 'corrupt']):
logger.error(f"FFmpeg stderr: {content}")
@ -545,7 +563,6 @@ class StreamManager:
actual_fps = None
if ffmpeg_fps is not None and ffmpeg_speed is not None and ffmpeg_speed > 0:
actual_fps = ffmpeg_fps / ffmpeg_speed
# Store in Redis if we have valid data
if any(x is not None for x in [ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_bitrate]):
self._update_ffmpeg_stats_in_redis(ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_bitrate)
@ -553,10 +570,51 @@ class StreamManager:
# Fix the f-string formatting
actual_fps_str = f"{actual_fps:.1f}" if actual_fps is not None else "N/A"
ffmpeg_bitrate_str = f"{ffmpeg_bitrate:.1f}" if ffmpeg_bitrate is not None else "N/A"
# Log the stats
logger.debug(f"FFmpeg stats - Speed: {ffmpeg_speed}x, FFmpeg FPS: {ffmpeg_fps}, "
f"Actual FPS: {actual_fps_str}, "
f"Bitrate: {ffmpeg_bitrate_str} kbps")
# If we have a valid speed, check for buffering
if ffmpeg_speed is not None and ffmpeg_speed < self.buffering_speed:
if self.buffering:
# Buffering is still ongoing, check for how long
if self.buffering_start_time is None:
self.buffering_start_time = time.time()
else:
buffering_duration = time.time() - self.buffering_start_time
if buffering_duration > self.buffering_timeout:
# Buffering timeout reached, log error and try next stream
logger.error(f"Buffering timeout reached for channel {self.channel_id} after {buffering_duration:.1f} seconds")
# Send next stream request
if self._try_next_stream():
logger.info(f"Switched to next stream for channel {self.channel_id} after buffering timeout")
# Reset buffering state
self.buffering = False
self.buffering_start_time = None
else:
logger.error(f"Failed to switch to next stream for channel {self.channel_id} after buffering timeout")
else:
# Buffering just started, set the flag and start timer
self.buffering = True
self.buffering_start_time = time.time()
logger.warning(f"Buffering started for channel {self.channel_id} - speed: {ffmpeg_speed}x")
# Log buffering warning
logger.debug(f"FFmpeg speed on channel {self.channel_id} is below {self.buffering_speed} ({ffmpeg_speed}x) - buffering detected")
# Set channel state to buffering
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.BUFFERING)
elif ffmpeg_speed is not None and ffmpeg_speed >= self.buffering_speed:
# Speed is good, check if we were buffering
if self.buffering:
# Reset buffering state
logger.info(f"Buffering ended for channel {self.channel_id} - speed: {ffmpeg_speed}x")
self.buffering = False
self.buffering_start_time = None
# Set channel state to active if speed is good
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.ACTIVE)
except Exception as e:
logger.debug(f"Error parsing FFmpeg stats: {e}")

2
apps/proxy/ts_proxy/url_utils.py
View file

@ -172,7 +172,7 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
# Get transcode info from the channel's stream profile
stream_profile = channel.get_stream_profile()
transcode = not (stream_profile.is_proxy() or stream_profile is None)
profile_value = str(stream_profile)
profile_value = stream_profile.id
return {
'url': stream_url,