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)