221 lines
7.9 KiB
Python
221 lines
7.9 KiB
Python
import os # Import the os module
|
|
import json
|
|
import requests
|
|
from cryptography.fernet import Fernet
|
|
|
|
# Define the path to the server settings file
|
|
RESOURCES_FOLDER = "resources"
|
|
SERVER_SETTINGS_FILE = os.path.join(RESOURCES_FOLDER, "server_settings.enc")
|
|
|
|
def load_key():
|
|
"""Load the encryption key from the key file."""
|
|
key_file_path = os.path.join(RESOURCES_FOLDER, "key.key")
|
|
if not os.path.exists(key_file_path):
|
|
raise FileNotFoundError("Encryption key file not found.")
|
|
with open(key_file_path, "rb") as key_file:
|
|
return key_file.read()
|
|
|
|
def decrypt_data(encrypted_data):
|
|
"""Decrypt the encrypted data using the loaded key."""
|
|
key = load_key() # Load the key from the file
|
|
fernet = Fernet(key)
|
|
return fernet.decrypt(encrypted_data).decode()
|
|
|
|
def check_server_settings():
|
|
"""
|
|
Load server settings from a configuration file or encrypted storage.
|
|
|
|
Returns:
|
|
dict: A dictionary containing server settings (e.g., server_url, username, password, token).
|
|
"""
|
|
settings_file = os.path.join("resources", "server_settings.enc")
|
|
if not os.path.exists(settings_file):
|
|
print("Settings file not found.")
|
|
return None
|
|
|
|
try:
|
|
with open(settings_file, "rb") as file:
|
|
encrypted_data = file.read()
|
|
decrypted_data = decrypt_data(encrypted_data)
|
|
settings = json.loads(decrypted_data)
|
|
return settings
|
|
except Exception as e:
|
|
print(f"Error loading settings: {str(e)}")
|
|
return None
|
|
|
|
def get_devices_from_server():
|
|
"""
|
|
Retrieve a list of devices from the Traccar server and create a mapping of device names to IDs.
|
|
|
|
Returns:
|
|
dict: A dictionary mapping device names to their IDs if the request is successful.
|
|
None: If the request fails.
|
|
"""
|
|
# Check if the server settings file exists
|
|
settings = check_server_settings()
|
|
if not settings:
|
|
return None
|
|
|
|
try:
|
|
# Extract server details
|
|
server_url = settings.get("server_url")
|
|
token = settings.get("token") # Optional, if token is used for authentication
|
|
|
|
if not server_url:
|
|
print("Error: Missing server URL in server_settings.enc.")
|
|
return None
|
|
|
|
# Ensure the server_url has a valid scheme
|
|
if not server_url.startswith("http://") and not server_url.startswith("https://"):
|
|
server_url = f"https://{server_url}" # Default to https:// if no scheme is provided
|
|
|
|
# Determine authentication method
|
|
headers = {"Authorization": f"Bearer {token}"} if token else None
|
|
|
|
if not token:
|
|
print("Error: Missing authentication details (token).")
|
|
return None
|
|
|
|
# Make a GET request to the /devices endpoint
|
|
response = requests.get(f"{server_url}/api/devices", headers=headers)
|
|
|
|
# Check the response status
|
|
if response.status_code == 200:
|
|
print("Devices retrieved successfully!")
|
|
devices = response.json() # Get the list of devices
|
|
|
|
# Create a mapping of device names to IDs
|
|
device_mapping = {device.get("name", "Unnamed Device"): device.get("id", "Unknown ID") for device in devices}
|
|
|
|
# Debugging: Print the mapping
|
|
for name, device_id in device_mapping.items():
|
|
print(f"Device Name: {name}, Device ID: {device_id}")
|
|
|
|
return device_mapping # Return the mapping of device names to IDs
|
|
else:
|
|
print(f"Error: {response.status_code} - {response.reason}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"Error retrieving devices: {str(e)}")
|
|
return None
|
|
|
|
def get_route_info(device_id, selected_date):
|
|
"""
|
|
Fetch route information for a specific device and date from the Traccar server.
|
|
|
|
Args:
|
|
device_id (int): The ID of the device.
|
|
selected_date (str): The selected date in the format 'YYYY-MM-DD'.
|
|
|
|
Returns:
|
|
list: The route information for the device on the selected date.
|
|
"""
|
|
# Load server settings
|
|
settings = check_server_settings()
|
|
if not settings:
|
|
print("Error: Unable to load server settings.")
|
|
return None
|
|
|
|
# Extract server details
|
|
server_url = settings.get("server_url")
|
|
token = settings.get("token")
|
|
|
|
if not server_url:
|
|
print("Error: Missing server URL in settings.")
|
|
return None
|
|
|
|
if not token:
|
|
print("Error: Missing token in settings.")
|
|
return None
|
|
|
|
# Ensure the server_url has a valid scheme
|
|
if not server_url.startswith("http://") and not server_url.startswith("https://"):
|
|
server_url = f"https://{server_url}" # Default to https:// if no scheme is provided
|
|
|
|
# Set the Authorization header with the token
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Convert the selected date to ISO 8601 format for the API
|
|
start_time = f"{selected_date}T00:00:00Z"
|
|
end_time = f"{selected_date}T23:59:59Z"
|
|
|
|
# API endpoint for fetching route reports
|
|
url = f"{server_url}/reports/route"
|
|
|
|
# Request payload
|
|
payload = {
|
|
"deviceId": device_id,
|
|
"from": start_time,
|
|
"to": end_time,
|
|
}
|
|
|
|
try:
|
|
# Log the payload for debugging
|
|
print(f"Request Payload: {payload}")
|
|
|
|
# Make the API request
|
|
response = requests.get(url, params=payload, headers=headers)
|
|
|
|
# Log the response status and content for debugging
|
|
print(f"Response Status Code: {response.status_code}")
|
|
print(f"Response Content: {response.text}")
|
|
|
|
# Check if the request was successful
|
|
if response.status_code == 200:
|
|
route = response.json()
|
|
print(f"Route for device {device_id} on {selected_date}:")
|
|
for position in route:
|
|
print(position)
|
|
return route
|
|
elif response.status_code == 400:
|
|
print("Bad Request: Please check the request payload and token.")
|
|
return None
|
|
else:
|
|
print(f"Failed to fetch route: {response.status_code} - {response.reason}")
|
|
return None
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Error fetching route: {str(e)}")
|
|
return None
|
|
|
|
def test_connection(server_url, username=None, password=None, token=None):
|
|
"""
|
|
Test the connection with the Traccar server.
|
|
|
|
Args:
|
|
server_url (str): The URL of the Traccar server.
|
|
username (str, optional): The username for basic authentication.
|
|
password (str, optional): The password for basic authentication.
|
|
token (str, optional): The token for bearer authentication.
|
|
|
|
Returns:
|
|
dict: A dictionary containing the connection status and message.
|
|
"""
|
|
if not server_url:
|
|
return {"status": False, "message": "Please provide the server URL."}
|
|
|
|
if not token and (not username or not password):
|
|
return {"status": False, "message": "Please provide either a token or username and password."}
|
|
|
|
try:
|
|
# Determine authentication method
|
|
headers = {"Authorization": f"Bearer {token}"} if token else None
|
|
auth = None if token else (username, password)
|
|
|
|
# Make a GET request to the server
|
|
response = requests.get(f"{server_url}/api/server", headers=headers, auth=auth, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
return {"status": True, "message": "Connection successful! Server is reachable."}
|
|
else:
|
|
return {"status": False, "message": f"Error: {response.status_code} - {response.reason}"}
|
|
except requests.exceptions.Timeout:
|
|
return {"status": False, "message": "Connection timed out. Please try again."}
|
|
except requests.exceptions.RequestException as e:
|
|
return {"status": False, "message": f"Connection failed: {str(e)}"}
|
|
|
|
# Call the function
|
|
test_device_id = 1 # Replace with the device ID from the spinner
|
|
test_date = "2025-06-01" # Replace with the selected date from the date picker
|
|
get_route_info(test_device_id, test_date)
|