mirror of
https://github.com/Horizontal42/f1-telemetry.git
synced 2026-11-10 15:22:03 +03:00
add profile mode, optional prompt, report cache, compare verdict
profile: cheap cross-lap corner tendencies + auto-picked laps, per tyre compound --no-prompt flag and GUI checkbox omit the embedded prompt; --force ignores the cache compare: deterministic fuel-corrected verdict section (fuel as a range, not a point) corner merge gap now scales with track length; race dedup keeps laps with no Lap field
This commit is contained in:
Andrew Pimenov
parent
917a66c224
commit
0661ff4322
9 changed files with 544 additions and 48 deletions
|
|
@ -4,12 +4,22 @@ import sys
|
|||
|
||||
from .parser import load_lap
|
||||
from .report_common import report_path
|
||||
from . import report_technique, report_setup, report_compare, report_race
|
||||
from . import report_technique, report_setup, report_compare, report_race, report_profile
|
||||
from .rename import rename_unprocessed
|
||||
|
||||
_RACES_DIR = os.path.normpath(os.path.join(os.path.dirname(__file__), '..', '..', 'races'))
|
||||
|
||||
|
||||
def is_fresh(report: str, sources: list[str]) -> bool:
|
||||
if not os.path.isfile(report):
|
||||
return False
|
||||
try:
|
||||
r_mtime = os.path.getmtime(report)
|
||||
return all(os.path.getmtime(s) <= r_mtime for s in sources if os.path.isfile(s))
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _print_result(abs_path: str, tokens: int) -> None:
|
||||
print(abs_path)
|
||||
print(f'~{tokens} tokens (budget 60k)')
|
||||
|
|
@ -48,14 +58,20 @@ def main() -> None:
|
|||
return
|
||||
|
||||
parser = argparse.ArgumentParser(prog='python -m telemetry')
|
||||
parser.add_argument('mode', choices=['technique', 'setup', 'compare', 'race', 'rename', 'gui'])
|
||||
parser.add_argument('mode', choices=['technique', 'setup', 'compare', 'race', 'profile', 'rename', 'gui'])
|
||||
parser.add_argument('files', nargs='*')
|
||||
parser.add_argument('--lang', choices=['ru', 'en'], default='ru')
|
||||
parser.add_argument('--no-prompt', action='store_true',
|
||||
help='omit the embedded analysis prompt (for agent pipelines)')
|
||||
parser.add_argument('--force', action='store_true',
|
||||
help='regenerate even if a fresh report already exists')
|
||||
args = parser.parse_args()
|
||||
|
||||
mode: str = args.mode
|
||||
files: list[str] = args.files
|
||||
lang: str = args.lang
|
||||
include_prompt: bool = not args.no_prompt
|
||||
force: bool = args.force
|
||||
|
||||
if mode == 'gui':
|
||||
from .gui import main as gui_main
|
||||
|
|
@ -75,38 +91,47 @@ def main() -> None:
|
|||
if mode == 'compare' and len(files) < 2:
|
||||
print('Error: compare requires 2+ files', file=sys.stderr)
|
||||
sys.exit(2)
|
||||
if mode == 'race' and len(files) != 1:
|
||||
print('Error: race requires exactly 1 directory', file=sys.stderr)
|
||||
if mode in ('race', 'profile') and len(files) != 1:
|
||||
print(f'Error: {mode} requires exactly 1 directory', file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
def _emit(out: str, sources: list[str], gen) -> None:
|
||||
if not force and is_fresh(out, sources):
|
||||
print(os.path.abspath(out))
|
||||
print('cached (up to date) — use --force to regenerate')
|
||||
return
|
||||
_print_result(*gen())
|
||||
|
||||
if mode == 'technique':
|
||||
lap = load_lap(files[0])
|
||||
out = report_path(files[0], 'technique')
|
||||
_print_result(*report_technique.generate(lap, out, lang))
|
||||
_emit(out, [files[0]],
|
||||
lambda: report_technique.generate(load_lap(files[0]), out, lang, include_prompt))
|
||||
|
||||
elif mode == 'setup':
|
||||
lap = load_lap(files[0])
|
||||
out = report_path(files[0], 'setup')
|
||||
_print_result(*report_setup.generate(lap, out, lang))
|
||||
_emit(out, [files[0]],
|
||||
lambda: report_setup.generate(load_lap(files[0]), out, lang, include_prompt))
|
||||
|
||||
elif mode == 'compare':
|
||||
laps = [load_lap(f) for f in files]
|
||||
stem_b = os.path.splitext(os.path.basename(files[1]))[0]
|
||||
out = report_path(files[0], 'compare', stem_b)
|
||||
_print_result(*report_compare.generate(laps, out, lang))
|
||||
_emit(out, files,
|
||||
lambda: report_compare.generate([load_lap(f) for f in files], out, lang, include_prompt))
|
||||
|
||||
elif mode == 'race':
|
||||
elif mode in ('race', 'profile'):
|
||||
race_dir = os.path.abspath(files[0])
|
||||
csv_files = sorted(
|
||||
f for f in os.listdir(race_dir) if f.lower().endswith('.csv')
|
||||
csv_paths = sorted(
|
||||
os.path.join(race_dir, f) for f in os.listdir(race_dir) if f.lower().endswith('.csv')
|
||||
)
|
||||
if not csv_files:
|
||||
if not csv_paths:
|
||||
print(f'Error: no CSV files found in {race_dir}', file=sys.stderr)
|
||||
sys.exit(2)
|
||||
laps = [load_lap(os.path.join(race_dir, f)) for f in csv_files]
|
||||
dirname = os.path.basename(race_dir.rstrip('/\\'))
|
||||
out = os.path.join(race_dir, 'reports', dirname + '_race.md')
|
||||
_print_result(*report_race.generate(laps, out, lang))
|
||||
suffix = '_race.md' if mode == 'race' else '_profile.md'
|
||||
out = os.path.join(race_dir, 'reports', dirname + suffix)
|
||||
gen_mod = report_race if mode == 'race' else report_profile
|
||||
_emit(out, csv_paths,
|
||||
lambda: gen_mod.generate([load_lap(p) for p in csv_paths], out, lang, include_prompt))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ from tkinter import filedialog, scrolledtext, ttk
|
|||
|
||||
from .parser import load_lap
|
||||
from .report_common import report_path
|
||||
from . import report_technique, report_setup, report_compare, report_race
|
||||
from . import report_technique, report_setup, report_compare, report_race, report_profile
|
||||
from .rename import rename_unprocessed
|
||||
|
||||
# project root = parent of the tool\ dir
|
||||
|
|
@ -27,6 +27,7 @@ class _App:
|
|||
|
||||
self._mode = tk.StringVar(value='technique')
|
||||
self._lang = tk.StringVar(value='ru')
|
||||
self._include_prompt = tk.BooleanVar(value=True)
|
||||
self._selection: list[str] = []
|
||||
self._last_dir: dict[str, str] = {}
|
||||
self._log_queue: queue.Queue[str | None] = queue.Queue()
|
||||
|
|
@ -41,7 +42,8 @@ class _App:
|
|||
mode_frame = ttk.LabelFrame(self._root, text='Mode')
|
||||
mode_frame.pack(fill='x', **pad)
|
||||
for label, value in [('Technique', 'technique'), ('Setup', 'setup'),
|
||||
('Compare', 'compare'), ('Race', 'race')]:
|
||||
('Compare', 'compare'), ('Race', 'race'),
|
||||
('Profile', 'profile')]:
|
||||
ttk.Radiobutton(mode_frame, text=label, variable=self._mode,
|
||||
value=value).pack(side='left', padx=6, pady=4)
|
||||
|
||||
|
|
@ -51,6 +53,8 @@ class _App:
|
|||
value='ru').pack(side='left', padx=6, pady=4)
|
||||
ttk.Radiobutton(lang_frame, text='EN', variable=self._lang,
|
||||
value='en').pack(side='left', padx=6, pady=4)
|
||||
ttk.Checkbutton(lang_frame, text='Включить промпт в отчёт / Include prompt',
|
||||
variable=self._include_prompt).pack(side='left', padx=16, pady=4)
|
||||
|
||||
input_frame = ttk.LabelFrame(self._root, text='Input')
|
||||
input_frame.pack(fill='x', **pad)
|
||||
|
|
@ -81,7 +85,7 @@ class _App:
|
|||
def _browse(self) -> None:
|
||||
mode = self._mode.get()
|
||||
init = self._last_dir.get(mode, _RACES_DIR)
|
||||
if mode == 'race':
|
||||
if mode in ('race', 'profile'):
|
||||
d = filedialog.askdirectory(title='Select race folder', initialdir=init)
|
||||
if d:
|
||||
self._selection = [d]
|
||||
|
|
@ -113,8 +117,8 @@ class _App:
|
|||
return f'{mode} requires exactly 1 file'
|
||||
if mode == 'compare' and len(sel) < 2:
|
||||
return 'compare requires 2+ files'
|
||||
if mode == 'race' and (len(sel) != 1 or not os.path.isdir(sel[0])):
|
||||
return 'race requires a directory'
|
||||
if mode in ('race', 'profile') and (len(sel) != 1 or not os.path.isdir(sel[0])):
|
||||
return f'{mode} requires a directory'
|
||||
return None
|
||||
|
||||
def _set_busy(self, busy: bool) -> None:
|
||||
|
|
@ -152,32 +156,33 @@ class _App:
|
|||
mode = self._mode.get()
|
||||
sel = list(self._selection)
|
||||
lang = self._lang.get()
|
||||
inc = self._include_prompt.get()
|
||||
self._set_busy(True)
|
||||
threading.Thread(target=self._worker_generate, args=(mode, sel, lang),
|
||||
threading.Thread(target=self._worker_generate, args=(mode, sel, lang, inc),
|
||||
daemon=True).start()
|
||||
|
||||
def _worker_generate(self, mode: str, sel: list[str], lang: str) -> None:
|
||||
def _worker_generate(self, mode: str, sel: list[str], lang: str, inc: bool) -> None:
|
||||
try:
|
||||
if mode == 'technique':
|
||||
lap = load_lap(sel[0])
|
||||
out = report_path(sel[0], 'technique')
|
||||
abs_path, tokens = report_technique.generate(lap, out, lang)
|
||||
abs_path, tokens = report_technique.generate(lap, out, lang, inc)
|
||||
self._push(f'✓ {abs_path} (~{tokens} tokens)')
|
||||
|
||||
elif mode == 'setup':
|
||||
lap = load_lap(sel[0])
|
||||
out = report_path(sel[0], 'setup')
|
||||
abs_path, tokens = report_setup.generate(lap, out, lang)
|
||||
abs_path, tokens = report_setup.generate(lap, out, lang, inc)
|
||||
self._push(f'✓ {abs_path} (~{tokens} tokens)')
|
||||
|
||||
elif mode == 'compare':
|
||||
laps = [load_lap(f) for f in sel]
|
||||
stem_b = os.path.splitext(os.path.basename(sel[1]))[0]
|
||||
out = report_path(sel[0], 'compare', stem_b)
|
||||
abs_path, tokens = report_compare.generate(laps, out, lang)
|
||||
abs_path, tokens = report_compare.generate(laps, out, lang, inc)
|
||||
self._push(f'✓ {abs_path} (~{tokens} tokens)')
|
||||
|
||||
elif mode == 'race':
|
||||
elif mode in ('race', 'profile'):
|
||||
race_dir = os.path.abspath(sel[0])
|
||||
csv_files = sorted(
|
||||
f for f in os.listdir(race_dir) if f.lower().endswith('.csv')
|
||||
|
|
@ -186,8 +191,11 @@ class _App:
|
|||
self._push(f'✗ No CSV files found in {race_dir}')
|
||||
return
|
||||
laps = [load_lap(os.path.join(race_dir, f)) for f in csv_files]
|
||||
out = _race_output_path(race_dir)
|
||||
abs_path, tokens = report_race.generate(laps, out, lang)
|
||||
dirname = os.path.basename(race_dir.rstrip('/\\'))
|
||||
suffix = '_race.md' if mode == 'race' else '_profile.md'
|
||||
out = os.path.join(race_dir, 'reports', dirname + suffix)
|
||||
gen_mod = report_race if mode == 'race' else report_profile
|
||||
abs_path, tokens = gen_mod.generate(laps, out, lang, inc)
|
||||
self._push(f'✓ {abs_path} (~{tokens} tokens)')
|
||||
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -8,6 +8,11 @@ from .report_common import (
|
|||
LEGEND_DIST, LEGEND_SPD, LEGEND_DT_SEG, LEGEND_DT_CUM,
|
||||
)
|
||||
|
||||
# seconds gained per kg of fuel burned — track-dependent in reality, so report a range
|
||||
# (rule of thumb across F1 22 circuits) instead of one false-precise number.
|
||||
FUEL_S_PER_KG_LOW = 0.03
|
||||
FUEL_S_PER_KG_HIGH = 0.05
|
||||
|
||||
|
||||
def _interp_laptime(lap: Lap, dist_query: float) -> float:
|
||||
dist = lap.ch.get('LapDistance', [])
|
||||
|
|
@ -118,6 +123,78 @@ def _corner_table(laps: list[Lap], corners: list[Corner]) -> str:
|
|||
return md_table(headers, rows)
|
||||
|
||||
|
||||
def _fuel_start(lap: Lap) -> float | None:
|
||||
raw = lap.track.get('Fuel at Start', lap.track.get('FuelAtStart', ''))
|
||||
raw = raw.strip().split()[0] if raw.strip() else ''
|
||||
try:
|
||||
return float(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _seg_time(lap: Lap, c: Corner) -> float | None:
|
||||
if not (lap.ch.get('LapDistance') and lap.ch.get('LapTime')):
|
||||
return None
|
||||
return _interp_laptime(lap, c.end_m) - _interp_laptime(lap, c.start_m)
|
||||
|
||||
|
||||
def _vmin(lap: Lap, c: Corner) -> float | None:
|
||||
d = lap.ch.get('LapDistance', [])
|
||||
spd = lap.ch.get('Speed', [])
|
||||
idxs = [i for i, v in enumerate(d) if c.start_m <= v <= c.end_m and i < len(spd)]
|
||||
return min(spd[i] for i in idxs) if idxs else None
|
||||
|
||||
|
||||
def _verdict(laps: list[Lap], corners: list[Corner]) -> str:
|
||||
# deterministic comparison of each non-reference lap vs lap A (lap 1)
|
||||
lap_a = laps[0]
|
||||
lines: list[str] = []
|
||||
for k in range(1, len(laps)):
|
||||
lap_k = laps[k]
|
||||
dt_total = laptime_s(lap_k) - laptime_s(lap_a)
|
||||
fa, fk = _fuel_start(lap_a), _fuel_start(lap_k)
|
||||
|
||||
lines.append(f'### L{k + 1} vs L1')
|
||||
lines.append(f'- Итоговая Δ laptime: **{dt_total:+.3f} с** (L{k + 1} − L1)')
|
||||
if fa is not None and fk is not None:
|
||||
d_fuel = fa - fk
|
||||
fuel_lo = d_fuel * FUEL_S_PER_KG_LOW
|
||||
fuel_hi = d_fuel * FUEL_S_PER_KG_HIGH
|
||||
# residual = laptime delta minus fuel benefit; wider fuel benefit → lower residual
|
||||
res_lo = dt_total - max(fuel_lo, fuel_hi)
|
||||
res_hi = dt_total - min(fuel_lo, fuel_hi)
|
||||
lines.append(
|
||||
f'- Топливо: L1 {fa:.1f} кг / L{k + 1} {fk:.1f} кг → '
|
||||
f'поправка ~{min(fuel_lo, fuel_hi):+.3f}…{max(fuel_lo, fuel_hi):+.3f} с '
|
||||
f'(0.03–0.05 с/кг). '
|
||||
f'**Остаток (не топливо): {res_lo:+.3f}…{res_hi:+.3f} с**'
|
||||
)
|
||||
|
||||
deltas: list[tuple[float, int, float | None, float | None]] = []
|
||||
for c in corners:
|
||||
ta, tk = _seg_time(lap_a, c), _seg_time(lap_k, c)
|
||||
if ta is None or tk is None:
|
||||
continue
|
||||
deltas.append((tk - ta, c.number, _vmin(lap_a, c), _vmin(lap_k, c)))
|
||||
|
||||
losses = sorted([d for d in deltas if d[0] > 0], key=lambda x: -x[0])[:5]
|
||||
gains = sorted([d for d in deltas if d[0] < 0], key=lambda x: x[0])[:3]
|
||||
|
||||
if losses:
|
||||
lines.append(f'- **Где L{k + 1} теряет** (топ |Δt seg|):')
|
||||
for dt, num, va, vk in losses:
|
||||
vm = f' | Vmin {va:.0f}→{vk:.0f}' if (va is not None and vk is not None) else ''
|
||||
lines.append(f' - T{num}: {dt:+.3f} с{vm}')
|
||||
if gains:
|
||||
lines.append(f'- **Где L{k + 1} быстрее**:')
|
||||
for dt, num, va, vk in gains:
|
||||
vm = f' | Vmin {va:.0f}→{vk:.0f}' if (va is not None and vk is not None) else ''
|
||||
lines.append(f' - T{num}: {dt:+.3f} с{vm}')
|
||||
lines.append('')
|
||||
|
||||
return '\n'.join(lines).rstrip()
|
||||
|
||||
|
||||
def _delta_trace(laps: list[Lap], corners: list[Corner]) -> tuple[str, list[float]]:
|
||||
lap_a = laps[0]
|
||||
dists = adaptive_points(lap_a, corners)
|
||||
|
|
@ -153,12 +230,13 @@ def _delta_trace(laps: list[Lap], corners: list[Corner]) -> tuple[str, list[floa
|
|||
return block, final_deltas
|
||||
|
||||
|
||||
def generate(laps: list[Lap], out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
||||
def generate(laps: list[Lap], out_path: str, lang: str = 'ru', include_prompt: bool = True) -> tuple[str, int]:
|
||||
lap_a = laps[0]
|
||||
corners = detect_corners(lap_a)
|
||||
|
||||
head_table = _header_table(laps)
|
||||
corner_table = _corner_table(laps, corners)
|
||||
verdict = _verdict(laps, corners)
|
||||
delta_block, final_deltas = _delta_trace(laps, corners)
|
||||
|
||||
for k in range(1, len(laps)):
|
||||
|
|
@ -169,6 +247,8 @@ def generate(laps: list[Lap], out_path: str, lang: str = 'ru') -> tuple[str, int
|
|||
parts = [
|
||||
'# Lap Comparison\n',
|
||||
head_table,
|
||||
'\n## Verdict (вычислено детерминированно)',
|
||||
verdict,
|
||||
'\n## Corners',
|
||||
corner_table,
|
||||
'\n## Delta trace',
|
||||
|
|
@ -176,7 +256,8 @@ def generate(laps: list[Lap], out_path: str, lang: str = 'ru') -> tuple[str, int
|
|||
'\n' + legend([LEGEND_DIST, LEGEND_SPD, LEGEND_DT_SEG, LEGEND_DT_CUM]),
|
||||
]
|
||||
text = '\n'.join(parts)
|
||||
prompt = load_prompt('compare', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
if include_prompt:
|
||||
prompt = load_prompt('compare', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
return write_report(out_path, text)
|
||||
|
|
|
|||
205
telemetry/report_profile.py
Normal file
205
telemetry/report_profile.py
Normal file
|
|
@ -0,0 +1,205 @@
|
|||
# Cheap full-field aggregate: scans every lap (corner tables only, no trace) to surface
|
||||
# chronic driver tendencies and auto-pick laps worth a deep dive. Track-agnostic:
|
||||
# corners are clustered by apex distance (not index) and tendencies are relative to the
|
||||
# session, so it works on any track / compound without per-track calibration.
|
||||
from .parser import Lap, laptime_s
|
||||
from .segments import detect_corners, Corner, merge_gap_for
|
||||
from .report_common import md_table, write_report, load_prompt
|
||||
from .report_race import _dedup_laps, _sort_laps
|
||||
|
||||
# Corners are clustered across laps by apex distance. The gap reuses the detector's own
|
||||
# track-adaptive merge gap (merge_gap_for), so the cluster count stays aligned with per-lap
|
||||
# corner numbers (profile Tn == technique Tn) on chicanes and long circuits alike.
|
||||
|
||||
MIN_LAPS_PER_COMPOUND = 3
|
||||
# chronic = relative outlier vs the rest of the session's corners, with a physics floor
|
||||
# so a benign track never raises false alarms.
|
||||
COAST_REL, COAST_FLOOR = 1.5, 35.0
|
||||
SPIN_REL, SPIN_FLOOR = 1.7, 4.0
|
||||
LOCK_REL, LOCK_FLOOR = 1.7, 5.0
|
||||
TR_MARGIN, TR_FLOOR = 8.0, 100.0
|
||||
|
||||
|
||||
def _median(vals: list[float]) -> float:
|
||||
if not vals:
|
||||
return 0.0
|
||||
s = sorted(vals)
|
||||
n = len(s)
|
||||
mid = n // 2
|
||||
return s[mid] if n % 2 else (s[mid - 1] + s[mid]) / 2.0
|
||||
|
||||
|
||||
def _is_valid(lap: Lap) -> bool:
|
||||
return lap.track.get('Valid', 'true').lower() != 'false'
|
||||
|
||||
|
||||
def _is_pit(lap: Lap) -> bool:
|
||||
return lap.track.get('Pitlap', 'false').lower() == 'true'
|
||||
|
||||
|
||||
def _lap_no(lap: Lap) -> str:
|
||||
return lap.track.get('Lap', '?')
|
||||
|
||||
|
||||
def _clean_laps(laps: list[Lap]) -> list[Lap]:
|
||||
return [l for l in laps if _is_valid(l) and not _is_pit(l) and laptime_s(l) > 0]
|
||||
|
||||
|
||||
def _by_compound(laps: list[Lap]) -> list[tuple[str, list[Lap]]]:
|
||||
groups: dict[str, list[Lap]] = {}
|
||||
for l in laps:
|
||||
groups.setdefault(l.track.get('Tyre', '?'), []).append(l)
|
||||
return sorted(groups.items(), key=lambda kv: -len(kv[1]))
|
||||
|
||||
|
||||
def _cluster_corners(per: list[tuple[Lap, list[Corner]]], gap: float) -> list[list[Corner]]:
|
||||
# group corners across laps by apex location, robust to a lap detecting one extra/fewer
|
||||
flat = sorted(
|
||||
((c.apex_m, c) for _, cs in per for c in cs),
|
||||
key=lambda x: x[0],
|
||||
)
|
||||
if not flat:
|
||||
return []
|
||||
|
||||
clusters: list[list[Corner]] = [[flat[0][1]]]
|
||||
last_apex = flat[0][0]
|
||||
for apex, c in flat[1:]:
|
||||
if apex - last_apex > gap:
|
||||
clusters.append([c])
|
||||
else:
|
||||
clusters[-1].append(c)
|
||||
last_apex = apex
|
||||
|
||||
min_support = max(2, len(per) // 2)
|
||||
return [cl for cl in clusters if len(cl) >= min_support]
|
||||
|
||||
|
||||
def _aggregate_compound(laps: list[Lap]) -> tuple[str, list[str]]:
|
||||
per = [(l, detect_corners(l)) for l in laps]
|
||||
clusters = _cluster_corners(per, merge_gap_for(laps[0]))
|
||||
if not clusters:
|
||||
return '', []
|
||||
|
||||
agg = []
|
||||
for num, cl in enumerate(clusters, 1):
|
||||
agg.append({
|
||||
'num': num,
|
||||
'vmin': _median([c.v_min for c in cl]),
|
||||
'coast': _median([c.coast_m for c in cl]),
|
||||
'lock': _median([c.lock_pct for c in cl]),
|
||||
'spin': _median([c.spin_pct for c in cl]),
|
||||
'tr': _median([c.rear_tyre_temp_peak for c in cl]),
|
||||
'lock_freq': sum(1 for c in cl if c.lockup) / len(cl) * 100,
|
||||
'spin_freq': sum(1 for c in cl if c.wheelspin) / len(cl) * 100,
|
||||
})
|
||||
|
||||
med_coast = _median([a['coast'] for a in agg])
|
||||
med_spin = _median([a['spin'] for a in agg])
|
||||
med_lockmag = _median([abs(a['lock']) for a in agg])
|
||||
med_tr = _median([a['tr'] for a in agg])
|
||||
|
||||
rows, chronic = [], []
|
||||
for a in agg:
|
||||
rows.append([
|
||||
f"T{a['num']}", f"{a['vmin']:.0f}", f"{a['coast']:.0f}",
|
||||
f"{a['lock']:.1f}", f"{a['spin']:.1f}",
|
||||
f"{a['lock_freq']:.0f}", f"{a['spin_freq']:.0f}", f"{a['tr']:.0f}",
|
||||
])
|
||||
if a['coast'] >= max(COAST_REL * med_coast, COAST_FLOOR):
|
||||
chronic.append(f"T{a['num']}: длинный медианный coast {a['coast']:.0f} м — осторожный переход тормоз→газ")
|
||||
if a['spin'] >= max(SPIN_REL * med_spin, SPIN_FLOOR) or a['spin_freq'] >= 25:
|
||||
chronic.append(f"T{a['num']}: склонность к пробуксовке на выходе (медиана spin {a['spin']:.1f}%) — ранний/резкий газ")
|
||||
if abs(a['lock']) >= max(LOCK_REL * med_lockmag, LOCK_FLOOR) or a['lock_freq'] >= 25:
|
||||
chronic.append(f"T{a['num']}: склонность к блокировке на входе (медиана lock {a['lock']:.1f}%) — поздняя/жёсткая педаль")
|
||||
if a['tr'] >= med_tr + TR_MARGIN and a['tr'] >= TR_FLOOR:
|
||||
chronic.append(f"T{a['num']}: высокая нагрузка на задние шины (медиана Tr {a['tr']:.0f}°C) — тепловой риск в длинном стинте")
|
||||
|
||||
table = md_table(
|
||||
['T', 'Vmin', 'coast m', 'lock%', 'spin%', 'LOCK %laps', 'SPIN %laps', 'Tr peak °C'],
|
||||
rows,
|
||||
)
|
||||
return table, chronic
|
||||
|
||||
|
||||
def _recommend_laps(all_clean: list[Lap]) -> str:
|
||||
def rear_peak(l: Lap) -> float:
|
||||
rl = l.ch.get('TyreTemperatureRearLeft', [])
|
||||
rr = l.ch.get('TyreTemperatureRearRight', [])
|
||||
return max((max(rl) if rl else 0), (max(rr) if rr else 0))
|
||||
|
||||
def worst_corner(l: Lap, attr: str, sign: float) -> float:
|
||||
cs = detect_corners(l)
|
||||
return max((sign * getattr(c, attr) for c in cs), default=0.0)
|
||||
|
||||
picks = [
|
||||
('быстрейший валидный', min(all_clean, key=laptime_s)),
|
||||
('худший валидный', max(all_clean, key=laptime_s)),
|
||||
('тепловой пик (задние)', max(all_clean, key=rear_peak)),
|
||||
('макс. пробуксовка', max(all_clean, key=lambda l: worst_corner(l, 'spin_pct', 1.0))),
|
||||
('макс. блокировка', max(all_clean, key=lambda l: worst_corner(l, 'lock_pct', -1.0))),
|
||||
]
|
||||
|
||||
seen: dict[str, str] = {}
|
||||
tyre: dict[str, str] = {}
|
||||
for reason, lap in picks:
|
||||
key = _lap_no(lap)
|
||||
seen[key] = (seen.get(key, '') + '; ' + reason).lstrip('; ')
|
||||
tyre[key] = lap.track.get('Tyre', '?')
|
||||
|
||||
rows = [[f'L{k}', tyre[k], seen[k]] for k in seen]
|
||||
return md_table(['Круг', 'Состав', 'Почему интересен'], rows)
|
||||
|
||||
|
||||
def generate(laps: list[Lap], out_path: str, lang: str = 'ru', include_prompt: bool = True) -> tuple[str, int]:
|
||||
laps = _sort_laps(_dedup_laps(laps))
|
||||
clean = _clean_laps(laps)
|
||||
|
||||
if not clean:
|
||||
return write_report(out_path, '_Нет валидных кругов для профилирования._')
|
||||
|
||||
first = laps[0]
|
||||
breakdown = {}
|
||||
for l in clean:
|
||||
t = l.track.get('Tyre', '?')
|
||||
breakdown[t] = breakdown.get(t, 0) + 1
|
||||
breakdown_str = ', '.join(f'{k}: {v}' for k, v in breakdown.items())
|
||||
|
||||
parts = [
|
||||
f'**Track:** {first.session.get("track", "?")} ',
|
||||
f'**Car:** {first.session.get("car", "?")} ',
|
||||
f'**Валидных кругов:** {len(clean)} из {len(laps)} ',
|
||||
f'**Составы (валидные круги):** {breakdown_str} ',
|
||||
]
|
||||
|
||||
profiled = False
|
||||
for comp, claps in _by_compound(clean):
|
||||
if len(claps) < MIN_LAPS_PER_COMPOUND:
|
||||
continue
|
||||
table, chronic = _aggregate_compound(claps)
|
||||
if not table:
|
||||
continue
|
||||
profiled = True
|
||||
parts.append(f'\n## Тенденции по поворотам — состав {comp} ({len(claps)} кругов)')
|
||||
parts.append(table)
|
||||
parts.append(f'\n### Хронические паттерны — {comp} (авто-детект, относительно сессии)')
|
||||
parts.append('\n'.join(f'- {c}' for c in chronic) if chronic
|
||||
else '- Явных хронических паттернов не выявлено.')
|
||||
|
||||
if not profiled:
|
||||
parts.append(f'\n_Недостаточно кругов на один состав для агрегации (нужно ≥{MIN_LAPS_PER_COMPOUND}).')
|
||||
|
||||
parts += [
|
||||
'\n## Рекомендованные круги для детального разбора',
|
||||
_recommend_laps(clean),
|
||||
'\n## Легенда',
|
||||
'- `coast m` = медианный накат без тормоза и полного газа (большой = осторожный переход)',
|
||||
'- `LOCK %laps` / `SPIN %laps` = доля кругов с флагом блокировки / пробуксовки в этом повороте',
|
||||
'- Повороты кластеризуются по дистанции апекса (а не по индексу) — устойчиво к разному числу детектируемых поворотов',
|
||||
'- Хроника считается ОТНОСИТЕЛЬНО медианы поворотов сессии (адаптивно к трассе) с физическим порогом-полом',
|
||||
]
|
||||
text = '\n'.join(parts)
|
||||
if include_prompt:
|
||||
prompt = load_prompt('profile', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
return write_report(out_path, text)
|
||||
|
|
@ -23,6 +23,37 @@ def _fuel_start(lap: Lap) -> float:
|
|||
return v[0] if v else 0.0
|
||||
|
||||
|
||||
def _dedup_laps(laps: list[Lap]) -> list[Lap]:
|
||||
def lap_num(lap: Lap) -> int:
|
||||
try:
|
||||
return int(lap.track.get('Lap', '0'))
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
# Only dedup laps that carry a real positive lap number. Laps without one (export
|
||||
# missing the field → all parse to 0) are kept verbatim, so a missing 'Lap' column
|
||||
# can never silently collapse the whole race to a single lap.
|
||||
numbered = [l for l in laps if lap_num(l) > 0]
|
||||
unnumbered = [l for l in laps if lap_num(l) <= 0]
|
||||
|
||||
best: dict[int, Lap] = {}
|
||||
for lap in numbered:
|
||||
n = lap_num(lap)
|
||||
cur = best.get(n)
|
||||
if cur is None:
|
||||
best[n] = lap
|
||||
continue
|
||||
lap_valid = lap.track.get('Valid', 'true').lower() != 'false'
|
||||
cur_valid = cur.track.get('Valid', 'true').lower() != 'false'
|
||||
if lap_valid != cur_valid:
|
||||
if lap_valid:
|
||||
best[n] = lap
|
||||
continue
|
||||
if laptime_s(lap) < laptime_s(cur):
|
||||
best[n] = lap
|
||||
return list(best.values()) + unnumbered
|
||||
|
||||
|
||||
def _sort_laps(laps: list[Lap]) -> list[Lap]:
|
||||
def lap_num(lap: Lap) -> int:
|
||||
try:
|
||||
|
|
@ -262,7 +293,8 @@ def _ers_table(laps: list[Lap]) -> str:
|
|||
return md_table(['Lap', 'ERS MJ', 'MGU-K MJ', 'MGU-H MJ'], rows)
|
||||
|
||||
|
||||
def generate(laps: list[Lap], out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
||||
def generate(laps: list[Lap], out_path: str, lang: str = 'ru', include_prompt: bool = True) -> tuple[str, int]:
|
||||
laps = _dedup_laps(laps)
|
||||
laps = _sort_laps(laps)
|
||||
stints, ages = _assign_stints(laps)
|
||||
|
||||
|
|
@ -280,7 +312,8 @@ def generate(laps: list[Lap], out_path: str, lang: str = 'ru') -> tuple[str, int
|
|||
_ers_table(laps),
|
||||
]
|
||||
text = '\n'.join(parts)
|
||||
prompt = load_prompt('race', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
if include_prompt:
|
||||
prompt = load_prompt('race', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
return write_report(out_path, text)
|
||||
|
|
|
|||
|
|
@ -220,7 +220,7 @@ def _suspension_section(lap: Lap, corners: list[Corner]) -> str | None:
|
|||
return rake_line + '\n\n' + roll_table
|
||||
|
||||
|
||||
def generate(lap: Lap, out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
||||
def generate(lap: Lap, out_path: str, lang: str = 'ru', include_prompt: bool = True) -> tuple[str, int]:
|
||||
corners = detect_corners(lap)
|
||||
susp = _suspension_section(lap, corners)
|
||||
|
||||
|
|
@ -249,7 +249,8 @@ def generate(lap: Lap, out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
|||
'\n' + legend(legend_entries),
|
||||
]
|
||||
text = '\n'.join(parts)
|
||||
prompt = load_prompt('setup', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
if include_prompt:
|
||||
prompt = load_prompt('setup', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
return write_report(out_path, text)
|
||||
|
|
|
|||
|
|
@ -150,7 +150,7 @@ def _trace_block(lap: Lap, corners: list[Corner]) -> str:
|
|||
return '```\n' + header + '\n' + '\n'.join(rows) + '\n```'
|
||||
|
||||
|
||||
def generate(lap: Lap, out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
||||
def generate(lap: Lap, out_path: str, lang: str = 'ru', include_prompt: bool = True) -> tuple[str, int]:
|
||||
corners = detect_corners(lap)
|
||||
ss = straights(corners, lap)
|
||||
|
||||
|
|
@ -181,7 +181,8 @@ def generate(lap: Lap, out_path: str, lang: str = 'ru') -> tuple[str, int]:
|
|||
]),
|
||||
]
|
||||
text = '\n'.join(parts)
|
||||
prompt = load_prompt('technique', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
if include_prompt:
|
||||
prompt = load_prompt('technique', lang)
|
||||
if prompt:
|
||||
text += '\n\n---\n\n' + prompt
|
||||
return write_report(out_path, text)
|
||||
|
|
|
|||
|
|
@ -4,7 +4,10 @@ from .parser import Lap
|
|||
|
||||
STEER_ON = 6.0
|
||||
BRAKE_ON = 5.0
|
||||
MERGE_GAP_M = 40.0
|
||||
MERGE_GAP_M = 40.0 # anchored at Zandvoort (~4257 m): tracks scale around this
|
||||
MERGE_GAP_REF_LEN = 4257.0
|
||||
MERGE_GAP_MIN = 28.0
|
||||
MERGE_GAP_MAX = 60.0
|
||||
MIN_LEN_M = 15.0
|
||||
BRAKE_MARK = 20.0
|
||||
FULL_THROTTLE = 98.0
|
||||
|
|
@ -12,6 +15,22 @@ LOCKUP_REL_PCT = -10.0
|
|||
SPIN_REL_PCT = 8.0
|
||||
|
||||
|
||||
def merge_gap_for(lap: Lap) -> float:
|
||||
"""Corner-merge / cluster gap scaled to track length so chicanes and long
|
||||
circuits self-tune instead of relying on a fixed metre value. Equals
|
||||
MERGE_GAP_M at the reference length; clamped at the extremes."""
|
||||
raw = lap.track.get('Tracklen', '')
|
||||
try:
|
||||
tracklen = float(raw)
|
||||
except (TypeError, ValueError):
|
||||
dist = lap.ch.get('LapDistance', [])
|
||||
tracklen = (dist[-1] - dist[0]) if len(dist) >= 2 else 0.0
|
||||
if tracklen <= 0:
|
||||
return MERGE_GAP_M
|
||||
g = MERGE_GAP_M * tracklen / MERGE_GAP_REF_LEN
|
||||
return min(max(g, MERGE_GAP_MIN), MERGE_GAP_MAX)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Corner:
|
||||
number: int
|
||||
|
|
@ -93,9 +112,10 @@ def detect_corners(lap: Lap) -> list[Corner]:
|
|||
if in_zone:
|
||||
zones.append((z_start, n - 1))
|
||||
|
||||
merge_gap = merge_gap_for(lap)
|
||||
merged: list[tuple[int, int]] = []
|
||||
for z in zones:
|
||||
if merged and (dist[z[0]] - dist[merged[-1][1]]) < MERGE_GAP_M:
|
||||
if merged and (dist[z[0]] - dist[merged[-1][1]]) < merge_gap:
|
||||
merged[-1] = (merged[-1][0], z[1])
|
||||
else:
|
||||
merged.append(list(z))
|
||||
|
|
|
|||
|
|
@ -1,9 +1,13 @@
|
|||
import os
|
||||
import dataclasses
|
||||
import pytest
|
||||
import telemetry.report_technique as rt
|
||||
import telemetry.report_setup as rs
|
||||
import telemetry.report_compare as rc
|
||||
import telemetry.report_race as rr
|
||||
import telemetry.report_profile as rp
|
||||
from telemetry.__main__ import is_fresh
|
||||
from telemetry.parser import Lap
|
||||
|
||||
|
||||
def _check(result, path):
|
||||
|
|
@ -57,6 +61,124 @@ def test_setup_output_contains_setup(sample_lap, tmp_path):
|
|||
assert "## Tyres" in text
|
||||
|
||||
|
||||
def test_dedup_laps_prefers_valid():
|
||||
invalid_fast = Lap(
|
||||
session={'laptime': '113.5'},
|
||||
track={'Lap': '8', 'Valid': 'false'},
|
||||
setup={},
|
||||
ch={},
|
||||
n=0,
|
||||
)
|
||||
valid_slow = Lap(
|
||||
session={'laptime': '115.2'},
|
||||
track={'Lap': '8', 'Valid': 'true'},
|
||||
setup={},
|
||||
ch={},
|
||||
n=0,
|
||||
)
|
||||
result = rr._dedup_laps([invalid_fast, valid_slow])
|
||||
assert len(result) == 1
|
||||
assert result[0].track['Valid'] == 'true'
|
||||
|
||||
|
||||
def _variant(lap, *, laptime=None, lap_no=None, fuel=None, tyre=None):
|
||||
session = dict(lap.session)
|
||||
track = dict(lap.track)
|
||||
if laptime is not None:
|
||||
session['laptime'] = str(laptime)
|
||||
if lap_no is not None:
|
||||
track['Lap'] = str(lap_no)
|
||||
if fuel is not None:
|
||||
track['Fuel at Start'] = str(fuel)
|
||||
if tyre is not None:
|
||||
track['Tyre'] = tyre
|
||||
return dataclasses.replace(lap, session=session, track=track)
|
||||
|
||||
|
||||
def test_include_prompt_toggle(sample_lap, tmp_path):
|
||||
with_p = str(tmp_path / "with.md")
|
||||
no_p = str(tmp_path / "no.md")
|
||||
rt.generate(sample_lap, with_p, 'ru', include_prompt=True)
|
||||
rt.generate(sample_lap, no_p, 'ru', include_prompt=False)
|
||||
a = open(with_p, encoding="utf-8").read()
|
||||
b = open(no_p, encoding="utf-8").read()
|
||||
assert len(b) < len(a)
|
||||
assert "## Trace" in b # data kept
|
||||
assert "Начинай анализ" not in b # prompt dropped
|
||||
assert "Начинай анализ" in a
|
||||
|
||||
|
||||
def test_compare_verdict_fuel_correction_range(sample_lap, tmp_path):
|
||||
lap_a = _variant(sample_lap, laptime=100.0, fuel=60.0)
|
||||
lap_b = _variant(sample_lap, laptime=99.0, fuel=40.0)
|
||||
out = str(tmp_path / "cmp.md")
|
||||
rc.generate([lap_a, lap_b], out, 'ru', include_prompt=False)
|
||||
text = open(out, encoding="utf-8").read()
|
||||
assert "## Verdict" in text
|
||||
# fuel benefit (60-40)*[0.03..0.05] = 0.6..1.0 ; residual = -1.0 - [1.0..0.6] = -2.0..-1.6
|
||||
assert "-2.000" in text and "-1.600" in text
|
||||
|
||||
|
||||
def test_profile_sections_and_pick(sample_lap, tmp_path):
|
||||
laps = [
|
||||
_variant(sample_lap, lap_no=1, laptime=80.0, tyre='C3'),
|
||||
_variant(sample_lap, lap_no=2, laptime=75.0, tyre='C3'),
|
||||
_variant(sample_lap, lap_no=3, laptime=78.0, tyre='C3'),
|
||||
]
|
||||
out = str(tmp_path / "prof.md")
|
||||
rp.generate(laps, out, 'ru', include_prompt=False)
|
||||
text = open(out, encoding="utf-8").read()
|
||||
assert "Тенденции по поворотам" in text
|
||||
assert "Рекомендованные круги" in text
|
||||
assert "L2" in text # fastest valid lap picked
|
||||
|
||||
|
||||
def test_profile_covers_multiple_compounds(sample_lap, tmp_path):
|
||||
laps = (
|
||||
[_variant(sample_lap, lap_no=i, laptime=80.0, tyre='C3') for i in range(1, 4)]
|
||||
+ [_variant(sample_lap, lap_no=i, laptime=82.0, tyre='C2') for i in range(4, 7)]
|
||||
)
|
||||
out = str(tmp_path / "prof_multi.md")
|
||||
rp.generate(laps, out, 'ru', include_prompt=False)
|
||||
text = open(out, encoding="utf-8").read()
|
||||
assert "состав C3" in text
|
||||
assert "состав C2" in text # secondary compound not ignored
|
||||
|
||||
|
||||
def test_merge_gap_scales_with_track_length(sample_lap):
|
||||
from telemetry.segments import merge_gap_for, MERGE_GAP_M, MERGE_GAP_MIN, MERGE_GAP_MAX
|
||||
# sample lap is Zandvoort (~4257 m) → anchor, unchanged
|
||||
assert abs(merge_gap_for(sample_lap) - MERGE_GAP_M) < 1.0
|
||||
# tight circuit clamps low, long circuit clamps high
|
||||
short = _variant(sample_lap)
|
||||
short.track['Tracklen'] = '1500'
|
||||
assert merge_gap_for(short) == MERGE_GAP_MIN
|
||||
long = _variant(sample_lap)
|
||||
long.track['Tracklen'] = '9000'
|
||||
assert merge_gap_for(long) == MERGE_GAP_MAX
|
||||
|
||||
|
||||
def test_dedup_keeps_laps_without_number():
|
||||
# export missing the 'Lap' field must NOT collapse the field to one lap
|
||||
laps = [
|
||||
Lap(session={'laptime': str(80 + i)}, track={'Tyre': 'C3'}, setup={}, ch={}, n=0)
|
||||
for i in range(5)
|
||||
]
|
||||
result = rr._dedup_laps(laps)
|
||||
assert len(result) == 5
|
||||
|
||||
|
||||
def test_is_fresh_cache(tmp_path):
|
||||
src = tmp_path / "src.csv"
|
||||
rep = tmp_path / "rep.md"
|
||||
src.write_text("x")
|
||||
rep.write_text("y")
|
||||
os.utime(rep, (src.stat().st_atime + 10, src.stat().st_mtime + 10))
|
||||
assert is_fresh(str(rep), [str(src)]) is True
|
||||
os.utime(src, (rep.stat().st_atime + 20, rep.stat().st_mtime + 20))
|
||||
assert is_fresh(str(rep), [str(src)]) is False
|
||||
|
||||
|
||||
class TestRegressionReports:
|
||||
"""Regression: report generators must not crash on valid single-lap input."""
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue