73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Rank passive CAN bytes against timestamped scan-tool coolant readings."""
|
|
|
|
import argparse
|
|
import math
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
|
|
def correlation(xs, ys):
|
|
if len(xs) < 3 or len(set(xs)) < 2 or len(set(ys)) < 2:
|
|
return 0.0
|
|
xm, ym = sum(xs) / len(xs), sum(ys) / len(ys)
|
|
numerator = sum((x - xm) * (y - ym) for x, y in zip(xs, ys))
|
|
scale = math.sqrt(sum((x - xm) ** 2 for x in xs) * sum((y - ym) ** 2 for y in ys))
|
|
return numerator / scale if scale else 0.0
|
|
|
|
|
|
def parse_capture(path):
|
|
frames = defaultdict(list)
|
|
temperatures = []
|
|
for line in path.read_text(errors="replace").splitlines():
|
|
parts = [part.strip() for part in line.split(",")]
|
|
try:
|
|
if len(parts) >= 3 and parts[0] == "TEMP":
|
|
temperatures.append((int(parts[1]), float(parts[2])))
|
|
elif len(parts) >= 6 and parts[0] == "CAN":
|
|
timestamp, can_id, dlc = int(parts[1]), parts[2].upper(), int(parts[4])
|
|
for index, value in enumerate(parts[5 : 5 + min(dlc, 8)]):
|
|
if value:
|
|
frames[(can_id, index)].append((timestamp, int(value, 16)))
|
|
except ValueError:
|
|
continue
|
|
return frames, sorted(temperatures)
|
|
|
|
|
|
def value_at(samples, timestamp):
|
|
latest = None
|
|
for sample_time, value in samples:
|
|
if sample_time > timestamp:
|
|
break
|
|
latest = value
|
|
return latest
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Rank CAN bytes against TEMP markers.")
|
|
parser.add_argument("capture", type=Path)
|
|
parser.add_argument("--limit", type=int, default=20)
|
|
args = parser.parse_args()
|
|
frames, temperatures = parse_capture(args.capture)
|
|
if len(temperatures) < 3:
|
|
parser.error("capture needs at least three TEMP markers")
|
|
|
|
ranked = []
|
|
for (can_id, byte_index), samples in frames.items():
|
|
pairs = [(value_at(samples, timestamp), temp) for timestamp, temp in temperatures]
|
|
pairs = [(value, temp) for value, temp in pairs if value is not None]
|
|
values, temps = [p[0] for p in pairs], [p[1] for p in pairs]
|
|
score = correlation(values, temps)
|
|
if len(values) >= 3 and len(set(values)) >= 2:
|
|
ranked.append((abs(score), score, can_id, byte_index, len(values), min(values), max(values)))
|
|
|
|
ranked.sort(reverse=True)
|
|
print("rank id byte corr samples raw-range")
|
|
for rank, (_, score, can_id, byte_index, count, low, high) in enumerate(ranked[:args.limit], 1):
|
|
print(f"{rank:>4} 0x{can_id:<4} {byte_index:>4} {score:>+7.3f} {count:>7} {low:>3}..{high:<3}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|