90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
"""Прогон геометрической части конвейера: сколько кандидатов и где.
|
||
|
||
Это опорный эксперимент: на `doubleT_obstacle` обязан стабильно находиться
|
||
реальный объект на ~55 м, на пустых бэгах — считается поток ложных кандидатов.
|
||
|
||
python tools/probe_detect.py --bag data/for_hackathon/doubleT_obstacle --frames 40 --verbose
|
||
python tools/probe_detect.py --all --frames 60
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import time
|
||
|
||
import numpy as np
|
||
|
||
import _bootstrap as B # noqa: F401
|
||
from flyguard import lamina
|
||
from flyguard.bag import Bag, find_bags
|
||
from flyguard.geometry import TrackFrame, fit_corridor, fit_rail_plane
|
||
from flyguard.lobula import find_candidates
|
||
from flyguard.retina import ScanLayout
|
||
|
||
|
||
def run(bag_path, n_frames: int, stride: int, fov: float, verbose: bool,
|
||
**kw) -> dict:
|
||
bag = Bag(bag_path)
|
||
layout = ScanLayout.calibrate([pc for _, pc in bag.frames(start=2, stop=14)])
|
||
cols = layout.column_slice(fov)
|
||
lay = layout.sub(cols)
|
||
|
||
plane = corridor = None
|
||
per_frame, times = [], []
|
||
hits = []
|
||
for k, (_, pc) in enumerate(bag.frames(stop=n_frames * stride, stride=stride)):
|
||
t0 = time.perf_counter()
|
||
img = layout.project(pc).crop(cols)
|
||
plane = fit_rail_plane(img, lay, prev=plane)
|
||
tf = TrackFrame(img, lay, plane)
|
||
corridor = fit_corridor(tf, prev=corridor)
|
||
lam = lamina.process(tf.r, tf.valid)
|
||
cands = find_candidates(tf, lam, corridor, **kw)
|
||
times.append(time.perf_counter() - t0)
|
||
|
||
per_frame.append(len(cands))
|
||
hits.extend(cands)
|
||
if verbose:
|
||
head = " ".join(
|
||
f"[{c.d:6.1f}м u{c.u:+5.2f} h{c.h:4.2f} {c.width:4.2f}×{c.height:4.2f}м "
|
||
f"n={c.n_rays:4d} gap={c.gap:6.1f} def={c.floor_deficit:6.1f}]"
|
||
for c in cands[:4])
|
||
print(f" кадр {k*stride:4d}: {len(cands):3d} канд. {head}")
|
||
|
||
n = np.array(per_frame)
|
||
ds = np.array([c.d for c in hits]) if hits else np.zeros(0)
|
||
res = dict(name=bag.path.name, frames=len(n),
|
||
mean=float(n.mean()) if n.size else 0.0,
|
||
median=float(np.median(n)) if n.size else 0.0,
|
||
p95=float(np.percentile(n, 95)) if n.size else 0.0,
|
||
zero=float(np.mean(n == 0)) if n.size else 0.0,
|
||
total=int(n.sum()),
|
||
t_ms=float(np.median(times) * 1e3),
|
||
far=int((ds > 60).sum()))
|
||
print(f"{res['name']:40s} кадров {res['frames']:4d} | кандидатов/кадр "
|
||
f"среднее {res['mean']:6.2f} медиана {res['median']:4.0f} p95 {res['p95']:5.0f} | "
|
||
f"пустых кадров {res['zero']:5.1%} | дальше 60 м {res['far']:5d} | "
|
||
f"{res['t_ms']:6.1f} мс/кадр")
|
||
return res
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
ap.add_argument("--bag")
|
||
ap.add_argument("--all", action="store_true")
|
||
ap.add_argument("--frames", type=int, default=50)
|
||
ap.add_argument("--stride", type=int, default=2)
|
||
ap.add_argument("--fov", type=float, default=25.0)
|
||
ap.add_argument("--verbose", action="store_true")
|
||
ap.add_argument("--h-lo", type=float, default=0.28)
|
||
ap.add_argument("--min-rays", type=int, default=4)
|
||
ap.add_argument("--half-width", type=float, default=1.7)
|
||
args = ap.parse_args()
|
||
|
||
kw = dict(h_lo=args.h_lo, min_rays=args.min_rays, half_width=args.half_width)
|
||
bags = find_bags(B.DATA / "for_hackathon") if args.all else [args.bag]
|
||
for b in bags:
|
||
run(b, args.frames, args.stride, args.fov, args.verbose, **kw)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|