#!/usr/bin/env python3 """ dAISy Catcher Statistics and Spectrogram Viewer ========================================== This program displays real-time AIS channel power levels and spectrogram data from a dAISy Catcher receiver. It provides multiple automatic scaling modes for optimal spectrogram visualization. SPECTROGRAM SCALING MODES ------------------------- 1. ADAPTIVE MODE (Default - Recommended) Command: --spec-scale adaptive - Automatically adjusts based on recent frames only - Provides the best dynamic range for current conditions - Ignores old data that might have different characteristics - Uses percentile-based scaling to ignore outliers Usage: python stats.py /dev/serial0 --spec-scale adaptive Fine-tuning: # Use last 100 frames for scaling python stats.py /dev/serial0 --spec-scale adaptive --spec-adaptive-window 100 # Adjust outlier rejection (default 5-95 percentile) python stats.py /dev/serial0 --spec-scale adaptive --spec-percentile-low 10 --spec-percentile-high 90 2. PERCENTILE MODE Command: --spec-scale percentile - Uses percentiles across ALL historical data - Good for comparing signal levels across the entire recording - Automatically ignores extreme outliers Usage: python stats.py /dev/serial0 --spec-scale percentile # With custom percentiles python stats.py /dev/serial0 --spec-scale percentile --spec-percentile-low 2 --spec-percentile-high 98 3. AUTO MODE Command: --spec-scale auto - Scales to the absolute min/max of all data - Can be problematic if there are strong outliers - Good when you want to see the full dynamic range Usage: python stats.py /dev/serial0 --spec-scale auto 4. FIXED MODE Command: --spec-scale fixed - Uses fixed vmin/vmax values you specify - Best for comparing multiple recording sessions - Ensures consistent color mapping Usage: # With default range (-120 to -40 dB/Hz) python stats.py /dev/serial0 --spec-scale fixed # With custom range python stats.py /dev/serial0 --spec-scale fixed --spec-vmin -130 --spec-vmax -30 COMPARISON OF MODES ------------------- Mode | Best For | Pros | Cons -----------|-----------------------------------|--------------------------|--------------------------- Adaptive | Real-time monitoring | Always optimal | Historical comparison harder Percentile | Long recordings with varying | Robust to outliers | May miss weak signals | conditions | | Auto | Capturing full dynamic range | Shows everything | Outliers can compress scale Fixed | Comparing multiple sessions | Consistent scale | Requires manual tuning COMPLETE EXAMPLE COMMANDS -------------------------- # For general AIS monitoring (recommended): python stats.py /dev/serial0 --spec-scale adaptive # For weak signal hunting: python stats.py /dev/serial0 --spec-scale adaptive --spec-percentile-low 20 --spec-percentile-high 95 # For consistent measurements across sessions: python stats.py /dev/serial0 --spec-scale fixed --spec-vmin -110 --spec-vmax -50 # For research/documentation: python stats.py /dev/serial0 --spec-scale adaptive --spec-adaptive-window 20 --spec-history 200 TIPS ---- 1. Start with adaptive mode - it works well for most situations 2. Adjust adaptive window based on how fast conditions change: - Fast changing: use smaller window (10-30 frames) - Stable conditions: use larger window (50-100 frames) 3. Use fixed mode when comparing data from different times/locations 4. Watch the scale values in the title to understand what you're seeing 5. Lower percentile values = more sensitive to noise 6. Higher percentile values = may clip strong signals TROUBLESHOOTING --------------- Problem: Spectrogram looks washed out Solution: Use tighter percentiles (e.g., 10-90 instead of 5-95) Problem: Can't see weak signals Solution: Use adaptive mode with smaller window and tighter percentiles Problem: Strong signals saturate the display Solution: Widen the percentile range or use fixed mode with appropriate limits Problem: Scale jumps around too much Solution: Increase adaptive window size or use percentile mode Author: Enhanced version with automatic scaling License: Same as original """ import argparse import logging import matplotlib.pyplot as plt import numpy as np import serial import socket import json import time from pyais import decode from collections import deque from typing import List, Dict, Optional, Deque from queue import Queue import signal # Reliably exit on CTRL-C exit_sig = Queue() def ctrl_c_handler(sig, frame): # Ctrl-C detected, initiating exit exit_sig.put(True) signal.signal(signal.SIGINT, ctrl_c_handler) # Configure logging - provides timestamps and severity levels logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # ============================================================================= # CONSTANTS - Serial Communication # ============================================================================= SERIAL_BAUD_RATE = 115200 # Standard baud rate for dAISy catcher SERIAL_TIMEOUT = 1 # Read timeout in seconds DEVICE_BOOT_DELAY = 0.5 # Wait time for device to fully boot COMMAND_DELAY = 0.1 # Delay between sending commands BUFFER_FLUSH_DELAY = 0.2 # Delay for buffer flushing operations # ============================================================================= # CONSTANTS - Plot Configuration # ============================================================================= POWER_PLOT_WIDTH = 12 # Width of power plot in inches POWER_PLOT_HEIGHT = 6 # Height of power plot in inches SPEC_PLOT_WIDTH = 12 # Width of spectrogram plot in inches SPEC_PLOT_HEIGHT = 8 # Height of spectrogram plot in inches BLOCK_MASK = 0xFFFF # 16-bit block number wraparound mask POWER_UPDATE_INTERVAL = 0.05 # Minimum seconds between power plot updates INITIAL_POWER_VALUE = -100 # Initial power value for empty data (dBm) # ============================================================================= # CONSTANTS - Spectrogram Scaling Configuration # ============================================================================= # Default scaling mode - 'auto' provides most reliable results for most uses SPEC_SCALE_MODE = 'auto' # Options: 'auto', 'fixed', 'percentile', 'adaptive' # Fixed mode parameters - used when SPEC_SCALE_MODE = 'fixed' SPEC_VMIN = -120 # Fixed minimum for 'fixed' mode (dB/Hz) SPEC_VMAX = -40 # Fixed maximum for 'fixed' mode (dB/Hz) # Percentile parameters - used in 'percentile' and 'adaptive' modes SPEC_PERCENTILE_LOW = 5 # Lower percentile for scaling (ignore bottom 5%) SPEC_PERCENTILE_HIGH = 95 # Upper percentile for scaling (ignore top 5%) # Adaptive mode parameter - how many recent frames to consider SPEC_ADAPTIVE_WINDOW = 50 # Number of recent frames for 'adaptive' mode # ============================================================================= # GLOBAL VARIABLES - Data Storage # ============================================================================= # Power data for both AIS channels (A and B) # Using deques for automatic size limiting and efficient append/pop ch1_power: Deque[float] # Channel A power levels ch2_power: Deque[float] # Channel B power levels ch1_updated = False # Flag: Channel A has new data ch2_updated = False # Flag: Channel B has new data last_block = 0 # Last processed block number (for synchronization) last_power_update = 0 # Timestamp of last power plot update (for rate limiting) # AIS message storage ais: List[Dict] = [] # List of received AIS messages ann_list: List = [] # List of plot annotations (for cleanup) # Spectrogram data storage spectrogram_data: Deque # Queue of spectrogram frames spectrogram_updated = False # Flag: New spectrogram data available freq_buckets = 16 # Number of frequency bins in spectrogram freq_range = np.linspace(-96, 96, freq_buckets) # Frequency range: -96 kHz to +96 kHz # ============================================================================= # HELPER FUNCTIONS - Block Number Management # ============================================================================= def is_block_in_range(block: int, first_block: int, last_block: int) -> bool: """Check if block is within valid range, handling 16-bit wraparound Block numbers are 16-bit values that wrap around from 65535 to 0. This function correctly handles the wraparound case when checking if a block number falls within a range. Args: block: Block number to check first_block: Start of valid range last_block: End of valid range Returns: True if block is in range, False otherwise Example: is_block_in_range(100, 50, 150) -> True (normal case) is_block_in_range(10, 65530, 20) -> True (wraparound case) """ if last_block >= first_block: # Normal case - no wraparound return first_block <= block <= last_block else: # Wraparound case - range crosses 65535->0 boundary return block >= first_block or block <= last_block # ============================================================================= # HELPER FUNCTIONS - AIS Marker Display # ============================================================================= def marker_pos(a: Dict) -> int: """Calculate marker position in the power plot Converts an AIS message's block number to a position in the power plot display (0 to DATA_BACKLOG). Args: a: AIS message dictionary containing 'block' field Returns: Position in the plot (0 to DATA_BACKLOG) """ global last_block, DATA_BACKLOG if a["block"] <= last_block: # Normal case - block is before or at current position return DATA_BACKLOG - (last_block - a["block"]) else: # Wraparound case - block number wrapped around return DATA_BACKLOG - (last_block - a["block"] + 65536) def marker_col(a: Dict) -> str: """Determine marker color based on AIS channel AIS uses two channels (A and B). This function assigns colors to markers based on which channel the message came from. Args: a: AIS message dictionary containing NMEA data Returns: Color string for the marker: - 'tab:blue' for Channel A - 'tab:orange' for Channel B - 'black' if channel cannot be determined """ if a["channel"] == 'A': return "tab:blue" elif a["channel"] == 'B': return "tab:orange" else: return "black" # ============================================================================= # PLOT UPDATE FUNCTIONS - Power Display # ============================================================================= def update_power_plot() -> None: """Update the power level plot with current data This function updates both the power traces (lines) and the AIS message markers (scatter points with annotations). It also handles cleanup of old annotations to prevent memory leaks. """ global line1, line2, ax_power, fig_power, scat_power, ann_list, ais # Update the power level traces for both channels line1.set_ydata(ch1_power) line2.set_ydata(ch2_power) # Recalculate plot limits based on new data ax_power.relim() ax_power.autoscale_view() # Clear old annotations to prevent memory leak for a in ann_list: a.remove() ann_list.clear() # Update AIS markers if we have any messages if len(ais) > 0: # Update scatter plot positions and colors scat_power.set_offsets([(marker_pos(a), a["signalpower"]) for a in ais]) scat_power.set_color([marker_col(a) for a in ais]) # Add MMSI annotations for each AIS message for a in ais: ann_list.append(ax_power.annotate(" %s" % a["mmsi"], xy=(marker_pos(a), a["signalpower"]), textcoords="data")) else: # No AIS messages - move marker off screen scat_power.set_offsets([[-1000, -1000]]) # Flush the drawing events to update the display fig_power.canvas.flush_events() # ============================================================================= # PLOT UPDATE FUNCTIONS - Spectrogram Display # ============================================================================= def calculate_spec_scale(spec_array: np.ndarray, mode: str) -> tuple: """Calculate vmin and vmax for spectrogram based on scaling mode This function determines the color scale limits for the spectrogram display. Different modes provide different tradeoffs between showing detail and maintaining consistent scales. Args: spec_array: Spectrogram data array (frames x frequency bins) mode: Scaling mode - 'auto', 'fixed', 'percentile', or 'adaptive' Returns: Tuple of (vmin, vmax) in dB/Hz Modes explained: - 'fixed': Use predefined SPEC_VMIN and SPEC_VMAX values - 'auto': Scale to full range of all data (min to max) - 'percentile': Use percentiles of all data to ignore outliers - 'adaptive': Use percentiles of recent frames only """ if mode == 'fixed': # Use predefined fixed values - best for comparing sessions return SPEC_VMIN, SPEC_VMAX elif mode == 'auto': # Use full range of all data - shows everything but sensitive to outliers return np.min(spec_array), np.max(spec_array) elif mode == 'percentile': # Use percentile-based scaling to ignore outliers across all data # Good for long recordings with varying conditions vmin = np.percentile(spec_array, SPEC_PERCENTILE_LOW) vmax = np.percentile(spec_array, SPEC_PERCENTILE_HIGH) return vmin, vmax elif mode == 'adaptive': # Use only recent frames for scaling - adapts to current conditions # Best for real-time monitoring with changing conditions recent_frames = min(SPEC_ADAPTIVE_WINDOW, len(spec_array)) recent_data = spec_array[-recent_frames:] vmin = np.percentile(recent_data, SPEC_PERCENTILE_LOW) vmax = np.percentile(recent_data, SPEC_PERCENTILE_HIGH) return vmin, vmax else: # Unknown mode - default to auto logger.warning(f"Unknown scaling mode '{mode}', defaulting to 'auto'") return np.min(spec_array), np.max(spec_array) def update_spectrogram_plot() -> None: """Update the spectrogram plot with current data This function updates the spectrogram display. On first call, it creates the plot elements. On subsequent calls, it updates the image data and scaling more efficiently without recreating the plot. The spectrogram shows frequency (vertical) vs time (horizontal) with color representing power spectral density in dB/Hz. """ global spectrogram_updated, ax_spec, fig_spec, SPEC_SCALE_MODE # Only update if we have enough data (need at least 2 frames) if len(spectrogram_data) > 1: # Convert deque to numpy array for plotting spec_array = np.array(list(spectrogram_data)) # Calculate scaling based on selected mode vmin, vmax = calculate_spec_scale(spec_array, SPEC_SCALE_MODE) if not hasattr(update_spectrogram_plot, 'image_obj'): # FIRST TIME SETUP - Create all plot elements ax_spec.clear() # Create the spectrogram image update_spectrogram_plot.image_obj = ax_spec.imshow( spec_array.T, # Transpose so frequency is vertical aspect='auto', # Automatic aspect ratio origin='lower', # Frequency increases upward extent=[0, len(spectrogram_data), freq_range[0], freq_range[-1]], cmap='viridis', # Color map (blue-green-yellow) interpolation='nearest', # No interpolation between pixels vmin=vmin, vmax=vmax) # Set color scale limits # Add axis labels and title ax_spec.set_xlabel('Time (frames)') ax_spec.set_ylabel('Frequency (kHz)') ax_spec.set_title('Spectrogram (Real-time)') ax_spec.grid(True, alpha=0.3) # Add reference lines for AIS channels # AIS Channel A is at -25 kHz, Channel B at +25 kHz ax_spec.axhline(y=-25, color='red', linestyle='--', alpha=0.7, label='AIS A (-25kHz)') ax_spec.axhline(y=25, color='orange', linestyle='--', alpha=0.7, label='AIS B (+25kHz)') ax_spec.legend(loc='upper right') # Add colorbar to show the power scale update_spectrogram_plot.colorbar = fig_spec.colorbar( update_spectrogram_plot.image_obj, ax=ax_spec, label='Power Spectral Density (dB/Hz)') else: # SUBSEQUENT UPDATES - Just update the data update_spectrogram_plot.image_obj.set_data(spec_array.T) # Update extent in case number of frames changed update_spectrogram_plot.image_obj.set_extent( [0, len(spectrogram_data), freq_range[0], freq_range[-1]]) # Update color scale limits update_spectrogram_plot.image_obj.set_clim(vmin=vmin, vmax=vmax) # Update title with current scale information scale_info = f'{SPEC_SCALE_MODE.capitalize()}' if SPEC_SCALE_MODE != 'fixed': # Show current vmin/vmax for non-fixed modes scale_info += f' [{vmin:.1f} to {vmax:.1f} dB/Hz]' ax_spec.set_title(f'Spectrogram ({len(spectrogram_data)} frames) - {scale_info}') # Flush drawing events to update display fig_spec.canvas.flush_events() # Clear the update flag spectrogram_updated = False # ============================================================================= # DEVICE CONFIGURATION # ============================================================================= def configure_device(ser: serial.Serial, enable_multicore: bool = True, lo_side: str = 'high', fe_attn: str = 'auto') -> None: """Configure the device with optimal settings for spectrogram analysis This function sends a series of configuration commands to the dAISy Catcher device to set it up for data collection. It includes proper delays and response handling to ensure reliable configuration. Args: ser: Serial connection to the device enable_multicore: Enable multicore processing (default: True) lo_side: 'high' or 'low' side LO (default: 'high') Configuration commands sent: co2 - Set output format to dAISy-catcher JSON cd1 = Increase debug level to get block info with AIS messages cm1/cm0 - Enable/disable multicore processing cs1/cs0 - Set LO to high-side/low-side cfa/cf0/cf1 - Configure frontend attenuator as1 - Start spectrogram analysis ap1 - Start signal power statistics """ logger.info("Configuring device...") # Wait for device to fully boot and flush all buffers time.sleep(DEVICE_BOOT_DELAY) ser.reset_input_buffer() ser.reset_output_buffer() time.sleep(BUFFER_FLUSH_DELAY) # Send a dummy newline to flush any spurious data from port opening ser.write(b"\r\n") time.sleep(BUFFER_FLUSH_DELAY) ser.reset_input_buffer() # Discard any response def read_response() -> List[str]: """Helper function to read all response lines until no more data Reads lines until a timeout occurs (no data for 1 second), indicating the device has finished responding. Handles both old format (OK/ERR) and new format (JSON responses without OK/ERR). """ lines = [] while True: line = ser.readline().decode("ascii", errors="ignore").strip() if line: lines.append(line) # Old format: Commands complete when we see OK or ERR if line.startswith("OK") or line.startswith("ERR"): break else: # readline() timed out - no more data coming # This is the normal completion for new firmware that doesn't send OK/ERR break return lines # Set output format to dAISy-catcher JSON (co2 command) logger.info("Setting output format to dAISy-catcher JSON...") ser.write(b"co2\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Enable debug output for block info (cd1 command) logger.info("Enabling debug output for block info...") ser.write(b"cd1\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Enable/disable multicore processing (cm1/cm0 command) if enable_multicore: logger.info("Enabling multicore processing...") ser.write(b"cm1\r\n") else: logger.info("Disabling multicore processing...") ser.write(b"cm0\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Set LO side (cs1 = high-side, cs0 = low-side) if lo_side == 'high': logger.info("Setting LO to high-side...") ser.write(b"cs1\r\n") else: logger.info("Setting LO to low-side...") ser.write(b"cs0\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Configure front-end attenuator (cfa = auto, cf0 = always disabled, cf1 = always enabled) if fe_attn == 'on': logger.info("Enabling front-end attenuator...") ser.write(b"cf1\r\n") elif fe_attn == 'off': logger.info("Disabling front-end attenuator...") ser.write(b"cf0\r\n") else: logger.info("Setting front-end attenuator to automatic...") ser.write(b"cfa\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Start spectrogram analysis (as1 command) logger.info("Starting spectrogram analysis...") ser.write(b"as1\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) # Start signal power statistics (ap1 command) logger.info("Starting signal power statistics...") ser.write(b"ap1\r\n") responses = read_response() for line in responses: logger.info(f" {line}") time.sleep(COMMAND_DELAY) logger.info("Device configured successfully!") # ============================================================================= # MESSAGE PROCESSING FUNCTIONS # ============================================================================= def process_ais_message(j: Dict, sock: Optional[socket.socket], udp_port: Optional[int], msg: str) -> None: """Process an AIS message Stores the AIS message and optionally broadcasts it via UDP. Args: j: Parsed JSON message dictionary sock: UDP socket for broadcasting (or None) udp_port: UDP port number (or None) msg: Original message string for UDP broadcast """ global ais # Add to our list of AIS messages m = dict() if "signalpower" in j: m["signalpower"] = j["signalpower"] elif "rssi" in j: m["signalpower"] = j["rssi"] if "block" in j: m["block"] = j["block"] elif "blk" in j: m["block"] = j["blk"] try: # Decode NMEA sentences to get MMSI parts = [] for n in j["nmea"]: parts.append(n) d = decode(*parts) m["mmsi"] = d.mmsi if ",A," in parts[0]: m["channel"] = 'A' elif ",B," in parts[0]: m["channel"] = 'B' except Exception as e: # Log but don't crash if NMEA decode fails logger.debug(f"Error decoding NMEA for annotation: {e}") ais.append(m) # Broadcast via UDP if enabled if sock and udp_port: try: sock.sendto(msg.encode("ascii"), ('127.0.0.1', udp_port)) except Exception as e: logger.error(f'Failed to UDP broadcast AIS message to port {udp_port}: {e}') def process_stats_message(j: Dict) -> None: """Process a statistics message Extracts power statistics data from the JSON message and updates the power plots. Data comes in as hex-encoded bytes that need to be decoded and scaled. Args: j: Parsed JSON message containing: - offset: Baseline power level - scale: Scale factor for byte values - channel: "A" or "B" - block: Block number for synchronization - data: Hex-encoded power data """ global ch1_power, ch2_power, ch1_updated, ch2_updated, last_block, ais global last_power_update, DATA_BACKLOG # Extract message fields offset = j["offset"] scale = j["scale"] channel = j["channel"] block = j["block"] data = bytearray.fromhex(j["data"]) # Decode and append power values # Formula: power = offset + (byte_value * scale) for d in data: p = offset + (d * scale) if channel == "A": ch1_power.append(p) ch1_updated = True elif channel == "B": ch2_power.append(p) ch2_updated = True # Only update plot when we have data from both channels if ch1_updated and ch2_updated: # Update last block number for marker positioning last_block = block + len(data) - 1 # Clean up old AIS data that's scrolled off the display first_block = (last_block - DATA_BACKLOG) & BLOCK_MASK ais = [a for a in ais if is_block_in_range(a["block"], first_block, last_block)] # Rate limit plot updates to avoid excessive CPU usage current_time = time.time() if current_time - last_power_update > POWER_UPDATE_INTERVAL: update_power_plot() last_power_update = current_time ch1_updated = False ch2_updated = False def process_spectrogram_message(j: Dict) -> None: """Process a spectrogram message Extracts spectrogram data from the JSON message and updates the spectrogram display. Data comes as hex-encoded power spectral density values across frequency bins. Args: j: Parsed JSON message containing: - offset: Baseline power level - scale: Scale factor for byte values - data: Hex-encoded spectral data - frames: Number of frames in this message """ global spectrogram_data, spectrogram_updated # Extract message fields offset = j["offset"] scale = j["scale"] data = j["data"] # Decode hex data to power spectral density levels (dB/Hz) # Formula: power = offset + (byte_value * scale) hex_data = bytearray.fromhex(data) power_levels = [offset + (d * scale) for d in hex_data] # Add to spectrogram data queue (automatically drops old frames due to maxlen) spectrogram_data.append(power_levels) spectrogram_updated = True # Update the display update_spectrogram_plot() # ============================================================================= # MAIN PROGRAM # ============================================================================= def main(): """Main program entry point""" global ch1_power, ch2_power, spectrogram_data, DATA_BACKLOG, SPECTROGRAM_HISTORY global line1, line2, ax_power, fig_power, scat_power, ax_spec, fig_spec global SPEC_SCALE_MODE, SPEC_VMIN, SPEC_VMAX, SPEC_PERCENTILE_LOW, SPEC_PERCENTILE_HIGH global SPEC_ADAPTIVE_WINDOW # ========================================================================= # COMMAND LINE ARGUMENT PARSING # ========================================================================= parser = argparse.ArgumentParser( description='Show dAISy Catcher power levels and spectrogram.', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=''' Examples: %(prog)s /dev/serial0 %(prog)s /dev/serial0 -u 10110 %(prog)s /dev/serial0 --spec-scale adaptive --spec-adaptive-window 100 %(prog)s COM3 --spec-scale fixed --spec-vmin -110 --spec-vmax -50 ''') # Required arguments parser.add_argument('serial_port', help='Serial port for communication. For example COM1 or /dev/serial0') # Optional arguments parser.add_argument('-u', dest='udp_port', type=int, help='Port for UDP broadcast') parser.add_argument('--backlog', type=int, default=1000, help='Number of data points to display (default: 1000)') parser.add_argument('--spec-history', type=int, default=100, help='Number of spectrogram frames to keep (default: 100)') parser.add_argument('--no-multicore', action='store_true', help='Disable multicore processing') parser.add_argument('--lo-side', choices=['high', 'low'], default='high', help='LO side selection (default: high)') parser.add_argument('--fe-attn', choices=['auto', 'on', 'off'], default='auto', help='Front-end attenuator (default: auto)') parser.add_argument('--debug', action='store_true', help='Enable debug logging') # Spectrogram scaling options scale_group = parser.add_argument_group('spectrogram scaling options') scale_group.add_argument('--spec-scale', choices=['auto', 'fixed', 'percentile', 'adaptive'], default=SPEC_SCALE_MODE, help='''Spectrogram scaling mode (default: adaptive) auto: scale to full data range fixed: use fixed vmin/vmax values percentile: scale using percentiles to ignore outliers adaptive: scale using recent frames only (recommended)''') scale_group.add_argument('--spec-vmin', type=float, default=-120, help='Minimum value for fixed scaling mode (default: -120 dB/Hz)') scale_group.add_argument('--spec-vmax', type=float, default=-40, help='Maximum value for fixed scaling mode (default: -40 dB/Hz)') scale_group.add_argument('--spec-percentile-low', type=float, default=5, help='Lower percentile for percentile/adaptive modes (default: 5)') scale_group.add_argument('--spec-percentile-high', type=float, default=95, help='Upper percentile for percentile/adaptive modes (default: 95)') scale_group.add_argument('--spec-adaptive-window', type=int, default=50, help='Number of recent frames for adaptive mode (default: 50)') args = parser.parse_args() # ========================================================================= # LOGGING CONFIGURATION # ========================================================================= if args.debug: logger.setLevel(logging.DEBUG) # ========================================================================= # INITIALIZE GLOBAL VARIABLES # ========================================================================= # Set data storage sizes from command line DATA_BACKLOG = args.backlog SPECTROGRAM_HISTORY = args.spec_history # Set spectrogram scaling parameters from command line SPEC_SCALE_MODE = args.spec_scale SPEC_VMIN = args.spec_vmin SPEC_VMAX = args.spec_vmax SPEC_PERCENTILE_LOW = args.spec_percentile_low SPEC_PERCENTILE_HIGH = args.spec_percentile_high SPEC_ADAPTIVE_WINDOW = args.spec_adaptive_window # Create power data deques with automatic size limiting ch1_power = deque([INITIAL_POWER_VALUE] * DATA_BACKLOG, maxlen=DATA_BACKLOG) ch2_power = deque([INITIAL_POWER_VALUE] * DATA_BACKLOG, maxlen=DATA_BACKLOG) spectrogram_data = deque(maxlen=SPECTROGRAM_HISTORY) # Log scaling configuration logger.info(f"Spectrogram scaling mode: {SPEC_SCALE_MODE}") if SPEC_SCALE_MODE == 'fixed': logger.info(f" Fixed range: {SPEC_VMIN} to {SPEC_VMAX} dB/Hz") elif SPEC_SCALE_MODE in ['percentile', 'adaptive']: logger.info(f" Percentiles: {SPEC_PERCENTILE_LOW}% to {SPEC_PERCENTILE_HIGH}%") if SPEC_SCALE_MODE == 'adaptive': logger.info(f" Adaptive window: {SPEC_ADAPTIVE_WINDOW} frames") # ========================================================================= # UDP SOCKET SETUP # ========================================================================= sock = None if args.udp_port is not None: try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) logger.info(f"UDP broadcast enabled on port {args.udp_port}") except Exception as e: logger.error(f'Failed to create socket for UDP broadcast: {e}') return 1 # ========================================================================= # MAIN EXECUTION # ========================================================================= try: with serial.Serial(args.serial_port, SERIAL_BAUD_RATE, timeout=SERIAL_TIMEOUT) as ser: # Configure the device configure_device(ser, enable_multicore=not args.no_multicore, lo_side=args.lo_side, fe_attn=args.fe_attn) # ================================================================ # MATPLOTLIB SETUP # ================================================================ plt.ion() # Enable interactive mode for real-time updates # Figure 1: Power levels plot fig_power = plt.figure(figsize=(POWER_PLOT_WIDTH, POWER_PLOT_HEIGHT)) ax_power = fig_power.add_subplot(111) # Create power level traces line1, = ax_power.plot(range(len(ch1_power)), ch1_power, label="Channel A") line2, = ax_power.plot(range(len(ch2_power)), ch2_power, label="Channel B") # Create scatter plot for AIS message markers scat_power = ax_power.scatter([], [], color="black", marker="<") # Configure power plot ax_power.set_xlabel("Blocks") ax_power.set_ylabel("Power (dBm)") ax_power.set_title("AIS Channel Power Levels") ax_power.legend(loc="upper left") ax_power.set_xticks(range(0, DATA_BACKLOG + 50, 50)) ax_power.grid() # Figure 2: Spectrogram plot fig_spec = plt.figure(figsize=(SPEC_PLOT_WIDTH, SPEC_PLOT_HEIGHT)) ax_spec = fig_spec.add_subplot(111) ax_spec.set_xlabel('Time (frames)') ax_spec.set_ylabel('Frequency (kHz)') ax_spec.set_title('Spectrogram - Waiting for data...') ax_spec.grid(True, alpha=0.3) plt.show() # ================================================================ # MAIN DATA PROCESSING LOOP # ================================================================ logger.info("Listening for data... (Press Ctrl+C to stop)") while exit_sig.empty(): try: # Read a line from the serial port msg = ser.readline().decode("ascii", errors="ignore") if msg.strip(): # Process JSON messages if msg.startswith("{"): try: j = json.loads(msg) # Route message to appropriate handler based on class if j["class"] == "AIS": # Echo received data to console print(msg, end='') process_ais_message(j, sock, args.udp_port, msg) elif j["class"] == "stats": process_stats_message(j) elif j["class"] == "spectrogram": process_spectrogram_message(j) else: # Echo received data to console print(msg, end='') except (ValueError, KeyError) as e: logger.error(f"JSON decoding error: {e} - Message: {msg.strip()}") except Exception as e: logger.exception(f"Unexpected error processing message: {e}") else: # Echo received data to console print(msg, end='') except Exception as e: # Error detected - log and continue logger.exception(f"Error in main loop: {e}") continue logger.info("\nStopping..."); # Send stop commands to AIS receiver try: ser.write(b"as0\r\n") # Stop spectrogram (as0 command) time.sleep(COMMAND_DELAY) ser.write(b"ap0\r\n") # Stop power stats (ap0 command) time.sleep(COMMAND_DELAY) except Exception as e: logger.error(f"Error during cleanup: {e}") # Close UDP socket if it was opened if sock: try: sock.close() except Exception as e: logger.error(f"Error closing socket: {e}") except serial.SerialException as e: logger.error(f"Serial port error: {e}") return 1 except Exception as e: logger.exception(f"Fatal error: {e}") return 1 return 0 # ============================================================================= # PROGRAM ENTRY POINT # ============================================================================= if __name__ == "__main__": exit(main())