Fix jwt-token
This commit is contained in:
229
add_stockholm_device.py
Normal file
229
add_stockholm_device.py
Normal file
@@ -0,0 +1,229 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Add Stockholm Castle Device Script
|
||||
Creates a fake detection device positioned over Stockholm Castle for testing purposes.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
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 - Optional for local testing
|
||||
USERNAME = os.getenv('TEST_USERNAME', 'admin')
|
||||
PASSWORD = os.getenv('TEST_PASSWORD', 'admin123')
|
||||
SKIP_AUTH = os.getenv('SKIP_AUTH', 'false').lower() == 'true' # Set to 'true' to skip authentication
|
||||
|
||||
# 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}")
|
||||
if SKIP_AUTH:
|
||||
print("⚠️ AUTHENTICATION DISABLED - Running in local test mode")
|
||||
|
||||
# Stockholm Castle coordinates
|
||||
STOCKHOLM_CASTLE_LAT = 59.326668
|
||||
STOCKHOLM_CASTLE_LON = 18.071944
|
||||
|
||||
# Global variable to store authentication token
|
||||
AUTH_TOKEN = None
|
||||
|
||||
def authenticate():
|
||||
"""Authenticate with the API and get access token"""
|
||||
global AUTH_TOKEN
|
||||
|
||||
if SKIP_AUTH:
|
||||
return True
|
||||
|
||||
login_data = {
|
||||
"username": USERNAME,
|
||||
"password": PASSWORD
|
||||
}
|
||||
|
||||
try:
|
||||
print(f"🔐 Authenticating as user: {USERNAME}")
|
||||
response = requests.post(f"{API_BASE_URL}/users/login", json=login_data, verify=False)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
AUTH_TOKEN = data.get('data', {}).get('token')
|
||||
if AUTH_TOKEN:
|
||||
print("✅ Authentication successful")
|
||||
return True
|
||||
else:
|
||||
print("❌ No token received in response")
|
||||
print(f"Response data: {data}")
|
||||
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 SKIP_AUTH:
|
||||
return {"Content-Type": "application/json"}
|
||||
|
||||
if AUTH_TOKEN:
|
||||
return {
|
||||
"Authorization": f"Bearer {AUTH_TOKEN}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
return {"Content-Type": "application/json"}
|
||||
|
||||
def get_next_device_id():
|
||||
"""Get the next available device ID"""
|
||||
try:
|
||||
if SKIP_AUTH:
|
||||
# Try without authentication first for local testing
|
||||
response = requests.get(f"{API_BASE_URL}/devices/map", verify=False)
|
||||
else:
|
||||
response = requests.get(f"{API_BASE_URL}/devices", headers=get_auth_headers(), verify=False)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
devices = data.get('data', [])
|
||||
if devices:
|
||||
max_id = max(device['id'] for device in devices)
|
||||
return max_id + 1
|
||||
else:
|
||||
return 1001 # Start with a high number for test devices
|
||||
else:
|
||||
print(f"⚠️ Could not fetch existing devices: {response.status_code}")
|
||||
return 1001 # Default starting ID
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error fetching devices: {e}")
|
||||
return 1001 # Default starting ID
|
||||
|
||||
def create_stockholm_device():
|
||||
"""Create a device positioned over Stockholm Castle"""
|
||||
|
||||
# Get next available device ID
|
||||
device_id = get_next_device_id()
|
||||
|
||||
# Device data for Stockholm Castle
|
||||
device_data = {
|
||||
"id": device_id,
|
||||
"name": f"Stockholm Castle Detector {device_id}",
|
||||
"geo_lat": STOCKHOLM_CASTLE_LAT,
|
||||
"geo_lon": STOCKHOLM_CASTLE_LON,
|
||||
"location_description": "Stockholm Castle (Stockholms slott), Sweden - Royal Palace detector for airspace monitoring",
|
||||
"heartbeat_interval": 300, # 5 minutes
|
||||
"firmware_version": "1.0.0-test",
|
||||
"installation_date": datetime.now().isoformat(),
|
||||
"notes": "Test device for Stockholm Castle airspace monitoring - Created by add_stockholm_device.py script"
|
||||
}
|
||||
|
||||
try:
|
||||
print("🏰 Creating Stockholm Castle detector device...")
|
||||
print(f"📍 Location: {STOCKHOLM_CASTLE_LAT:.6f}, {STOCKHOLM_CASTLE_LON:.6f}")
|
||||
print(f"🆔 Device ID: {device_id}")
|
||||
|
||||
response = requests.post(f"{API_BASE_URL}/devices", json=device_data, headers=get_auth_headers(), verify=False)
|
||||
|
||||
if response.status_code == 201:
|
||||
created_device = response.json()
|
||||
print("✅ Stockholm Castle device created successfully!")
|
||||
print(f"📱 Device Name: {device_data['name']}")
|
||||
print(f"📍 Coordinates: {STOCKHOLM_CASTLE_LAT:.6f}, {STOCKHOLM_CASTLE_LON:.6f}")
|
||||
print(f"🏰 Location: Stockholm Castle, Sweden")
|
||||
print(f"🆔 Device ID: {device_id}")
|
||||
print("\n🎯 Device is ready for drone detection testing!")
|
||||
|
||||
return created_device
|
||||
else:
|
||||
print(f"❌ Failed to create device: {response.status_code}")
|
||||
print(f"Response: {response.text}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error creating device: {e}")
|
||||
return None
|
||||
|
||||
def approve_device_if_needed(device_id):
|
||||
"""Approve the device if the approval system is enabled"""
|
||||
try:
|
||||
approval_data = {"approved": True}
|
||||
response = requests.post(f"{API_BASE_URL}/devices/{device_id}/approve",
|
||||
json=approval_data, headers=get_auth_headers(), verify=False)
|
||||
|
||||
if response.status_code == 200:
|
||||
print(f"✅ Device {device_id} approved successfully")
|
||||
return True
|
||||
else:
|
||||
print(f"⚠️ Device approval response: {response.status_code}")
|
||||
# This might fail if approval isn't required, which is fine
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"⚠️ Device approval error (might not be required): {e}")
|
||||
return True
|
||||
|
||||
def main():
|
||||
"""Main function to create Stockholm Castle device"""
|
||||
print("=" * 70)
|
||||
print("🏰 STOCKHOLM CASTLE DEVICE CREATOR")
|
||||
print("=" * 70)
|
||||
print("Creating a test detector device positioned over Stockholm Castle")
|
||||
print("This device can be used for drone detection testing and simulation")
|
||||
print("=" * 70)
|
||||
|
||||
# Authenticate first (unless skipped)
|
||||
if not SKIP_AUTH and not authenticate():
|
||||
print("❌ Authentication failed. Cannot proceed.")
|
||||
print("Please check:")
|
||||
print("1. Is the server running?")
|
||||
print("2. Are the credentials correct?")
|
||||
print(f" Username: {USERNAME}")
|
||||
print("3. Set TEST_USERNAME and TEST_PASSWORD environment variables if needed")
|
||||
print("4. Or set SKIP_AUTH=true to skip authentication for local testing")
|
||||
return
|
||||
|
||||
# Create the device
|
||||
device = create_stockholm_device()
|
||||
|
||||
if device:
|
||||
device_id = device.get('data', {}).get('id') or device.get('id')
|
||||
if device_id:
|
||||
# Try to approve the device if approval system is enabled
|
||||
approve_device_if_needed(device_id)
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
print("✅ STOCKHOLM CASTLE DEVICE CREATED SUCCESSFULLY")
|
||||
print("=" * 70)
|
||||
print("Usage instructions:")
|
||||
print("1. The device is now available in your dashboard")
|
||||
print("2. You can use this device ID in test scripts")
|
||||
print("3. Run drone detection tests targeting Stockholm Castle area")
|
||||
print("4. Monitor the device status and detections on the map view")
|
||||
print("\nDevice Details:")
|
||||
print(f"🆔 Device ID: {device_id}")
|
||||
print(f"📍 Latitude: {STOCKHOLM_CASTLE_LAT:.6f}")
|
||||
print(f"📍 Longitude: {STOCKHOLM_CASTLE_LON:.6f}")
|
||||
print(f"🏰 Location: Stockholm Castle, Sweden")
|
||||
print("=" * 70)
|
||||
else:
|
||||
print("\n❌ Failed to create Stockholm Castle device")
|
||||
print("Please check the error messages above and try again")
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n⚠️ Stockholm Castle device creation interrupted by user")
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error during device creation: {e}")
|
||||
Reference in New Issue
Block a user