app.py

import streamlit as st
import cv2
import requests
import base64
import tempfile
import os
import time
from typing import Generator, Tuple

# --------------------------
# Configuration
# --------------------------
API_KEY = os.getenv("GEMINI_API_KEY")  # Fetch API key from Hugging Face secrets
API_URL = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key={API_KEY}"
SYSTEM_PROMPT = '''
You are a next-generation AI-driven military surveillance officer, embedded in a real-time battlefield monitoring system. Your core function is to analyze live battlefield footage, identify threats, and provide precise, structured intelligence reports for tactical decision-making.

Your analysis must be accurate, number-driven, and predictive. Focus on exact troop counts, vehicle tracking, movement patterns, and enemy strategies. Every report must be structured, concise, and mission-critical.

🔴 Mission-Critical Intelligence Goals:
1️⃣ Enemy Force Tracking (Exact Numbers & Movements)

Identify and count enemy troops, vehicles, and support assets.
Categorize unit types (infantry, mechanized, special forces, UAVs, artillery).
Track movement direction, speed, formation type (spread-out, tight, advancing, retreating).
Detect supply chains, reinforcements, and potential weak points.
2️⃣ Strategic Threat Assessment

Assign a Threat Level (Low, Moderate, High, Critical) based on:
Numerical strength vs. friendly forces
Equipment capability (heavy weapons, drones, armored units)
Terrain control & tactical positioning
Identify sniper nests, ambush zones, flanking attempts.
Recognize enemy command centers & key leadership units.
3️⃣ Real-Time Vehicle & Artillery Monitoring

Detect and count tanks, armored personnel carriers (APCs), artillery, UAVs, supply trucks.
Differentiate between stationary vs. mobile units.
Identify heavy artillery positions and potential missile launch sites.
Track drones and air support operations.
4️⃣ Tactical Terrain Intelligence

Identify strategic high ground, choke points, supply routes.
Recognize defensive fortifications, trenches, camouflaged positions.
Determine obstacles affecting movement (bridges, rivers, urban zones, minefields).
5️⃣ Predictive Battlefield Analytics

Forecast enemy movements & possible attack vectors.
Detect supply vulnerabilities, reinforcements, and fallback positions.
Predict next-phase engagements based on movement patterns.
Provide recommended counter-maneuvers and precision strikes.
🟢 Tactical Intelligence Report Format (Example)
🛰️ SITUATION UPDATE: Enemy forces detected (450 infantry, 32 armored vehicles, 6 UAVs, 4 artillery units) advancing eastward at 8 km/h toward Hill 72.
⚠️ THREAT ASSESSMENT: High-risk engagement. Enemy equipped with anti-armor weaponry, air support.
📡 OPERATIONAL INTELLIGENCE: Flanking maneuver detected—50 soldiers moving through dense foliage to breach our western perimeter.
🚨 VEHICLE & ARTILLERY STATUS:
Tanks: 12 moving in formation
APCs: 20 transporting troops
UAVs: 6 scouting ahead
Artillery: 4 positioned near enemy HQ
🔺 COMMAND RECOMMENDATION: Deploy airstrikes on artillery, engage flanking troops with ambush teams, reposition anti-tank defenses along eastern approach.
🟠 Intelligence Prioritization
✅ Numbers First: Every report must include exact troop & vehicle counts.
✅ Tactical Clarity: No vague descriptions—only clear, structured battle intelligence.
✅ Predict & Neutralize: AI must anticipate enemy next moves and suggest counter-tactics.
✅ Mission-Focused Communication: Short, clear, actionable updates—no unnecessary details.




'''

# --------------------------
# Gemini API Client
# --------------------------
class GeminiClient:
    def __init__(self, api_key: str, api_url: str):
        self.api_key = api_key
        self.api_url = api_url
        self.session = requests.Session()
        self.timeout = 30

    def analyze_frame(self, frame_b64: str, timestamp: int) -> str:
        """Send frame to Gemini API for analysis."""
        payload = {
            "contents": [{
                "parts": [
                    {"text": f"Analyze this battlefield image from {timestamp} seconds:"},
                    {"inline_data": {"mime_type": "image/jpeg", "data": frame_b64}}
                ]
            }],
            "systemInstruction": {
                "parts": [{"text": SYSTEM_PROMPT}]
            }
        }

        try:
            response = self.session.post(
                self.api_url,
                json=payload,
                timeout=self.timeout,
                headers={"Content-Type": "application/json"}
            )
            response.raise_for_status()
            return self._parse_response(response.json())
        except requests.exceptions.RequestException as e:
            return f"Analysis error: {str(e)}"

    @staticmethod
    def _parse_response(response: dict) -> str:
        """Extract response text from Gemini API response."""
        try:
            return response["candidates"][0]["content"]["parts"][0]["text"]
        except (KeyError, IndexError):
            return "No analysis available"

# --------------------------
# Helper Functions
# --------------------------
def frame_to_base64(frame):
    """Convert an image frame to base64 format."""
    _, buffer = cv2.imencode(".jpg", frame)
    return base64.b64encode(buffer).decode("utf-8")

def extract_video_frame(video_path, timestamp):
    """Extract a frame at a specific timestamp from the video."""
    cap = cv2.VideoCapture(video_path)
    cap.set(cv2.CAP_PROP_POS_MSEC, timestamp * 1000)
    success, frame = cap.read()
    cap.release()
    return frame if success else None

def download_video(video_url):
    """Download video from URL to a temporary file."""
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as temp_file:
            response = requests.get(video_url, stream=True, timeout=30)
            response.raise_for_status()
            for chunk in response.iter_content(chunk_size=8192):
                temp_file.write(chunk)
            return temp_file.name
    except Exception as e:
        st.error(f"Video download failed: {str(e)}")
        return None

# --------------------------
# Streamlit App
# --------------------------
st.title("🎥 Military Surveillance AI")
st.write("Upload a battlefield surveillance video to analyze.")

video_url = st.text_input("Enter Video URL:")
if st.button("Analyze Video") and video_url:
    st.info("Downloading video...")
    video_path = download_video(video_url)
    
    if video_path:
        client = GeminiClient(API_KEY, API_URL)
        st.success("Video downloaded successfully!")
        
        st.info("Processing video and analyzing frames...")
        log = []
        
        # Extract frames and analyze
        for timestamp in range(10, 40, 10):  # Analyze at 10s, 20s, 30s
            frame = extract_video_frame(video_path, timestamp)
            if frame is None:
                log.append(f"[{timestamp}s] Frame extraction failed ❌")
                continue
            
            analysis = client.analyze_frame(frame_to_base64(frame), timestamp)
            log.append(f"[{timestamp}s] {analysis}")
            st.write(f"### Timestamp: {timestamp}s")
            st.image(frame, caption=f"Frame at {timestamp}s", use_column_width=True)
            st.write(analysis)
            time.sleep(2)  # Simulate processing delay
            
        st.success("Analysis complete! ✅")
        st.text_area("Summary of Analysis:", "\n".join(log), height=200)