mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-10-04 13:24:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Report partial log corruption in route characterization

Browse Source
This commit is contained in:
firestar5683
2026-09-18 22:49:47 -07:00
parent 31f2d0f612
commit 90bd992ced
2 changed files with 79 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
roadscore/tools/characterize_routes.py
+53 -49
View File
@@ -2,6 +2,7 @@
import argparse
from collections import Counter
import json
import warnings
from pathlib import Path
import numpy as np
from openpilot.tools.lib.logreader import LogReader
@@ -25,54 +26,57 @@ def characterize(folder):
if not files:
continue
row = {'segment': int(segment.name.rsplit('--', 1)[1]), 'readable': False}
try:
for event in LogReader(str(files[0])):
kind = event.which()
services[kind] += 1
t = event.logMonoTime / 1e9
first = t if first is None else min(first, t)
last = t if last is None else max(last, t)
if kind == 'carState':
car = event.carState
speed = float(car.vEgo)
speeds.append(speed)
if previous_car is not None and speed > .5:
moving_seconds += max(0, min(.2, t-previous_car))
previous_car = t
signals = (bool(car.leftBlinker), bool(car.rightBlinker))
counts['turn_signal_activations'] += sum(now and not before for now, before in zip(signals, previous_signals))
previous_signals = signals
stopped = bool(car.standstill)
if previous_standstill is not None and stopped != previous_standstill:
counts['stop_events' if stopped else 'resume_events'] += 1
previous_standstill = stopped
elif kind == 'modelV2':
model = event.modelV2
samples = [abs(float(y))/max(speed, 3) for dt, y in zip(model.orientationRate.t, model.orientationRate.z) if 1 <= dt <= 5]
if samples and speed > 3:
curvature = max(samples)
curvatures.append(curvature)
if curvature > 1/150 and not active_curve:
counts['curve_proxy_events'] += 1
strongest.append({'log_monotonic_seconds': t, 'max_curvature_per_meter': curvature})
active_curve = True
elif curvature < 1/300:
active_curve = False
lane = str(model.meta.laneChangeState)
if lane != previous_lane and lane not in ('off', '0'):
counts['lane_change_state_transitions'] += 1
previous_lane = lane
elif kind == 'navInstruction' and event.valid:
nav = event.navInstruction
key = (str(nav.maneuverType), str(nav.maneuverModifier))
distance = float(nav.maneuverDistance)
if key != nav_key or (previous_nav_distance is not None and distance > previous_nav_distance+100):
counts['navigation_maneuvers'] += 1
counts['arrival_messages'] += nav.maneuverType == 'arrive'
nav_key, previous_nav_distance = key, distance
row['readable'] = True
except Exception as error:
row.update(error_type=type(error).__name__, error=str(error))
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
try:
for event in LogReader(str(files[0])):
kind = event.which()
services[kind] += 1
t = event.logMonoTime / 1e9
first = t if first is None else min(first, t)
last = t if last is None else max(last, t)
if kind == 'carState':
car = event.carState
speed = float(car.vEgo)
speeds.append(speed)
if previous_car is not None and speed > .5:
moving_seconds += max(0, min(.2, t-previous_car))
previous_car = t
signals = (bool(car.leftBlinker), bool(car.rightBlinker))
counts['turn_signal_activations'] += sum(now and not before for now, before in zip(signals, previous_signals))
previous_signals = signals
stopped = bool(car.standstill)
if previous_standstill is not None and stopped != previous_standstill:
counts['stop_events' if stopped else 'resume_events'] += 1
previous_standstill = stopped
elif kind == 'modelV2':
model = event.modelV2
samples = [abs(float(y))/max(speed, 3) for dt, y in zip(model.orientationRate.t, model.orientationRate.z) if 1 <= dt <= 5]
if samples and speed > 3:
curvature = max(samples)
curvatures.append(curvature)
if curvature > 1/150 and not active_curve:
counts['curve_proxy_events'] += 1
strongest.append({'log_monotonic_seconds': t, 'max_curvature_per_meter': curvature})
active_curve = True
elif curvature < 1/300:
active_curve = False
lane = str(model.meta.laneChangeState)
if lane != previous_lane and lane not in ('off', '0'):
counts['lane_change_state_transitions'] += 1
previous_lane = lane
elif kind == 'navInstruction' and event.valid:
nav = event.navInstruction
key = (str(nav.maneuverType), str(nav.maneuverModifier))
distance = float(nav.maneuverDistance)
if key != nav_key or (previous_nav_distance is not None and distance > previous_nav_distance+100):
counts['navigation_maneuvers'] += 1
counts['arrival_messages'] += nav.maneuverType == 'arrive'
nav_key, previous_nav_distance = key, distance
row['readable'] = True
except Exception as error:
row.update(error_type=type(error).__name__, error=str(error))
row["warnings"] = [str(w.message) for w in caught]
segments.append(row)
def distribution(values):
return dict(zip(['min', 'p50', 'p90', 'p99', 'max'], map(float, np.percentile(values, [0, 50, 90, 99, 100])))) if values else None
@@ -85,7 +89,7 @@ def characterize(folder):
'services': dict(services), 'segments': segments,
'required_messages_present': all(services[x] > 0 for x in ['modelV2', 'carState', 'roadEncodeIdx']),
'navigation_present': services['navInstruction'] > 0,
'all_available_logs_readable': bool(segments) and all(s['readable'] for s in segments)}
'all_available_logs_readable': bool(segments) and all(s['readable'] and not s.get('warnings') for s in segments)}
if __name__ == '__main__':
roadscore/tools/test_characterize_routes.py
+26
View File
@@ -0,0 +1,26 @@
import tempfile
import unittest
import warnings
from pathlib import Path
from unittest.mock import patch
from characterize_routes import characterize
class CharacterizationTests(unittest.TestCase):
def test_parser_warning_prevents_clean_log_claim(self):
def warned_reader(_):
warnings.warn('Corrupted events detected', RuntimeWarning)
return iter(())
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
segment = root / 'fixture--0'
segment.mkdir()
(segment / 'rlog').touch()
with patch('characterize_routes.LogReader', side_effect=warned_reader):
result = characterize(root)
self.assertFalse(result['all_available_logs_readable'])
self.assertEqual(result['segments'][0]['warnings'], ['Corrupted events detected'])
if __name__ == '__main__':
unittest.main()