#!/usr/bin/env python3 """ rtcom.py - Range Test Termux Companion App WebUI for LXMF-CLI rangetest plugin Exposes the range test HTML map via Flask on localhost:8033 Features: - Live HTML map with auto-refresh - Current GPS position tracking (red arrow like a navigator) - Real-time point updates - Mobile-optimized interface Usage: python3 rtcom.py Then open: http://localhost:8033 on a phone web browser """ import os import sys import json import time import shutil import subprocess from datetime import datetime from flask import Flask, render_template_string, jsonify, send_file, request from threading import Thread, Lock app = Flask(__name__) # Configuration PORT = 8033 HOST = '0.0.0.0' # Listen on all interfaces for LAN access GPS_UPDATE_INTERVAL = 1 # seconds between GPS updates HTML_REFRESH_INTERVAL = 5 # seconds between HTML file checks # Global state current_gps = { 'latitude': None, 'longitude': None, 'accuracy': None, 'speed': None, 'altitude': None, 'bearing': None, 'provider': None, 'timestamp': None, 'available': False } gps_lock = Lock() # File paths (matching rangetest_client.py) STORAGE_DIR = os.path.expanduser('/data/data/com.termux/files/home/lxmf-cli/lxmf_client_storage') HTML_FILE = os.path.join(STORAGE_DIR, 'rangetest.html') JSON_FILE = os.path.join(STORAGE_DIR, 'rangetest.json') def is_termux(): """Check if running on Termux""" return os.path.exists('/data/data/com.termux') def get_current_gps(): """Get current GPS position (non-blocking)""" if not is_termux(): return None try: # Try GPS first (quick 2s timeout) result = subprocess.run( ['termux-location', '-p', 'gps', '-r', 'once'], capture_output=True, text=True, timeout=2, env=os.environ.copy() ) if result.returncode == 0 and result.stdout.strip(): data = json.loads(result.stdout.strip()) if 'latitude' in data and 'longitude' in data: lat = data.get('latitude') lon = data.get('longitude') if lat and lon and (abs(lat) > 0.001 or abs(lon) > 0.001): return data # Fallback to network (1s timeout) result = subprocess.run( ['termux-location', '-p', 'network', '-r', 'once'], capture_output=True, text=True, timeout=1, env=os.environ.copy() ) if result.returncode == 0 and result.stdout.strip(): data = json.loads(result.stdout.strip()) if 'latitude' in data and 'longitude' in data: lat = data.get('latitude') lon = data.get('longitude') if lat and lon and (abs(lat) > 0.001 or abs(lon) > 0.001): return data except Exception as e: print(f"[GPS] Error: {e}") return None def gps_updater(): """Background thread to continuously update GPS position""" global current_gps print(f"[GPS Updater] Started (interval: {GPS_UPDATE_INTERVAL}s)") while True: try: gps_data = get_current_gps() with gps_lock: if gps_data: current_gps = { 'latitude': gps_data.get('latitude'), 'longitude': gps_data.get('longitude'), 'accuracy': gps_data.get('accuracy', 0), 'speed': gps_data.get('speed', 0), 'altitude': gps_data.get('altitude', 0), 'bearing': gps_data.get('bearing', None), 'provider': gps_data.get('provider', 'unknown'), 'timestamp': datetime.now().isoformat(), 'available': True } else: current_gps['available'] = False except Exception as e: print(f"[GPS Updater] Error: {e}") time.sleep(GPS_UPDATE_INTERVAL) def get_logged_points(): """Get logged points from JSON file""" try: if os.path.exists(JSON_FILE): with open(JSON_FILE, 'r') as f: data = json.load(f) return data.get('points', []) except Exception as e: print(f"[Points] Error reading JSON: {e}") return [] # Enhanced HTML template with live GPS tracking NAVIGATOR_TEMPLATE = ''' Range Test Navigator - Live
πŸ“ Current Position: Waiting for GPS...
Logged Points: 0
Distance: 0 km
Avg RSSI: N/A
Avg SNR: N/A
Last Update: Live
''' @app.route('/') def index(): """Serve the live navigator page""" return render_template_string( NAVIGATOR_TEMPLATE, gps_interval=GPS_UPDATE_INTERVAL, html_interval=HTML_REFRESH_INTERVAL ) @app.route('/api/current_gps') def api_current_gps(): """API endpoint for current GPS position""" with gps_lock: return jsonify(current_gps) @app.route('/api/logged_points') def api_logged_points(): """API endpoint for logged points""" points = get_logged_points() return jsonify({'points': points}) @app.route('/map') def static_map(): """Serve the original static HTML map""" if os.path.exists(HTML_FILE): return send_file(HTML_FILE) else: return "Range test map not found. Start logging points first!", 404 @app.route('/api/export_map', methods=['POST']) def export_map(): """Export current map HTML to /sdcard/Download with timestamp""" try: if not is_termux(): return jsonify({'success': False, 'error': 'Export only works on Termux/Android'}) download_dir = '/sdcard/Download' if not os.path.exists(download_dir): return jsonify({'success': False, 'error': '/sdcard/Download not found'}) # Generate timestamp filename timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'rangetest_{timestamp}.html' dest_path = os.path.join(download_dir, filename) # Check if source HTML exists if not os.path.exists(HTML_FILE): return jsonify({'success': False, 'error': 'No range test data to export'}) # Copy HTML file to Download folder import shutil shutil.copy(HTML_FILE, dest_path) return jsonify({'success': True, 'filename': filename, 'path': dest_path}) except Exception as e: return jsonify({'success': False, 'error': str(e)}) @app.route('/api/send_lxmf_command', methods=['POST']) def send_lxmf_command(): """Send command to LXMF-CLI via command file""" try: data = request.get_json() contact_index = data.get('contact_index') command = data.get('command') if not contact_index or not command: return jsonify({'success': False, 'error': 'Missing required fields'}) # Build message based on command if command == 'rt': # Start range test: s # rt N D ping_count = data.get('ping_count', 10) ping_delay = data.get('ping_delay', 5) message = f"s {contact_index} rt {ping_count} {ping_delay}" elif command == 'rs': # Stop range test: s # rs message = f"s {contact_index} rs" else: return jsonify({'success': False, 'error': 'Unknown command'}) # Write command to file as a ONE-TIME TRIGGER command_file = os.path.join(STORAGE_DIR, 'rtcom_command.txt') # Write command with open(command_file, 'w') as f: f.write(message + '\n') # Flush to ensure file is written import time time.sleep(0.1) # Note: The bridge plugin will clear the file after reading it # This ensures the command is only executed once return jsonify({ 'success': True, 'message': message, 'note': 'Command sent to LXMF-CLI via rtcom_bridge plugin' }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) def print_banner(): """Print startup banner""" print("\n" + "="*70) print("πŸ—ΊοΈ rtcom - Range Test Termux Companion App WebUI") print("="*70) print(f"πŸ“‘ Listening on: http://{HOST}:{PORT}") print(f"🌐 Local access: http://localhost:{PORT}") # Try to get local IP try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) local_ip = s.getsockname()[0] s.close() print(f"πŸ“± LAN access: http://{local_ip}:{PORT}") except: pass print(f"\nπŸ“‚ Data directory: {STORAGE_DIR}") print(f"πŸ”„ GPS updates: every {GPS_UPDATE_INTERVAL}s") print(f"πŸ”„ Map refresh: every {HTML_REFRESH_INTERVAL}s") if is_termux(): print(f"\nβœ… Termux detected - GPS tracking enabled") else: print(f"\n⚠️ Not on Termux - GPS tracking disabled") print(f"\nπŸ“ Endpoints:") print(f" / Live navigator with current position") print(f" /map Original static map") print(f" /api/current_gps Current GPS JSON") print(f" /api/logged_points Logged points JSON") print(f" /api/export_map Export map to /sdcard/Download (POST)") print("\nπŸ’‘ Press Ctrl+C to stop") print("="*70 + "\n") def main(): """Main entry point""" print_banner() # Check if storage directory exists if not os.path.exists(STORAGE_DIR): print(f"⚠️ Warning: Storage directory not found: {STORAGE_DIR}") print(f" Make sure LXMF-CLI is running from the correct directory") print(f" Expected: /data/data/com.termux/files/home/lxmf-cli/") # Start GPS updater thread (only on Termux) if is_termux(): gps_thread = Thread(target=gps_updater, daemon=True) gps_thread.start() else: print("[Info] GPS tracking disabled (not on Termux)") # Start Flask server try: app.run(host=HOST, port=PORT, debug=False, threaded=True) except KeyboardInterrupt: print("\n\nπŸ›‘ Server stopped by user") except Exception as e: print(f"\n❌ Server error: {e}") sys.exit(1) if __name__ == '__main__': main()