mirror of
https://github.com/gnss-sdr/gnss-sdr
synced 2026-10-11 18:21:51 +00:00
While testing, I found the old version of this script had an old bug.
For some reason, the old script calculated "mean()" of PRN numbers in a tracking channel. E.g., channel 1 tracked PRN 3 for few seconds, then tracked PRN 16 for the rest. The script would add legend PRN 9 (3+16)/2 = .... Fixed that and added "PRN-wise tracking indicator" in addition to the existing "channel-wise tracking indicator" Signed-off-by: minhaj <minhaj.sixbyte@gmail.com>
This commit is contained in:
minhaj
@@ -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 <prefix><channel>.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.
|
||||
|
||||
Reference in New Issue
Block a user