src/demo/app.py

# app.py  ─────────────────────────────────────────────────────────
"""
Laban Movement Analysis – modernised Gradio Space
Author: Csaba (BladeSzaSza)
"""
import gradio as gr
import os
from pathlib import Path
# from backend.gradio_labanmovementanalysis import LabanMovementAnalysis
from backend.gradio_labanmovementanalysis import LabanMovementAnalysis
from gradio_overlay_video import OverlayVideo

# Import agent API if available
# Initialize agent API if available
agent_api = None
try:
    from gradio_labanmovementanalysis.agent_api import (
        LabanAgentAPI,
        PoseModel,
        MovementDirection,
        MovementIntensity
    )
    agent_api = LabanAgentAPI()
    HAS_AGENT_API = True
except Exception as e:
    print(f"Warning: Agent API not available: {e}")
    agent_api = None
    HAS_AGENT_API = False
# Initialize components
try:
    analyzer = LabanMovementAnalysis(
        enable_visualization=True
    )
    print("✅ Core features initialized successfully")
except Exception as e:
    print(f"Warning: Some features may not be available: {e}")
    analyzer = LabanMovementAnalysis()


def process_video_enhanced(video_input, model, enable_viz, include_keypoints):
    """Enhanced video processing with all new features."""
    if not video_input:
        return {"error": "No video provided"}, None
    
    try:
        # Handle both file upload and URL input
        video_path = video_input.name if hasattr(video_input, 'name') else video_input
        
        json_result, viz_result = analyzer.process_video(
            video_path,
            model=model,
            enable_visualization=enable_viz,
            include_keypoints=include_keypoints
        )
        return json_result, viz_result
    except Exception as e:
        error_result = {"error": str(e)}
        return error_result, None

def process_video_standard(video : str, model : str, include_keypoints : bool) -> dict:
    """
    Processes a video file using the specified pose estimation model and returns movement analysis results.

    Args:
        video (str): Path to the video file to be analyzed.
        model (str): The name of the pose estimation model to use (e.g., "mediapipe-full", "movenet-thunder", etc.).
        include_keypoints (bool): Whether to include raw keypoint data in the output.

    Returns:
        dict:
            - A dictionary containing the movement analysis results in JSON format, or an error message if processing fails.


    Notes:
        - Visualization is disabled in this standard processing function.
        - If the input video is None, both return values will be None.
        - If an error occurs during processing, the first return value will be a dictionary with an "error" key.
    """
    if video is None:
        return None
    try:
        json_output, _ = analyzer.process_video(
            video,
            model=model,
            enable_visualization=False,
            include_keypoints=include_keypoints
        )
        return json_output
    except (RuntimeError, ValueError, OSError) as e:
        return {"error": str(e)}

def process_video_for_agent(video, model, output_format="summary"):
    """Process video with agent-friendly output format."""
    if not HAS_AGENT_API or agent_api is None:
        return {"error": "Agent API not available"}
    
    if not video:
        return {"error": "No video provided"}
    
    try:
        model_enum = PoseModel(model)
        result = agent_api.analyze(video, model=model_enum, generate_visualization=False)
        
        if output_format == "summary":
            return {"summary": agent_api.get_movement_summary(result)}
        elif output_format == "structured":
            return {
                "success": result.success,
                "direction": result.dominant_direction.value,
                "intensity": result.dominant_intensity.value,
                "speed": result.dominant_speed,
                "fluidity": result.fluidity_score,
                "expansion": result.expansion_score,
                "segments": len(result.movement_segments)
            }
        else:  # json
            return result.raw_data
    except Exception as e:
        return {"error": str(e)}

# Batch processing removed due to MediaPipe compatibility issues

# process_standard_for_agent is now imported from backend

# Movement filtering removed due to MediaPipe compatibility issues

# Import agentic analysis functions from backend
try:
    from gradio_labanmovementanalysis.agentic_analysis import (
        generate_agentic_analysis,
        process_standard_for_agent
    )
except ImportError:
    # Fallback if backend module is not available
    def generate_agentic_analysis(json_data, analysis_type, filter_direction="any", filter_intensity="any", filter_min_fluidity=0.0, filter_min_expansion=0.0):
        return {"error": "Agentic analysis backend not available"}
    
    def process_standard_for_agent(json_data, output_format="summary"):
        return {"error": "Agent conversion backend not available"}

# ── 4.  Build UI ─────────────────────────────────────────────────
def create_demo() -> gr.Blocks:
    with gr.Blocks(
        title="Laban Movement Analysis",
        theme='gstaff/sketch',
        fill_width=True,
    ) as demo:
        # gr.api(process_video_standard, api_name="process_video") 
        # ── Hero banner ──
        gr.Markdown(
            """
            # 🩰 Laban Movement Analysis 
            
            Pose estimation • AI action recognition • Movement Analysis 
            """
        )
        with gr.Tabs():
            # Tab 1: Standard Analysis
            with gr.Tab("🎭 Standard Analysis"):
                gr.Markdown("""
                ### Upload a video file to analyze movement using traditional LMA metrics with pose estimation.
                """)
                # ── Workspace ──
                with gr.Row(equal_height=True):
                    # Input column
                    with gr.Column(scale=1, min_width=260):
                        
                        analyze_btn_enh = gr.Button("🚀 Analyze Movement", variant="primary", size="lg")
                        video_in = gr.Video(label="Upload Video", sources=["upload"], format="mp4")
                        # URL input option
                        url_input_enh = gr.Textbox(
                            label="Or Enter Video URL",
                            placeholder="YouTube URL, Vimeo URL, or direct video URL",
                            info="Leave file upload empty to use URL"
                        )
                       
                        gr.Markdown("**Model Selection**")
                        
                        model_sel = gr.Dropdown(
                            choices=[
                                # MediaPipe variants
                                "mediapipe-lite", "mediapipe-full", "mediapipe-heavy",
                                # MoveNet variants
                                "movenet-lightning", "movenet-thunder",
                                # YOLO v8 variants
                                "yolo-v8-n", "yolo-v8-s", "yolo-v8-m", "yolo-v8-l", "yolo-v8-x",
                                # YOLO v11 variants
                                "yolo-v11-n", "yolo-v11-s", "yolo-v11-m", "yolo-v11-l", "yolo-v11-x"
                            ],
                            value="mediapipe-full",
                            label="Advanced Pose Models",
                            info="15 model variants available"
                        )
                        
                        with gr.Accordion("Analysis Options", open=False):
                            enable_viz = gr.Radio([("Create", 1), ("Dismiss", 0)], value=1, label="Visualization")
                            include_kp = gr.Radio([("Include", 1), ("Exclude", 0)], value=1, label="Raw Keypoints")

                        gr.Examples(
                            examples=[
                                ["examples/balette.mp4"],
                                ["https://www.youtube.com/shorts/RX9kH2l3L8U"],
                                ["https://vimeo.com/815392738"],
                                ["https://vimeo.com/548964931"],
                                ["https://videos.pexels.com/video-files/5319339/5319339-uhd_1440_2560_25fps.mp4"],
                            ],
                            inputs=url_input_enh,
                            label="Examples"
                        )


                    # Output column
                    with gr.Column(scale=2, min_width=320):
                        viz_out = gr.Video(label="Annotated Video", scale=1, height=400)
                        with gr.Accordion("Raw JSON", open=True):
                            json_out = gr.JSON(label="Movement Analysis", elem_classes=["json-output"])

                # Wiring
                def process_enhanced_input(file_input, url_input, model, enable_viz, include_keypoints):
                    """Process either file upload or URL input."""
                    video_source = file_input if file_input else url_input
                    [json_out, viz_out] = process_video_enhanced(video_source, model, enable_viz, include_keypoints)
                    overlay_video.value = (None, json_out)
                    return [json_out, viz_out]
                
                analyze_btn_enh.click(
                    fn=process_enhanced_input,
                    inputs=[video_in, url_input_enh, model_sel, enable_viz, include_kp],
                    outputs=[json_out, viz_out],
                    api_name="analyze_enhanced"
                )

            with gr.Tab("🎬 Overlayed Visualisation"):
                gr.Markdown(
                    "# 🩰 Interactive Pose Visualization\n"
                    "## See the movement analysis in action with an interactive overlay. "
                    "Analyze video @ 🎬 Standard Analysis tab"
                )
                with gr.Row(equal_height=True, min_height=240):
                    with gr.Column(scale=1):
                        overlay_video = OverlayVideo(
                                value=(None, json_out),
                                autoplay=True,
                                interactive=False
                            )
                
                       
                # Update overlay when JSON changes
                def update_overlay(json_source):
                    """Update overlay video with JSON data from analysis or upload."""
                    if json_source:
                        return OverlayVideo(value=("", json_source), autoplay=True, interactive=False)
                    return OverlayVideo(value=("", None), autoplay=True, interactive=False)
                
                # Connect JSON output from analysis to overlay
                json_out.change(
                    fn=update_overlay,
                    inputs=[json_out],
                    outputs=[overlay_video]
                )

            # Tab 3: Agentic Analysis
            with gr.Tab("🤖 Agentic Analysis"):
                gr.Markdown("""
                ### Intelligent Movement Interpretation
                AI-powered analysis using the processed data from the Standard Analysis tab.
                """)
                
                with gr.Row(equal_height=True):
                    # Left column - Video display (sourced from first tab)
                    with gr.Column(scale=1, min_width=400):
                        gr.Markdown("**Source Video** *(from Standard Analysis)*")
                        agentic_video_display = gr.Video(
                            label="Analyzed Video", 
                            interactive=False,
                            height=350
                        )
                        
                        # Model info display (sourced from first tab)
                        gr.Markdown("**Model Used** *(from Standard Analysis)*")
                        agentic_model_display = gr.Textbox(
                            label="Pose Model",
                            interactive=False,
                            value="No analysis completed yet"
                        )
                    
                    # Right column - Analysis options and output
                    with gr.Column(scale=1, min_width=400):
                        gr.Markdown("**Analysis Type**")
                        agentic_analysis_type = gr.Radio(
                            choices=[
                                ("🎯 SUMMARY", "summary"),
                                ("📊 STRUCTURED", "structured"), 
                                ("🔍 MOVEMENT FILTERS", "movement_filters")
                            ],
                            value="summary",
                            label="Choose Analysis",
                            info="Select the type of intelligent analysis"
                        )
                        
                        # Movement filters options (shown when movement_filters is selected)
                        with gr.Group(visible=False) as movement_filter_options:
                            gr.Markdown("**Filter Criteria**")
                            filter_direction = gr.Dropdown(
                                choices=["any", "up", "down", "left", "right", "forward", "backward", "stationary"],
                                value="any",
                                label="Dominant Direction"
                            )
                            filter_intensity = gr.Dropdown(
                                choices=["any", "low", "medium", "high"],
                                value="any", 
                                label="Movement Intensity"
                            )
                            filter_min_fluidity = gr.Slider(0.0, 1.0, 0.0, label="Minimum Fluidity Score")
                            filter_min_expansion = gr.Slider(0.0, 1.0, 0.0, label="Minimum Expansion Score")
                        
                        analyze_agentic_btn = gr.Button("🚀 Generate Analysis", variant="primary", size="lg")
                        
                        # Output display
                        with gr.Accordion("Analysis Results", open=True):
                            agentic_output = gr.JSON(label="Intelligent Analysis Results")

                # Show/hide movement filter options based on selection
                def toggle_filter_options(analysis_type):
                    return gr.Group(visible=(analysis_type == "movement_filters"))
                
                agentic_analysis_type.change(
                    fn=toggle_filter_options,
                    inputs=[agentic_analysis_type],
                    outputs=[movement_filter_options]
                )
                
                # Update video display when standard analysis completes
                def update_agentic_video_display(video_input, url_input, model):
                    """Update agentic tab with video and model from standard analysis."""
                    video_source = video_input if video_input else url_input
                    return video_source, f"Model: {model}"
                
                # Link to standard analysis inputs
                video_in.change(
                    fn=update_agentic_video_display,
                    inputs=[video_in, url_input_enh, model_sel],
                    outputs=[agentic_video_display, agentic_model_display]
                )
                
                url_input_enh.change(
                    fn=update_agentic_video_display,
                    inputs=[video_in, url_input_enh, model_sel],
                    outputs=[agentic_video_display, agentic_model_display]
                )
                
                model_sel.change(
                    fn=update_agentic_video_display,
                    inputs=[video_in, url_input_enh, model_sel],
                    outputs=[agentic_video_display, agentic_model_display]
                )
                
                # Hook up the Generate Analysis button
                def process_agentic_analysis(json_data, analysis_type, filter_direction, filter_intensity, filter_min_fluidity, filter_min_expansion):
                    """Process agentic analysis based on user selection."""
                    return generate_agentic_analysis(
                        json_data, 
                        analysis_type, 
                        filter_direction, 
                        filter_intensity, 
                        filter_min_fluidity, 
                        filter_min_expansion
                    )
                
                analyze_agentic_btn.click(
                    fn=process_agentic_analysis,
                    inputs=[
                        json_out,  # JSON data from standard analysis
                        agentic_analysis_type,
                        filter_direction,
                        filter_intensity, 
                        filter_min_fluidity,
                        filter_min_expansion
                    ],
                    outputs=[agentic_output],
                    api_name="analyze_agentic"
                )
                
                # Auto-update agentic analysis when JSON changes and analysis type is summary
                def auto_update_summary(json_data, analysis_type):
                    """Auto-update with summary when new analysis is available."""
                    if json_data and analysis_type == "summary":
                        return generate_agentic_analysis(json_data, "summary")
                    return None
                
                json_out.change(
                    fn=auto_update_summary,
                    inputs=[json_out, agentic_analysis_type],
                    outputs=[agentic_output]
                )

            # Tab 4: About
            with gr.Tab("ℹ️ About"):
                gr.Markdown("""
                # 🩰 Developer Journey: Laban Movement Analysis
                
                ## 🎯 Project Vision
                
                Created to bridge the gap between traditional **Laban Movement Analysis (LMA)** principles and modern **AI-powered pose estimation**, this platform represents a comprehensive approach to understanding human movement through technology.
                
                ## 🛠️ Technical Architecture
                
                ### **Core Foundation**
                - **15 Pose Estimation Models** from diverse sources and frameworks
                - **Multi-format Video Processing** with URL support (YouTube, Vimeo, direct links)
                - **Real-time Analysis Pipeline** with configurable model selection
                - **MCP-Compatible API** for AI agent integration
                
                ### **Pose Model Ecosystem**
                ```
                📊 MediaPipe Family (Google)     → 3 variants (lite/full/heavy)
                ⚡ MoveNet Family (TensorFlow)   → 2 variants (lightning/thunder)  
                🎯 YOLO v8 Family (Ultralytics) → 5 variants (n/s/m/l/x)
                🔥 YOLO v11 Family (Ultralytics)→ 5 variants (n/s/m/l/x)
                ```
                
                ## 🎨 Innovation Highlights
                
                ### **1. Custom Gradio Component: `gradio_overlay_video`**
                - **Layered Visualization**: Controlled overlay of pose data on original video
                - **Interactive Controls**: Frame-by-frame analysis with movement metrics
                - **Synchronized Playback**: Real-time correlation between video and data
                
                ### **2. Agentic Analysis Engine**
                Beyond raw pose detection, we've developed intelligent interpretation layers:
                
                - **🎯 SUMMARY**: Narrative movement interpretation with temporal pattern analysis
                - **📊 STRUCTURED**: Comprehensive quantitative breakdowns with statistical insights
                - **🔍 MOVEMENT FILTERS**: Advanced pattern detection with customizable criteria
                
                ### **3. Temporal Pattern Recognition**
                - **Movement Consistency Tracking**: Direction and intensity variation analysis
                - **Complexity Scoring**: Multi-dimensional movement sophistication metrics
                - **Sequence Detection**: Continuous movement pattern identification
                - **Laban Integration**: Professional movement quality assessment using LMA principles
                
                ## 📈 Processing Pipeline
                
                ```mermaid
                Video Input → Pose Detection → LMA Analysis → JSON Output
                     ↓              ↓              ↓           ↓
                URL/Upload → 15 Models → Temporal → Visualization
                     ↓              ↓        Patterns    ↓
                Preprocessing → Keypoints → Metrics → Agentic Analysis
                ```
                
                ## 🎭 Laban Movement Analysis Integration
                
                Our implementation translates raw pose coordinates into meaningful movement qualities:
                
                - **Effort Qualities**: Intensity, speed, and flow characteristics
                - **Space Usage**: Expansion patterns and directional preferences  
                - **Temporal Dynamics**: Rhythm, acceleration, and movement consistency
                - **Quality Assessment**: Fluidity scores and movement sophistication
                
                ## 🔬 Technical Achievements
                
                ### **Multi-Source Model Integration**
                Successfully unified models from different frameworks:
                - Google's MediaPipe (BlazePose architecture)
                - TensorFlow's MoveNet (lightweight and accurate variants)
                - Ultralytics' YOLO ecosystem (object detection adapted for pose)
                
                ### **Real-Time Processing Capabilities**
                - **Streaming Support**: Frame-by-frame processing with temporal continuity
                - **Memory Optimization**: Efficient handling of large video files
                - **Error Recovery**: Graceful handling of pose detection failures
                
                ### **Agent-Ready Architecture**
                - **MCP Server Integration**: Compatible with AI agent workflows
                - **Structured API**: RESTful endpoints for programmatic access
                - **Flexible Output Formats**: JSON, visualization videos, and metadata
                
                ## 🌟 Future Roadmap
                
                - **3D Pose Integration**: Depth-aware movement analysis
                - **Multi-Person Tracking**: Ensemble and group movement dynamics
                - **Real-Time Streaming**: Live movement analysis capabilities
                - **Machine Learning Enhancement**: Custom models trained on movement data
                
                ## 🔧 Built With
                
                - **Frontend**: Gradio 5.33+ with custom Svelte components
                - **Backend**: Python with FastAPI and async processing
                - **Computer Vision**: MediaPipe, TensorFlow, PyTorch, Ultralytics
                - **Analysis**: NumPy, OpenCV, custom Laban algorithms
                - **Deployment**: Hugging Face Spaces with Docker support
                
                ---
                
                ### 👨‍💻 Created by **Csaba Bolyós**
                
                *Combining classical movement analysis with cutting-edge AI to unlock new possibilities in human movement understanding.*
                
                **Connect:**  
                [GitHub](https://github.com/bladeszasza) • [Hugging Face](https://huggingface.co/BladeSzaSza) • [LinkedIn](https://www.linkedin.com/in/csaba-bolyós-00a11767/)
                
                ---
                
                > *"Movement is a language. Technology helps us understand what the body is saying."*
                """)
         
        # Footer
        with gr.Row():
            gr.Markdown(
                """
                **Built by Csaba Bolyós**  
                [GitHub](https://github.com/bladeszasza) • [HF](https://huggingface.co/BladeSzaSza) • [LinkedIn](https://www.linkedin.com/in/csaba-bolyós-00a11767/)
                """
            )
    return demo


if __name__ == "__main__":
    demo = create_demo()
    demo.launch(server_name="0.0.0.0",
                share=True,
                server_port=int(os.getenv("PORT", 7860)),
                mcp_server=True)