import asyncio import logging from datetime import datetime, timezone from app.database import SessionLocal, engine from app.models.processing_job import ProcessingJob from app.models.stream import Stream from app.config import settings from app.sse import publish_event import subprocess import os from sqlalchemy.orm import Session from sqlalchemy import text from app.services.video_processor import VideoProcessor import threading import shutil logging.basicConfig(level=settings.LOG_LEVEL) logger = logging.getLogger(__name__) def check_gpu_available() -> tuple[bool, str]: """Check if GPU is available for hardware acceleration""" # Check if ffmpeg with NVENC support is available try: result = subprocess.run( ["ffmpeg", "-encoders"], capture_output=True, text=True, timeout=5 ) if "h264_nvenc" in result.stdout: logger.info("NVIDIA NVENC GPU encoder found") return True, "h264_nvenc" except (subprocess.TimeoutExpired, FileNotFoundError): pass # Check for Intel Quick Sync try: result = subprocess.run( ["ffmpeg", "-encoders"], capture_output=True, text=True, timeout=5 ) if "h264_qsv" in result.stdout: logger.info("Intel Quick Sync Video encoder found") return True, "h264_qsv" except (subprocess.TimeoutExpired, FileNotFoundError): pass # Check for AMD AMF try: result = subprocess.run( ["ffmpeg", "-encoders"], capture_output=True, text=True, timeout=5 ) if "h264_amf" in result.stdout: logger.info("AMD AMF encoder found") return True, "h264_amf" except (subprocess.TimeoutExpired, FileNotFoundError): pass logger.info("No GPU encoder found, using CPU encoding") return False, "libx264" # Cache GPU availability # GPU_AVAILABLE, GPU_ENCODER = check_gpu_available() # Disabled for Windows - requires Ubuntu Server with GPU for GPU passthrough GPU_AVAILABLE, GPU_ENCODER = False, "libx264" class StreamDaemon: def __init__(self): self.running = False async def start(self): logger.info("Starting StreamA2026 Daemon...") self.running = True # Resume interrupted processing jobs await self.resume_interrupted_jobs() # Start background tasks tasks = [ self.process_jobs(), self.health_check(), self.cleanup_old_sessions() ] await asyncio.gather(*tasks) async def stop(self): logger.info("Stopping StreamA2026 Daemon...") self.running = False async def resume_interrupted_jobs(self): """Check for interrupted processing jobs and resume them""" try: from app.redis_client import redis_client import json import os db = SessionLocal() try: # Find streams with status='processing' (stuck due to backend restart) stuck_streams = db.query(Stream).filter(Stream.status == 'processing').all() for stream in stuck_streams: logger.info(f"Found stuck processing job for stream {stream.id}") # Count existing segments hls_dir = f"{settings.STORAGE_PATH}/hls/{stream.id}" if os.path.exists(hls_dir): segment_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts')] current_segments = len(segment_files) logger.info(f"Found {current_segments} existing segments for stream {stream.id}") else: current_segments = 0 logger.info(f"No existing segments found for stream {stream.id}") # Set state to interrupted with segment count state_key = f"processing_state:{stream.id}" state_data = { 'status': 'interrupted', 'last_successful_segment': current_segments } redis_client.set(state_key, json.dumps(state_data)) def process_in_background(): try: processor = VideoProcessor() processor.process_video(str(stream.id)) logger.info(f"Resumed processing completed for stream {stream.id}") except Exception as e: logger.error(f"Resumed processing failed for stream {stream.id}: {e}") thread = threading.Thread(target=process_in_background) thread.start() finally: db.close() except Exception as e: logger.error(f"Failed to resume interrupted jobs: {e}") async def process_jobs(self): while self.running: try: db = SessionLocal() jobs = db.query(ProcessingJob).filter( ProcessingJob.status == "pending" ).limit(1).all() for job in jobs: await self.process_job(job, db) db.close() await asyncio.sleep(5) except Exception as e: logger.error(f"Error processing jobs: {e}") await asyncio.sleep(10) async def process_job(self, job: ProcessingJob, db: Session): logger.info(f"Processing job {job.id} for stream {job.stream_id}") # Update job status job.status = "processing" job.started_at = datetime.now(timezone.utc) job.progress = 0 db.commit() stream = db.query(Stream).filter(Stream.id == job.stream_id).first() if not stream or not stream.file_path: job.status = "failed" job.error_message = "Stream or file not found" job.completed_at = datetime.now(timezone.utc) db.commit() return try: if job.job_type == "hls" or job.job_type == "transcode": await self.convert_to_hls(stream, job, db) elif job.job_type == "dash": await self.convert_to_dash(stream, job, db) elif job.job_type == "thumbnail": await self.generate_thumbnail(stream, job, db) job.status = "completed" job.progress = 100 job.completed_at = datetime.now(timezone.utc) db.commit() # Publish event publish_event("job_completed", { "job_id": str(job.id), "stream_id": str(stream.id), "job_type": job.job_type }) except Exception as e: logger.error(f"Error processing job {job.id}: {e}") job.status = "failed" job.error_message = str(e) job.completed_at = datetime.now(timezone.utc) db.commit() publish_event("job_failed", { "job_id": str(job.id), "stream_id": str(stream.id), "error": str(e) }) async def convert_to_hls(self, stream: Stream, job: ProcessingJob, db: Session): output_dir = os.path.join(settings.FFMPEG_CACHE_PATH, f"hls_{stream.id}") os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, "playlist.m3u8") # Use GPU encoder if available video_encoder = GPU_ENCODER if GPU_AVAILABLE else "libx264" if GPU_AVAILABLE: logger.info(f"Using GPU encoder: {video_encoder}") cmd = [ "ffmpeg", "-i", stream.file_path, "-c:v", video_encoder, "-c:a", "aac", "-f", "hls", "-hls_time", "2", "-hls_playlist_type", "vod", "-y", # Overwrite output files output_path ] # Add GPU-specific options for NVENC if GPU_AVAILABLE and video_encoder == "h264_nvenc": cmd.extend(["-preset", "p6", "-tune", "ll", "-rc", "vbr", "-b:v", "5M", "-maxrate", "10M", "-bufsize", "20M"]) logger.info(f"Running FFmpeg command: {' '.join(cmd)}") process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode == 0: stream.hls_path = output_path stream.status = "ready" stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_total = 0 stream.thumbnail_progress_message = None stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = None db.commit() logger.info(f"HLS conversion completed for stream {stream.id}") else: error_msg = stderr.decode() if stderr else "Unknown error" logger.error(f"FFmpeg conversion failed: {error_msg}") raise Exception(f"FFmpeg conversion failed: {error_msg}") async def convert_to_dash(self, stream: Stream, job: ProcessingJob, db: Session): output_dir = os.path.join(settings.FFMPEG_CACHE_PATH, f"dash_{stream.id}") os.makedirs(output_dir, exist_ok=True) output_path = os.path.join(output_dir, "manifest.mpd") # Use GPU encoder if available video_encoder = GPU_ENCODER if GPU_AVAILABLE else "libx264" cmd = [ "ffmpeg", "-i", stream.file_path, "-c:v", video_encoder, "-c:a", "aac", "-f", "dash", output_path ] # Add GPU-specific options for NVENC if GPU_AVAILABLE and video_encoder == "h264_nvenc": cmd.extend(["-preset", "p6", "-tune", "ll", "-rc", "vbr", "-b:v", "5M"]) process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode == 0: stream.dash_path = output_path db.commit() else: error_msg = stderr.decode() if stderr else "Unknown error" raise Exception(f"FFmpeg DASH conversion failed: {error_msg}") async def generate_thumbnail(self, stream: Stream, job: ProcessingJob, db: Session): os.makedirs(settings.STORAGE_PATH, exist_ok=True) output_path = os.path.join(settings.STORAGE_PATH, f"thumb_{stream.id}.jpg") cmd = [ "ffmpeg", "-i", stream.file_path, "-ss", "00:00:01", "-vframes", "1", "-y", # Overwrite output file output_path ] process = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await process.communicate() if process.returncode == 0: stream.thumbnail_url = output_path db.commit() else: error_msg = stderr.decode() if stderr else "Unknown error" raise Exception(f"Thumbnail generation failed: {error_msg}") async def health_check(self): while self.running: try: # Check database connection db = SessionLocal() try: db.execute(text("SELECT 1")) finally: db.close() # Check redis connection from app.redis_client import redis_client redis_client.ping() logger.debug("Health check passed") publish_event("health_check", {"status": "healthy"}) except Exception as e: logger.error(f"Health check failed: {e}") publish_event("health_check", {"status": "unhealthy", "error": str(e)}) await asyncio.sleep(30) break # Exit after first check async def cleanup_old_sessions(self): while self.running: try: from datetime import timedelta from app.models.session import Session as SessionModel db = SessionLocal() expired = db.query(SessionModel).filter( SessionModel.expires_at < datetime.now(timezone.utc) ).all() for session in expired: db.delete(session) db.commit() db.close() logger.info(f"Cleaned up {len(expired)} expired sessions") except Exception as e: logger.error(f"Error cleaning up sessions: {e}") await asyncio.sleep(3600) # Run every hour async def main(): daemon = StreamDaemon() try: await daemon.start() except KeyboardInterrupt: await daemon.stop() if __name__ == "__main__": asyncio.run(main())