diff --git a/telemetry/__main__.py b/telemetry/__main__.py index c4f43df..5c41079 100644 --- a/telemetry/__main__.py +++ b/telemetry/__main__.py @@ -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__': diff --git a/telemetry/gui.py b/telemetry/gui.py index a6aae7b..eaebefe 100644 --- a/telemetry/gui.py +++ b/telemetry/gui.py @@ -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: diff --git a/telemetry/report_compare.py b/telemetry/report_compare.py index 38d9f48..a358091 100644 --- a/telemetry/report_compare.py +++ b/telemetry/report_compare.py @@ -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) diff --git a/telemetry/report_profile.py b/telemetry/report_profile.py new file mode 100644 index 0000000..e7e8979 --- /dev/null +++ b/telemetry/report_profile.py @@ -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) diff --git a/telemetry/report_race.py b/telemetry/report_race.py index 12adfad..a2aaa98 100644 --- a/telemetry/report_race.py +++ b/telemetry/report_race.py @@ -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) diff --git a/telemetry/report_setup.py b/telemetry/report_setup.py index 3f298b6..3212209 100644 --- a/telemetry/report_setup.py +++ b/telemetry/report_setup.py @@ -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) diff --git a/telemetry/report_technique.py b/telemetry/report_technique.py index ec5f09e..b6d6519 100644 --- a/telemetry/report_technique.py +++ b/telemetry/report_technique.py @@ -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) diff --git a/telemetry/segments.py b/telemetry/segments.py index ea7821c..7fcdaaf 100644 --- a/telemetry/segments.py +++ b/telemetry/segments.py @@ -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)) diff --git a/tests/test_reports.py b/tests/test_reports.py index 9b3ad81..30c30b5 100644 --- a/tests/test_reports.py +++ b/tests/test_reports.py @@ -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."""