fly316/PX4-Autopilot
7
0
Fork 0
mirror of https://gitee.com/mirrors_PX4/PX4-Autopilot.git synced 2026-10-11 07:13:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

code simplifier

Browse Source
This commit is contained in:
Jacob Dahl
2026-01-27 15:39:49 -09:00
parent 2d7b3a83c3
commit 597611e7a5
1 changed files with 115 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tools/bdshot_analyzer.py
+115 -292
View File
@@ -39,7 +39,6 @@ import csv
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional, Tuple
import numpy as np
@@ -106,33 +105,25 @@ class GapStats:
std_us: float
def parse_csv(filepath: Path, channel_a_idx: int, channel_b_idx: int,
verbose: bool = False) -> Tuple[List[Tuple[float, int]], List[Tuple[float, int]]]:
def parse_csv(filepath: Path, channel_a_idx: int, channel_b_idx: int, verbose: bool = False):
"""
Parse Saleae CSV export and extract edges for each channel.
Returns:
Tuple of (edges_a, edges_b) where each is a list of (timestamp, new_state) tuples.
"""
edges_a = []
edges_b = []
prev_state_a = None
prev_state_b = None
edges_a, edges_b = [], []
prev_a, prev_b = None, None
ch_a_col, ch_b_col = channel_a_idx + 1, channel_b_idx + 1 # +1 for time column
with open(filepath, 'r', newline='') as f:
reader = csv.reader(f)
# Read header
header = next(reader)
if verbose:
print(f"CSV Header: {header}")
# Determine column indices
# Header format: "Time [s],Channel0,Channel1" or "Time [s],CH1,CH5" etc.
time_col = 0
ch_a_col = channel_a_idx + 1 # +1 because time is column 0
ch_b_col = channel_b_idx + 1
print(f"Using columns: time=0, ch_a={ch_a_col} ({header[ch_a_col]}), "
f"ch_b={ch_b_col} ({header[ch_b_col]})")
if ch_a_col >= len(header) or ch_b_col >= len(header):
raise ValueError(
@@ -140,37 +131,22 @@ def parse_csv(filepath: Path, channel_a_idx: int, channel_b_idx: int,
f"Requested channel_a={channel_a_idx}, channel_b={channel_b_idx}"
)
if verbose:
print(f"Using columns: time={time_col}, ch_a={ch_a_col} ({header[ch_a_col]}), "
f"ch_b={ch_b_col} ({header[ch_b_col]})")
for row_num, row in enumerate(reader, start=2):
try:
timestamp = float(row[time_col])
state_a = int(row[ch_a_col])
state_b = int(row[ch_b_col])
timestamp = float(row[0])
state_a, state_b = int(row[ch_a_col]), int(row[ch_b_col])
# Record edges (state changes)
if prev_state_a is not None and state_a != prev_state_a:
# Record edges (state changes) or initial LOW state
if (prev_a is not None and state_a != prev_a) or (prev_a is None and state_a == 0):
edges_a.append((timestamp, state_a))
elif prev_state_a is None:
# First row - record initial state as an edge if LOW
if state_a == 0:
edges_a.append((timestamp, state_a))
if prev_state_b is not None and state_b != prev_state_b:
if (prev_b is not None and state_b != prev_b) or (prev_b is None and state_b == 0):
edges_b.append((timestamp, state_b))
elif prev_state_b is None:
if state_b == 0:
edges_b.append((timestamp, state_b))
prev_state_a = state_a
prev_state_b = state_b
prev_a, prev_b = state_a, state_b
except (ValueError, IndexError) as e:
if verbose:
print(f"Warning: Skipping malformed row {row_num}: {e}")
continue
if verbose:
print(f"Parsed {len(edges_a)} edges for channel A, {len(edges_b)} edges for channel B")
@@ -178,21 +154,12 @@ def parse_csv(filepath: Path, channel_a_idx: int, channel_b_idx: int,
return edges_a, edges_b
def detect_transactions(edges: List[Tuple[float, int]], idle_threshold_us: float,
verbose: bool = False) -> List[BDShotTransaction]:
def detect_transactions(edges, idle_threshold_us: float, verbose: bool = False):
"""
Detect BDShot transactions from edge list.
A transaction starts with a falling edge after an idle period and ends
when the line returns HIGH and stays HIGH for > idle_threshold_us.
Args:
edges: List of (timestamp, new_state) tuples
idle_threshold_us: Minimum idle time in microseconds to consider transaction complete
verbose: Print debug information
Returns:
List of BDShotTransaction objects
"""
if not edges:
return []
@@ -200,31 +167,21 @@ def detect_transactions(edges: List[Tuple[float, int]], idle_threshold_us: float
idle_threshold_s = idle_threshold_us / 1_000_000
transactions = []
tx_start = None
last_rising_edge = None
last_rising = None
for i, (timestamp, state) in enumerate(edges):
if state == 0: # Falling edge
if tx_start is None:
# Check if this is after an idle period
if last_rising_edge is None or (timestamp - last_rising_edge) > idle_threshold_s:
tx_start = timestamp
else: # Rising edge (state == 1)
last_rising_edge = timestamp
# Check if next edge is far enough away to consider this transaction complete
if tx_start is not None:
# Look ahead to see if there's a gap
if i + 1 < len(edges):
next_timestamp = edges[i + 1][0]
gap = next_timestamp - timestamp
if gap > idle_threshold_s:
# Transaction complete
transactions.append(BDShotTransaction(start=tx_start, end=timestamp))
tx_start = None
else:
# Last edge in file - complete the transaction
transactions.append(BDShotTransaction(start=tx_start, end=timestamp))
tx_start = None
if state == 0: # Falling edge - start transaction if idle
if tx_start is None and (last_rising is None or timestamp - last_rising > idle_threshold_s):
tx_start = timestamp
elif tx_start is not None: # Rising edge with active transaction
last_rising = timestamp
# Complete transaction if next edge is far away or this is the last edge
next_gap = edges[i + 1][0] - timestamp if i + 1 < len(edges) else float('inf')
if next_gap > idle_threshold_s:
transactions.append(BDShotTransaction(start=tx_start, end=timestamp))
tx_start = None
else:
last_rising = timestamp
if verbose:
print(f"Detected {len(transactions)} transactions")
@@ -235,21 +192,12 @@ def detect_transactions(edges: List[Tuple[float, int]], idle_threshold_us: float
return transactions
def compute_channel_stats(transactions: List[BDShotTransaction], channel_name: str,
expected_interval_us: float) -> Tuple[ChannelStats, np.ndarray]:
"""
Compute frame interval statistics for a channel.
Returns:
Tuple of (ChannelStats, intervals_array_us)
"""
def compute_channel_stats(transactions, channel_name: str, expected_interval_us: float):
"""Compute frame interval statistics for a channel."""
if len(transactions) < 2:
raise ValueError(f"Need at least 2 transactions to compute intervals, got {len(transactions)}")
# Compute intervals between transaction starts
starts = np.array([tx.start for tx in transactions])
intervals_s = np.diff(starts)
intervals_us = intervals_s * 1_000_000
intervals_us = np.diff([tx.start for tx in transactions]) * 1_000_000
stats = ChannelStats(
channel_name=channel_name,
@@ -264,57 +212,31 @@ def compute_channel_stats(transactions: List[BDShotTransaction], channel_name: s
return stats, intervals_us
def compute_inter_channel_gaps(transactions_a: List[BDShotTransaction],
transactions_b: List[BDShotTransaction],
expected_interval_us: float,
verbose: bool = False) -> Tuple[GapStats, np.ndarray]:
"""
Compute inter-channel gap statistics (CH_A end -> CH_B start).
Args:
transactions_a: Channel A transactions
transactions_b: Channel B transactions
expected_interval_us: Expected loop interval (used to filter out wrong matches)
verbose: Print debug information
Returns:
Tuple of (GapStats, gaps_array_us)
"""
def compute_inter_channel_gaps(transactions_a, transactions_b, expected_interval_us: float,
verbose: bool = False):
"""Compute inter-channel gap statistics (CH_A end -> CH_B start)."""
gaps = []
# Maximum reasonable gap - should be much less than loop period
max_reasonable_gap_us = expected_interval_us * 0.5
b_idx = 0
matched = 0
skipped = 0
for tx_a in transactions_a:
# Find first CH_B transaction that starts after this CH_A ends
while b_idx < len(transactions_b) and transactions_b[b_idx].start <= tx_a.end:
b_idx += 1
if b_idx >= len(transactions_b):
break
tx_b = transactions_b[b_idx]
gap_us = (tx_b.start - tx_a.end) * 1_000_000
# Sanity check: gap should be positive and less than half the loop period
gap_us = (transactions_b[b_idx].start - tx_a.end) * 1_000_000
if 0 < gap_us < max_reasonable_gap_us:
gaps.append(gap_us)
matched += 1
else:
skipped += 1
if verbose:
print(f"Inter-channel gap matching: {matched} matched, {skipped} skipped")
print(f"Inter-channel gap matching: {len(gaps)} matched")
if not gaps:
raise ValueError("No valid inter-channel gaps found")
gaps_arr = np.array(gaps)
stats = GapStats(
count=len(gaps_arr),
min_us=float(np.min(gaps_arr)),
@@ -326,106 +248,56 @@ def compute_inter_channel_gaps(transactions_a: List[BDShotTransaction],
return stats, gaps_arr
def _plot_histogram(ax, data: np.ndarray, title: str, xlabel: str,
expected_value: Optional[float] = None,
stats_text: Optional[str] = None):
def _plot_histogram(ax, data, title: str, xlabel: str, expected_value=None, stats_text=None):
"""Plot a histogram on the given axes."""
# Determine bin count based on data range
data_range = np.max(data) - np.min(data)
if data_range > 0:
bin_count = min(100, max(20, int(len(data) / 100)))
else:
bin_count = 20
bin_count = min(100, max(20, len(data) // 100)) if np.ptp(data) > 0 else 20
ax.hist(data, bins=bin_count, edgecolor='black', alpha=0.7)
ax.set_title(title, fontsize=12, fontweight='bold')
ax.set_xlabel(xlabel, fontsize=10)
ax.set_ylabel('Count', fontsize=10)
# Add vertical line for expected/mean value
if expected_value is not None:
ax.axvline(expected_value, color='red', linestyle='--', linewidth=2,
label=f'Expected: {expected_value:.2f} us')
ax.axvline(np.mean(data), color='green', linestyle='-', linewidth=2,
label=f'Mean: {np.mean(data):.2f} us')
mean_val = np.mean(data)
ax.axvline(mean_val, color='green', linestyle='-', linewidth=2,
label=f'Mean: {mean_val:.2f} us')
# Add statistics text box
if stats_text:
props = dict(boxstyle='round', facecolor='wheat', alpha=0.8)
ax.text(0.98, 0.97, stats_text, transform=ax.transAxes, fontsize=9,
verticalalignment='top', horizontalalignment='right', bbox=props,
family='monospace')
verticalalignment='top', horizontalalignment='right',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8), family='monospace')
ax.legend(loc='upper left', fontsize=8)
ax.grid(True, alpha=0.3)
def generate_combined_histogram(
intervals_a: np.ndarray, intervals_b: np.ndarray, gaps: np.ndarray,
stats_a: 'ChannelStats', stats_b: 'ChannelStats', gap_stats: 'GapStats',
expected_interval_us: float, dshot_rate: int
):
def generate_combined_histogram(intervals_a, intervals_b, gaps, stats_a, stats_b,
gap_stats, expected_interval_us: float, dshot_rate: int):
"""Generate a combined figure with all histograms."""
if not HAS_MATPLOTLIB:
return None
# Create figure with GridSpec: 2 rows, 2 columns
# Top row: Channel A (left), Channel B (right)
# Bottom row: Inter-channel gap (spans both columns)
fig = plt.figure(figsize=(14, 10))
gs = fig.add_gridspec(2, 2, height_ratios=[1, 1], hspace=0.3, wspace=0.25)
ax_a, ax_b, ax_gap = fig.add_subplot(gs[0, 0]), fig.add_subplot(gs[0, 1]), fig.add_subplot(gs[1, :])
ax_a = fig.add_subplot(gs[0, 0])
ax_b = fig.add_subplot(gs[0, 1])
ax_gap = fig.add_subplot(gs[1, :]) # Span both columns
def stats_text(s):
return format_stats_text(s.min_us, s.max_us, s.mean_us, s.std_us, s.count)
# Channel A intervals
_plot_histogram(
ax_a, intervals_a,
f"Channel A Frame Intervals",
"Interval (us)",
expected_value=expected_interval_us,
stats_text=format_stats_text(
stats_a.min_us, stats_a.max_us, stats_a.mean_us,
stats_a.std_us, stats_a.count
)
)
_plot_histogram(ax_a, intervals_a, "Channel A Frame Intervals", "Interval (us)",
expected_value=expected_interval_us, stats_text=stats_text(stats_a))
_plot_histogram(ax_b, intervals_b, "Channel B Frame Intervals", "Interval (us)",
expected_value=expected_interval_us, stats_text=stats_text(stats_b))
# Channel B intervals
_plot_histogram(
ax_b, intervals_b,
f"Channel B Frame Intervals",
"Interval (us)",
expected_value=expected_interval_us,
stats_text=format_stats_text(
stats_b.min_us, stats_b.max_us, stats_b.mean_us,
stats_b.std_us, stats_b.count
)
)
# Inter-channel gaps
if len(gaps) > 0:
_plot_histogram(
ax_gap, gaps,
"Inter-Channel Gap (CH_A End -> CH_B Start)",
"Gap (us)",
stats_text=format_stats_text(
gap_stats.min_us, gap_stats.max_us, gap_stats.mean_us,
gap_stats.std_us, gap_stats.count
)
)
_plot_histogram(ax_gap, gaps, "Inter-Channel Gap (CH_A End -> CH_B Start)",
"Gap (us)", stats_text=stats_text(gap_stats))
else:
ax_gap.text(0.5, 0.5, "No inter-channel gap data available",
ha='center', va='center', transform=ax_gap.transAxes)
ax_gap.set_title("Inter-Channel Gap (CH_A End -> CH_B Start)")
# Add overall title
fig.suptitle(f"BDShot Timing Analysis (DShot{dshot_rate})",
fontsize=14, fontweight='bold', y=0.98)
fig.suptitle(f"BDShot Timing Analysis (DShot{dshot_rate})", fontsize=14, fontweight='bold', y=0.98)
return fig
@@ -441,168 +313,119 @@ def format_stats_text(min_val: float, max_val: float, mean_val: float,
def print_report(input_file: Path, dshot_rate: int, loop_rate: float,
capture_duration: float, tx_count_a: int, tx_count_b: int,
stats_a: ChannelStats, stats_b: ChannelStats,
gap_stats: GapStats):
stats_a: ChannelStats, stats_b: ChannelStats, gap_stats: GapStats):
"""Print the analysis report to console."""
bit_period = DSHOT_BIT_PERIODS.get(dshot_rate, 3.33)
expected_interval = 1_000_000 / loop_rate
print()
print("BDShot Timing Analysis")
print("=" * 60)
print(f"Input: {input_file}")
print(f"DShot Rate: {dshot_rate} (bit period: {bit_period:.2f} us)")
print(f"Expected Loop Rate: {loop_rate:.0f} Hz ({expected_interval:.2f} us interval)")
print(f"Capture Duration: {capture_duration:.2f} s")
print(f"Transactions Detected: CH_A={tx_count_a}, CH_B={tx_count_b}")
print()
print(f"""
BDShot Timing Analysis
{"=" * 60}
Input: {input_file}
DShot Rate: {dshot_rate} (bit period: {bit_period:.2f} us)
Expected Loop Rate: {loop_rate:.0f} Hz ({expected_interval:.2f} us interval)
Capture Duration: {capture_duration:.2f} s
Transactions Detected: CH_A={tx_count_a}, CH_B={tx_count_b}
""")
# Channel A stats
print(f"Channel A Frame Intervals")
print("-" * 40)
print(f" Count: {stats_a.count}")
print(f" Min: {stats_a.min_us:.2f} us ({stats_a.min_jitter_us:+.2f} us from nominal)")
print(f" Max: {stats_a.max_us:.2f} us ({stats_a.max_jitter_us:+.2f} us from nominal)")
print(f" Mean: {stats_a.mean_us:.2f} us ({stats_a.mean_jitter_us:+.2f} us from nominal)")
print(f" Std Dev: {stats_a.std_us:.2f} us")
print()
def print_channel_stats(name: str, s: ChannelStats):
print(f"{name} Frame Intervals")
print("-" * 40)
print(f" Count: {s.count}")
print(f" Min: {s.min_us:.2f} us ({s.min_jitter_us:+.2f} us from nominal)")
print(f" Max: {s.max_us:.2f} us ({s.max_jitter_us:+.2f} us from nominal)")
print(f" Mean: {s.mean_us:.2f} us ({s.mean_jitter_us:+.2f} us from nominal)")
print(f" Std Dev: {s.std_us:.2f} us\n")
# Channel B stats
print(f"Channel B Frame Intervals")
print("-" * 40)
print(f" Count: {stats_b.count}")
print(f" Min: {stats_b.min_us:.2f} us ({stats_b.min_jitter_us:+.2f} us from nominal)")
print(f" Max: {stats_b.max_us:.2f} us ({stats_b.max_jitter_us:+.2f} us from nominal)")
print(f" Mean: {stats_b.mean_us:.2f} us ({stats_b.mean_jitter_us:+.2f} us from nominal)")
print(f" Std Dev: {stats_b.std_us:.2f} us")
print()
print_channel_stats("Channel A", stats_a)
print_channel_stats("Channel B", stats_b)
# Inter-channel gap stats
print(f"Inter-Channel Gap (CH_A End -> CH_B Start)")
print("-" * 40)
print(f" Count: {gap_stats.count}")
print(f" Min: {gap_stats.min_us:.2f} us")
print(f" Max: {gap_stats.max_us:.2f} us")
print(f" Mean: {gap_stats.mean_us:.2f} us")
print(f" Std Dev: {gap_stats.std_us:.2f} us")
print()
print(f"""Inter-Channel Gap (CH_A End -> CH_B Start)
{"-" * 40}
Count: {gap_stats.count}
Min: {gap_stats.min_us:.2f} us
Max: {gap_stats.max_us:.2f} us
Mean: {gap_stats.mean_us:.2f} us
Std Dev: {gap_stats.std_us:.2f} us
""")
def fatal(msg: str):
"""Print error and exit."""
print(f"Error: {msg}", file=sys.stderr)
sys.exit(1)
def main():
parser = argparse.ArgumentParser(
description="Analyze BDShot timing from Saleae Logic Analyzer CSV exports",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s capture.csv
%(prog)s capture.csv --dshot-rate 300 --loop-rate 800
%(prog)s capture.csv --channel-a 0 --channel-b 1
"""
epilog="Examples:\n %(prog)s capture.csv\n %(prog)s capture.csv --dshot-rate 300 --loop-rate 800"
)
parser.add_argument('input_file', type=Path,
help='Path to Saleae CSV export')
parser.add_argument('--dshot-rate', type=int, default=300,
choices=[150, 300, 600, 1200],
parser.add_argument('input_file', type=Path, help='Path to Saleae CSV export')
parser.add_argument('--dshot-rate', type=int, default=300, choices=[150, 300, 600, 1200],
help='DShot variant (default: 300)')
parser.add_argument('--loop-rate', type=float, default=800,
help='Expected loop rate in Hz (default: 800)')
parser.add_argument('--channel-a', type=int, default=0,
help='CSV column index for first channel (default: 0)')
parser.add_argument('--channel-b', type=int, default=1,
help='CSV column index for second channel (default: 1)')
parser.add_argument('--no-plots', action='store_true',
help='Disable histogram display')
parser.add_argument('--verbose', '-v', action='store_true',
help='Enable verbose output for debugging')
parser.add_argument('--loop-rate', type=float, default=800, help='Expected loop rate in Hz (default: 800)')
parser.add_argument('--channel-a', type=int, default=0, help='CSV column index for first channel (default: 0)')
parser.add_argument('--channel-b', type=int, default=1, help='CSV column index for second channel (default: 1)')
parser.add_argument('--no-plots', action='store_true', help='Disable histogram display')
parser.add_argument('--verbose', '-v', action='store_true', help='Enable verbose output for debugging')
args = parser.parse_args()
# Validate input file
if not args.input_file.exists():
print(f"Error: Input file not found: {args.input_file}", file=sys.stderr)
sys.exit(1)
fatal(f"Input file not found: {args.input_file}")
# Calculate timing parameters
bit_period_us = DSHOT_BIT_PERIODS.get(args.dshot_rate, 3.33)
# Idle threshold must be longer than the turnaround gap (~25-30us) between
# command and response frames, but shorter than the inter-transaction gap.
# Use 50us minimum to handle all DShot rates safely.
idle_threshold_us = max(50.0, bit_period_us * 15)
idle_threshold_us = max(50.0, bit_period_us * 15) # Must exceed turnaround gap (~25-30us)
expected_interval_us = 1_000_000 / args.loop_rate
if args.verbose:
print(f"Bit period: {bit_period_us:.2f} us")
print(f"Idle threshold: {idle_threshold_us:.2f} us")
print(f"Expected interval: {expected_interval_us:.2f} us")
print()
print(f"Bit period: {bit_period_us:.2f} us\nIdle threshold: {idle_threshold_us:.2f} us\n"
f"Expected interval: {expected_interval_us:.2f} us\n")
# Parse CSV
print(f"Parsing {args.input_file}...")
try:
edges_a, edges_b = parse_csv(
args.input_file, args.channel_a, args.channel_b, args.verbose
)
edges_a, edges_b = parse_csv(args.input_file, args.channel_a, args.channel_b, args.verbose)
except Exception as e:
print(f"Error parsing CSV: {e}", file=sys.stderr)
sys.exit(1)
fatal(f"parsing CSV: {e}")
if not edges_a or not edges_b:
print("Error: No edges found in one or both channels", file=sys.stderr)
sys.exit(1)
fatal("No edges found in one or both channels")
# Detect transactions
print("Detecting transactions...")
transactions_a = detect_transactions(edges_a, idle_threshold_us, args.verbose)
transactions_b = detect_transactions(edges_b, idle_threshold_us, args.verbose)
if len(transactions_a) < 2:
print(f"Error: Insufficient transactions in channel A ({len(transactions_a)})",
file=sys.stderr)
sys.exit(1)
if len(transactions_b) < 2:
print(f"Error: Insufficient transactions in channel B ({len(transactions_b)})",
file=sys.stderr)
sys.exit(1)
for name, txns in [("A", transactions_a), ("B", transactions_b)]:
if len(txns) < 2:
fatal(f"Insufficient transactions in channel {name} ({len(txns)})")
# Calculate capture duration
all_starts = [tx.start for tx in transactions_a] + [tx.start for tx in transactions_b]
all_ends = [tx.end for tx in transactions_a] + [tx.end for tx in transactions_b]
capture_duration = max(all_ends) - min(all_starts)
all_txns = transactions_a + transactions_b
capture_duration = max(tx.end for tx in all_txns) - min(tx.start for tx in all_txns)
# Compute statistics
print("Computing statistics...")
stats_a, intervals_a = compute_channel_stats(
transactions_a, "Channel A", expected_interval_us
)
stats_b, intervals_b = compute_channel_stats(
transactions_b, "Channel B", expected_interval_us
)
stats_a, intervals_a = compute_channel_stats(transactions_a, "Channel A", expected_interval_us)
stats_b, intervals_b = compute_channel_stats(transactions_b, "Channel B", expected_interval_us)
try:
gap_stats, gaps = compute_inter_channel_gaps(
transactions_a, transactions_b, expected_interval_us, args.verbose
)
gap_stats, gaps = compute_inter_channel_gaps(transactions_a, transactions_b,
expected_interval_us, args.verbose)
except ValueError as e:
print(f"Warning: {e}", file=sys.stderr)
gap_stats = GapStats(count=0, min_us=0, max_us=0, mean_us=0, std_us=0)
gaps = np.array([])
# Print report
print_report(
args.input_file, args.dshot_rate, args.loop_rate,
capture_duration, len(transactions_a), len(transactions_b),
stats_a, stats_b, gap_stats
)
print_report(args.input_file, args.dshot_rate, args.loop_rate, capture_duration,
len(transactions_a), len(transactions_b), stats_a, stats_b, gap_stats)
# Generate and show histogram
if not args.no_plots and HAS_MATPLOTLIB:
generate_combined_histogram(
intervals_a, intervals_b, gaps,
stats_a, stats_b, gap_stats,
expected_interval_us, args.dshot_rate
)
generate_combined_histogram(intervals_a, intervals_b, gaps, stats_a, stats_b,
gap_stats, expected_interval_us, args.dshot_rate)
plt.show()