diff --git a/utils/python/plot_tracking_quality_indicators.py b/utils/python/plot_tracking_quality_indicators.py index 994937b38..8bdfb3593 100755 --- a/utils/python/plot_tracking_quality_indicators.py +++ b/utils/python/plot_tracking_quality_indicators.py @@ -3,7 +3,13 @@ plot_tracking_quality_indicators.py Reads GNSS-SDR tracking dump binary files and plots C/N0 plus carrier-lock - quality indicators across channels. + quality indicators against elapsed time. + + Two layouts are available (see --style): + per-satellite : one subplot per PRN, merged across channels (default). A + satellite that is handed between channels (lost lock and + re-acquired) appears as a single continuous trace. + per-channel : one line per channel on a single axes. File format: {input_path}/{file_prefix}{channel}.dat @@ -20,6 +26,7 @@ """ import argparse +import math from pathlib import Path import matplotlib.pyplot as plt @@ -55,6 +62,12 @@ def parse_args(): "include a directory and extension; the matching .dat " "files are read, resolved against --input-path.", ) + parser.add_argument( + "--sampling-frequency", + type=float, + default=4000000.0, + help="Signal sampling frequency in Hz.", + ) parser.add_argument( "--channels", type=int, @@ -76,6 +89,20 @@ def parse_args(): "C/N0 legend label. The tracking dump stores only the PRN, so the " "signal is supplied here (default: 1C).", ) + parser.add_argument( + "--style", + choices=("per-satellite", "per-channel"), + default="per-satellite", + help="per-satellite: one subplot per PRN, merged across channels " + "(default). per-channel: one line per channel on a single axes.", + ) + parser.add_argument( + "--subplot-cols", + type=int, + default=3, + help="Columns in the per-satellite subplot grid; rows grow to fit " + "the number of satellites (default: 3).", + ) parser.add_argument( "--show", action="store_true", @@ -94,8 +121,88 @@ def read_tracking_dumps(args): return dumps -def mean_prn_label(tracking): - return str(round(np.mean(tracking["PRN"]))) +def prn_label(tracking): + # Distinct PRNs the channel tracked, in order of appearance. A channel that + # is reassigned mid-run (lost lock and re-acquired) tracks more than one + # satellite; + prns = dict.fromkeys(int(p) for p in tracking["PRN"] if int(p)) + return "/".join(str(p) for p in prns) + + +def time_seconds(tracking, sampling_frequency): + return np.asarray(tracking["PRN_start_sample"], dtype=float) / sampling_frequency + + +def series_by_prn(tracking_list, key, sampling_frequency): + # Merge each PRN's samples across all channels, sorted by time, so a + # satellite handed between channels becomes one continuous series. + # Assumes no two channels track the same PRN simultaneously + parts = {} + for tracking in tracking_list: + time_s = time_seconds(tracking, sampling_frequency) + prn = np.asarray(tracking["PRN"]) + values = np.asarray(tracking[key], dtype=float) + for p in np.unique(prn): + if not p: + continue + sel = prn == p + parts.setdefault(int(p), []).append((time_s[sel], values[sel])) + series = {} + for p, chunks in parts.items(): + time_s = np.concatenate([c[0] for c in chunks]) + values = np.concatenate([c[1] for c in chunks]) + order = np.argsort(time_s, kind="stable") + series[p] = (time_s[order], values[order]) + return series + + +def plot_per_satellite(tracking_list, spec, args): + series = series_by_prn(tracking_list, spec["key"], args.sampling_frequency) + prns = sorted(series) + ncols = max(1, args.subplot_cols) + nrows = max(1, math.ceil(len(prns) / ncols)) + fig, axes = plt.subplots( + nrows, + ncols, + figsize=(4.0 * ncols, 2.3 * nrows), + sharex=True, + squeeze=False, + ) + fig.canvas.manager.set_window_title(f"{spec['ylabel']} per satellite") + for ax in axes.flat: # hide unused cells (e.g. 10 sats in a 4x3 grid) + ax.set_visible(False) + for ax, p in zip(axes.flat, prns): + ax.set_visible(True) + time_s, values = series[p] + ax.plot(time_s, values, linewidth=0.8) + ax.set_title(f"PRN {p}", fontsize=9) + ax.grid(alpha=0.3) + fig.suptitle(f"{spec['ylabel']} per satellite") + fig.supxlabel("Time(s)") + fig.supylabel(spec["ylabel"]) + fig.tight_layout() + fig.savefig(args.fig_path / f"{spec['stem']}.{args.output_format}") + if not args.show: + plt.close(fig) + + +def plot_per_channel(tracking_list, spec, args): + fig = plt.figure() + fig.canvas.manager.set_window_title(f"{spec['ylabel']} per channel") + plt.title(f"{spec['ylabel']} per channel") + for tracking in tracking_list: + time_s = time_seconds(tracking, args.sampling_frequency) + plt.plot( + time_s, + tracking[spec["key"]], + label=f"{spec['legend_prefix']} {prn_label(tracking)}", + ) + plt.xlabel("Time(s)") + plt.ylabel(spec["ylabel"]) + plt.legend() + plt.savefig(args.fig_path / f"{spec['stem']}.{args.output_format}") + if not args.show: + plt.close(fig) def main(): @@ -105,31 +212,25 @@ def main(): gnss_tracking = read_tracking_dumps(args) - plt.figure() - plt.gcf().canvas.manager.set_window_title( - "Carrier lock test output for all the channels" - ) - plt.title("Carrier lock test output for all the channels") - carrier_lock_names = [] - for tracking in gnss_tracking: - plt.plot(tracking["carrier_lock_test"]) - carrier_lock_names.append(f"SV {mean_prn_label(tracking)}") - plt.legend(carrier_lock_names) - plt.savefig(args.fig_path / f"carrier_lock_test.{args.output_format}") - if not args.show: - plt.close() - - plt.figure() - plt.gcf().canvas.manager.set_window_title("Carrier CN0 output for all channels") - plt.title("Carrier CN0 output for all channels") - cn0_names = [] - for tracking in gnss_tracking: - plt.plot(tracking["CN0_SNV_dB_Hz"]) - cn0_names.append(f"{args.signal_type} PRN {mean_prn_label(tracking)}") - plt.legend(cn0_names) - plt.savefig(args.fig_path / f"CN0_SNV_dB_Hz.{args.output_format}") - if not args.show: - plt.close() + specs = [ + { + "key": "carrier_lock_test", + "ylabel": "Carrier lock test", + "stem": "carrier_lock_test", + "legend_prefix": "SV", + }, + { + "key": "CN0_SNV_dB_Hz", + "ylabel": "C/N0 (dB-Hz)", + "stem": "CN0_SNV_dB_Hz", + "legend_prefix": f"{args.signal_type} PRN", + }, + ] + for spec in specs: + if args.style == "per-channel": + plot_per_channel(gnss_tracking, spec, args) + else: + plot_per_satellite(gnss_tracking, spec, args) # Show all saved figures with a single plt.show() to avoid the repeated # show()/close() cycle that can crash interactive backends on macOS.