working version
This commit is contained in:
@ -5,7 +5,7 @@ This application reads JPEG frames from shared memory and streams them
|
||||
to web browsers via Server-Sent Events (SSE) or WebSocket (H.264 stream).
|
||||
"""
|
||||
|
||||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import HTMLResponse, StreamingResponse, Response
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
@ -15,7 +15,6 @@ import time
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from typing import Set
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
# Global state
|
||||
@ -26,11 +25,10 @@ last_frame_header = None
|
||||
frame_lock = asyncio.Lock()
|
||||
|
||||
# Video streaming state
|
||||
active_websocket_clients: Set[WebSocket] = set()
|
||||
streaming_task = None
|
||||
streaming_active = False
|
||||
TCP_HOST = '127.0.0.1'
|
||||
TCP_PORT = 8888
|
||||
VIDEO_FIFO_PATH = '/tmp/beacon_video_stream' # Named pipe for H.264 stream
|
||||
video_fifo_file = None # Keep FIFO open to prevent C++ blocking
|
||||
|
||||
|
||||
async def init_reader():
|
||||
@ -46,23 +44,77 @@ async def init_reader():
|
||||
reader = None
|
||||
|
||||
|
||||
async def init_video_fifo():
|
||||
"""Initialize video FIFO - open it to unblock C++ writer"""
|
||||
global video_fifo_file
|
||||
|
||||
# Wait for C++ to create FIFO
|
||||
for i in range(30):
|
||||
if os.path.exists(VIDEO_FIFO_PATH):
|
||||
break
|
||||
print(f"Waiting for FIFO to be created by C++ ({i+1}/30)...")
|
||||
await asyncio.sleep(1)
|
||||
|
||||
if not os.path.exists(VIDEO_FIFO_PATH):
|
||||
print(f"WARNING: Video FIFO not found at {VIDEO_FIFO_PATH}")
|
||||
return
|
||||
|
||||
try:
|
||||
print(f"Opening FIFO {VIDEO_FIFO_PATH} for reading (will block until C++ opens for writing)...")
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Open file in BLOCKING mode - this will unblock C++ when it opens for writing
|
||||
def open_fifo():
|
||||
return open(VIDEO_FIFO_PATH, 'rb', buffering=0)
|
||||
|
||||
video_fifo_file = await loop.run_in_executor(None, open_fifo)
|
||||
print(f"FIFO opened successfully - C++ should now be writing")
|
||||
except Exception as e:
|
||||
print(f"Failed to open video FIFO: {e}")
|
||||
|
||||
|
||||
async def cleanup_reader():
|
||||
"""Clean up shared memory reader"""
|
||||
global reader
|
||||
global reader, video_fifo_file
|
||||
async with reader_lock:
|
||||
if reader:
|
||||
reader.close()
|
||||
reader = None
|
||||
|
||||
# Close video FIFO
|
||||
if video_fifo_file:
|
||||
try:
|
||||
video_fifo_file.close()
|
||||
print("Video FIFO closed")
|
||||
except:
|
||||
pass
|
||||
video_fifo_file = None
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Lifespan context manager for startup and shutdown events"""
|
||||
global streaming_task, streaming_active
|
||||
|
||||
# Startup
|
||||
await init_reader()
|
||||
await init_video_fifo()
|
||||
|
||||
# Start HLS streaming task
|
||||
streaming_task = asyncio.create_task(stream_video_from_fifo())
|
||||
print("Started HLS video streaming task")
|
||||
|
||||
print("FastAPI application started")
|
||||
yield
|
||||
|
||||
# Shutdown
|
||||
streaming_active = False
|
||||
if streaming_task:
|
||||
try:
|
||||
await asyncio.wait_for(streaming_task, timeout=5.0)
|
||||
except asyncio.TimeoutError:
|
||||
print("HLS streaming task did not stop gracefully")
|
||||
|
||||
await cleanup_reader()
|
||||
print("FastAPI application shutdown")
|
||||
|
||||
@ -207,123 +259,228 @@ async def latest_frame():
|
||||
return Response(content="No frame available", status_code=404)
|
||||
|
||||
|
||||
async def stream_video_from_tcp():
|
||||
async def stream_video_from_fifo():
|
||||
"""
|
||||
Read MPEG-TS video stream from TCP socket and broadcast via WebSocket.
|
||||
This runs as a background task.
|
||||
Read raw H.264 from FIFO and convert to HLS format via ffmpeg.
|
||||
This runs as a background task and generates HLS playlist + segments.
|
||||
"""
|
||||
global streaming_active
|
||||
global streaming_active, video_fifo_file
|
||||
import subprocess
|
||||
|
||||
print(f"Starting video stream reader from TCP: {TCP_HOST}:{TCP_PORT}")
|
||||
print(f"Starting HLS video stream generator from FIFO: {VIDEO_FIFO_PATH}")
|
||||
|
||||
# Wait a bit for C++ application to start TCP server
|
||||
await asyncio.sleep(2)
|
||||
ffmpeg_process = None
|
||||
hls_dir = "hls_output"
|
||||
|
||||
# Create HLS output directory if it doesn't exist
|
||||
if not os.path.exists(hls_dir):
|
||||
os.makedirs(hls_dir)
|
||||
|
||||
try:
|
||||
# Connect to TCP server
|
||||
print(f"Connecting to TCP server {TCP_HOST}:{TCP_PORT}...")
|
||||
# Use the globally opened FIFO file
|
||||
if video_fifo_file is None:
|
||||
print("ERROR: FIFO not opened during startup")
|
||||
return
|
||||
|
||||
reader, writer = await asyncio.open_connection(TCP_HOST, TCP_PORT)
|
||||
fifo = video_fifo_file
|
||||
|
||||
print("Connected to TCP server, starting stream...")
|
||||
print("Connected to FIFO, starting ffmpeg H.264->HLS converter...")
|
||||
|
||||
# Start ffmpeg to convert raw H.264 to HLS format
|
||||
# -probesize 32 -analyzeduration 0: minimal probing for low latency startup
|
||||
# -f h264: input format is raw H.264 byte-stream (Annex B from GStreamer)
|
||||
# -r 10: input framerate (from config.ini StreamFps)
|
||||
# -i pipe:0: read from stdin
|
||||
# -c:v copy: copy video codec without re-encoding (use hardware-encoded H.264)
|
||||
# -f hls: output HLS format
|
||||
# -hls_time 1: 1 second per segment (low latency)
|
||||
# -hls_list_size 5: keep 5 segments in playlist
|
||||
# -hls_flags delete_segments+append_list+omit_endlist: live streaming flags
|
||||
# -hls_segment_type mpegts: use MPEG-TS segments
|
||||
# -start_number 0: start segment numbering from 0
|
||||
ffmpeg_process = subprocess.Popen(
|
||||
['ffmpeg',
|
||||
'-probesize', '32',
|
||||
'-analyzeduration', '0',
|
||||
'-f', 'h264',
|
||||
'-r', '10', # Must match GStreamer StreamFps from config.ini
|
||||
'-i', 'pipe:0',
|
||||
'-c:v', 'copy',
|
||||
'-f', 'hls',
|
||||
'-hls_time', '1',
|
||||
'-hls_list_size', '5',
|
||||
'-hls_flags', 'delete_segments+append_list+omit_endlist',
|
||||
'-hls_segment_type', 'mpegts',
|
||||
'-hls_segment_filename', f'{hls_dir}/segment_%03d.ts',
|
||||
'-start_number', '0',
|
||||
f'{hls_dir}/playlist.m3u8'],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE
|
||||
)
|
||||
|
||||
streaming_active = True
|
||||
chunk_size = 32768 # 32KB chunks for MPEG-TS
|
||||
chunk_size = 65536 # 64KB chunks
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
while streaming_active and active_websocket_clients:
|
||||
bytes_written = 0
|
||||
|
||||
print(f"Reading raw H.264 from FIFO and generating HLS segments...")
|
||||
|
||||
async def read_from_fifo():
|
||||
"""Read from FIFO and pipe to ffmpeg stdin"""
|
||||
nonlocal bytes_written
|
||||
try:
|
||||
# Read chunk from TCP
|
||||
data = await asyncio.wait_for(reader.read(chunk_size), timeout=5.0)
|
||||
while streaming_active:
|
||||
def read_chunk():
|
||||
return fifo.read(chunk_size)
|
||||
|
||||
if not data:
|
||||
print("No data from TCP, stream may have ended")
|
||||
break
|
||||
data = await loop.run_in_executor(None, read_chunk)
|
||||
if not data:
|
||||
print("FIFO EOF reached")
|
||||
break
|
||||
|
||||
# Broadcast binary data to all connected WebSocket clients
|
||||
disconnected_clients = set()
|
||||
for client in active_websocket_clients.copy():
|
||||
if ffmpeg_process and ffmpeg_process.stdin:
|
||||
try:
|
||||
ffmpeg_process.stdin.write(data)
|
||||
ffmpeg_process.stdin.flush()
|
||||
bytes_written += len(data)
|
||||
|
||||
# Log progress every 1MB
|
||||
if bytes_written % (1024 * 1024) < chunk_size:
|
||||
print(f"Written {bytes_written / 1024 / 1024:.2f} MB to ffmpeg for HLS encoding")
|
||||
except Exception as e:
|
||||
print(f"Error writing to ffmpeg stdin: {e}")
|
||||
break
|
||||
finally:
|
||||
if ffmpeg_process and ffmpeg_process.stdin:
|
||||
try:
|
||||
await client.send_bytes(data)
|
||||
except Exception as e:
|
||||
print(f"Error sending to client: {e}")
|
||||
disconnected_clients.add(client)
|
||||
ffmpeg_process.stdin.close()
|
||||
print("Closed ffmpeg stdin")
|
||||
except:
|
||||
pass
|
||||
|
||||
# Remove disconnected clients
|
||||
active_websocket_clients.difference_update(disconnected_clients)
|
||||
async def read_ffmpeg_stderr():
|
||||
"""Read and log ffmpeg stderr for debugging"""
|
||||
line_count = 0
|
||||
try:
|
||||
while streaming_active:
|
||||
def read_err():
|
||||
line = ffmpeg_process.stderr.readline()
|
||||
return line.decode('utf-8', errors='ignore') if line else None
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
# No data in 5 seconds, continue waiting
|
||||
continue
|
||||
line = await loop.run_in_executor(None, read_err)
|
||||
if not line:
|
||||
break
|
||||
|
||||
line_count += 1
|
||||
|
||||
# Log first 30 lines to see HLS setup, then only errors/warnings
|
||||
if line_count <= 30:
|
||||
# Skip empty lines and progress updates
|
||||
if line.strip() and not line.startswith('frame='):
|
||||
print(f"FFmpeg[{line_count}]: {line.strip()}")
|
||||
elif 'error' in line.lower() or 'warning' in line.lower():
|
||||
print(f"FFmpeg: {line.strip()}")
|
||||
except Exception as e:
|
||||
print(f"Error reading from TCP: {e}")
|
||||
break
|
||||
print(f"Error reading ffmpeg stderr: {e}")
|
||||
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
# Run both tasks concurrently
|
||||
await asyncio.gather(
|
||||
read_from_fifo(),
|
||||
read_ffmpeg_stderr(),
|
||||
return_exceptions=True
|
||||
)
|
||||
|
||||
except ConnectionRefusedError:
|
||||
print(f"ERROR: Could not connect to TCP server {TCP_HOST}:{TCP_PORT}")
|
||||
except FileNotFoundError:
|
||||
print(f"ERROR: Could not open FIFO {VIDEO_FIFO_PATH}")
|
||||
print("Make sure C++ application is running first")
|
||||
except Exception as e:
|
||||
print(f"ERROR in video streaming: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
# Cleanup ffmpeg process
|
||||
if ffmpeg_process:
|
||||
try:
|
||||
ffmpeg_process.terminate()
|
||||
ffmpeg_process.wait(timeout=2)
|
||||
except:
|
||||
try:
|
||||
ffmpeg_process.kill()
|
||||
except:
|
||||
pass
|
||||
|
||||
streaming_active = False
|
||||
print("Video streaming stopped")
|
||||
|
||||
|
||||
@app.websocket("/ws/video")
|
||||
async def websocket_video_endpoint(websocket: WebSocket):
|
||||
"""WebSocket endpoint for video streaming"""
|
||||
global streaming_task, streaming_active
|
||||
@app.get("/hls/playlist.m3u8")
|
||||
async def get_hls_playlist():
|
||||
"""Serve HLS playlist manifest"""
|
||||
playlist_path = "hls_output/playlist.m3u8"
|
||||
|
||||
await websocket.accept()
|
||||
print(f"Client connected to video stream: {id(websocket)}")
|
||||
|
||||
# Add client to active set
|
||||
active_websocket_clients.add(websocket)
|
||||
|
||||
# Start streaming task if not already running
|
||||
if not streaming_active:
|
||||
streaming_task = asyncio.create_task(stream_video_from_tcp())
|
||||
print("Started video streaming task")
|
||||
if not os.path.exists(playlist_path):
|
||||
return Response(content="HLS playlist not ready yet", status_code=404)
|
||||
|
||||
try:
|
||||
# Keep connection alive and handle client messages
|
||||
while True:
|
||||
# Wait for client messages (like ping/pong or control messages)
|
||||
try:
|
||||
data = await asyncio.wait_for(websocket.receive_text(), timeout=1.0)
|
||||
# Handle client messages if needed
|
||||
if data == "ping":
|
||||
await websocket.send_text("pong")
|
||||
except asyncio.TimeoutError:
|
||||
# No message received, continue
|
||||
continue
|
||||
with open(playlist_path, 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
except WebSocketDisconnect:
|
||||
print(f"Client disconnected from video stream: {id(websocket)}")
|
||||
return Response(
|
||||
content=content,
|
||||
media_type="application/vnd.apple.mpegurl",
|
||||
headers={
|
||||
"Cache-Control": "no-cache, no-store, must-revalidate",
|
||||
"Pragma": "no-cache",
|
||||
"Expires": "0",
|
||||
"Access-Control-Allow-Origin": "*"
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"WebSocket error: {e}")
|
||||
finally:
|
||||
# Remove client from active set
|
||||
active_websocket_clients.discard(websocket)
|
||||
print(f"Error serving HLS playlist: {e}")
|
||||
return Response(content=f"Error: {e}", status_code=500)
|
||||
|
||||
# Stop streaming if no more clients
|
||||
if not active_websocket_clients:
|
||||
streaming_active = False
|
||||
print("No more clients, stopping stream")
|
||||
|
||||
@app.get("/hls/{segment_name}")
|
||||
async def get_hls_segment(segment_name: str):
|
||||
"""Serve HLS video segments"""
|
||||
# Security: only allow .ts files
|
||||
if not segment_name.endswith('.ts'):
|
||||
return Response(content="Invalid segment name", status_code=400)
|
||||
|
||||
segment_path = f"hls_output/{segment_name}"
|
||||
|
||||
if not os.path.exists(segment_path):
|
||||
return Response(content="Segment not found", status_code=404)
|
||||
|
||||
try:
|
||||
with open(segment_path, 'rb') as f:
|
||||
content = f.read()
|
||||
|
||||
return Response(
|
||||
content=content,
|
||||
media_type="video/MP2T",
|
||||
headers={
|
||||
"Cache-Control": "public, max-age=3600",
|
||||
"Access-Control-Allow-Origin": "*"
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"Error serving HLS segment {segment_name}: {e}")
|
||||
return Response(content=f"Error: {e}", status_code=500)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
"""Health check endpoint"""
|
||||
hls_playlist_exists = os.path.exists("hls_output/playlist.m3u8")
|
||||
|
||||
return {
|
||||
"status": "healthy",
|
||||
"active_websocket_clients": len(active_websocket_clients),
|
||||
"streaming_active": streaming_active,
|
||||
"tcp_endpoint": f"{TCP_HOST}:{TCP_PORT}"
|
||||
"hls_playlist_ready": hls_playlist_exists,
|
||||
"video_fifo": VIDEO_FIFO_PATH
|
||||
}
|
||||
|
||||
|
||||
@ -332,7 +489,7 @@ if __name__ == '__main__':
|
||||
|
||||
print("Starting FastAPI server on http://0.0.0.0:5000")
|
||||
print(f"Video stream will be available at ws://0.0.0.0:5000/ws/video")
|
||||
print(f"Expecting C++ TCP stream at {TCP_HOST}:{TCP_PORT}")
|
||||
print(f"Expecting C++ H.264 stream via FIFO at {VIDEO_FIFO_PATH}")
|
||||
|
||||
uvicorn.run(
|
||||
app,
|
||||
|
||||
Reference in New Issue
Block a user