209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Device Status Debug Script
|
|
Checks the actual device data to debug why the device shows as offline
|
|
"""
|
|
|
|
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
|
|
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')
|
|
SKIP_AUTH = os.getenv('SKIP_AUTH', 'false').lower() == 'true'
|
|
|
|
if BASE_PATH and not API_BASE_URL.endswith('/api'):
|
|
domain = API_BASE_URL.replace('/api', '').replace('/drones/api', '').replace('/uggla/api', '')
|
|
API_BASE_URL = f"{domain}{BASE_PATH}/api"
|
|
|
|
# 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:
|
|
print("🔓 Authentication skipped (SKIP_AUTH=true)")
|
|
return True
|
|
|
|
try:
|
|
login_data = {
|
|
"username": USERNAME,
|
|
"password": PASSWORD
|
|
}
|
|
|
|
response = requests.post(f"{API_BASE_URL}/users/login", json=login_data, verify=False, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('success') and 'data' in data and 'token' in data['data']:
|
|
AUTH_TOKEN = data['data']['token']
|
|
print(f"✅ Authentication successful")
|
|
return True
|
|
else:
|
|
print(f"❌ Invalid login response: {data}")
|
|
return False
|
|
else:
|
|
print(f"❌ Authentication failed: HTTP {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Authentication error: {e}")
|
|
return False
|
|
|
|
def get_auth_headers():
|
|
"""Get headers with authentication token"""
|
|
headers = {
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
if not SKIP_AUTH and AUTH_TOKEN:
|
|
headers['Authorization'] = f'Bearer {AUTH_TOKEN}'
|
|
|
|
return headers
|
|
|
|
def debug_device_status():
|
|
"""Debug device status calculation"""
|
|
print("🔍 DEVICE STATUS DEBUG")
|
|
print("=" * 50)
|
|
|
|
try:
|
|
# Get all devices
|
|
response = requests.get(f"{API_BASE_URL}/devices", headers=get_auth_headers(), verify=False, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('success') and data.get('data'):
|
|
devices = data['data']
|
|
|
|
# Find Stockholm device (device ID 1001)
|
|
stockholm_device = None
|
|
for device in devices:
|
|
if device.get('id') == 1001 or 'Stockholm' in device.get('name', ''):
|
|
stockholm_device = device
|
|
break
|
|
|
|
if stockholm_device:
|
|
print(f"📱 Found Stockholm Device:")
|
|
print(f" ID: {stockholm_device['id']}")
|
|
print(f" Name: {stockholm_device.get('name', 'N/A')}")
|
|
print(f" Active: {stockholm_device.get('is_active', False)}")
|
|
print(f" Approved: {stockholm_device.get('is_approved', False)}")
|
|
print(f" Coordinates: {stockholm_device.get('geo_lat', 'N/A')}, {stockholm_device.get('geo_lon', 'N/A')}")
|
|
print(f" Location: {stockholm_device.get('location_description', 'N/A')}")
|
|
print(f" Last Heartbeat: {stockholm_device.get('last_heartbeat', 'Never')}")
|
|
print(f" Heartbeat Interval: {stockholm_device.get('heartbeat_interval', 'Not set')} seconds")
|
|
print(f" Firmware Version: {stockholm_device.get('firmware_version', 'N/A')}")
|
|
print(f" Installation Date: {stockholm_device.get('installation_date', 'N/A')}")
|
|
print(f" Notes: {stockholm_device.get('notes', 'N/A')}")
|
|
print(f" Created: {stockholm_device.get('created_at', 'N/A')}")
|
|
print(f" Updated: {stockholm_device.get('updated_at', 'N/A')}")
|
|
|
|
# Calculate status manually
|
|
if stockholm_device.get('last_heartbeat'):
|
|
from datetime import datetime
|
|
now = datetime.now()
|
|
last_heartbeat = datetime.fromisoformat(stockholm_device['last_heartbeat'].replace('Z', '+00:00'))
|
|
time_diff = (now - last_heartbeat.replace(tzinfo=None)).total_seconds()
|
|
|
|
expected_interval = stockholm_device.get('heartbeat_interval', 300)
|
|
threshold = expected_interval * 2
|
|
|
|
print(f"\n🔢 Status Calculation:")
|
|
print(f" Current Time: {now}")
|
|
print(f" Last Heartbeat: {last_heartbeat}")
|
|
print(f" Time Since Last: {time_diff:.1f} seconds")
|
|
print(f" Expected Interval: {expected_interval} seconds")
|
|
print(f" Online Threshold: {threshold} seconds")
|
|
print(f" Should be Online: {time_diff < threshold}")
|
|
|
|
# Check device stats
|
|
if 'stats' in stockholm_device:
|
|
stats = stockholm_device['stats']
|
|
print(f"\n📊 Device Stats:")
|
|
print(f" Status: {stats.get('status', 'Unknown')}")
|
|
print(f" Time Since Last Heartbeat: {stats.get('time_since_last_heartbeat', 'Unknown')} seconds")
|
|
print(f" Detections 24h: {stats.get('detections_24h', 0)}")
|
|
else:
|
|
print("❌ Stockholm device not found")
|
|
print("Available devices:")
|
|
for device in devices:
|
|
coords = f"({device.get('geo_lat', 'N/A')}, {device.get('geo_lon', 'N/A')})"
|
|
status = "✅ Active" if device.get('is_active') else "❌ Inactive"
|
|
approved = "✅ Approved" if device.get('is_approved') else "⚠️ Pending"
|
|
print(f" - ID {device['id']}: {device.get('name', 'Unnamed')} at {coords} | {status} | {approved}")
|
|
else:
|
|
print(f"❌ No devices found")
|
|
else:
|
|
print(f"❌ HTTP Error: {response.status_code}")
|
|
print(f" Response: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
|
|
def send_test_heartbeat():
|
|
"""Send a test heartbeat to see if it gets processed"""
|
|
print(f"\n💓 Sending Test Heartbeat")
|
|
print("-" * 30)
|
|
|
|
try:
|
|
payload = {
|
|
'type': 'heartbeat',
|
|
'key': '1001' # Stockholm device key
|
|
}
|
|
|
|
print(f"📡 Sending to: {API_BASE_URL}/detectors")
|
|
print(f"📦 Payload: {json.dumps(payload)}")
|
|
|
|
response = requests.post(
|
|
f"{API_BASE_URL}/detectors",
|
|
json=payload,
|
|
headers=get_auth_headers(),
|
|
verify=False,
|
|
timeout=10
|
|
)
|
|
|
|
print(f"📊 Response: {response.status_code}")
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
print(f"✅ Success: {data.get('message', 'OK')}")
|
|
else:
|
|
print(f"❌ Error: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error sending heartbeat: {e}")
|
|
|
|
def main():
|
|
print(f"🔗 API URL: {API_BASE_URL}")
|
|
print(f"⏰ Test Time: {datetime.now()}")
|
|
print()
|
|
|
|
# Authenticate first
|
|
print("🔐 Authenticating...")
|
|
if not authenticate():
|
|
print("❌ Authentication failed, tests may not work")
|
|
return
|
|
|
|
debug_device_status()
|
|
send_test_heartbeat()
|
|
|
|
print(f"\n⏳ Waiting 2 seconds...")
|
|
import time
|
|
time.sleep(2)
|
|
|
|
print(f"\n🔄 Checking status after heartbeat:")
|
|
debug_device_status()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|