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)