ske087/traccar_animation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated view

Browse Source
This commit is contained in:
ske087
2025-06-05 13:25:19 +03:00
parent 26640678cb
commit 418ebc6f49
19 changed files with 124737 additions and 8263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

300
utils.py
View File

@@ -1,210 +1,74 @@
import os # Import the os module
import os
import json
import requests
from cryptography.fernet import Fernet
# Define the path to the server settings file
RESOURCES_FOLDER = "resources"
CREDENTIALS_FILE = os.path.join(RESOURCES_FOLDER, "credentials.enc")
KEY_FILE = os.path.join(RESOURCES_FOLDER, "key.key")
SERVER_SETTINGS_FILE = os.path.join(RESOURCES_FOLDER, "server_settings.enc")
# --- Encryption Utilities ---
def generate_key():
"""Generate and save a key for encryption."""
if not os.path.exists(KEY_FILE):
key = Fernet.generate_key()
with open(KEY_FILE, "wb") as key_file:
key_file.write(key)
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:
"""Load the encryption key."""
with open(KEY_FILE, "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
def encrypt_data(data):
"""Encrypt data using the encryption key."""
key = load_key()
fernet = Fernet(key)
return fernet.decrypt(encrypted_data).decode()
return fernet.encrypt(data.encode())
def decrypt_data(data):
"""Decrypt data using the encryption key."""
key = load_key()
fernet = Fernet(key)
return fernet.decrypt(data).decode()
# --- Server Settings ---
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.")
"""Load and decrypt server settings from file."""
if not os.path.exists(SERVER_SETTINGS_FILE):
return None
try:
with open(settings_file, "rb") as file:
with open(SERVER_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)}")
print(f"Failed to load server settings: {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 save_server_settings(settings_data):
"""Encrypt and save server settings."""
encrypted_data = encrypt_data(json.dumps(settings_data))
with open(SERVER_SETTINGS_FILE, "wb") as file:
file.write(encrypted_data)
# --- Traccar Server Connection ---
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.
Returns: dict with 'status' (bool) and 'message' (str)
"""
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:
@@ -214,7 +78,91 @@ def test_connection(server_url, username=None, password=None, token=None):
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)
# --- Device Fetching ---
def get_devices_from_server():
"""Retrieve a mapping of device names to IDs from the Traccar server."""
settings = check_server_settings()
if not settings:
return None
server_url = settings.get("server_url")
token = settings.get("token")
if not server_url or not token:
return None
headers = {"Authorization": f"Bearer {token}"}
try:
response = requests.get(f"{server_url}/api/devices", headers=headers)
if response.status_code == 200:
devices = response.json()
return {device.get("name", "Unnamed Device"): device.get("id", "Unknown ID") for device in devices}
else:
print(f"Error: {response.status_code} - {response.reason}")
return None
except Exception as e:
print(f"Error retrieving devices: {str(e)}")
return None
# --- Route Saving ---
def save_route_to_file(route_name, positions, base_folder="resources/projects"):
"""
Save the given positions as a route in resources/projects/<route_name>/positions.json.
Returns (success, message, file_path)
"""
if not route_name:
return False, "Please enter a route name.", None
if not positions:
return False, "No positions to save.", None
folder_path = os.path.join(base_folder, route_name)
os.makedirs(folder_path, exist_ok=True)
file_path = os.path.join(folder_path, "positions.json")
try:
with open(file_path, "w") as f:
json.dump(positions, f, indent=2)
return True, f"Route '{route_name}' saved!", file_path
except Exception as e:
return False, f"Failed to save route: {str(e)}", None
def fetch_positions(server_url, token, device_id, from_time, to_time):
"""
Fetch positions from the Traccar API.
Returns (positions, error_message)
"""
url = f"{server_url}/api/reports/route"
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
params = {
"deviceId": device_id,
"from": from_time,
"to": to_time
}
try:
response = requests.get(url, params=params, headers=headers, timeout=15)
if response.status_code == 200:
return response.json(), None
elif response.status_code == 400:
return None, "Bad Request: Please check the request payload and token."
else:
return None, f"Failed: {response.status_code} - {response.reason}"
except requests.exceptions.RequestException as e:
return None, f"Error fetching positions: {str(e)}"
def fetch_positions_for_selected_day(settings, device_mapping, device_name, start_date, end_date, start_hour, end_hour):
"""
Fetch positions for the selected day/device using Traccar API.
Returns (positions, error_message)
"""
if not settings:
return [], "Server settings not found."
server_url = settings.get("server_url")
token = settings.get("token")
device_id = device_mapping.get(device_name)
if not device_id:
return [], "Device ID not found."
from_time = f"{start_date}T{start_hour}:00:00Z"
to_time = f"{end_date}T{end_hour}:59:59Z"
positions, error = fetch_positions(server_url, token, device_id, from_time, to_time)
if error:
return [], error
return positions, None