246 lines
8.8 KiB
Python
246 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Orlan Military Drone Test Script
|
|
Simulates an Orlan military drone approaching from long distance to a target facility.
|
|
This test demonstrates the critical alert system for high-threat drones.
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
import math
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# Disable SSL warnings for self-signed certificates
|
|
import warnings
|
|
warnings.filterwarnings('ignore', message='Unverified HTTPS request')
|
|
|
|
# Configuration from environment variables
|
|
# Tests default to localhost:3002 for local development
|
|
API_BASE_URL = os.getenv('API_BASE_URL', 'http://localhost:3002/api')
|
|
BASE_PATH = os.getenv('VITE_BASE_PATH', '').rstrip('/')
|
|
|
|
# Authentication configuration
|
|
USERNAME = os.getenv('TEST_USERNAME', 'admin')
|
|
PASSWORD = os.getenv('TEST_PASSWORD', 'admin123')
|
|
|
|
# If BASE_PATH is set, construct the full URL
|
|
if BASE_PATH and not API_BASE_URL.endswith('/api'):
|
|
# Extract domain from API_BASE_URL and add base path
|
|
domain = API_BASE_URL.replace('/api', '').replace('/drones/api', '').replace('/uggla/api', '')
|
|
API_BASE_URL = f"{domain}{BASE_PATH}/api"
|
|
|
|
print(f"🔗 Using API Base URL: {API_BASE_URL}")
|
|
|
|
# Global variable to store authentication token
|
|
AUTH_TOKEN = None
|
|
|
|
def authenticate():
|
|
"""Authenticate with the API and get access token"""
|
|
global AUTH_TOKEN
|
|
|
|
login_data = {
|
|
"username": USERNAME,
|
|
"password": PASSWORD
|
|
}
|
|
|
|
try:
|
|
print(f"🔐 Authenticating as user: {USERNAME}")
|
|
response = requests.post(f"{API_BASE_URL}/auth/login", json=login_data, verify=False)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
AUTH_TOKEN = data.get('token')
|
|
if AUTH_TOKEN:
|
|
print("✅ Authentication successful")
|
|
return True
|
|
else:
|
|
print("❌ No token received in response")
|
|
return False
|
|
else:
|
|
print(f"❌ Authentication failed: {response.status_code}")
|
|
print(f"Response: {response.text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Authentication error: {e}")
|
|
return False
|
|
|
|
def get_auth_headers():
|
|
"""Get headers with authentication token"""
|
|
if AUTH_TOKEN:
|
|
return {
|
|
"Authorization": f"Bearer {AUTH_TOKEN}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
return {"Content-Type": "application/json"}
|
|
|
|
def haversine_distance(lat1, lon1, lat2, lon2):
|
|
"""Calculate distance between two points in kilometers"""
|
|
R = 6371 # Earth's radius in kilometers
|
|
|
|
dlat = math.radians(lat2 - lat1)
|
|
dlon = math.radians(lon2 - lon1)
|
|
|
|
a = (math.sin(dlat/2) * math.sin(dlat/2) +
|
|
math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *
|
|
math.sin(dlon/2) * math.sin(dlon/2))
|
|
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
|
|
|
|
return R * c
|
|
|
|
def rssi_from_distance(distance_km):
|
|
"""Calculate RSSI based on distance"""
|
|
# Orlan drones have powerful transmission systems
|
|
# Base RSSI at 1km = -60 dBm (stronger than consumer drones)
|
|
base_rssi = -60
|
|
path_loss_exponent = 2.5 # Lower path loss due to better equipment
|
|
|
|
if distance_km < 0.001: # Less than 1 meter
|
|
return -30
|
|
|
|
rssi = base_rssi - (20 * path_loss_exponent * math.log10(distance_km))
|
|
return max(rssi, -100) # Cap at -100 dBm
|
|
|
|
def fetch_devices():
|
|
"""Fetch devices from API"""
|
|
try:
|
|
response = requests.get(f"{API_BASE_URL}/devices/map", verify=False)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return data.get('data', [])
|
|
except Exception as e:
|
|
print(f"Error fetching devices: {e}")
|
|
return []
|
|
|
|
def send_detection(device, drone_lat, drone_lon, distance_km, step, total_steps):
|
|
"""Send a detection to the API"""
|
|
rssi = rssi_from_distance(distance_km)
|
|
|
|
# Use the EXACT standard payload format - NEVER change this!
|
|
detection_data = {
|
|
"device_id": device["id"],
|
|
"geo_lat": drone_lat,
|
|
"geo_lon": drone_lon,
|
|
"device_timestamp": int(time.time() * 1000), # Current timestamp in milliseconds
|
|
"drone_type": 1, # Orlan/Military type
|
|
"rssi": int(rssi),
|
|
"freq": 24, # Orlan operates on 2.4 GHz (24 = 2400 MHz)
|
|
"drone_id": 1000 + step # Unique drone ID for tracking
|
|
}
|
|
|
|
try:
|
|
response = requests.post(f"{API_BASE_URL}/detections", json=detection_data, verify=False)
|
|
if response.status_code == 201:
|
|
print(f"🚨 ORLAN DETECTION {step}/{total_steps}: Distance={distance_km:.2f}km, RSSI={rssi:.0f}dBm")
|
|
return True
|
|
else:
|
|
print(f"Failed to send detection: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"Error sending detection: {e}")
|
|
return False
|
|
|
|
def run_orlan_approach_scenario():
|
|
"""Run the Orlan approach scenario"""
|
|
print("=" * 60)
|
|
print("🚨 ORLAN MILITARY DRONE APPROACH SIMULATION")
|
|
print("=" * 60)
|
|
print("This simulation demonstrates:")
|
|
print("- Long-range Orlan drone detection")
|
|
print("- Critical alert escalation")
|
|
print("- Automatic threat assessment")
|
|
print("- Real-time approach tracking")
|
|
print("=" * 60)
|
|
|
|
# Fetch devices
|
|
devices = fetch_devices()
|
|
if not devices:
|
|
print("❌ No devices found!")
|
|
return
|
|
|
|
# Use the first device as target (usually Arlanda)
|
|
target_device = devices[0]
|
|
print(f"🎯 Target: {target_device['name']}")
|
|
print(f"📍 Location: {target_device['geo_lat']:.4f}, {target_device['geo_lon']:.4f}")
|
|
|
|
# Starting position: 15km northeast of target (simulating approach from hostile territory)
|
|
start_distance = 15.0 # km
|
|
angle = math.radians(45) # 45 degrees (northeast)
|
|
|
|
# Calculate starting position
|
|
start_lat = target_device['geo_lat'] + (start_distance / 111.0) * math.cos(angle)
|
|
start_lon = target_device['geo_lon'] + (start_distance / (111.0 * math.cos(math.radians(target_device['geo_lat'])))) * math.sin(angle)
|
|
|
|
print(f"🛫 Orlan starting position: {start_lat:.4f}, {start_lon:.4f}")
|
|
print(f"📏 Initial distance: {start_distance:.1f}km")
|
|
print()
|
|
|
|
# Simulation parameters
|
|
total_steps = 30
|
|
final_distance = 0.05 # 50 meters final approach
|
|
|
|
print("Starting approach simulation...")
|
|
print()
|
|
|
|
for step in range(1, total_steps + 1):
|
|
# Calculate current distance (exponential approach for realistic acceleration)
|
|
progress = step / total_steps
|
|
# Use exponential curve for more realistic approach pattern
|
|
distance_km = start_distance * (1 - progress) ** 2 + final_distance * progress ** 2
|
|
|
|
# Calculate current position
|
|
current_lat = start_lat + (target_device['geo_lat'] - start_lat) * progress
|
|
current_lon = start_lon + (target_device['geo_lon'] - start_lon) * progress
|
|
|
|
# Send detection
|
|
success = send_detection(target_device, current_lat, current_lon, distance_km, step, total_steps)
|
|
|
|
if not success:
|
|
print(f"❌ Failed to send detection at step {step}")
|
|
continue
|
|
|
|
# Show threat escalation messages
|
|
if step == 1:
|
|
print(" 🔍 Initial long-range detection - monitoring")
|
|
elif distance_km < 10 and step < 10:
|
|
print(" ⚠️ Entering medium-range surveillance zone")
|
|
elif distance_km < 5:
|
|
print(" 🚨 HIGH ALERT: Orlan approaching critical zone")
|
|
elif distance_km < 1:
|
|
print(" 🔥 IMMEDIATE THREAT: Orlan within facility perimeter")
|
|
elif distance_km < 0.2:
|
|
print(" 💥 CRITICAL: Orlan directly overhead - TAKE COVER")
|
|
|
|
# Variable delay based on distance (faster updates as it gets closer)
|
|
if distance_km > 10:
|
|
delay = 3.0 # 3 seconds for long-range
|
|
elif distance_km > 5:
|
|
delay = 2.0 # 2 seconds for medium-range
|
|
elif distance_km > 1:
|
|
delay = 1.5 # 1.5 seconds for close-range
|
|
else:
|
|
delay = 1.0 # 1 second for critical proximity
|
|
|
|
time.sleep(delay)
|
|
|
|
print()
|
|
print("=" * 60)
|
|
print("🚨 ORLAN APPROACH SIMULATION COMPLETED")
|
|
print("=" * 60)
|
|
print("Summary:")
|
|
print(f"- Total detections sent: {total_steps}")
|
|
print(f"- Distance covered: {start_distance - final_distance:.1f}km")
|
|
print(f"- Target facility: {target_device['name']}")
|
|
print("- All alerts should have triggered critical notifications")
|
|
print("- Check the dashboard for real-time tracking")
|
|
print("=" * 60)
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
run_orlan_approach_scenario()
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ Simulation interrupted by user")
|
|
except Exception as e:
|
|
print(f"\n❌ Error during simulation: {e}")
|