#!/usr/bin/env python3 """ Pager Decoder - POCSAG/FLEX decoder using RTL-SDR and multimon-ng """ import subprocess import shutil import re import threading import queue import pty import os import select from flask import Flask, render_template_string, jsonify, request, Response app = Flask(__name__) # Global process management current_process = None output_queue = queue.Queue() process_lock = threading.Lock() # Logging settings logging_enabled = False log_file_path = 'pager_messages.log' HTML_TEMPLATE = ''' INTERCEPT // Signal Intelligence

INTERCEPT

Signal Intelligence // POCSAG & FLEX Decoder

Device

rtl_fm: {{ 'OK' if tools.rtl_fm else 'Missing' }} multimon-ng: {{ 'OK' if tools.multimon else 'Missing' }}

Frequency

Frequency (MHz)

Protocols

POCSAG-512 POCSAG-1200 POCSAG-2400 FLEX

Settings

Gain (dB, 0 = auto)

Squelch Level

PPM Correction

Logging

Enable message logging

Log file path

Decoded Messages

Messages: 0

POCSAG: 0

FLEX: 0

Configure settings and click "Start Decoding" to begin.

Idle

''' def check_tool(name): """Check if a tool is installed.""" return shutil.which(name) is not None def detect_devices(): """Detect RTL-SDR devices.""" devices = [] if not check_tool('rtl_test'): return devices try: result = subprocess.run( ['rtl_test', '-t'], capture_output=True, text=True, timeout=5 ) output = result.stderr + result.stdout # Parse device info device_pattern = r'(\d+):\s+(.+?)(?:,\s*SN:\s*(\S+))?$' for line in output.split('\n'): line = line.strip() match = re.match(device_pattern, line) if match: devices.append({ 'index': int(match.group(1)), 'name': match.group(2).strip().rstrip(','), 'serial': match.group(3) or 'N/A' }) if not devices: found_match = re.search(r'Found (\d+) device', output) if found_match: count = int(found_match.group(1)) for i in range(count): devices.append({ 'index': i, 'name': f'RTL-SDR Device {i}', 'serial': 'Unknown' }) except Exception: pass return devices def parse_multimon_output(line): """Parse multimon-ng output line.""" # POCSAG formats: # POCSAG512: Address: 1234567 Function: 0 Alpha: Message here # POCSAG1200: Address: 1234567 Function: 0 Numeric: 123-456-7890 # POCSAG2400: Address: 1234567 Function: 0 (no message) # FLEX formats: # FLEX: NNNN-NN-NN NN:NN:NN NNNN/NN/C NN.NNN [NNNNNNN] ALN Message here # FLEX|NNNN-NN-NN|NN:NN:NN|NNNN/NN/C|NN.NNN|NNNNNNN|ALN|Message line = line.strip() # POCSAG parsing - with message content pocsag_match = re.match( r'(POCSAG\d+):\s*Address:\s*(\d+)\s+Function:\s*(\d+)\s+(Alpha|Numeric):\s*(.*)', line ) if pocsag_match: return { 'protocol': pocsag_match.group(1), 'address': pocsag_match.group(2), 'function': pocsag_match.group(3), 'msg_type': pocsag_match.group(4), 'message': pocsag_match.group(5).strip() or '[No Message]' } # POCSAG parsing - address only (no message content) pocsag_addr_match = re.match( r'(POCSAG\d+):\s*Address:\s*(\d+)\s+Function:\s*(\d+)\s*$', line ) if pocsag_addr_match: return { 'protocol': pocsag_addr_match.group(1), 'address': pocsag_addr_match.group(2), 'function': pocsag_addr_match.group(3), 'msg_type': 'Tone', 'message': '[Tone Only]' } # FLEX parsing (standard format) flex_match = re.match( r'FLEX[:\|]\s*[\d\-]+[\s\|]+[\d:]+[\s\|]+([\d/A-Z]+)[\s\|]+([\d.]+)[\s\|]+\[?(\d+)\]?[\s\|]+(\w+)[\s\|]+(.*)', line ) if flex_match: return { 'protocol': 'FLEX', 'address': flex_match.group(3), 'function': flex_match.group(1), 'msg_type': flex_match.group(4), 'message': flex_match.group(5).strip() or '[No Message]' } # Simple FLEX format flex_simple = re.match(r'FLEX:\s*(.+)', line) if flex_simple: return { 'protocol': 'FLEX', 'address': 'Unknown', 'function': '', 'msg_type': 'Unknown', 'message': flex_simple.group(1).strip() } return None def stream_decoder(master_fd, process): """Stream decoder output to queue using PTY for unbuffered output.""" global current_process try: output_queue.put({'type': 'status', 'text': 'started'}) buffer = "" while True: try: ready, _, _ = select.select([master_fd], [], [], 1.0) except Exception: break if ready: try: data = os.read(master_fd, 1024) if not data: break buffer += data.decode('utf-8', errors='replace') while '\n' in buffer: line, buffer = buffer.split('\n', 1) line = line.strip() if not line: continue parsed = parse_multimon_output(line) if parsed: from datetime import datetime parsed['timestamp'] = datetime.now().strftime('%H:%M:%S') output_queue.put({'type': 'message', **parsed}) log_message(parsed) else: output_queue.put({'type': 'raw', 'text': line}) except OSError: break if process.poll() is not None: break except Exception as e: output_queue.put({'type': 'error', 'text': str(e)}) finally: try: os.close(master_fd) except: pass process.wait() output_queue.put({'type': 'status', 'text': 'stopped'}) with process_lock: current_process = None @app.route('/') def index(): tools = { 'rtl_fm': check_tool('rtl_fm'), 'multimon': check_tool('multimon-ng') } devices = detect_devices() return render_template_string(HTML_TEMPLATE, tools=tools, devices=devices) @app.route('/devices') def get_devices(): return jsonify(detect_devices()) @app.route('/start', methods=['POST']) def start_decoding(): global current_process with process_lock: if current_process: return jsonify({'status': 'error', 'message': 'Already running'}) data = request.json freq = data.get('frequency', '929.6125') gain = data.get('gain', '0') squelch = data.get('squelch', '0') ppm = data.get('ppm', '0') device = data.get('device', '0') protocols = data.get('protocols', ['POCSAG512', 'POCSAG1200', 'POCSAG2400', 'FLEX']) # Clear queue while not output_queue.empty(): try: output_queue.get_nowait() except: break # Build multimon-ng decoder arguments decoders = [] for proto in protocols: if proto == 'POCSAG512': decoders.extend(['-a', 'POCSAG512']) elif proto == 'POCSAG1200': decoders.extend(['-a', 'POCSAG1200']) elif proto == 'POCSAG2400': decoders.extend(['-a', 'POCSAG2400']) elif proto == 'FLEX': decoders.extend(['-a', 'FLEX']) # Build rtl_fm command # rtl_fm -d -f M -M fm -s 22050 -g -p -l - | multimon-ng -t raw -a POCSAG512 -a POCSAG1200 -a FLEX -f alpha - rtl_cmd = [ 'rtl_fm', '-d', str(device), '-f', f'{freq}M', '-M', 'fm', '-s', '22050', ] if gain and gain != '0': rtl_cmd.extend(['-g', str(gain)]) if ppm and ppm != '0': rtl_cmd.extend(['-p', str(ppm)]) if squelch and squelch != '0': rtl_cmd.extend(['-l', str(squelch)]) rtl_cmd.append('-') multimon_cmd = ['multimon-ng', '-t', 'raw'] + decoders + ['-f', 'alpha', '-'] # Log the command being run full_cmd = ' '.join(rtl_cmd) + ' | ' + ' '.join(multimon_cmd) print(f"Running: {full_cmd}") try: # Create pipe: rtl_fm | multimon-ng # Use PTY for multimon-ng to get unbuffered output rtl_process = subprocess.Popen( rtl_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # Start a thread to monitor rtl_fm stderr for errors def monitor_rtl_stderr(): for line in rtl_process.stderr: err_text = line.decode('utf-8', errors='replace').strip() if err_text: print(f"[RTL_FM] {err_text}", flush=True) output_queue.put({'type': 'raw', 'text': f'[rtl_fm] {err_text}'}) rtl_stderr_thread = threading.Thread(target=monitor_rtl_stderr) rtl_stderr_thread.daemon = True rtl_stderr_thread.start() # Create a pseudo-terminal for multimon-ng output # This tricks it into thinking it's connected to a terminal, # which disables output buffering master_fd, slave_fd = pty.openpty() multimon_process = subprocess.Popen( multimon_cmd, stdin=rtl_process.stdout, stdout=slave_fd, stderr=slave_fd, close_fds=True ) os.close(slave_fd) # Close slave fd in parent process rtl_process.stdout.close() # Allow rtl_process to receive SIGPIPE current_process = multimon_process current_process._rtl_process = rtl_process # Store reference to kill later current_process._master_fd = master_fd # Store for cleanup # Start output thread with PTY master fd thread = threading.Thread(target=stream_decoder, args=(master_fd, multimon_process)) thread.daemon = True thread.start() # Send the command info to the client output_queue.put({'type': 'info', 'text': f'Command: {full_cmd}'}) return jsonify({'status': 'started', 'command': full_cmd}) except FileNotFoundError as e: return jsonify({'status': 'error', 'message': f'Tool not found: {e.filename}'}) except Exception as e: return jsonify({'status': 'error', 'message': str(e)}) @app.route('/stop', methods=['POST']) def stop_decoding(): global current_process with process_lock: if current_process: # Kill rtl_fm process first if hasattr(current_process, '_rtl_process'): try: current_process._rtl_process.terminate() current_process._rtl_process.wait(timeout=2) except: try: current_process._rtl_process.kill() except: pass # Close PTY master fd if hasattr(current_process, '_master_fd'): try: os.close(current_process._master_fd) except: pass # Kill multimon-ng current_process.terminate() try: current_process.wait(timeout=2) except subprocess.TimeoutExpired: current_process.kill() current_process = None return jsonify({'status': 'stopped'}) return jsonify({'status': 'not_running'}) @app.route('/status') def get_status(): """Check if decoder is currently running.""" with process_lock: if current_process and current_process.poll() is None: return jsonify({'running': True, 'logging': logging_enabled, 'log_file': log_file_path}) return jsonify({'running': False, 'logging': logging_enabled, 'log_file': log_file_path}) @app.route('/logging', methods=['POST']) def toggle_logging(): """Toggle message logging.""" global logging_enabled, log_file_path data = request.json if 'enabled' in data: logging_enabled = data['enabled'] if 'log_file' in data and data['log_file']: log_file_path = data['log_file'] return jsonify({'logging': logging_enabled, 'log_file': log_file_path}) def log_message(msg): """Log a message to file if logging is enabled.""" if not logging_enabled: return try: with open(log_file_path, 'a') as f: from datetime import datetime timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') f.write(f"{timestamp} | {msg.get('protocol', 'UNKNOWN')} | {msg.get('address', '')} | {msg.get('message', '')}\n") except Exception as e: print(f"[ERROR] Failed to log message: {e}", flush=True) @app.route('/killall', methods=['POST']) def kill_all(): """Kill all rtl_fm and multimon-ng processes.""" global current_process killed = [] try: result = subprocess.run(['pkill', '-f', 'rtl_fm'], capture_output=True) if result.returncode == 0: killed.append('rtl_fm') except: pass try: result = subprocess.run(['pkill', '-f', 'multimon-ng'], capture_output=True) if result.returncode == 0: killed.append('multimon-ng') except: pass with process_lock: current_process = None return jsonify({'status': 'killed', 'processes': killed}) @app.route('/stream') def stream(): def generate(): import json while True: try: msg = output_queue.get(timeout=1) yield f"data: {json.dumps(msg)}\n\n" except queue.Empty: yield f"data: {json.dumps({'type': 'keepalive'})}\n\n" response = Response(generate(), mimetype='text/event-stream') response.headers['Cache-Control'] = 'no-cache' response.headers['X-Accel-Buffering'] = 'no' response.headers['Connection'] = 'keep-alive' return response def main(): print("=" * 50) print(" Pager Decoder") print(" POCSAG / FLEX using RTL-SDR + multimon-ng") print("=" * 50) print() print("Open http://localhost:5050 in your browser") print() print("Press Ctrl+C to stop") print() app.run(host='0.0.0.0', port=5050, debug=False, threaded=True) if __name__ == '__main__': main()