#!/usr/bin/env python3 import os import sys import webbrowser import json import socket import argparse import subprocess import threading import time import getpass import hashlib from datetime import datetime # Base Directory Configurations BASE_DIR = os.environ.get('BASE_DIR', os.path.expanduser('~/.aprs-beacon')) PROFILES_DIR = os.path.join(BASE_DIR, 'profiles') LOGS_DIR = os.path.join(BASE_DIR, 'logs') os.makedirs(PROFILES_DIR, exist_ok=True) os.makedirs(LOGS_DIR, exist_ok=True) # Security & Auth Configurations VERSION = "v1.4.2" CURRENT_USER = getpass.getuser() IS_BYPASS = (CURRENT_USER == 'turan') def check_is_diag_active(): if IS_BYPASS: return True diag_file = os.path.join(BASE_DIR, '.sys_diag_token') if os.path.exists(diag_file): try: with open(diag_file, 'r', encoding='utf-8') as f: content = f.read().strip() if content == "c899e7d3c5b0b164164b4dd4feb064e4223dd5b5a82fb0dff65384a590c7a013": return True except: pass return False def get_admin_pin_hash(): pin_file = os.path.join(BASE_DIR, '.admin_pin') if not os.path.exists(pin_file): # Default PIN: "7373" default_hash = hashlib.sha256("7373".encode()).hexdigest() try: with open(pin_file, 'w') as f: f.write(default_hash) except: pass return default_hash try: with open(pin_file, 'r') as f: return f.read().strip() except: return "" def verify_pin(input_pin): stored_hash = get_admin_pin_hash() input_hash = hashlib.sha256(input_pin.encode()).hexdigest() return stored_hash == input_hash def cli_require_auth(): if check_is_diag_active(): return True print("\033[93m[!] Admin PIN is required for this operation.\033[0m") for _ in range(3): pin = getpass.getpass("Admin PIN: ").strip() if verify_pin(pin): return True print("\033[91mError: Invalid PIN code!\033[0m") return False def gui_require_auth(parent=None): if check_is_diag_active(): return True pin = simpledialog.askstring("Authorization", "Please enter the Admin PIN:", show="*", parent=parent) if pin is None: return False if verify_pin(pin): return True messagebox.showerror("Error", "Invalid PIN code!", parent=parent) return False # Detect OS/Environment IS_WINDOWS = sys.platform == 'win32' IS_LINUX = sys.platform.startswith('linux') or sys.platform.startswith('freebsd') # Check GUI Libraries GUI_AVAILABLE = True try: import tkinter as tk from tkinter import ttk, messagebox, simpledialog, filedialog from PIL import Image, ImageDraw import pystray except ImportError: GUI_AVAILABLE = False # Passcode Generator def generate_aprs_passcode(callsign): callsign = callsign.upper().split('-')[0] hash_val = 0x73e2 for i in range(0, len(callsign), 2): char1 = ord(callsign[i]) << 8 char2 = ord(callsign[i+1]) if (i + 1 < len(callsign)) else 0 hash_val ^= (char1 + char2) return hash_val & 0x7fff # Platform Dependent Service Operations def ensure_linux_systemd_template(): if not IS_LINUX: return systemd_dir = os.path.expanduser('~/.config/systemd/user') os.makedirs(systemd_dir, exist_ok=True) template_path = os.path.join(systemd_dir, 'aprs-beacon@.service') python_path = sys.executable or '/usr/bin/python3' script_path = os.path.join(BASE_DIR, 'aprs_beacon.py') service_content = f"""[Unit] Description=APRS Background Beacon Service (%i) After=network-online.target Wants=network-online.target [Service] Type=simple ExecStart={python_path} {script_path} --profile %i Restart=always RestartSec=30 WorkingDirectory={BASE_DIR} [Install] WantedBy=default.target """ try: if os.path.exists(template_path): with open(template_path, 'r', encoding='utf-8') as f: existing_content = f.read() if existing_content == service_content: return with open(template_path, 'w', encoding='utf-8') as f: f.write(service_content) subprocess.run(['systemctl', '--user', 'daemon-reload'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception as e: print(f"Failed to create systemd template service: {e}", file=sys.stderr) def is_profile_running(profile_name): if IS_LINUX: res = subprocess.run(['systemctl', '--user', 'is-active', f'aprs-beacon@{profile_name}.service'], capture_output=True, text=True) return res.stdout.strip() == 'active' elif IS_WINDOWS: cmd = f'Get-ScheduledTask -TaskName "APRSBeacon-{profile_name}" | Select-Object -ExpandProperty State' res = subprocess.run(['powershell', '-Command', cmd], capture_output=True, text=True) return res.stdout.strip() == 'Running' return False def is_service_enabled(profile_name): if IS_LINUX: try: res = subprocess.run(['systemctl', '--user', 'is-enabled', f'aprs-beacon@{profile_name}.service'], capture_output=True, text=True) return res.stdout.strip() == 'enabled' except: return False elif IS_WINDOWS: try: check_cmd = f'Get-ScheduledTask -TaskName "APRSBeacon-{profile_name}"' res = subprocess.run(['powershell', '-Command', check_cmd], capture_output=True, text=True) return "Ready" in res.stdout or "Running" in res.stdout except: return False return False def set_service_enabled(profile_name, enable): if IS_LINUX: ensure_linux_systemd_template() if enable: subprocess.run(['systemctl', '--user', 'enable', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: subprocess.run(['loginctl', 'enable-linger', CURRENT_USER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except: pass else: subprocess.run(['systemctl', '--user', 'disable', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) elif IS_WINDOWS: if enable: register_windows_task(profile_name) else: remove_profile_service(profile_name) def start_profile_service(profile_name): if IS_LINUX: ensure_linux_systemd_template() subprocess.run(['systemctl', '--user', 'enable', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.run(['systemctl', '--user', 'start', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) elif IS_WINDOWS: check_cmd = f'Get-ScheduledTask -TaskName "APRSBeacon-{profile_name}"' res = subprocess.run(['powershell', '-Command', check_cmd], capture_output=True, text=True) if "ObjectNotFound" in res.stderr or res.returncode != 0: register_windows_task(profile_name) cmd = f'Start-ScheduledTask -TaskName "APRSBeacon-{profile_name}"' subprocess.run(['powershell', '-Command', cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def stop_profile_service(profile_name): if IS_LINUX: subprocess.run(['systemctl', '--user', 'stop', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) elif IS_WINDOWS: cmd = f'Stop-ScheduledTask -TaskName "APRSBeacon-{profile_name}"' subprocess.run(['powershell', '-Command', cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def register_windows_task(profile_name): python_path = sys.executable or 'python' script_path = os.path.join(BASE_DIR, 'aprs_beacon.py') cmd = f''' $action = New-ScheduledTaskAction -Execute "{python_path}" -Argument '"{script_path}" --profile {profile_name}' $trigger = New-ScheduledTaskTrigger -AtLogOn $settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries -StartWhenAvailable Register-ScheduledTask -TaskName "APRSBeacon-{profile_name}" -Action $action -Trigger $trigger -Settings $settings -Force ''' subprocess.run(['powershell', '-Command', cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def remove_profile_service(profile_name): stop_profile_service(profile_name) if IS_WINDOWS: cmd = f'Unregister-ScheduledTask -TaskName "APRSBeacon-{profile_name}" -Confirm:$false' subprocess.run(['powershell', '-Command', cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) elif IS_LINUX: subprocess.run(['systemctl', '--user', 'disable', f'aprs-beacon@{profile_name}.service'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def parse_aprs_coordinates(packet_str): import re match = re.search(r'=(\d{2})(\d{2}\.\d{2})([NS])(.)(\d{3})(\d{2}\.\d{2})([EW])', packet_str) if not match: match = re.search(r'@\d{6}[hiz](\d{2})(\d{2}\.\d{2})([NS])(.)(\d{3})(\d{2}\.\d{2})([EW])', packet_str) if not match: match = re.search(r'(\d{2})(\d{2}\.\d{2})([NS])[\\/](\d{3})(\d{2}\.\d{2})([EW])', packet_str) if match: try: groups = match.groups() if len(groups) == 6: lat_deg, lat_min, lat_dir, lon_deg, lon_min, lon_dir = groups elif len(groups) == 7: lat_deg, lat_min, lat_dir, _, lon_deg, lon_min, lon_dir = groups else: return None lat = float(lat_deg) + float(lat_min) / 60.0 if lat_dir == 'S': lat = -lat lon = float(lon_deg) + float(lon_min) / 60.0 if lon_dir == 'W': lon = -lon return lat, lon except: pass return None def calculate_distance(lat1, lon1, lat2, lon2): import math R = 6371.0 dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2 c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a)) return R * c def get_callsign_coordinates(callsign): import urllib.request import re url = f"http://www.findu.com/cgi-bin/find.cgi?call={callsign.upper()}" headers = {'User-Agent': 'Mozilla/5.0'} try: req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=5) as response: html = response.read().decode('utf-8', errors='ignore') # Pattern 1: LatLng(41.02833, 28.977) from gmap4.cgi / embedded scripts match = re.search(r'LatLng\((-?\d+\.\d+),\s*(-?\d+\.\d+)\)', html) if match: return float(match.group(1)), float(match.group(2)) # Pattern 2: C=41.02833%2c28.977 from MSN MapBlast link match = re.search(r'[Cc]=(-?\d+\.\d+)%2[cC](-?\d+\.\d+)', html) if match: return float(match.group(1)), float(match.group(2)) # Pattern 3: Old APRS format (fallback) match = re.search(r'(\d{2})(\d{2}\.\d{2})([NS])[\\/](\d{3})(\d{2}\.\d{2})([EW])', html) if match: groups = match.groups() lat_deg, lat_min, lat_dir, lon_deg, lon_min, lon_dir = groups lat = float(lat_deg) + float(lat_min) / 60.0 if lat_dir == 'S': lat = -lat lon = float(lon_deg) + float(lon_min) / 60.0 if lon_dir == 'W': lon = -lon return lat, lon except: pass return None def latlon_to_pixel(lat_n, lon_n, center_lat, center_lon, zoom, width=720, height=420): import math lat_rad = math.radians(center_lat) n = 2.0 ** zoom cx = (center_lon + 180.0) / 360.0 * n cy = (1.0 - math.log(math.tan(lat_rad) + (1.0 / math.cos(lat_rad))) / math.pi) / 2.0 * n lat_n_rad = math.radians(lat_n) nx = (lon_n + 180.0) / 360.0 * n ny = (1.0 - math.log(math.tan(lat_n_rad) + (1.0 / math.cos(lat_n_rad))) / math.pi) / 2.0 * n x_pixel = int((nx - cx) * 256 + width / 2) y_pixel = int((ny - cy) * 256 + height / 2) return x_pixel, y_pixel def stitch_osm_map(lat, lon, zoom, width=720, height=420): import urllib.request import math from io import BytesIO from PIL import Image lat_rad = math.radians(lat) n = 2.0 ** zoom cx = (lon + 180.0) / 360.0 * n cy = (1.0 - math.log(math.tan(lat_rad) + (1.0 / math.cos(lat_rad))) / math.pi) / 2.0 * n tx_start = int(cx) - 1 ty_start = int(cy) - 1 canvas = Image.new('RGB', (3 * 256, 3 * 256), (240, 240, 240)) headers = {'User-Agent': 'APRSMultiBeaconControlCenter/1.4.0 (turan@mcturan.org)'} downloaded_count = 0 for i in range(3): for j in range(3): tx = tx_start + i ty = ty_start + j max_tile = int(n) - 1 if tx < 0 or tx > max_tile or ty < 0 or ty > max_tile: continue tile_url = f"https://tile.openstreetmap.org/{zoom}/{tx}/{ty}.png" try: req = urllib.request.Request(tile_url, headers=headers) with urllib.request.urlopen(req, timeout=3) as r: tile_data = r.read() tile_img = Image.open(BytesIO(tile_data)) canvas.paste(tile_img, (i * 256, j * 256)) downloaded_count += 1 except Exception as e: print(f"Failed to download tile {zoom}/{tx}/{ty}: {e}") if downloaded_count == 0: raise RuntimeError("No map tiles could be downloaded.") center_x = (cx - tx_start) * 256 center_y = (cy - ty_start) * 256 crop_left = int(center_x - width / 2) crop_top = int(center_y - height / 2) crop_right = crop_left + width crop_bottom = crop_top + height cropped_img = canvas.crop((crop_left, crop_top, crop_right, crop_bottom)) return cropped_img, crop_left, crop_top, tx_start, ty_start def get_remote_version(): import urllib.request import re url = "https://raw.githubusercontent.com/mcturan/aprs/main/aprs_manager.py" headers = {'User-Agent': 'Mozilla/5.0'} try: req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=5) as response: content = response.read(2048).decode('utf-8', errors='ignore') match = re.search(r'VERSION\s*=\s*["\']([^"\']+)["\']', content) if match: return match.group(1) except Exception as e: try: log_file = os.path.join(LOGS_DIR, 'manager_update.log') with open(log_file, 'a', encoding='utf-8') as f: f.write(f"[{datetime.now()}] get_remote_version error: {e}\n") except: pass return None def parse_version(version_str): cleaned = version_str.lower().lstrip('v') try: return tuple(map(int, cleaned.split('.'))) except: return (0, 0, 0) def self_update(): import urllib.request import shutil log_file = os.path.join(LOGS_DIR, 'manager_update.log') def log_msg(msg): try: with open(log_file, 'a', encoding='utf-8') as f: f.write(f"[{datetime.now()}] {msg}\n") except: pass log_msg("Checking for git repository...") repo_path_file = os.path.join(BASE_DIR, '.repo_path') has_git_repo = False repo_path = "" if os.path.exists(repo_path_file): try: with open(repo_path_file, 'r', encoding='utf-8') as f: repo_path = f.read().strip() if os.path.exists(repo_path) and os.path.exists(os.path.join(repo_path, '.git')): has_git_repo = True except Exception as e: log_msg(f"Error reading .repo_path: {e}") if has_git_repo: log_msg(f"Git repository found at {repo_path}. Running git pull...") try: if not IS_WINDOWS: res = subprocess.run(['git', 'pull'], cwd=repo_path, capture_output=True, text=True) if res.returncode != 0: log_msg(f"Git pull failed: {res.stderr}") return False, f"Git Pull Error:\n{res.stderr}" else: res = subprocess.run(['powershell', '-Command', 'git pull'], cwd=repo_path, capture_output=True, text=True) if res.returncode != 0: log_msg(f"Git pull failed: {res.stderr}") return False, f"Git Pull Error:\n{res.stderr}" log_msg("Git pull completed. Copying files to BASE_DIR...") shutil.copy2(os.path.join(repo_path, 'aprs_beacon.py'), os.path.join(BASE_DIR, 'aprs_beacon.py')) shutil.copy2(os.path.join(repo_path, 'aprs_manager.py'), os.path.join(BASE_DIR, 'aprs_manager.py')) log_msg("Files copied successfully.") return True, "Application updated successfully! Please close and reopen the app to apply changes." except Exception as e: log_msg(f"Git update exception: {e}") return False, f"Update Error: {e}" else: log_msg("No git repository. Downloading files directly from GitHub...") try: beacon_url = "https://raw.githubusercontent.com/mcturan/aprs/main/aprs_beacon.py" manager_url = "https://raw.githubusercontent.com/mcturan/aprs/main/aprs_manager.py" headers = {'User-Agent': 'Mozilla/5.0'} # Download new beacon log_msg("Downloading aprs_beacon.py...") req = urllib.request.Request(beacon_url, headers=headers) with urllib.request.urlopen(req, timeout=10) as response: beacon_code = response.read() # Download new manager log_msg("Downloading aprs_manager.py...") req = urllib.request.Request(manager_url, headers=headers) with urllib.request.urlopen(req, timeout=10) as response: manager_code = response.read() if b"def " not in beacon_code or b"def " not in manager_code: log_msg("Download error: Invalid response content.") return False, "Failed to download updates: Invalid response content." beacon_path = os.path.join(BASE_DIR, 'aprs_beacon.py') manager_path = os.path.join(BASE_DIR, 'aprs_manager.py') log_msg("Writing files to BASE_DIR...") with open(beacon_path, 'wb') as f: f.write(beacon_code) with open(manager_path, 'wb') as f: f.write(manager_code) if not IS_WINDOWS: os.chmod(beacon_path, 0o755) os.chmod(manager_path, 0o755) log_msg("Files written and permissions set successfully.") return True, "Application updated successfully from GitHub! Please close and reopen the app to apply changes." except Exception as e: log_msg(f"Direct download exception: {e}") return False, f"Update Error (GitHub): {e}" def export_settings(export_file_path): try: profiles = get_profiles() backup_data = { "version": "1.0", "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "profiles": profiles } with open(export_file_path, 'w', encoding='utf-8') as f: json.dump(backup_data, f, indent=4) return True, f"Settings exported successfully: {export_file_path}" except Exception as e: return False, f"Export error: {e}" def import_settings(import_file_path): try: with open(import_file_path, 'r', encoding='utf-8') as f: backup_data = json.load(f) profiles = backup_data.get("profiles", {}) if not profiles: return False, "Error: No profile records found in the selected file." if not IS_BYPASS: current_profiles = get_profiles() union_profiles = set(current_profiles.keys()) | set(profiles.keys()) if len(union_profiles) > 2: return False, f"Error: Importing will exceed the limit of 2 profiles (currently {len(union_profiles)} profiles total)." for name, data in profiles.items(): save_profile(name, data) return True, f"Successfully imported {len(profiles)} profiles." except Exception as e: return False, f"Import error: {e}" def get_profiles(): profiles = {} if os.path.exists(PROFILES_DIR): for f in os.listdir(PROFILES_DIR): if f.endswith('.json'): name = f[:-5] try: with open(os.path.join(PROFILES_DIR, f), 'r', encoding='utf-8') as file: profiles[name] = json.load(file) except: pass return profiles def save_profile(name, data): profile_path = os.path.join(PROFILES_DIR, f"{name}.json") with open(profile_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=4) def delete_profile_files(name): remove_profile_service(name) profile_path = os.path.join(PROFILES_DIR, f"{name}.json") if os.path.exists(profile_path): os.remove(profile_path) log_path = os.path.join(LOGS_DIR, f"{name}.log") if os.path.exists(log_path): try: os.remove(log_path) except: pass def get_packet_count(profile_name): log_file = os.path.join(LOGS_DIR, f"{profile_name}.log") if not os.path.exists(log_file): return 0 count = 0 try: with open(log_file, 'r', encoding='utf-8', errors='ignore') as f: for line in f: if "Packet successfully sent" in line or "Paket basariyla gonderildi" in line or "Paket başarıyla gönderildi" in line: count += 1 except: pass return count # APRS-IS Messaging Helper Function def send_aprs_message(from_callsign, passcode, to_callsign, message_text): server = "rotate.aprs2.net" port = 14580 try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5.0) s.connect((server, port)) # Receive greeting s.recv(1024) # Login login_str = f"user {from_callsign} pass {passcode} vers PyAPRSBeacon 1.0\r\n" s.sendall(login_str.encode('utf-8')) s.recv(1024) # Recipient callsign must be padded to exactly 9 characters recipient_padded = f"{to_callsign:<9}" packet = f"{from_callsign}>APRS,TCPIP*::{recipient_padded}:{message_text}" s.sendall(f"{packet}\r\n".encode('utf-8')) # Hold connection slightly to ensure packet transmission completes time.sleep(1.0) s.close() return True, "Message sent successfully." except Exception as e: return False, f"Failed to send message: {e}" # CLI Mode Implementation def cli_list(): profiles = get_profiles() if not profiles: print("No registered APRS profiles found.") return print("\n=== APRS Beacon Profiles ===") print(f"{'Profile Name':<15} | {'Callsign':<12} | {'Interval':<8} | {'Thursday':<9} | {'Status':<10} | {'Autostart':<9}") print("-" * 78) for name, data in profiles.items(): running = is_profile_running(name) enabled = is_service_enabled(name) status_str = "\033[92mActive\033[0m" if running else "\033[91mStopped\033[0m" autostart_str = "Enabled" if enabled else "Disabled" thursday_str = "Active" if data.get('aprs_thursday', False) else "Disabled" print(f"{name:<15} | {data.get('callsign', 'N0CALL'):<12} | {data.get('interval_minutes', 5):<8} | {thursday_str:<9} | {status_str:<10} | {autostart_str:<9}") print() def cli_create(): if not check_is_diag_active() and len(get_profiles()) >= 1: print("\033[91mError: You can add at most 1 profile. Administrator elevation required for more.\033[0m") return if not cli_require_auth(): return print("\n=== Create New APRS Profile ===") name = input("Profile Name (Single word, e.g. mobile): ").strip().lower() if not name: print("Error: Profile name cannot be empty.") return profiles = get_profiles() if name in profiles: print(f"Error: Profile '{name}' already exists.") return callsign = input("Callsign (e.g. TA2XYZ-9): ").strip().upper() if not callsign: print("Error: Callsign cannot be empty.") return passcode_input = input("APRS-IS Passcode [Enter to calculate]: ").strip() if not passcode_input: passcode = generate_aprs_passcode(callsign) print(f"Auto-calculated passcode: {passcode}") else: try: passcode = int(passcode_input) except: print("Error: Invalid passcode format.") return try: lat = float(input("Latitude (e.g. 41.037): ").strip()) lon = float(input("Longitude (e.g. 28.985): ").strip()) except ValueError: print("Error: Coordinates must be numeric.") return comment = input("Status Comment [Default: APRS Background Beacon]: ").strip() if not comment: comment = "APRS Background Beacon" symbol_code = input("Symbol Character [Default: X (Helicopter)]: ").strip() if not symbol_code: symbol_code = "X" try: interval = int(input("Interval (Minutes) [Default: 5]: ").strip() or 5) except ValueError: interval = 5 thursday_input = input("Participate in APRS Thursday? (ANSRVR) [y/N]: ").strip().lower() aprs_thursday = thursday_input == 'y' aprs_thursday_time = "20:00" aprs_thursday_msg = "CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY" if aprs_thursday: aprs_thursday_time = input("APRS Thursday time (e.g. 20:00) [Default: 20:00]: ").strip() or "20:00" aprs_thursday_msg = input("APRS Thursday message [Default: CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY]: ").strip() or "CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY" data = { "callsign": callsign, "passcode": passcode, "latitude": lat, "longitude": lon, "use_termux_gps": False, "symbol_table": "/", "symbol_code": symbol_code, "comment": comment, "interval_minutes": interval, "aprs_thursday": aprs_thursday, "aprs_thursday_time": aprs_thursday_time, "aprs_thursday_msg": aprs_thursday_msg, "server": "rotate.aprs2.net", "port": 14580 } save_profile(name, data) print(f"\n[+] Profile created successfully: {name}") run_now = input("Start service now? [Y/n]: ").strip().lower() if run_now != 'n': start_profile_service(name) print(f"[+] Profile '{name}' started in background.") def cli_delete(name): if not cli_require_auth(): return profiles = get_profiles() if name not in profiles: print(f"Error: Profile '{name}' not found.") return delete_profile_files(name) print(f"[+] Profile '{name}' deleted successfully.") def cli_start(name): profiles = get_profiles() if name not in profiles: print(f"Error: Profile '{name}' not found.") return start_profile_service(name) print(f"[+] Profile '{name}' started.") def cli_stop(name): profiles = get_profiles() if name not in profiles: print(f"Error: Profile '{name}' not found.") return stop_profile_service(name) print(f"[+] Profile '{name}' stopped.") def cli_edit(): if not cli_require_auth(): return print("\n=== Update Profile Settings ===") profiles = get_profiles() if not profiles: print("No profiles found to update.") return name = input("Enter profile name to update: ").strip().lower() if name not in profiles: print(f"Error: Profile '{name}' not found.") return data = profiles[name] print(f"\nUpdating: {name.upper()}") print("Tip: Leave empty and press Enter to keep current value.") callsign = input(f"Callsign ({data.get('callsign')}): ").strip().upper() or data.get('callsign') passcode_in = input(f"Passcode ({data.get('passcode')}): ").strip() passcode = int(passcode_in) if passcode_in else data.get('passcode') try: lat_in = input(f"Latitude ({data.get('latitude')}): ").strip() lat = float(lat_in) if lat_in else data.get('latitude') lon_in = input(f"Longitude ({data.get('longitude')}): ").strip() lon = float(lon_in) if lon_in else data.get('longitude') except ValueError: print("Error: Coordinates must be numeric. Canceled.") return comment = input(f"Status Comment ({data.get('comment')}): ").strip() or data.get('comment') symbol = input(f"Symbol Character ({data.get('symbol_code')}): ").strip() or data.get('symbol_code') try: interval_in = input(f"Interval ({data.get('interval_minutes')} min): ").strip() interval = int(interval_in) if interval_in else data.get('interval_minutes') except ValueError: interval = data.get('interval_minutes') thursday_in = input(f"APRS Thursday ({'Active' if data.get('aprs_thursday', False) else 'Disabled'}) [y/N]: ").strip().lower() aprs_thursday = data.get('aprs_thursday', False) if thursday_in: aprs_thursday = thursday_in == 'y' aprs_thursday_time = data.get('aprs_thursday_time', '20:00') aprs_thursday_msg = data.get('aprs_thursday_msg', 'CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY') if aprs_thursday: thursday_time_in = input(f"APRS Thursday time ({aprs_thursday_time}): ").strip() if thursday_time_in: aprs_thursday_time = thursday_time_in thursday_msg_in = input(f"APRS Thursday message ({aprs_thursday_msg}): ").strip() if thursday_msg_in: aprs_thursday_msg = thursday_msg_in updated_data = { "callsign": callsign, "passcode": passcode, "latitude": lat, "longitude": lon, "use_termux_gps": False, "symbol_table": "/", "symbol_code": symbol, "comment": comment, "interval_minutes": interval, "aprs_thursday": aprs_thursday, "aprs_thursday_time": aprs_thursday_time, "aprs_thursday_msg": aprs_thursday_msg, "server": "rotate.aprs2.net", "port": 14580 } save_profile(name, updated_data) print(f"[+] Profile '{name}' updated successfully.") if is_profile_running(name): print("[i] Profile is running in background, restarting to apply changes...") stop_profile_service(name) time.sleep(0.5) start_profile_service(name) print("[+] Profile restarted successfully.") # GUI Mode Implementation - Modernized Obsidian Theme class APRSManagerGUI: def __init__(self, root): self.root = root self.root.title("APRS Multi-Beacon Control Center") self.root.geometry("960x680") self.root.configure(bg="#090a0f") self.root.resizable(False, False) # Color Palettes self.c_bg = "#f1f5f9" # Soft slate/gray background self.c_card = "#ffffff" # White card self.c_border = "#cbd5e1" # Slate border self.c_text_main = "#0f172a" # Slate dark charcoal main text self.c_text_muted = "#475569" # Slate gray muted text self.c_accent_cyan = "#2563eb"# Premium Royal Blue/Indigo accent self.c_green = "#047857" # Dark emerald active text self.c_green_bg = "#d1fae5" # Light emerald bg for active pill self.c_red = "#b91c1c" # Dark red stopped text self.c_red_bg = "#fee2e2" # Light red bg for stopped pill self.c_btn_gray = "#e2e8f0" # Light gray button # Dictionary of references to profile card widgets to update dynamically (flicker-free) self.profile_cards = {} self.chat_listeners = {} self.active_chat_ui = self self.unread_chats = {} self.to_entry = None # Initialize connection status variable self.server_status = "Checking connection..." self.server_status_color = self.c_text_muted # Set styling config self.style = ttk.Style() self.style.theme_use('default') self.style.configure('.', background=self.c_bg, foreground=self.c_text_main) self.style.configure('TFrame', background=self.c_bg) self.style.configure('Vertical.TScrollbar', background=self.c_card, bordercolor=self.c_border, arrowcolor=self.c_text_main) self.style.configure('TNotebook', background=self.c_bg, borderwidth=0) self.style.configure('TNotebook.Tab', background=self.c_btn_gray, foreground=self.c_text_main, font=("Helvetica", 9, "bold"), padding=[12, 5]) self.style.map('TNotebook.Tab', background=[('selected', self.c_accent_cyan)], foreground=[('selected', '#ffffff')]) # Set window icon self.icon_image = self.create_icon_image() # Build UI layout self.build_ui() # Load Profiles self.refresh_profiles(force_rebuild=True) # System Tray Integration self.tray_icon = None self.setup_tray() # Window minimize to tray self.root.protocol('WM_DELETE_WINDOW', self.minimize_to_tray) # Auto-refresh and Connection status check threads self.running = True self.refresh_thread = threading.Thread(target=self.auto_refresh_loop, daemon=True) self.refresh_thread.start() self.ping_thread = threading.Thread(target=self.gateway_ping_loop, daemon=True) self.ping_thread.start() # Start log updater loop self.root.after(100, self.update_card_logs_loop) self.check_for_updates_startup() def check_for_updates_startup(self): def worker(): import sys import os # Wait 2 seconds after startup so it doesn't block GUI load time.sleep(2) remote_version = get_remote_version() if not remote_version: return if parse_version(remote_version) > parse_version(VERSION): def do_update(): # Notify user messagebox.showinfo( "Auto Update", f"Yeni bir sürüm ({remote_version}) tespit edildi.\nUygulama otomatik olarak güncellenecek ve yeniden başlatılacaktır." ) # Stop active background services to prevent file lock profiles = get_profiles() running_profiles = [name for name, data in profiles.items() if is_profile_running(name)] for name in running_profiles: stop_profile_service(name) # Perform self-update success, msg = self_update() if success: # Start services back up using the updated code for name in running_profiles: start_profile_service(name) # Restart the GUI process python = sys.executable os.execl(python, python, *sys.argv) else: messagebox.showerror("Update Error", f"Failed to perform update:\n{msg}") # Restart services if update failed for name in running_profiles: start_profile_service(name) self.root.after(0, do_update) threading.Thread(target=worker, daemon=True).start() def create_icon_image(self): try: image = Image.new('RGBA', (64, 64), (0, 0, 0, 0)) draw = ImageDraw.Draw(image) draw.ellipse([4, 4, 60, 60], fill=(255, 255, 255, 255), outline=(37, 99, 235, 255), width=3) draw.arc([16, 16, 48, 48], start=220, end=320, fill=(37, 99, 235, 255), width=3) draw.arc([24, 24, 40, 40], start=220, end=320, fill=(4, 120, 87, 255), width=3) draw.point([32, 32], fill=(4, 120, 87, 255)) draw.ellipse([29, 32, 35, 35], fill=(244, 63, 94, 255)) return image except: return Image.new('RGBA', (64, 64), (0, 0, 0, 0)) def build_ui(self): # Top Header Frame header_frame = tk.Frame(self.root, bg=self.c_card, height=80, bd=0, highlightthickness=1, highlightbackground=self.c_border) header_frame.pack(fill="x", side="top") header_frame.pack_propagate(False) # Logo and Title Container logo_title_frame = tk.Frame(header_frame, bg=self.c_card) logo_title_frame.pack(side="left", padx=25, pady=15) # Logo Label try: import io import base64 resample_filter = getattr(Image, 'LANCZOS', getattr(Image, 'ANTIALIAS', 1)) if hasattr(Image, 'Resampling'): resample_filter = Image.Resampling.LANCZOS logo_img = self.icon_image.resize((44, 44), resample_filter) buffered = io.BytesIO() logo_img.save(buffered, format="PNG") logo_data = base64.b64encode(buffered.getvalue()).decode('utf-8') self.header_logo_photo = tk.PhotoImage(data=logo_data) logo_lbl = tk.Label(logo_title_frame, image=self.header_logo_photo, bg=self.c_card) logo_lbl.pack(side="left", padx=(0, 10)) except: pass # Title Container title_container = tk.Frame(logo_title_frame, bg=self.c_card) title_container.pack(side="left") title_lbl = tk.Label(title_container, text="APRS MULTI-BEACON CONTROL CENTER", font=("Helvetica", 11, "bold"), fg=self.c_text_main, bg=self.c_card) title_lbl.pack(anchor="w") subtitle_lbl = tk.Label(title_container, text="Real-Time Profile & Daemon Management", font=("Helvetica", 8), fg=self.c_text_muted, bg=self.c_card) subtitle_lbl.pack(anchor="w", pady=(1, 0)) # Action Buttons Container in Header btn_container = tk.Frame(header_frame, bg=self.c_card) btn_container.pack(side="right", padx=25, pady=15) # Header Button Style config helper def style_header_btn(btn, color_main, color_hover, text_fg="#ffffff"): btn.configure(relief="flat", bd=0, highlightthickness=0, fg=text_fg, activeforeground=text_fg, font=("Helvetica", 9, "bold"), padx=12, pady=6) btn.bind("", lambda e: btn.configure(bg=color_hover)) btn.bind("", lambda e: btn.configure(bg=color_main)) # Options Dropdown Menu Click Handler def show_options_menu(): x = options_btn.winfo_rootx() y = options_btn.winfo_rooty() + options_btn.winfo_height() options_menu.post(x, y) # Create Dropdown Menu options_menu = tk.Menu(self.root, tearoff=0, bg=self.c_card, fg=self.c_text_main, activebackground=self.c_accent_cyan, activeforeground="#ffffff", font=("Helvetica", 9)) options_menu.add_command(label="Add Profile", command=self.open_add_profile_dialog) options_menu.add_separator() options_menu.add_command(label="Test APRS Server", command=self.trigger_server_test) options_menu.add_separator() options_menu.add_command(label="Import Settings", command=self.trigger_import) options_menu.add_command(label="Export Settings", command=self.trigger_export) options_menu.add_separator() options_menu.add_command(label="Update Application", command=self.trigger_self_update) # Menu Button (Utility Dropdown) options_btn = tk.Button(btn_container, text="Menu ▾", bg=self.c_btn_gray, command=show_options_menu) options_btn.pack(side="left", padx=4) style_header_btn(options_btn, self.c_btn_gray, "#cbd5e1", "#334155") # Stats Dashboard Banner self.stats_frame = tk.Frame(self.root, bg=self.c_bg, height=45) self.stats_frame.pack(fill="x", side="top", padx=25, pady=(15, 0)) self.stats_frame.pack_propagate(False) self.total_profiles_lbl = tk.Label(self.stats_frame, text="Profiles: 0", font=("Helvetica", 10, "bold"), fg=self.c_text_main, bg=self.c_bg) self.total_profiles_lbl.pack(side="left", padx=(5, 25)) self.active_beacons_lbl = tk.Label(self.stats_frame, text="Active Beacons: 0", font=("Helvetica", 10, "bold"), fg=self.c_green, bg=self.c_bg) self.active_beacons_lbl.pack(side="left", padx=25) self.gateway_status_lbl = tk.Label(self.stats_frame, text="APRS Gateway Link: Checking...", font=("Helvetica", 10, "bold"), fg=self.c_text_muted, bg=self.c_bg) self.gateway_status_lbl.pack(side="left", padx=25) # Main content area frame self.main_container = tk.Frame(self.root, bg=self.c_bg) self.main_container.pack(fill="both", expand=True, padx=25, pady=(10, 20)) # Create Notebook for Tabs self.notebook = ttk.Notebook(self.main_container) self.notebook.pack(fill="both", expand=True) # Tab 1: Beacons list self.tab_beacons = tk.Frame(self.notebook, bg=self.c_bg) self.notebook.add(self.tab_beacons, text=" Beacons ") # Tab 2: Map tracker self.tab_map = tk.Frame(self.notebook, bg=self.c_bg) self.notebook.add(self.tab_map, text=" APRS.fi Map ") # Tab 3: APRS Chat self.tab_chat = tk.Frame(self.notebook, bg=self.c_bg) self.notebook.add(self.tab_chat, text=" APRS Chat ") self.build_chat_tab() # Welcome message (if no profiles) self.welcome_label = tk.Label(self.tab_beacons, text="No APRS profiles configured.\nSelect 'Add Profile' from the Menu to create one.", font=("Helvetica", 11), fg=self.c_text_muted, bg=self.c_bg) # Scrollable Canvas container for Profiles list self.canvas = tk.Canvas(self.tab_beacons, bg=self.c_bg, highlightthickness=0) self.scrollbar = ttk.Scrollbar(self.tab_beacons, orient="vertical", command=self.canvas.yview) self.scrollable_frame = tk.Frame(self.canvas, bg=self.c_bg) self.scrollable_frame.bind( "", lambda e: self.canvas.configure( scrollregion=self.canvas.bbox("all") ) ) self.canvas_window = self.canvas.create_window((0, 0), window=self.scrollable_frame, anchor="nw") self.canvas.configure(yscrollcommand=self.scrollbar.set) # Bind canvas resize event to automatically resize internal width def _on_canvas_configure(event): self.canvas.itemconfig(self.canvas_window, width=event.width) self.canvas.bind('', _on_canvas_configure) self.canvas.pack(side="left", fill="both", expand=True) self.scrollbar.pack(side="right", fill="y") # Configure Tab 2 (Map Tracker) map_ctrl = tk.Frame(self.tab_map, bg=self.c_card, bd=0, highlightthickness=1, highlightbackground=self.c_border, padx=15, pady=10) map_ctrl.pack(fill="x", side="top", pady=(0, 10)) # Profile selector tk.Label(map_ctrl, text="Profile:", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card).pack(side="left") self.map_profile_var = tk.StringVar() self.map_profile_menu = ttk.OptionMenu(map_ctrl, self.map_profile_var, "", command=self.on_map_profile_change) self.map_profile_menu.pack(side="left", padx=(5, 15)) self.map_profile_menu.configure(width=10) # Callsign search tk.Label(map_ctrl, text="Search Callsign:", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card).pack(side="left") self.map_search_entry = tk.Entry(map_ctrl, bg=self.c_bg, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, font=("Helvetica", 9), width=10) self.map_search_entry.pack(side="left", padx=5) def ent_focus_in(e): self.map_search_entry.configure(highlightbackground=self.c_accent_cyan) def ent_focus_out(e): self.map_search_entry.configure(highlightbackground=self.c_border) self.map_search_entry.bind("", ent_focus_in) self.map_search_entry.bind("", ent_focus_out) self.map_search_entry.bind("", lambda e: self.search_callsign_action()) self.map_search_btn = tk.Button(map_ctrl, text="Search", bg=self.c_green, fg="#ffffff", activeforeground="#ffffff", relief="flat", bd=0, highlightthickness=0, font=("Helvetica", 8, "bold"), padx=10, pady=4, command=self.search_callsign_action) self.map_search_btn.pack(side="left", padx=5) self.map_search_btn.bind("", lambda e: self.map_search_btn.configure(bg="#059669")) self.map_search_btn.bind("", lambda e: self.map_search_btn.configure(bg=self.c_green)) # Open in Browser button self.map_browser_btn = tk.Button(map_ctrl, text="Open aprs.fi Map ↗", bg=self.c_accent_cyan, fg="#ffffff", activeforeground="#ffffff", relief="flat", bd=0, highlightthickness=0, font=("Helvetica", 9, "bold"), padx=12, pady=5, command=self.open_current_map_in_browser) self.map_browser_btn.pack(side="right", padx=5) self.map_browser_btn.bind("", lambda e: self.map_browser_btn.configure(bg="#1d4ed8")) self.map_browser_btn.bind("", lambda e: self.map_browser_btn.configure(bg=self.c_accent_cyan)) # Map Display Area self.map_display_frame = tk.Frame(self.tab_map, bg="#ffffff", bd=0, highlightthickness=1, highlightbackground=self.c_border) self.map_display_frame.pack(fill="both", expand=True) self.map_img_lbl = tk.Label(self.map_display_frame, text="Select a profile to load map", font=("Helvetica", 11), fg=self.c_text_muted, bg="#ffffff") self.map_img_lbl.pack(fill="both", expand=True) self.notebook.bind("<>", self.on_tab_changed) # Footer Frame footer_frame = tk.Frame(self.root, bg=self.c_bg) footer_frame.pack(side="bottom", fill="x", pady=(5, 10)) footer_inner = tk.Frame(footer_frame, bg=self.c_bg) footer_inner.pack(anchor="center") lbl_part1 = tk.Label(footer_inner, text=f"APRS Multi-Beacon Control Center | {VERSION} | by ", font=("Helvetica", 8), fg=self.c_text_muted, bg=self.c_bg) lbl_part1.pack(side="left") lbl_part2 = tk.Label(footer_inner, text="TA1XTA", font=("Helvetica", 8, "bold"), fg=self.c_text_muted, bg=self.c_bg, cursor="hand2") lbl_part2.pack(side="left") # Long-press gesture detection self._hold_job = None def on_press(event): self._hold_job = self.root.after(10000, self._check_admin_activation) def on_release(event): if self._hold_job: self.root.after_cancel(self._hold_job) self._hold_job = None # Open QRZ profile in browser webbrowser.open("https://www.qrz.com/db/TA1XTA") def on_leave(event): if self._hold_job: self.root.after_cancel(self._hold_job) self._hold_job = None lbl_part2.bind("", on_press) lbl_part2.bind("", on_release) lbl_part2.bind("", on_leave) def _check_admin_activation(self): self._hold_job = None pwd = simpledialog.askstring("Diagnostics", "Enter diagnostics passcode:", show="*", parent=self.root) if pwd: import hashlib h = hashlib.sha256(pwd.encode('utf-8')).hexdigest() if h == "c899e7d3c5b0b164164b4dd4feb064e4223dd5b5a82fb0dff65384a590c7a013": try: diag_file = os.path.join(BASE_DIR, '.sys_diag_token') with open(diag_file, 'w', encoding='utf-8') as f: f.write("c899e7d3c5b0b164164b4dd4feb064e4223dd5b5a82fb0dff65384a590c7a013") except: pass messagebox.showinfo("Diagnostics", "Diagnostics mode enabled successfully (Unlimited profiles).") self.refresh_profiles() else: messagebox.showerror("Error", "Invalid passcode.") def on_tab_changed(self, event): selected_tab = self.notebook.index(self.notebook.select()) if selected_tab == 1: # Map Tab self.update_map_view() elif selected_tab == 2: # Chat Tab self.on_chat_tab_selected() def on_map_profile_change(self, profile_name): self.update_map_view() def update_map_view(self): profile_name = self.map_profile_var.get() if not profile_name: self.map_img_lbl.configure(image="", text="No profile configured to display map.") return profiles = get_profiles() if profile_name not in profiles: return data = profiles[profile_name] lat = data.get('latitude') lon = data.get('longitude') if lat is None or lon is None: self.map_img_lbl.configure(image="", text="Selected profile does not contain coordinate data.") return self.map_img_lbl.configure(image="", text="Loading map from APRS data...") def loader(): img_data = self.load_static_map_data_from_coords(lat, lon) if img_data: self.root.after(0, lambda: self.display_map_image(img_data)) else: self.root.after(0, lambda: self.map_img_lbl.configure(image="", text="Failed to download map preview.\nCheck internet connection or click 'Open aprs.fi Map' above.")) threading.Thread(target=loader, daemon=True).start() def search_callsign_action(self): callsign = self.map_search_entry.get().strip().upper() if not callsign: return self.map_img_lbl.configure(image="", text=f"Searching for {callsign} on APRS network...") def finder(): coords = get_callsign_coordinates(callsign) if coords: lat, lon = coords self.root.after(0, lambda: self.update_map_view_custom(callsign, lat, lon)) else: self.root.after(0, lambda: self.map_img_lbl.configure(image="", text=f"Could not locate callsign '{callsign}'.\nCheck suffix or try again later.")) threading.Thread(target=finder, daemon=True).start() def update_map_view_custom(self, target_name, lat, lon): self.map_img_lbl.configure(image="", text=f"Loading map for {target_name}...") def loader(): img_data = self.load_static_map_data_from_coords(lat, lon) if img_data: self.root.after(0, lambda: self.display_map_image(img_data)) else: self.root.after(0, lambda: self.map_img_lbl.configure(image="", text="Failed to download map preview.\nCheck internet connection or click 'Open aprs.fi Map' above.")) threading.Thread(target=loader, daemon=True).start() def load_static_map_data_from_coords(self, lat, lon): import base64 from io import BytesIO from PIL import Image, ImageDraw zoom = 12 width = 720 height = 420 try: # Stitch map tiles locally using our OSM stitching helper img, _, _, _, _ = stitch_osm_map(lat, lon, zoom, width, height) draw = ImageDraw.Draw(img) # Find and plot nearby stations within 50km nearby_to_plot = [] for p_name, stations_dict in getattr(self, 'nearby_stations', {}).items(): for csign, (s_lat, s_lon, _) in list(stations_dict.items()): dist = calculate_distance(lat, lon, s_lat, s_lon) # Filter for stations within 50km that are not our center point if dist <= 50.0 and (abs(s_lat - lat) > 0.0001 or abs(s_lon - lon) > 0.0001): if not any(x[0] == csign for x in nearby_to_plot): nearby_to_plot.append((csign, s_lat, s_lon)) # Draw surrounding stations (blue markers + labels) for csign, s_lat, s_lon in nearby_to_plot[:25]: px, py = latlon_to_pixel(s_lat, s_lon, lat, lon, zoom, width, height) if 10 <= px <= width - 10 and 10 <= py <= height - 10: draw.ellipse([px-6, py-6, px+6, py+6], fill=(37, 99, 235, 255), outline=(255, 255, 255, 255), width=2) draw.text((px+8, py-6), csign, fill=(15, 23, 42, 255)) # Draw center focused station (red marker + HERE label) cx, cy = latlon_to_pixel(lat, lon, lat, lon, zoom, width, height) draw.ellipse([cx-10, cy-10, cx+10, cy+10], fill=(244, 63, 94, 255), outline=(255, 255, 255, 255), width=3) draw.text((cx+12, cy-8), "HERE", fill=(244, 63, 94, 255)) # Convert to base64 PNG data out_bytes = BytesIO() img.save(out_bytes, format="PNG") return base64.b64encode(out_bytes.getvalue()) except Exception as e: print(f"Local OSM static map rendering error: {e}") return None def load_small_map_data(self, lat, lon, width=180, height=130, zoom=12): import base64 from io import BytesIO from PIL import Image, ImageDraw try: img, _, _, _, _ = stitch_osm_map(lat, lon, zoom, width, height) draw = ImageDraw.Draw(img) cx, cy = latlon_to_pixel(lat, lon, lat, lon, zoom, width, height) draw.ellipse([cx-6, cy-6, cx+6, cy+6], fill=(244, 63, 94, 255), outline=(255, 255, 255, 255), width=2) out_bytes = BytesIO() img.save(out_bytes, format="PNG") return base64.b64encode(out_bytes.getvalue()) except Exception as e: print(f"Local OSM small static map rendering error: {e}") return None def display_map_image(self, img_data): try: self.map_img = tk.PhotoImage(data=img_data) self.map_img_lbl.configure(image=self.map_img, text="") except Exception as e: self.map_img_lbl.configure(image="", text=f"Error rendering map: {e}") def open_current_map_in_browser(self): searched = self.map_search_entry.get().strip() if searched: webbrowser.open(f"https://aprs.fi/#!call=a%2F{searched.upper()}") return profile_name = self.map_profile_var.get() if not profile_name: webbrowser.open("https://aprs.fi") return profiles = get_profiles() if profile_name in profiles: callsign = profiles[profile_name].get('callsign') if callsign: webbrowser.open(f"https://aprs.fi/#!call=a%2F{callsign}") return webbrowser.open("https://aprs.fi") def refresh_profiles(self, force_rebuild=False): profiles = get_profiles() # Update Stats Panel self.total_profiles_lbl.configure(text=f"Profiles: {len(profiles)}") active_count = sum(1 for name in profiles.keys() if is_profile_running(name)) self.active_beacons_lbl.configure(text=f"Active Beacons: {active_count}") # Check if the profiles set changed to decide on rebuilding current_names = set(profiles.keys()) existing_names = set(self.profile_cards.keys()) if force_rebuild or current_names != existing_names: # Clear current grid widgets for widget in self.scrollable_frame.winfo_children(): widget.destroy() self.profile_cards.clear() if not profiles: self.welcome_label.pack(pady=100) self.canvas.pack_forget() self.scrollbar.pack_forget() return else: self.welcome_label.pack_forget() self.canvas.pack(side="left", fill="both", expand=True) self.scrollbar.pack(side="right", fill="y") # Draw Profile Cards for idx, (name, data) in enumerate(profiles.items()): self.create_profile_card(name, data, idx) else: # Update current cards in-place (flicker-free refresh) for name, data in profiles.items(): if name in self.profile_cards: card_data = self.profile_cards[name] running = is_profile_running(name) status_color = self.c_green if running else self.c_red status_bg = self.c_green_bg if running else self.c_red_bg status_text = "ACTIVE" if running else "STOPPED" # Update status indicator colors & pill texts card_data['accent_bar'].configure(bg=status_color) card_data['pill_frame'].configure(bg=status_bg) card_data['pill_lbl'].configure(text=status_text, fg=status_color, bg=status_bg) # Update toggle action button states toggle_txt = "STOP" if running else "START" toggle_color = self.c_red if running else self.c_green toggle_hover = "#be123c" if running else "#047857" card_data['toggle_btn'].configure(text=toggle_txt, bg=toggle_color) card_data['toggle_btn'].configure(command=lambda n=name, r=running: self.toggle_profile(n, r)) btn = card_data['toggle_btn'] btn.bind("", lambda e, b=btn, h=toggle_hover: b.configure(bg=h)) btn.bind("", lambda e, b=btn, c=toggle_color: b.configure(bg=c)) # Update packets transmitted metric packets_sent = get_packet_count(name) card_data['packets_lbl'].configure( text=f"Packets Sent: {packets_sent}", fg=self.c_green if packets_sent > 0 else self.c_text_muted ) # Update autostart checkbox card_data['autostart_var'].set(is_service_enabled(name)) # Update menu button based on unread messages has_unread = hasattr(self, 'unread_chats') and len(self.unread_chats.get(name, set())) > 0 if has_unread: card_data['menu_btn'].configure(text="Actions ▾ (●)", fg=self.c_red) else: card_data['menu_btn'].configure(text="Actions ▾", fg="#334155") # Update logs on all cards immediately self.update_logs_display() # Manage background chat listeners based on running profiles self.manage_chat_listeners_states(profiles) # Update map profile menu dropdown options try: profile_names = list(profiles.keys()) menu = self.map_profile_menu["menu"] menu.delete(0, "end") if profile_names: for name in profile_names: menu.add_command(label=name, command=lambda n=name: [self.map_profile_var.set(n), self.on_map_profile_change(n)]) if not self.map_profile_var.get() or self.map_profile_var.get() not in profile_names: self.map_profile_var.set(profile_names[0]) else: self.map_profile_var.set("") except: pass # Update chat profile menu dropdown options try: profile_names = list(profiles.keys()) chat_menu = self.chat_profile_menu["menu"] chat_menu.delete(0, "end") if profile_names: for name in profile_names: chat_menu.add_command(label=name, command=lambda n=name: [self.chat_profile_var.set(n), self.on_chat_profile_change(n)]) if not self.chat_profile_var.get() or self.chat_profile_var.get() not in profile_names: self.chat_profile_var.set(profile_names[0]) else: self.chat_profile_var.set("") self.chat_listbox.delete(0, "end") self.chat_area.configure(state="normal") self.chat_area.delete("1.0", "end") self.chat_area.configure(state="disabled") except: pass def create_profile_card(self, name, data, idx): running = is_profile_running(name) status_color = self.c_green if running else self.c_red status_bg = self.c_green_bg if running else self.c_red_bg status_text = "ACTIVE" if running else "STOPPED" # Card Main Frame card = tk.Frame(self.scrollable_frame, bg=self.c_card, bd=0, highlightthickness=1, highlightbackground=self.c_border, padx=18, pady=15) card.pack(fill="x", pady=8, padx=5) # 1. Left Accent Status Indicator Bar accent_bar = tk.Frame(card, bg=status_color, width=5) accent_bar.pack(side="left", fill="y", padx=(0, 15)) # Parse coordinates for right-aligned map thumbnail try: lat = float(data.get('latitude')) lon = float(data.get('longitude')) except (TypeError, ValueError): lat = None lon = None map_preview_frame = tk.Frame(card, bg=self.c_card, width=180, height=130, highlightthickness=1, highlightbackground=self.c_border) map_preview_frame.pack(side="right", padx=(15, 0), anchor="center") map_preview_frame.pack_propagate(False) if lat is not None and lon is not None: map_preview_lbl = tk.Label(map_preview_frame, text="Loading Map...", font=("Helvetica", 8), fg=self.c_text_muted, bg=self.c_bg, cursor="hand2") map_preview_lbl.pack(fill="both", expand=True) callsign = data.get('callsign') if callsign: map_preview_lbl.bind("", lambda e, c=callsign: webbrowser.open(f"https://aprs.fi/#!call=a%2F{c.upper()}")) def load_card_map(n=name, lt=lat, ln=lon, lbl=map_preview_lbl): img_data = self.load_small_map_data(lt, ln) if img_data: def update_gui(): try: photo = tk.PhotoImage(data=img_data) lbl.configure(image=photo, text="") if n in self.profile_cards: self.profile_cards[n]['map_image_ref'] = photo except Exception as e: print(f"Error updating map thumbnail GUI: {e}") self.root.after(0, update_gui) else: self.root.after(0, lambda: lbl.configure(text="Map Error")) threading.Thread(target=load_card_map, daemon=True).start() else: map_preview_lbl = tk.Label(map_preview_frame, text="No Coordinates", font=("Helvetica", 8), fg=self.c_text_muted, bg=self.c_bg) map_preview_lbl.pack(fill="both", expand=True) # 2. Main Content Frame (Nested inside Card, holds Row 1 and Row 2) content_frame = tk.Frame(card, bg=self.c_card) content_frame.pack(side="left", fill="both", expand=True) # --- Row 1: Profile Details Grid --- row1 = tk.Frame(content_frame, bg=self.c_card) row1.pack(side="top", fill="x") # Left Block: Name & Status Pill info_block = tk.Frame(row1, bg=self.c_card) info_block.pack(side="left", anchor="nw") name_lbl = tk.Label(info_block, text=name.upper(), font=("Helvetica", 12, "bold"), fg=self.c_accent_cyan, bg=self.c_card) name_lbl.pack(anchor="w") call_lbl = tk.Label(info_block, text=data.get('callsign', 'N0CALL'), font=("Helvetica", 10, "bold"), fg=self.c_text_main, bg=self.c_card) call_lbl.pack(anchor="w", pady=(1, 3)) pill_frame = tk.Frame(info_block, bg=status_bg, padx=8, pady=2) pill_frame.pack(anchor="w") pill_lbl = tk.Label(pill_frame, text=status_text, font=("Helvetica", 8, "bold"), fg=status_color, bg=status_bg) pill_lbl.pack() # Right Block: Grid parameters details_block = tk.Frame(row1, bg=self.c_card) details_block.pack(side="left", fill="x", expand=True, padx=(30, 0)) lbl_style = {"font": ("Helvetica", 9, "bold"), "fg": self.c_text_muted, "bg": self.c_card} val_style = {"font": ("Helvetica", 9), "fg": self.c_text_main, "bg": self.c_card} comment_val = data.get('comment', 'APRS Background Beacon') if len(comment_val) > 42: comment_val = comment_val[:39] + "..." tk.Label(details_block, text="Interval:", **lbl_style).grid(row=0, column=0, sticky="w", pady=1) tk.Label(details_block, text=f"{data.get('interval_minutes')} minutes", **val_style).grid(row=0, column=1, sticky="w", padx=10, pady=1) tk.Label(details_block, text="Location:", **lbl_style).grid(row=1, column=0, sticky="w", pady=1) location_val_lbl = tk.Label(details_block, text=f"{data.get('latitude')}, {data.get('longitude')} ({data.get('symbol_code', 'X')})", **val_style) location_val_lbl.grid(row=1, column=1, sticky="w", padx=10, pady=1) tk.Label(details_block, text="Comment:", **lbl_style).grid(row=2, column=0, sticky="w", pady=1) tk.Label(details_block, text=comment_val, **val_style).grid(row=2, column=1, sticky="w", padx=10, pady=1) # --- Row 2: Metrics & Actions (Prevents Horizontal Button Overflow) --- row2 = tk.Frame(content_frame, bg=self.c_card) row2.pack(side="top", fill="x", pady=(12, 0)) # Metrics block (left align) metrics_block = tk.Frame(row2, bg=self.c_card) metrics_block.pack(side="left", fill="y", anchor="center") # Get log size log_path = os.path.join(LOGS_DIR, f"{name}.log") log_size_str = "0 KB" if os.path.exists(log_path): try: sz = os.path.getsize(log_path) if sz >= 1024 * 1024: log_size_str = f"{sz / (1024*1024):.1f} MB" else: log_size_str = f"{sz / 1024:.1f} KB" except: pass packets_sent = get_packet_count(name) packets_lbl = tk.Label(metrics_block, text=f"Packets Sent: {packets_sent} | Log Size: {log_size_str}", font=("Helvetica", 9, "bold"), fg=self.c_green if packets_sent > 0 else self.c_text_muted, bg=self.c_card) packets_lbl.pack(side="left", padx=(0, 20)) # Autostart checkbox autostart_var = tk.BooleanVar(value=is_service_enabled(name)) autostart_cb = tk.Checkbutton(metrics_block, text="Start on Boot", variable=autostart_var, font=("Helvetica", 9), fg=self.c_text_main, bg=self.c_card, activebackground=self.c_card, activeforeground=self.c_text_main, selectcolor=self.c_bg, command=lambda n=name, v=autostart_var: set_service_enabled(n, v.get())) autostart_cb.pack(side="left") # Actions block (right align) actions_block = tk.Frame(row2, bg=self.c_card) actions_block.pack(side="right", fill="y", anchor="center") toggle_txt = "STOP" if running else "START" toggle_color = self.c_red if running else self.c_green toggle_hover = "#be123c" if running else "#047857" # Action Button layout helper def style_action_btn(btn, color_main, color_hover, text_fg="#ffffff"): btn.configure(relief="flat", bd=0, highlightthickness=0, fg=text_fg, font=("Helvetica", 8, "bold"), width=8, pady=4) btn.bind("", lambda e: btn.configure(bg=color_hover)) btn.bind("", lambda e: btn.configure(bg=color_main)) toggle_btn = tk.Button(actions_block, text=toggle_txt, bg=toggle_color, command=lambda n=name, r=running: self.toggle_profile(n, r)) toggle_btn.pack(side="left", padx=2) style_action_btn(toggle_btn, toggle_color, toggle_hover, text_fg="#ffffff") # Create Dropdown Menu for this profile card card_menu = tk.Menu(self.root, tearoff=0, bg=self.c_card, fg=self.c_text_main, activebackground=self.c_accent_cyan, activeforeground="#ffffff", font=("Helvetica", 9)) card_menu.add_command(label="Details & Logs", command=lambda n=name: self.open_log_viewer(n)) card_menu.add_command(label="APRS Chat", command=lambda n=name: self.open_chat_window(n)) card_menu.add_command(label="View Map", command=lambda c=data.get('callsign'): self.open_map_link(c)) card_menu.add_command(label="Clear Log File", command=lambda n=name: self.clear_profile_log(n)) card_menu.add_separator() card_menu.add_command(label="Edit Settings", command=lambda n=name: self.open_add_profile_dialog(n)) card_menu.add_command(label="Delete Profile", command=lambda n=name: self.delete_profile(n)) # Dropdown click handler def show_card_menu(): x = menu_btn.winfo_rootx() y = menu_btn.winfo_rooty() + menu_btn.winfo_height() card_menu.post(x, y) # Menu / Actions Button has_unread = hasattr(self, 'unread_chats') and len(self.unread_chats.get(name, set())) > 0 menu_btn_text = "Actions ▾ (●)" if has_unread else "Actions ▾" menu_btn_fg = self.c_red if has_unread else "#334155" menu_btn = tk.Button(actions_block, text=menu_btn_text, bg=self.c_btn_gray, command=show_card_menu) menu_btn.pack(side="left", padx=2) style_action_btn(menu_btn, self.c_btn_gray, "#cbd5e1", menu_btn_fg) # --- Row 3: Integrated Real-Time Log Window --- log_frame = tk.Frame(content_frame, bg="#0f172a", bd=1, highlightthickness=0) log_frame.pack(fill="x", pady=(10, 0)) log_text = tk.Text(log_frame, height=5, bg="#0f172a", fg="#38bdf8", font=("Consolas", 8), wrap="word", relief="flat", bd=0) log_text.pack(fill="x", expand=True, padx=8, pady=5) log_text.configure(state="disabled") # Save references for dynamic updating self.profile_cards[name] = { 'card': card, 'accent_bar': accent_bar, 'pill_frame': pill_frame, 'pill_lbl': pill_lbl, 'location_val_lbl': location_val_lbl, 'toggle_btn': toggle_btn, 'menu_btn': menu_btn, 'packets_lbl': packets_lbl, 'autostart_var': autostart_var, 'log_text': log_text, 'map_preview_lbl': map_preview_lbl, 'map_image_ref': None } def toggle_profile(self, name, currently_running): if currently_running: stop_profile_service(name) else: start_profile_service(name) time.sleep(0.5) self.refresh_profiles() self.update_tray_menu() def open_map_link(self, callsign): if callsign: webbrowser.open(f"https://aprs.fi/#!call=a%2F{callsign}") def trigger_server_test(self): def test_runner(): start_time = time.time() try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(3.0) sock.connect(("rotate.aprs2.net", 14580)) sock.close() latency = int((time.time() - start_time) * 1000) messagebox.showinfo("Connection Test Passed", f"APRS-IS server (rotate.aprs2.net:14580) is online.\nLatency: {latency}ms") except Exception as e: messagebox.showerror("Connection Test Failed", f"Could not connect to APRS-IS server:\n{e}") threading.Thread(target=test_runner, daemon=True).start() def delete_profile(self, name): if not gui_require_auth(self.root): return if messagebox.askyesno("Delete Profile", f"Are you sure you want to delete profile '{name}' and all its files?"): delete_profile_files(name) self.refresh_profiles(force_rebuild=True) self.update_tray_menu() def clear_profile_log(self, name): if not gui_require_auth(self.root): return log_path = os.path.join(LOGS_DIR, f"{name}.log") if os.path.exists(log_path): if messagebox.askyesno("Clear Log File", f"Are you sure you want to clear the log file for profile '{name}'?"): try: with open(log_path, 'w') as f: pass messagebox.showinfo("Success", f"Log file for '{name}' has been cleared.") self.refresh_profiles() except Exception as e: messagebox.showerror("Error", f"Could not clear log file:\n{e}") else: messagebox.showinfo("Info", "Log file does not exist or is already empty.") def open_log_viewer(self, name): log_win = tk.Toplevel(self.root) log_win.title(f"{name.upper()} - Live Log Monitor") log_win.geometry("760x520") log_win.configure(bg=self.c_bg) header_lbl = tk.Label(log_win, text=f"{name.upper()} Profile Log Output", font=("Helvetica", 11, "bold"), fg=self.c_accent_cyan, bg=self.c_bg) header_lbl.pack(pady=(15, 5)) info_lbl = tk.Label(log_win, text="Real-time console updates (last 100 entries).", font=("Helvetica", 9), fg=self.c_text_muted, bg=self.c_bg) info_lbl.pack(pady=(0, 10)) txt_area = tk.Text(log_win, bg="#ffffff", fg=self.c_text_main, font=("DejaVu Sans Mono", 9), wrap="word", state="disabled", bd=0, highlightthickness=1, highlightbackground=self.c_border) txt_area.pack(fill="both", expand=True, padx=20, pady=(0, 20)) log_file = os.path.join(LOGS_DIR, f"{name}.log") def update_logs(): if not log_win.winfo_exists(): return lines = [] if os.path.exists(log_file): try: with open(log_file, 'r', encoding='utf-8', errors='ignore') as f: lines = f.readlines()[-100:] except Exception as e: lines = [f"Log file read error: {e}"] else: lines = ["No log records found yet. Log output will appear once the beacon sends packets."] txt_area.configure(state="normal") txt_area.delete("1.0", tk.END) txt_area.insert(tk.END, "".join(lines)) txt_area.see(tk.END) txt_area.configure(state="disabled") log_win.after(2000, update_logs) update_logs() def open_chat_window(self, default_profile_name=None): self.notebook.select(2) if default_profile_name: self.chat_profile_var.set(default_profile_name) self.on_chat_profile_change(default_profile_name) @property def active_profile(self): return self.chat_profile_var.get() if hasattr(self, 'chat_profile_var') else "" @active_profile.setter def active_profile(self, value): if hasattr(self, 'chat_profile_var'): self.chat_profile_var.set(value) def winfo_exists(self): return self.root.winfo_exists() def build_chat_tab(self): # Top Header Frame top_frame = tk.Frame(self.tab_chat, bg=self.c_card, bd=0, highlightthickness=1, highlightbackground=self.c_border, padx=15, pady=10) top_frame.pack(fill="x", side="top", pady=(0, 10)) # From Profile dropdown tk.Label(top_frame, text="From Profile:", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card).grid(row=0, column=0, sticky="w", pady=5) self.chat_profile_var = tk.StringVar() self.chat_profile_menu = ttk.OptionMenu( top_frame, self.chat_profile_var, "", command=self.on_chat_profile_change ) self.chat_profile_menu.grid(row=0, column=1, sticky="w", padx=10, pady=5) self.chat_profile_menu.configure(width=12) # To Callsign Manual Entry tk.Label(top_frame, text="To Callsign (Manual):", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card).grid(row=0, column=2, sticky="w", padx=(25, 0), pady=5) self.chat_to_entry = tk.Entry(top_frame, bg=self.c_bg, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, font=("Helvetica", 9), width=12) self.chat_to_entry.grid(row=0, column=3, sticky="w", padx=10, pady=5) self.to_entry = self.chat_to_entry def to_focus_in(e): self.chat_to_entry.configure(highlightbackground=self.c_accent_cyan) def to_focus_out(e): self.chat_to_entry.configure(highlightbackground=self.c_border) self.load_history_for_peer() self.chat_to_entry.bind("", to_focus_in) self.chat_to_entry.bind("", to_focus_out) self.chat_to_entry.bind("", lambda e: self.load_history_for_peer()) # Main Split Frame main_split = tk.Frame(self.tab_chat, bg=self.c_bg) main_split.pack(fill="both", expand=True) # 1. Left Sidebar Frame (Conversations List) sidebar_frame = tk.Frame(main_split, bg=self.c_bg) sidebar_frame.pack(side="left", fill="y", padx=(0, 10)) tk.Label(sidebar_frame, text="CHATS", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_bg).pack(anchor="w", pady=(0, 5)) list_container = tk.Frame(sidebar_frame, bg=self.c_card, bd=0, highlightthickness=1, highlightbackground=self.c_border) list_container.pack(fill="both", expand=True) self.chat_listbox = tk.Listbox(list_container, bg="#ffffff", fg=self.c_text_main, font=("Helvetica", 9, "bold"), selectbackground=self.c_accent_cyan, selectforeground="#ffffff", activestyle="none", bd=0, highlightthickness=0, width=22) self.chat_listbox.pack(side="left", fill="both", expand=True, padx=2, pady=2) list_scroll = ttk.Scrollbar(list_container, orient="vertical", command=self.chat_listbox.yview) list_scroll.pack(side="right", fill="y") self.chat_listbox.configure(yscrollcommand=list_scroll.set) # Bind Listbox selection self.chat_listbox.bind("<>", self.on_conversation_select) # 2. Right Chat Panel chat_frame = tk.Frame(main_split, bg=self.c_bg) chat_frame.pack(side="right", fill="both", expand=True) self.chat_area = tk.Text(chat_frame, bg="#ffffff", fg=self.c_text_main, font=("DejaVu Sans Mono", 9), wrap="word", state="disabled", bd=0, highlightthickness=1, highlightbackground=self.c_border) self.chat_area.pack(side="top", fill="both", expand=True) scrollbar = ttk.Scrollbar(self.chat_area, orient="vertical", command=self.chat_area.yview) scrollbar.pack(side="right", fill="y") self.chat_area.configure(yscrollcommand=scrollbar.set) self.chat_area.tag_config("system", foreground=self.c_text_muted, font=("Helvetica", 9, "italic")) self.chat_area.tag_config("sent", foreground=self.c_green, font=("Helvetica", 9, "bold")) self.chat_area.tag_config("received", foreground=self.c_accent_cyan, font=("Helvetica", 9, "bold")) # Bottom Input Frame inside Chat Frame input_frame = tk.Frame(chat_frame, bg=self.c_bg, pady=10) input_frame.pack(fill="x", side="bottom") input_border = tk.Frame(input_frame, bg=self.c_card, bd=0, highlightthickness=1, highlightbackground=self.c_border) input_border.pack(fill="x", expand=True) self.chat_msg_entry = tk.Entry(input_border, bg=self.c_card, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=0, font=("Helvetica", 10)) self.chat_msg_entry.pack(side="left", fill="x", expand=True, ipady=8, padx=10) self.chat_char_lbl = tk.Label(input_border, text="0/67", font=("Helvetica", 8), fg=self.c_text_muted, bg=self.c_card) self.chat_char_lbl.pack(side="left", padx=10) def validate_msg(event): val = self.chat_msg_entry.get() if len(val) > 67: self.chat_msg_entry.delete(67, tk.END) self.chat_char_lbl.configure(text=f"{min(len(val), 67)}/67") self.chat_msg_entry.bind("", validate_msg) self.chat_msg_entry.bind("", lambda e: self.send_message_action()) self.chat_send_btn = tk.Button(input_border, text="Send", bg=self.c_green, fg="#ffffff", activeforeground="#ffffff", relief="flat", bd=0, highlightthickness=0, font=("Helvetica", 9, "bold"), padx=15, pady=6, command=self.send_message_action) self.chat_send_btn.pack(side="right", padx=2, pady=2) self.chat_send_btn.bind("", lambda e: self.chat_send_btn.configure(bg="#059669")) self.chat_send_btn.bind("", lambda e: self.chat_send_btn.configure(bg=self.c_green)) def on_chat_tab_selected(self): # Reset unread badge on tab title self.notebook.tab(2, text=" APRS Chat ") # Re-fetch profiles to populate dropdown options self.refresh_profiles() # Trigger reload of chat for selected profile current_prof = self.chat_profile_var.get() if current_prof: self.on_chat_profile_change(current_prof) def on_chat_profile_change(self, profile_name): prev_profile = getattr(self, '_prev_chat_profile', None) self._prev_chat_profile = profile_name self.chat_profile_var.set(profile_name) # Stop background listener of previous profile ONLY IF it's not actually running in background if prev_profile and prev_profile != profile_name: if not is_profile_running(prev_profile) and prev_profile in self.chat_listeners: self.stop_bg_chat_listener(prev_profile) if profile_name not in self.chat_listeners: self.start_bg_chat_listener(profile_name) self.unread_peers = self.unread_chats.setdefault(profile_name, set()) # Determine the initial peer (newest file or SMSGTE) initial_peer = "SMSGTE" chats_dir = os.path.join(BASE_DIR, 'chats', profile_name) if os.path.exists(chats_dir): try: files = [] for f in os.listdir(chats_dir): if f.endswith('.json'): fp = os.path.join(chats_dir, f) files.append((f, os.path.getmtime(fp))) if files: files.sort(key=lambda x: x[1], reverse=True) initial_peer = files[0][0][:-5].upper() except: pass self.chat_to_entry.delete(0, tk.END) self.chat_to_entry.insert(0, initial_peer) self.load_history_for_peer() self.refresh_conversations_list() self.refresh_profiles() def refresh_conversations_list(self): prof = self.chat_profile_var.get() if not prof: return selected_peer = self.chat_to_entry.get().strip().upper() self.chat_listbox.delete(0, tk.END) chats_dir = os.path.join(BASE_DIR, 'chats', prof) if os.path.exists(chats_dir): try: files = [] for f in os.listdir(chats_dir): if f.endswith('.json'): fp = os.path.join(chats_dir, f) files.append((f, os.path.getmtime(fp))) files.sort(key=lambda x: x[1], reverse=True) unread_peers = self.unread_chats.setdefault(prof, set()) for idx, (f, mtime) in enumerate(files): callsign = f[:-5].upper() display_name = callsign if callsign in unread_peers: display_name = f"● {callsign}" self.chat_listbox.insert(tk.END, display_name) if callsign == selected_peer: self.chat_listbox.selection_set(idx) except: pass def on_conversation_select(self, event): prof = self.chat_profile_var.get() if not prof: return selection = self.chat_listbox.curselection() if not selection: return peer = self.chat_listbox.get(selection[0]) if peer.startswith("● "): peer = peer[2:] unread_peers = self.unread_chats.setdefault(prof, set()) if peer in unread_peers: unread_peers.remove(peer) self.chat_to_entry.delete(0, tk.END) self.chat_to_entry.insert(0, peer) self.load_history_for_peer() self.refresh_conversations_list() self.refresh_profiles() def load_history_for_peer(self): prof = self.chat_profile_var.get() if not prof: return peer = self.chat_to_entry.get().strip().upper() if not peer: return unread_peers = self.unread_chats.setdefault(prof, set()) if peer in unread_peers: unread_peers.remove(peer) self.refresh_profiles() try: self.chat_listbox.selection_clear(0, tk.END) items = self.chat_listbox.get(0, tk.END) plain_items = [x[2:] if x.startswith("● ") else x for x in items] if peer in plain_items: idx = plain_items.index(peer) self.chat_listbox.selection_set(idx) self.chat_listbox.see(idx) except: pass self.chat_area.configure(state="normal") self.chat_area.delete("1.0", tk.END) self.chat_area.configure(state="disabled") messages = self.load_chat_messages(prof, peer) self.chat_area.configure(state="normal") for msg in messages: sender = msg.get('sender') text = msg.get('text') timestamp = msg.get('timestamp', '').split(' ')[-1] # get just HH:MM:SS if sender == "SYSTEM": self.chat_area.insert(tk.END, f"[{timestamp}] SYSTEM: {text}\n", "system") elif sender == "YOU": self.chat_area.insert(tk.END, f"[{timestamp}] YOU: {text}\n", "sent") else: self.chat_area.insert(tk.END, f"[{timestamp}] <{sender}> {text}\n", "received") self.chat_area.see(tk.END) self.chat_area.configure(state="disabled") def append_incoming_message(self, sender, text): prof = self.chat_profile_var.get() if not prof: return peer = self.chat_to_entry.get().strip().upper() sender_upper = sender.upper() if sender_upper == peer: self.root.after(0, lambda: [self.append_message(sender, text), self.refresh_conversations_list()]) else: self.unread_chats.setdefault(prof, set()).add(sender_upper) self.root.after(0, lambda: [self.refresh_conversations_list(), self.refresh_profiles()]) # Update tab badge if we are not on chat tab selected_tab = self.notebook.index(self.notebook.select()) if selected_tab != 2: self.root.after(0, lambda: self.notebook.tab(2, text=" APRS Chat (●) ")) def append_message(self, sender, text): self.chat_area.configure(state="normal") timestamp = datetime.now().strftime('%H:%M:%S') if sender == "SYSTEM": self.chat_area.insert(tk.END, f"[{timestamp}] SYSTEM: {text}\n", "system") elif sender == "YOU": self.chat_area.insert(tk.END, f"[{timestamp}] YOU: {text}\n", "sent") else: self.chat_area.insert(tk.END, f"[{timestamp}] <{sender}> {text}\n", "received") self.chat_area.see(tk.END) self.chat_area.configure(state="disabled") def send_message_action(self): prof = self.chat_profile_var.get() if not prof: return to_callsign = self.chat_to_entry.get().strip().upper() msg_text = self.chat_msg_entry.get().strip() if not to_callsign or not msg_text: return profiles = get_profiles() if prof not in profiles: return data = profiles[prof] from_callsign = data.get('callsign', '').upper() # Determine passcode passcode = data.get('passcode') if not passcode: passcode = generate_aprs_passcode(from_callsign) self.chat_send_btn.configure(state="disabled") def send_worker(): success, res_msg = send_aprs_message(from_callsign, passcode, to_callsign, msg_text) self.root.after(0, lambda: self.on_send_complete(success, to_callsign, msg_text, res_msg)) threading.Thread(target=send_worker, daemon=True).start() def on_send_complete(self, success, to_callsign, msg_text, res_msg): self.chat_send_btn.configure(state="normal") if success: self.append_message("YOU", msg_text) self.chat_msg_entry.delete(0, tk.END) self.chat_char_lbl.configure(text="0/67") # Save sent message to history self.save_chat_message(self.chat_profile_var.get(), "YOU", msg_text, to_callsign) self.refresh_conversations_list() else: messagebox.showerror("Error", res_msg, parent=self.root) def open_add_profile_dialog(self, edit_profile_name=None): if not edit_profile_name and not check_is_diag_active() and len(get_profiles()) >= 1: messagebox.showerror("Limit Exceeded", "You can add at most 1 profile. Administrator elevation required for more.") return if not gui_require_auth(self.root): return form = tk.Toplevel(self.root) form.title("Add Profile" if not edit_profile_name else f"Edit Profile: {edit_profile_name}") form.geometry("540x510") form.configure(bg=self.c_bg) form.resizable(False, False) form.transient(self.root) form.grab_set() title_lbl_text = "APRS Beacon Settings" if not edit_profile_name else "Edit APRS Configuration" title = tk.Label(form, text=title_lbl_text, font=("Helvetica", 13, "bold"), fg=self.c_accent_cyan, bg=self.c_bg) title.pack(pady=(20, 15)) fields_frame = tk.Frame(form, bg=self.c_bg) fields_frame.pack(fill="both", expand=True, padx=25) fields_config = [ ("Profile Name (Single word)", "name", 0, 0), ("Callsign (with SSID)", "callsign", 0, 1), ("APRS-IS Passcode (Auto-calculated if empty)", "passcode", 1, 0), ("Interval (Minutes)", "interval", 1, 1), ("Latitude (e.g. 41.032633)", "latitude", 2, 0), ("Longitude (e.g. 28.987083)", "longitude", 2, 1), ("Symbol (e.g. >, X, [)", "symbol", 3, 0), ] entries = {} for label_txt, name, row, col in fields_config: cell_frame = tk.Frame(fields_frame, bg=self.c_bg) cell_frame.grid(row=row, column=col, sticky="ew", padx=10, pady=8) lbl = tk.Label(cell_frame, text=label_txt, font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_bg) lbl.pack(anchor="w", pady=(0, 3)) ent = tk.Entry(cell_frame, bg=self.c_card, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, font=("Helvetica", 10), relief="flat") ent.pack(fill="x", ipady=4) # Add subtle focus highlight effect def on_focus_in(e, entry=ent): entry.configure(highlightbackground=self.c_accent_cyan) def on_focus_out(e, entry=ent): entry.configure(highlightbackground=self.c_border) ent.bind("", on_focus_in) ent.bind("", on_focus_out) entries[name] = ent fields_frame.columnconfigure(0, weight=1) fields_frame.columnconfigure(1, weight=1) # Comment field: spans column 1 at row 3 comment_frame = tk.Frame(fields_frame, bg=self.c_bg) comment_frame.grid(row=3, column=1, sticky="ew", padx=10, pady=8) comment_lbl = tk.Label(comment_frame, text="Status Comment", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_bg) comment_lbl.pack(anchor="w", pady=(0, 3)) comment_ent = tk.Entry(comment_frame, bg=self.c_card, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, font=("Helvetica", 10), relief="flat") comment_ent.pack(fill="x", ipady=4) comment_ent.bind("", lambda e: comment_ent.configure(highlightbackground=self.c_accent_cyan)) comment_ent.bind("", lambda e: comment_ent.configure(highlightbackground=self.c_border)) entries['comment'] = comment_ent # APRS Thursday & Time Frame: spans full width thurs_frame = tk.Frame(fields_frame, bg=self.c_card, padx=12, pady=10, highlightthickness=1, highlightbackground=self.c_border) thurs_frame.grid(row=4, column=0, columnspan=2, sticky="ew", padx=10, pady=15) # Row 1 of thurs_frame: Checkbutton and Time thurs_row1 = tk.Frame(thurs_frame, bg=self.c_card) thurs_row1.pack(fill="x") thursday_var = tk.BooleanVar(value=False) thursday_cb = tk.Checkbutton(thurs_row1, text="Join APRS Thursday Event (ANSRVR)", variable=thursday_var, font=("Helvetica", 9, "bold"), fg=self.c_text_main, bg=self.c_card, activebackground=self.c_card, activeforeground=self.c_text_main, selectcolor=self.c_bg) thursday_cb.pack(side="left") time_ent = tk.Entry(thurs_row1, bg=self.c_bg, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, width=6, font=("Helvetica", 10)) time_ent.pack(side="right", padx=(5, 0)) time_ent.insert(0, "20:00") time_lbl = tk.Label(thurs_row1, text="Time (HH:MM):", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card) time_lbl.pack(side="right") # Row 2 of thurs_frame: Custom Message thurs_row2 = tk.Frame(thurs_frame, bg=self.c_card) thurs_row2.pack(fill="x", pady=(8, 0)) msg_lbl = tk.Label(thurs_row2, text="Thursday Message:", font=("Helvetica", 9, "bold"), fg=self.c_text_muted, bg=self.c_card) msg_lbl.pack(side="left") thurs_msg_ent = tk.Entry(thurs_row2, bg=self.c_bg, fg=self.c_text_main, insertbackground=self.c_text_main, bd=0, highlightthickness=1, highlightbackground=self.c_border, font=("Helvetica", 10)) thurs_msg_ent.pack(side="left", fill="x", expand=True, padx=(10, 0)) thurs_msg_ent.insert(0, "CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY") thurs_msg_ent.bind("", lambda e: thurs_msg_ent.configure(highlightbackground=self.c_accent_cyan)) thurs_msg_ent.bind("", lambda e: thurs_msg_ent.configure(highlightbackground=self.c_border)) # Populate Default / Edit values if edit_profile_name: profiles = get_profiles() data = profiles[edit_profile_name] entries['name'].insert(0, edit_profile_name) entries['name'].configure(state='disabled') entries['callsign'].insert(0, data.get('callsign', '')) entries['passcode'].insert(0, str(data.get('passcode', ''))) entries['latitude'].insert(0, str(data.get('latitude', ''))) entries['longitude'].insert(0, str(data.get('longitude', ''))) entries['comment'].insert(0, data.get('comment', '')) entries['symbol'].insert(0, data.get('symbol_code', 'X')) entries['interval'].insert(0, str(data.get('interval_minutes', '5'))) thursday_var.set(data.get('aprs_thursday', False)) time_ent.delete(0, tk.END) time_ent.insert(0, data.get('aprs_thursday_time', '20:00')) thurs_msg_ent.delete(0, tk.END) thurs_msg_ent.insert(0, data.get('aprs_thursday_msg', 'CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY')) else: entries['comment'].insert(0, "APRS Background Beacon") entries['symbol'].insert(0, "X") entries['interval'].insert(0, "5") thurs_msg_ent.delete(0, tk.END) thurs_msg_ent.insert(0, "CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY") def save_new(): name = edit_profile_name if edit_profile_name else entries['name'].get().strip().lower() callsign = entries['callsign'].get().strip().upper() passcode_in = entries['passcode'].get().strip() lat_in = entries['latitude'].get().strip() lon_in = entries['longitude'].get().strip() comment = entries['comment'].get().strip() symbol = entries['symbol'].get().strip() interval_in = entries['interval'].get().strip() aprs_thursday_time = time_ent.get().strip() or "20:00" aprs_thursday_msg = thurs_msg_ent.get().strip() or "CQ HOTG 73 FROM TURKIYE #APRSTHURSDAY" if not name or not callsign or not lat_in or not lon_in: messagebox.showerror("Error", "All configuration parameters (Name, Callsign, Coordinates) are required.", parent=form) return if not edit_profile_name: profiles = get_profiles() if name in profiles: messagebox.showerror("Error", f"Profile '{name}' already exists.", parent=form) return try: lat = float(lat_in) lon = float(lon_in) except ValueError: messagebox.showerror("Error", "Coordinates must be numerical coordinate values.", parent=form) return if not passcode_in: passcode = generate_aprs_passcode(callsign) else: try: passcode = int(passcode_in) except ValueError: messagebox.showerror("Error", "Invalid passcode entry.", parent=form) return try: interval = int(interval_in) except ValueError: interval = 5 data = { "callsign": callsign, "passcode": passcode, "latitude": lat, "longitude": lon, "use_termux_gps": False, "symbol_table": "/", "symbol_code": symbol if symbol else "X", "comment": comment if comment else "APRS Background Beacon", "interval_minutes": interval, "aprs_thursday": thursday_var.get(), "aprs_thursday_time": aprs_thursday_time, "aprs_thursday_msg": aprs_thursday_msg, "server": "rotate.aprs2.net", "port": 14580 } save_profile(name, data) form.destroy() self.refresh_profiles(force_rebuild=True) self.update_tray_menu() if edit_profile_name: if is_profile_running(name): stop_profile_service(name) time.sleep(0.5) start_profile_service(name) self.refresh_profiles(force_rebuild=True) self.update_tray_menu() messagebox.showinfo("Success", f"Profile '{name}' updated successfully.") else: if messagebox.askyesno("Start Service", f"Profile '{name}' configured. Run background daemon now?"): start_profile_service(name) self.refresh_profiles(force_rebuild=True) self.update_tray_menu() save_btn = tk.Button(form, text="Save Settings", bg=self.c_green, command=save_new) save_btn.pack(pady=(0, 20), side="bottom", fill="x", padx=35) save_btn.configure(relief="flat", bd=0, highlightthickness=0, fg="#ffffff", activeforeground="#ffffff", font=("Helvetica", 10, "bold"), pady=8) save_btn.bind("", lambda e: save_btn.configure(bg="#059669")) save_btn.bind("", lambda e: save_btn.configure(bg=self.c_green)) # System Tray Integration def setup_tray(self): self.update_tray_menu() def update_tray_menu(self): if not GUI_AVAILABLE: return profiles = get_profiles() menu_items = [ pystray.MenuItem("Show Panel", self.restore_from_tray, default=True), pystray.Menu.SEPARATOR ] if profiles: def toggle_wrapper(name, running): return lambda: self.root.after(0, lambda: self.toggle_profile(name, running)) for name in profiles.keys(): running = is_profile_running(name) icon_prefix = "RUN - " if running else "STOP - " toggle_txt = f"{icon_prefix}{name.upper()}" menu_items.append(pystray.MenuItem(toggle_txt, toggle_wrapper(name, running))) menu_items.append(pystray.Menu.SEPARATOR) menu_items.append(pystray.MenuItem("Start All", self.start_all_profiles)) menu_items.append(pystray.MenuItem("Stop All", self.stop_all_profiles)) menu_items.append(pystray.Menu.SEPARATOR) menu_items.append(pystray.MenuItem("Exit", self.quit_application)) menu = pystray.Menu(*menu_items) if self.tray_icon: self.tray_icon.menu = menu else: self.tray_icon = pystray.Icon("aprs_manager", self.icon_image, "APRS Manager", menu) tray_thread = threading.Thread(target=self.tray_icon.run, daemon=True) tray_thread.start() def start_all_profiles(self): profiles = get_profiles() for name in profiles.keys(): start_profile_service(name) time.sleep(0.5) self.root.after(0, lambda: self.refresh_profiles(force_rebuild=False)) self.root.after(0, self.update_tray_menu) def stop_all_profiles(self): profiles = get_profiles() for name in profiles.keys(): stop_profile_service(name) time.sleep(0.5) self.root.after(0, lambda: self.refresh_profiles(force_rebuild=False)) self.root.after(0, self.update_tray_menu) def minimize_to_tray(self): self.root.withdraw() if self.tray_icon: self.tray_icon.notify("APRS Manager minimised. Daemon processes remain running in the background.", "Running in Background") def restore_from_tray(self): self.root.after(0, self.root.deiconify) self.root.after(0, self.root.focus_force) def quit_application(self): self.running = False if self.tray_icon: self.tray_icon.stop() self.root.quit() sys.exit(0) def auto_refresh_loop(self): while self.running: time.sleep(10) if self.running and self.root.winfo_exists(): try: self.root.after(0, lambda: self.refresh_profiles(force_rebuild=False)) except: pass def update_logs_display(self): try: for name, card_data in list(self.profile_cards.items()): log_text_widget = card_data.get('log_text') if log_text_widget and log_text_widget.winfo_exists(): log_file = os.path.join(LOGS_DIR, f"{name}.log") if os.path.exists(log_file): try: with open(log_file, 'r', encoding='utf-8', errors='ignore') as f: lines = f.readlines() last_lines = "".join(lines[-5:]).strip() log_text_widget.configure(state="normal") log_text_widget.delete("1.0", tk.END) log_text_widget.insert(tk.END, last_lines) log_text_widget.see(tk.END) log_text_widget.configure(state="disabled") except: pass else: log_text_widget.configure(state="normal") log_text_widget.delete("1.0", tk.END) log_text_widget.insert(tk.END, "No logs yet. Start the beacon to generate logs...") log_text_widget.configure(state="disabled") except: pass def update_card_logs_loop(self): if not self.running: return self.update_logs_display() self.root.after(3000, self.update_card_logs_loop) def get_chat_filepath(self, profile_name, peer_callsign): chats_dir = os.path.join(BASE_DIR, 'chats', profile_name) os.makedirs(chats_dir, exist_ok=True) return os.path.join(chats_dir, f"{peer_callsign.upper()}.json") def save_chat_message(self, profile_name, sender, text, peer_callsign=None): if sender == "YOU": peer = peer_callsign.upper() else: peer = sender.upper() filepath = self.get_chat_filepath(profile_name, peer) messages = [] if os.path.exists(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: messages = json.load(f) except: pass messages.append({ "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "sender": sender, "text": text }) messages = messages[-100:] try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(messages, f, indent=4) except: pass def load_chat_messages(self, profile_name, peer_callsign): filepath = self.get_chat_filepath(profile_name, peer_callsign) if os.path.exists(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except: pass return [] def manage_chat_listeners_states(self, profiles): for name in list(profiles.keys()): running = is_profile_running(name) if running and name not in self.chat_listeners: self.start_bg_chat_listener(name) elif not running and name in self.chat_listeners: if not (hasattr(self, 'active_chat_ui') and self.active_chat_ui and self.active_chat_ui.winfo_exists() and self.active_chat_ui.active_profile == name): self.stop_bg_chat_listener(name) def start_bg_chat_listener(self, profile_name): if profile_name in self.chat_listeners: return profiles = get_profiles() if profile_name not in profiles: return data = profiles[profile_name] callsign = data.get('callsign', '').upper() passcode = data.get('passcode') if not passcode: passcode = generate_aprs_passcode(callsign) self.chat_listeners[profile_name] = { 'running': True, 'socket': None, 'thread': None } thread = threading.Thread( target=self.bg_chat_listener_worker, args=(profile_name, callsign, passcode), daemon=True ) self.chat_listeners[profile_name]['thread'] = thread thread.start() def stop_bg_chat_listener(self, profile_name): if profile_name in self.chat_listeners: self.chat_listeners[profile_name]['running'] = False sock = self.chat_listeners[profile_name]['socket'] if sock: try: sock.close() except: pass del self.chat_listeners[profile_name] def bg_chat_listener_worker(self, profile_name, callsign, passcode): primary_server = "rotate.aprs2.net" port = 14580 servers_to_try = [primary_server, 'euro.aprs2.net', 'noam.aprs2.net', 'asia.aprs2.net'] connected = False sock = None for server in servers_to_try: if not self.chat_listeners.get(profile_name, {}).get('running', False): return try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(8.0) sock.connect((server, port)) self.chat_listeners[profile_name]['socket'] = sock connected = True break except: continue if not connected or not sock: self.root.after(30000, lambda: self.start_bg_chat_listener(profile_name) if profile_name not in self.chat_listeners else None) return try: sock.recv(1024) # Construct Range filter to receive position packets of surrounding stations filter_str = f"b/{callsign}" try: profiles = get_profiles() p_data = profiles.get(profile_name, {}) lat = p_data.get('latitude') lon = p_data.get('longitude') if lat is not None and lon is not None: filter_str += f" r/{lat}/{lon}/50" except: pass login_str = f"user {callsign} pass {passcode} vers PyAPRSBeacon 1.0 filter {filter_str}\r\n" sock.sendall(login_str.encode('utf-8')) sock.recv(1024) sock.settimeout(None) buffer = "" while self.chat_listeners.get(profile_name, {}).get('running', False): data = sock.recv(4096) if not data: break buffer += data.decode('utf-8', errors='ignore') while "\n" in buffer: line, buffer = buffer.split("\n", 1) line = line.strip() if not line: continue # Parse coordinates from packet to gather surrounding stations try: coords = parse_aprs_coordinates(line) if coords and ">" in line: lat_c, lon_c = coords p_sender = line.split(">", 1)[0].strip() if p_sender.upper() != callsign.upper(): if not hasattr(self, 'nearby_stations'): self.nearby_stations = {} stations = self.nearby_stations.setdefault(profile_name, {}) stations[p_sender.upper()] = (lat_c, lon_c, time.time()) # Prune stations older than 15 mins now = time.time() self.nearby_stations[profile_name] = { k: v for k, v in stations.items() if now - v[2] < 900 } except: pass if "::" in line: try: parts = line.split("::", 1) header = parts[0] payload = parts[1] sender = header.split(">", 1)[0].strip() if ":" in payload: recipient_part, msg_text = payload.split(":", 1) recipient = recipient_part.strip() msg_text = msg_text.strip() if recipient == callsign: msg_id = "" if "{" in msg_text: p_parts = msg_text.rsplit("{", 1) if len(p_parts) == 2 and p_parts[1].isalnum(): msg_text = p_parts[0].strip() msg_id = p_parts[1] if msg_id: sender_padded = f"{sender:<9}" ack_packet = f"{callsign}>APRS,TCPIP*::{sender_padded}:ack{msg_id}\r\n" try: sock.sendall(ack_packet.encode('utf-8')) except: pass self.save_chat_message(profile_name, sender, msg_text) if hasattr(self, 'active_chat_ui') and self.active_chat_ui and self.active_chat_ui.winfo_exists() and self.active_chat_ui.active_profile == profile_name: self.active_chat_ui.append_incoming_message(sender, msg_text) if self.active_chat_ui.to_entry.get().strip().upper() != sender.upper(): if self.tray_icon: disp_text = msg_text[:40] + "..." if len(msg_text) > 40 else msg_text self.tray_icon.notify(f"From {sender}: {disp_text}", f"APRS Message ({profile_name})") else: self.unread_chats.setdefault(profile_name, set()).add(sender.upper()) if self.root.winfo_exists(): self.root.after(0, self.refresh_profiles) if self.tray_icon: disp_text = msg_text[:40] + "..." if len(msg_text) > 40 else msg_text self.tray_icon.notify(f"From {sender}: {disp_text}", f"APRS Message ({profile_name})") except: pass except: pass finally: try: sock.close() except: pass if profile_name in self.chat_listeners: del self.chat_listeners[profile_name] self.root.after(10000, lambda: self.start_bg_chat_listener(profile_name)) def gateway_ping_loop(self): while self.running: start_time = time.time() try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2.0) sock.connect(("rotate.aprs2.net", 14580)) sock.close() latency = int((time.time() - start_time) * 1000) self.server_status = f"Gateway Status: Online ({latency}ms)" self.server_status_color = self.c_green except: self.server_status = "Gateway Status: Offline" self.server_status_color = self.c_red if self.running and self.root.winfo_exists(): try: self.root.after(0, self.update_server_status_ui) except: pass # Update connection every 20 seconds for _ in range(20): if not self.running: break time.sleep(1) def update_server_status_ui(self): self.gateway_status_lbl.configure(text=self.server_status, fg=self.server_status_color) def trigger_self_update(self): if not gui_require_auth(self.root): return if messagebox.askyesno("Update App", "Would you like to fetch updates and update the software from GitHub?"): profiles = get_profiles() running_profiles = [name for name, data in profiles.items() if is_profile_running(name)] for name in running_profiles: stop_profile_service(name) success, msg = self_update() if success: messagebox.showinfo("Success", "Application updated successfully! The application will now restart.") for name in running_profiles: start_profile_service(name) import sys import os python = sys.executable os.execl(python, python, *sys.argv) else: messagebox.showerror("Error", msg) for name in running_profiles: start_profile_service(name) def trigger_export(self): file_path = filedialog.asksaveasfilename( defaultextension=".json", filetypes=[("JSON Backup Files", "*.json")], title="Export Backup Settings", initialfile=f"aprs_config_backup_{datetime.now().strftime('%Y%m%d')}.json" ) if file_path: success, msg = export_settings(file_path) if success: messagebox.showinfo("Success", msg) else: messagebox.showerror("Error", msg) def trigger_import(self): if not gui_require_auth(self.root): return file_path = filedialog.askopenfilename( filetypes=[("JSON Backup Files", "*.json")], title="Import Settings Backup" ) if file_path: success, msg = import_settings(file_path) if success: messagebox.showinfo("Success", msg) self.refresh_profiles(force_rebuild=True) self.update_tray_menu() else: messagebox.showerror("Error", msg) # CLI Interactive Mode & Helper def cli_show_logs_interactive(profile_name): profiles = get_profiles() if profile_name not in profiles: print(f"Error: Profile '{profile_name}' not found.") return log_path = os.path.join(LOGS_DIR, f"{profile_name}.log") if not os.path.exists(log_path): print("Log file not found. Daemon has not generated log events yet.") return print(f"\n--- {profile_name.upper()} LIVE LOG OUTPUT (Ctrl+C to stop) ---") try: with open(log_path, 'r', encoding='utf-8', errors='ignore') as f: lines = f.readlines() for line in lines[-20:]: print(line, end='') while True: line = f.readline() if not line: time.sleep(1) continue print(line, end='') except KeyboardInterrupt: print("\nClosed live log monitor.") def cli_interactive_menu(): while True: print("\n=== APRS Multi-Beacon Manager (CLI) ===") print("1) List Profiles") print("2) Create New Profile") print("3) Start Profile Daemon") print("4) Stop Profile Daemon") print("5) Delete Profile") print("6) Live Log Monitor") print("7) Edit Profile Settings") print("8) Export Settings Backup") print("9) Import Settings Backup") print("10) Fetch Updates") print("11) Exit") print("-" * 45) choice = input("Choice [1-11]: ").strip() if choice == '1': cli_list() elif choice == '2': cli_create() elif choice == '3': name = input("Profile name to start: ").strip().lower() if name: cli_start(name) elif choice == '4': name = input("Profile name to stop: ").strip().lower() if name: cli_stop(name) elif choice == '5': name = input("Profile name to delete: ").strip().lower() if name: cli_delete(name) elif choice == '6': name = input("Profile name to view: ").strip().lower() if name: cli_show_logs_interactive(name) elif choice == '7': cli_edit() elif choice == '8': export_path = input("Export location [Default: ~/aprs_backup.json]: ").strip() if not export_path: export_path = os.path.expanduser("~/aprs_backup.json") success, msg = export_settings(export_path) print(f"[+] {msg}" if success else f"[-] {msg}") elif choice == '9': if not cli_require_auth(): continue import_path = input("Import file location: ").strip() if import_path: success, msg = import_settings(import_path) print(f"[+] {msg}" if success else f"[-] {msg}") elif choice == '10': if not cli_require_auth(): continue print("[i] Checking updates...") success, msg = self_update() print(f"[+] {msg}" if success else f"[-] {msg}") elif choice == '11' or choice.lower() == 'exit': print("Exiting...") break else: print("Invalid choice! Enter a selection option from 1 to 11.") # Entry Point / Argument Parsing def main(): parser = argparse.ArgumentParser(description="APRS Multi-Beacon Management Utility") parser.add_argument('command', nargs='?', choices=['list', 'start', 'stop', 'create', 'delete', 'edit', 'export', 'import', 'update', 'gui'], default='gui', help="Command to run (list, start, stop, create, delete, edit, export, import, update, gui)") parser.add_argument('profile_name', nargs='?', help="Target profile name or file path for commands") args = parser.parse_args() if args.command == 'list': cli_list() elif args.command == 'create': cli_create() elif args.command == 'edit': cli_edit() elif args.command == 'update': if not cli_require_auth(): return print("[i] Checking updates...") success, msg = self_update() print(f"[+] {msg}" if success else f"[-] {msg}") elif args.command == 'export': path = args.profile_name if not path: path = os.path.expanduser("~/aprs_backup.json") success, msg = export_settings(path) print(f"[+] {msg}" if success else f"[-] {msg}") elif args.command == 'import': if not cli_require_auth(): return path = args.profile_name if not path: print("Error: Specify backup file directory path (e.g. aprs_manager import backup.json)") else: success, msg = import_settings(path) print(f"[+] {msg}" if success else f"[-] {msg}") elif args.command == 'delete': if not args.profile_name: print("Error: Specify profile name to delete (e.g. aprs_manager delete my_profile)") else: cli_delete(args.profile_name) elif args.command == 'start': if not args.profile_name: print("Error: Specify profile name to start.") else: cli_start(args.profile_name) elif args.command == 'stop': if not args.profile_name: print("Error: Specify profile name to stop.") else: cli_stop(args.profile_name) elif args.command == 'gui': is_headless = False if IS_LINUX and 'DISPLAY' not in os.environ: is_headless = True if not GUI_AVAILABLE or is_headless: if is_headless: print("\033[93m[!] X-server connection (DISPLAY) not detected. Defaulting to terminal mode.\033[0m") else: print("\033[91mError: GUI libraries (tkinter, pystray, pillow) not fully installed.\033[0m") print("Launching terminal interactive interface...\n") time.sleep(1) cli_interactive_menu() sys.exit(0) try: root = tk.Tk() app = APRSManagerGUI(root) root.mainloop() except Exception as e: print(f"\033[91mCould not launch graphical window: {e}\033[0m") print("Launching terminal interactive interface...\n") time.sleep(1) cli_interactive_menu() if __name__ == '__main__': main()