Brainrot_Muxa/tools/diagnose_fp.py
Данил Омелечко b8e95eccbe ML-ядро детектора: конвейер на схемах мозга дрозофилы
Ретина, ламина, медулла, лобула, грибовидное тело, веерное тело,
центральный комплекс, нисходящие нейроны. Обучение памяти тоннеля и
считывания MBON, оценка leave-one-bag-out, полигон дальности, 24 теста.

Реальный объект на 55 м — 98.9 % кадров, ложных 7.5 трека на км,
кадр обрабатывается за 33 мс на CPU.
2026-09-21 17:24:25 +03:00

108 lines
5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Разбор ложных тревог на одном бэге: кто именно сработал и почему.
Память обучается **без** проверяемого бэга (как в evaluate.py), затем конвейер
гоняется по нему и каждая подтверждённая цель записывается вместе с признаками
породившего её кандидата. На выходе — сводка по трекам: где стоял, сколько
кадров жил, какие у него размеры, новизна и наполненность габарита.
python tools/diagnose_fp.py --bag roundT_doubleT --device cuda
"""
from __future__ import annotations
import argparse
from collections import defaultdict
import numpy as np
import _bootstrap as B # noqa: F401
from flyguard.bag import Bag, find_bags
from flyguard.pipeline import FlyGuard, Params
from evaluate import OBSTACLE_BAG, TRUE_D, train_excluding
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--root", default=str(B.DATA / "for_hackathon"))
ap.add_argument("--cache", default=str(B.CACHE / "tune_candidates.npz"))
ap.add_argument("--extra-cache", default=str(B.CACHE / "new_data_candidates.npz"))
ap.add_argument("--bag", default="roundT_doubleT")
ap.add_argument("--limit", type=int, default=250)
ap.add_argument("--device", default="cpu")
args = ap.parse_args()
from pathlib import Path
d = np.load(args.cache, allow_pickle=True)
per_bag = {str(k): d[f"X_{k}"].astype(np.float32) for k in d["names"]}
extra = (np.load(args.extra_cache)["X"].astype(np.float32)
if Path(args.extra_cache).exists() else None)
bag_path = next(p for p in find_bags(args.root) if p.name == args.bag)
memory = train_excluding(per_bag, extra, args.bag, 0.4, args.device)
fg = FlyGuard(Params(), memory=memory)
bag = Bag(bag_path)
is_obs = args.bag == OBSTACLE_BAG
per_track = defaultdict(list)
s_world = 0.0
n = 0
for k, (_, pc) in enumerate(bag.frames(stop=args.limit)):
res = fg.process(pc)
if res is None:
continue
n += 1
s_world += res.ego.ds if res.ego else 0.0
for o in res.decision.objects:
if is_obs and TRUE_D[0] < o.distance < TRUE_D[1]:
continue
# найти кандидата, породивший эту цель — ближайший по дальности
best, err = None, 1e9
for c in res.candidates:
e = abs(c.d - o.distance)
if e < err:
best, err = c, e
per_track[o.track_id].append(dict(
frame=k, s=s_world, d=o.distance, u=o.lateral, h=o.height,
w=o.width, sv=o.size_v, conf=o.confidence, nov=o.novelty,
rays=o.n_rays,
cd=best.d if best else np.nan,
cu=best.u if best else np.nan,
ch=best.h if best else np.nan,
chmin=best.h_min if best else np.nan,
cdepth=best.depth if best else np.nan,
ccont=best.containment if best else np.nan,
cgap=best.gap if best else np.nan,
caz=best.az_deg if best else np.nan,
cel=best.el_deg if best else np.nan,
cnov=best.novelty if best else np.nan,
cinten=best.inten if best else np.nan,
crings=best.n_rings if best else np.nan,
ccols=best.n_cols if best else np.nan,
cfloor=best.floor_deficit if best else np.nan,
cshadow=best.shadow if best else np.nan,
))
print(f"бэг {args.bag}: {n} кадров, путь {s_world:.0f} м, "
f"ложных треков {len(per_track)}\n")
hdr = (f"{'трек':>5s} {'кадров':>6s} {'кадры':>9s} {'d,м':>10s} {'u,м':>7s} "
f"{'h,м':>6s} {'hmin':>6s} {'shxsv':>11s} {'глуб':>5s} {'напол':>6s} "
f"{'зазор':>6s} {'лучей':>6s} {'кольц':>5s} {'стлб':>5s} "
f"{'нов':>5s} {'инт':>5s} {'az°':>7s} {'el°':>6s} {'пол.деф':>7s}")
print(hdr)
print("-" * len(hdr))
rows = sorted(per_track.items(), key=lambda kv: -len(kv[1]))
for tid, recs in rows:
a = {k: np.array([r[k] for r in recs], float) for k in recs[0]}
med = lambda k: float(np.nanmedian(a[k]))
print(f"{tid:5d} {len(recs):6d} {int(a['frame'][0]):4d}-{int(a['frame'][-1]):<4d} "
f"{med('d'):5.1f}→{a['d'][-1]:4.1f} {med('u'):7.2f} "
f"{med('h'):6.2f} {med('chmin'):6.2f} "
f"{med('w'):5.2f}x{med('sv'):<5.2f} {med('cdepth'):5.2f} "
f"{med('ccont'):6.2f} {med('cgap'):6.2f} {med('rays'):6.0f} "
f"{med('crings'):5.0f} {med('ccols'):5.0f} "
f"{med('nov'):5.2f} {med('cinten'):5.0f} "
f"{med('caz'):7.2f} {med('cel'):6.2f} {med('cfloor'):7.2f}")
if __name__ == "__main__":
main()