RTSP playbackMode and Files Duration Check (#1888)

Co-authored-by: Huiling Bao <huiling.bao@intel.com>

Author: d-rushma

education-ai-suite/smart-classroom/api/endpoints.py

@@ -1,11 +1,12 @@
 import asyncio
 from typing import Optional
 from fastapi import Header, UploadFile
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, FileResponse
 from fastapi import APIRouter, FastAPI, File, HTTPException, status
 from dto.transcription_dto import TranscriptionRequest
 from dto.summarizer_dto import SummaryRequest
 from dto.video_analytics_dto import VideoAnalyticsRequest
+from dto.video_metadata_dto import VideoDurationRequest
 from pipeline import Pipeline
 import json, os
 import subprocess, re
@@ -22,6 +23,7 @@
 from components.va.va_pipeline_service import VideoAnalyticsPipelineService, PipelineOptions
 from utils.session_manager import generate_session_id
 from dto.search_dto import SearchRequest
+from utils.session_state_manager import SessionState
 import logging
 logger = logging.getLogger(__name__)
 
@@ -523,6 +525,126 @@ async def stream_statistics():
 
     return StreamingResponse(stream_statistics(), media_type="application/json")
 
+@router.post("/mark-video-usage")
+def mark_video_usage(
+    session_id: str = Header(None, alias="X-Session-ID")
+):
+    """
+    Mark that a video is being used in the current session.
+
+    """
+    if not session_id:
+        raise HTTPException(
+            status_code=400,
+            detail="X-Session-ID header is required"
+        )
+    
+    try:
+        with SessionState._lock:
+            if session_id not in SessionState._sessions:
+                SessionState._sessions[session_id] = {}
+            SessionState._sessions[session_id]['has_video'] = True
+        
+        logger.info(f"Session {session_id}: Video usage marked")
+        return JSONResponse(
+            status_code=200,
+            content={"status": "success", "message": "Video usage marked for session"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Session {session_id}: Error marking video usage: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error marking video usage: {e}"
+        )
+
+@router.post("/store-video-duration")
+def store_video_duration(
+    request: VideoDurationRequest,
+    session_id: str = Header(None, alias="X-Session-ID")
+):
+    """
+    Store video duration 
+
+    """
+    if not session_id:
+        raise HTTPException(
+            status_code=400,
+            detail="X-Session-ID header is required"
+        )
+    
+    try:
+        duration = request.duration
+        
+        if not duration or duration <= 0:
+            raise HTTPException(
+                status_code=400,
+                detail="Invalid duration: duration must be greater than 0"
+            )
+        
+        # Store the video duration in session state
+        SessionState.set_video_duration(session_id, duration)
+        
+        return JSONResponse(
+            status_code=200,
+            content={"status": "success", "message": f"Video duration stored: {duration:.2f}s"}
+        )
+        
+    except HTTPException as http_exc:
+        raise http_exc
+    except Exception as e:
+        logger.error(f"Session {session_id}: Error storing video duration: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error storing video duration: {e}"
+        )
+
+@router.post("/store-audio-duration")
+def store_audio_duration(
+    request: VideoDurationRequest,
+    session_id: str = Header(None, alias="X-Session-ID")
+):
+    """
+    Store audio duration 
+
+    """
+    
+    if not session_id:
+        raise HTTPException(
+            status_code=400,
+            detail="X-Session-ID header is required"
+        )
+    
+    try:
+        duration = request.duration
+        
+        if not duration or duration <= 0:
+            raise HTTPException(
+                status_code=400,
+                detail="Invalid duration: duration must be greater than 0"
+            )
+        
+        SessionState.set_audio_duration(session_id, duration)
+
+        with SessionState._lock:
+            if session_id not in SessionState._sessions:
+                SessionState._sessions[session_id] = {}
+            SessionState._sessions[session_id]['has_audio'] = True
+
+        return JSONResponse(
+            status_code=200,
+            content={"status": "success", "message": f"Audio duration stored: {duration:.2f}s"}
+        )
+        
+    except HTTPException as http_exc:
+        raise http_exc
+    except Exception as e:
+        logger.error(f"Session {session_id}: Error storing audio duration: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error storing audio duration: {e}"
+        )
+
 @router.post("/content-segmentation")
 def content_segmentation(request: SummaryRequest):
     """
@@ -534,17 +656,22 @@ def content_segmentation(request: SummaryRequest):
         raise HTTPException(status_code=429, detail="Session Active, Try Later")
 
     pipeline = Pipeline(request.session_id)
+    
+    # Log session state before validation
+    session_state = SessionState.get_session_state(request.session_id)
+    logger.info(f"📋 Content-segmentation request for session: {request.session_id}")
+    logger.info(f"   Session state: {session_state}")
 
     try:
         contents_json = pipeline.run_content_segmentation()
-        logger.info("content segmentation generated successfully.")
-        JSONResponse(content={"session_id": request.session_id})
+        logger.info("✅ content segmentation generated successfully.")
+        return JSONResponse(content={"session_id": request.session_id})
 
     except HTTPException as http_exc:
         raise http_exc
 
     except Exception as e:
-        logger.exception(f"Error during content segmentation: {e}")
+        logger.exception(f"❌ Error during content segmentation: {e}")
         raise HTTPException(
             status_code=500,
             detail=f"content segmentation failed: {e}"
@@ -576,5 +703,136 @@ def search_content(request: SearchRequest):
             detail=f"Search failed: {e}"
         )
 
+@router.get("/check-recorded-videos")
+def check_recorded_videos(x_session_id: Optional[str] = Header(None)):
+    """
+    Check which video files were saved for a session after RTSP recording.
+    Returns the priority-ordered available video (back > board > front).
+
+    """
+    if not x_session_id:
+        raise HTTPException(
+            status_code=400, detail="Missing required header: x-session-id"
+        )
+    
+    try:
+        project_config = RuntimeConfig.get_section("Project")
+        base_path = os.path.join(
+            project_config.get("location"),
+            project_config.get("name"),
+            x_session_id
+        )
+        
+        if not os.path.exists(base_path):
+            logger.warn(f"Session path does not exist: {base_path}")
+            return JSONResponse(
+                content={
+                    "session_id": x_session_id,
+                    "back": None,
+                    "board": None,
+                    "front": None,
+                    "selected_video": None,
+                    "message": "No session path found"
+                },
+                status_code=200
+            )
+        
+        # Check which videos exist
+        videos = {
+            "back": None,
+            "board": None, 
+            "front": None,
+        }
+        
+        back_path = os.path.join(base_path, "back.mp4")
+        content_path = os.path.join(base_path, "content.mp4") 
+        front_path = os.path.join(base_path, "front.mp4")
+        
+        if os.path.exists(back_path):
+            videos["back"] = back_path
+        if os.path.exists(content_path):
+            videos["board"] = content_path  
+        if os.path.exists(front_path):
+            videos["front"] = front_path
+        
+        # Select highest priority video (back > board > front)
+        selected_video = None
+        if videos["back"]:
+            selected_video = "back"
+        elif videos["board"]:
+            selected_video = "board"
+        elif videos["front"]:
+            selected_video = "front"
+        
+        return JSONResponse(
+            content={
+                "session_id": x_session_id,
+                "back": videos["back"],
+                "board": videos["board"],
+                "front": videos["front"],
+                "selected_video": selected_video,
+                "selected_path": videos[selected_video] if selected_video else None
+            },
+            status_code=200
+        )
+        
+    except Exception as e:
+        logger.error(f"Error checking recorded videos: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/recorded-video/{videoType}")
+def get_recorded_video(videoType: str, x_session_id: Optional[str] = Header(None), session_id: Optional[str] = None):
+    """
+    Stream a recorded video file (back.mp4, board.mp4, or front.mp4).
+
+    """
+    # Accept session ID from either header or query parameter
+    actual_session_id = x_session_id or session_id
+    if not actual_session_id:
+        raise HTTPException(
+            status_code=400, detail="Missing required session ID: x-session-id header or ?session_id query parameter"
+        )
+    
+    if videoType not in ['back', 'board', 'front']:
+        raise HTTPException(
+            status_code=400, detail=f"Invalid videoType: {videoType}. Must be 'back', 'board', or 'front'"
+        )
+    
+    try:
+        backend_video_type = "content" if videoType == "board" else videoType
+        
+        project_config = RuntimeConfig.get_section("Project")
+        video_path = os.path.join(
+            project_config.get("location"),
+            project_config.get("name"),
+            actual_session_id,
+            f"{backend_video_type}.mp4"
+        )
+        
+        if not os.path.exists(video_path):
+            raise HTTPException(
+                status_code=404, 
+                detail=f"Video file not found: {videoType}.mp4"
+            )
+        
+        logger.info(f"Serving video file: {video_path} for session {actual_session_id}")
+        
+        file_response = FileResponse(
+            path=video_path,
+            media_type="video/mp4",
+            filename=f"{videoType}.mp4"
+        )
+        file_response.headers["Accept-Ranges"] = "bytes"
+        file_response.headers["Access-Control-Allow-Origin"] = "*"
+        file_response.headers["Cache-Control"] = "no-cache"
+        
+        return file_response
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error serving recorded video: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
 def register_routes(app: FastAPI):
     app.include_router(router)

education-ai-suite/smart-classroom/dto/video_metadata_dto.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class VideoMetadataRequest(BaseModel):
+    session_id: str
+    video_file_path: str
+
+class VideoDurationRequest(BaseModel):
+    duration: float
+

education-ai-suite/smart-classroom/pipeline.py

@@ -16,7 +16,8 @@
 from pathlib import Path
 import json
 from utils.faiss_content_search import FaissContentSearcher
-
+from utils.media_validation_service import MediaValidationService
+from utils.session_state_manager import SessionState
 import time
 logger = logging.getLogger(__name__)
 
@@ -194,6 +195,19 @@ def run_content_segmentation(self):
 
         transcription_path = os.path.join(session_dir, "transcription.txt")
 
+        session_state = SessionState.get_session_state(self.session_id)
+        # VALIDATION: Check media duration match before processing
+        is_valid, error_msg = MediaValidationService.validate_duration_match(self.session_id)
+        
+        if not is_valid:
+            SessionState.clear_session(self.session_id)
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=error_msg
+            )
+
+        logger.info(f"✅ Validation passed - proceeding with content segmentation")
+
         try:
             transcript_text = StorageManager.read_text_file(transcription_path)
 
@@ -256,6 +270,9 @@ def run_content_segmentation(self):
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=f"Error during topic segmentation: {e}"
             )
+        finally:
+            # Clean up session state after processing
+            SessionState.clear_session(self.session_id)
 
 
     def search_content(self, query: str, top_k: int = 5):