diff --git a/sample_detection.json b/sample_detection.json new file mode 100644 index 0000000..821ac5f --- /dev/null +++ b/sample_detection.json @@ -0,0 +1,12 @@ +{ + "device_id": 1, + "drone_id": 2001, + "drone_type": 1, + "rssi": -72, + "freq": 2450, + "geo_lat": 59.3293, + "geo_lon": 18.0686, + "device_timestamp": 1692252000000, + "confidence_level": 0.92, + "signal_duration": 3200 +} diff --git a/test_drone_curl.sh b/test_drone_curl.sh new file mode 100644 index 0000000..ccfbf9e --- /dev/null +++ b/test_drone_curl.sh @@ -0,0 +1,58 @@ +#!/bin/bash +# Simple bash script to test drone detection API with curl + +API_BASE="http://selfservice.cqers.com/drones/api" +# Alternative for local testing: "http://localhost:3002/api" + +echo "🚁 Drone Detection API Test Script" +echo "==================================" + +# Test API health +echo "πŸ” Testing API health..." +curl -s "$API_BASE/health" | jq '.' || echo "Health check failed" +echo "" + +# Send test detection data +echo "πŸ“‘ Sending test drone detection..." + +# Generate timestamp +TIMESTAMP=$(date +%s)000 + +# Test detection payload +DETECTION_DATA='{ + "device_id": 1, + "drone_id": 1001, + "drone_type": 1, + "rssi": -65, + "freq": 2400, + "geo_lat": 59.3293, + "geo_lon": 18.0686, + "device_timestamp": '$TIMESTAMP', + "confidence_level": 0.85, + "signal_duration": 2500 +}' + +echo "Payload:" +echo "$DETECTION_DATA" | jq '.' +echo "" + +# Send the detection +echo "Sending detection..." +RESPONSE=$(curl -s -w "\nHTTP_STATUS:%{http_code}" \ + -X POST "$API_BASE/detections" \ + -H "Content-Type: application/json" \ + -d "$DETECTION_DATA") + +# Parse response +HTTP_BODY=$(echo "$RESPONSE" | sed '$d') +HTTP_STATUS=$(echo "$RESPONSE" | tail -n1 | sed 's/HTTP_STATUS://') + +echo "Response Status: $HTTP_STATUS" +echo "Response Body:" +echo "$HTTP_BODY" | jq '.' 2>/dev/null || echo "$HTTP_BODY" + +if [ "$HTTP_STATUS" = "201" ]; then + echo "βœ… Detection sent successfully!" +else + echo "❌ Detection failed with status $HTTP_STATUS" +fi diff --git a/test_drone_data.py b/test_drone_data.py new file mode 100644 index 0000000..1c1c115 --- /dev/null +++ b/test_drone_data.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +""" +Drone Detection Test Script +Sends simulated drone detection data to the API for testing purposes. +""" + +import requests +import json +import time +import random +from datetime import datetime + +# Configuration +API_BASE_URL = "http://selfservice.cqers.com/drones/api" +# Alternative for local testing: "http://localhost:3002/api" + +# Device configurations (simulating different detector devices) +DEVICES = [ + { + "id": 1, + "name": "Stockholm Central Detector", + "lat": 59.3293, + "lon": 18.0686 + }, + { + "id": 2, + "name": "Arlanda Airport Detector", + "lat": 59.6519, + "lon": 17.9186 + }, + { + "id": 3, + "name": "GΓΆteborg Harbor Detector", + "lat": 57.7089, + "lon": 11.9746 + } +] + +# Drone types +DRONE_TYPES = { + 0: "Unknown", + 1: "Commercial DJI", + 2: "Racing Drone", + 3: "Fixed Wing", + 4: "Military/Surveillance", + 5: "Custom Build" +} + +def generate_detection_data(device): + """Generate realistic drone detection data""" + + # Random position near the device (within ~5km radius) + lat_offset = random.uniform(-0.045, 0.045) # ~5km latitude + lon_offset = random.uniform(-0.09, 0.09) # ~5km longitude + + detection = { + "device_id": device["id"], + "drone_id": random.randint(1000, 9999), + "drone_type": random.randint(0, 5), + "rssi": random.randint(-90, -30), # Signal strength in dBm + "freq": random.choice([2400, 2450, 5800, 5850]), # Common drone frequencies + "geo_lat": device["lat"] + lat_offset, + "geo_lon": device["lon"] + lon_offset, + "device_timestamp": int(time.time() * 1000), # Unix timestamp in ms + "confidence_level": round(random.uniform(0.6, 1.0), 2), + "signal_duration": random.randint(500, 5000) # Duration in ms + } + + return detection + +def send_detection(detection_data): + """Send detection data to the API""" + url = f"{API_BASE_URL}/detections" + headers = { + "Content-Type": "application/json" + } + + try: + response = requests.post(url, json=detection_data, headers=headers, timeout=10) + + if response.status_code == 201: + print(f"βœ… Detection sent successfully: Device {detection_data['device_id']}, " + f"Drone {detection_data['drone_id']}, RSSI {detection_data['rssi']}dBm") + return True + else: + print(f"❌ Failed to send detection: {response.status_code} - {response.text}") + return False + + except requests.exceptions.RequestException as e: + print(f"❌ Network error: {e}") + return False + +def test_single_detection(): + """Send a single test detection""" + device = random.choice(DEVICES) + detection = generate_detection_data(device) + + print("🚁 Sending single test detection...") + print(f"πŸ“ Device: {device['name']}") + print(f"πŸ›°οΈ Drone ID: {detection['drone_id']}") + print(f"πŸ“‘ RSSI: {detection['rssi']}dBm") + print(f"πŸ“Š Frequency: {detection['freq']}MHz") + print(f"🎯 Confidence: {detection['confidence_level']}") + + return send_detection(detection) + +def simulate_continuous_detections(duration_minutes=5, interval_seconds=10): + """Simulate continuous drone detections for testing""" + print(f"πŸ”„ Starting continuous simulation for {duration_minutes} minutes...") + print(f"⏱️ Sending detection every {interval_seconds} seconds") + print("Press Ctrl+C to stop\n") + + start_time = time.time() + end_time = start_time + (duration_minutes * 60) + detection_count = 0 + + try: + while time.time() < end_time: + # Generate 1-3 simultaneous detections from different devices + num_detections = random.randint(1, 3) + + for _ in range(num_detections): + device = random.choice(DEVICES) + detection = generate_detection_data(device) + + if send_detection(detection): + detection_count += 1 + + # Small delay between simultaneous detections + time.sleep(0.5) + + time.sleep(interval_seconds) + + except KeyboardInterrupt: + print("\n⏹️ Simulation stopped by user") + + elapsed = time.time() - start_time + print(f"\nπŸ“ˆ Simulation completed:") + print(f" β€’ Duration: {elapsed/60:.1f} minutes") + print(f" β€’ Detections sent: {detection_count}") + print(f" β€’ Average rate: {detection_count/(elapsed/60):.1f} detections/minute") + +def test_api_health(): + """Test if the API is responding""" + url = f"{API_BASE_URL}/health" + + try: + response = requests.get(url, timeout=5) + if response.status_code == 200: + print("βœ… API health check passed") + return True + else: + print(f"❌ API health check failed: {response.status_code}") + return False + except requests.exceptions.RequestException as e: + print(f"❌ Cannot reach API: {e}") + return False + +def main(): + print("🚁 Drone Detection System - Test Script") + print("=" * 50) + + # Test API connectivity + if not test_api_health(): + print("Cannot connect to API. Please check if the system is running.") + return + + print("\nChoose test mode:") + print("1. Send single test detection") + print("2. Simulate continuous detections") + print("3. Send burst of 10 detections") + + try: + choice = input("\nEnter choice (1-3): ").strip() + + if choice == "1": + test_single_detection() + + elif choice == "2": + duration = input("Duration in minutes (default: 5): ").strip() + interval = input("Interval in seconds (default: 10): ").strip() + + duration = int(duration) if duration else 5 + interval = int(interval) if interval else 10 + + simulate_continuous_detections(duration, interval) + + elif choice == "3": + print("πŸš€ Sending burst of 10 detections...") + success_count = 0 + + for i in range(10): + device = random.choice(DEVICES) + detection = generate_detection_data(device) + + if send_detection(detection): + success_count += 1 + + time.sleep(1) # 1 second between detections + + print(f"\nπŸ“Š Burst completed: {success_count}/10 detections sent successfully") + + else: + print("Invalid choice") + + except KeyboardInterrupt: + print("\nπŸ‘‹ Goodbye!") + except ValueError: + print("Invalid input") + +if __name__ == "__main__": + main()