2829 lines
126 KiB
Python
Executable File
2829 lines
126 KiB
Python
Executable File
#!/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("<Enter>", lambda e: btn.configure(bg=color_hover))
|
||
btn.bind("<Leave>", 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(
|
||
"<Configure>",
|
||
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('<Configure>', _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("<FocusIn>", ent_focus_in)
|
||
self.map_search_entry.bind("<FocusOut>", ent_focus_out)
|
||
self.map_search_entry.bind("<Return>", 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("<Enter>", lambda e: self.map_search_btn.configure(bg="#059669"))
|
||
self.map_search_btn.bind("<Leave>", 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("<Enter>", lambda e: self.map_browser_btn.configure(bg="#1d4ed8"))
|
||
self.map_browser_btn.bind("<Leave>", 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("<<NotebookTabChanged>>", 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("<ButtonPress-1>", on_press)
|
||
lbl_part2.bind("<ButtonRelease-1>", on_release)
|
||
lbl_part2.bind("<Leave>", 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("<Enter>", lambda e, b=btn, h=toggle_hover: b.configure(bg=h))
|
||
btn.bind("<Leave>", 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("<Button-1>", 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("<Enter>", lambda e: btn.configure(bg=color_hover))
|
||
btn.bind("<Leave>", 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("<FocusIn>", to_focus_in)
|
||
self.chat_to_entry.bind("<FocusOut>", to_focus_out)
|
||
self.chat_to_entry.bind("<Return>", 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("<<ListboxSelect>>", 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("<KeyRelease>", validate_msg)
|
||
self.chat_msg_entry.bind("<Return>", 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("<Enter>", lambda e: self.chat_send_btn.configure(bg="#059669"))
|
||
self.chat_send_btn.bind("<Leave>", 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("<FocusIn>", on_focus_in)
|
||
ent.bind("<FocusOut>", 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("<FocusIn>", lambda e: comment_ent.configure(highlightbackground=self.c_accent_cyan))
|
||
comment_ent.bind("<FocusOut>", 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("<FocusIn>", lambda e: thurs_msg_ent.configure(highlightbackground=self.c_accent_cyan))
|
||
thurs_msg_ent.bind("<FocusOut>", 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("<Enter>", lambda e: save_btn.configure(bg="#059669"))
|
||
save_btn.bind("<Leave>", 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()
|