forked from Dan4ick/Lidar_Muxa
160 lines
7.4 KiB
Python
160 lines
7.4 KiB
Python
"""Кривые дальности обнаружения по размеру предмета.
|
||
|
||
Читает результат `tools/make_benchmark.py` и строит то, что требует ТЗ, п. 5:
|
||
с какой дистанции предмет уверенно виден и как это зависит от его размера.
|
||
|
||
Считаются две величины:
|
||
|
||
* **вероятность обнаружения в поясе дальности** — доля кадров, в которых
|
||
подтверждённый трек совпал с истинным положением предмета;
|
||
* **рабочая дальность** — самая дальняя точка, начиная с которой вероятность
|
||
устойчиво держится выше порога при сближении. Именно она отвечает на вопрос
|
||
«за сколько метров поезд увидел препятствие».
|
||
|
||
python tools/plot_benchmark.py --in artifacts/benchmark.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import matplotlib
|
||
matplotlib.use("Agg")
|
||
import matplotlib.pyplot as plt # noqa: E402
|
||
|
||
import numpy as np # noqa: E402
|
||
|
||
import _bootstrap as B # noqa: F401,E402
|
||
|
||
NL = chr(10)
|
||
|
||
EDGES = np.array([0, 15, 25, 40, 55, 70, 90, 110, 135, 160, 190, 230], float)
|
||
MIN_VISIBLE_RAYS = 2 # меньше — предмет физически не освещён лучами
|
||
|
||
|
||
def curve(d: np.ndarray, hit: np.ndarray):
|
||
"""Вероятность обнаружения по поясам дальности."""
|
||
idx = np.digitize(d, EDGES) - 1
|
||
p, n, centres = [], [], []
|
||
for b in range(len(EDGES) - 1):
|
||
m = idx == b
|
||
if m.sum() < 5:
|
||
continue
|
||
p.append(hit[m].mean())
|
||
n.append(int(m.sum()))
|
||
centres.append(0.5 * (EDGES[b] + EDGES[b + 1]))
|
||
return np.array(centres), np.array(p), np.array(n)
|
||
|
||
|
||
def working_range(centres: np.ndarray, p: np.ndarray, thr: float = 0.5) -> float:
|
||
"""Дальняя граница устойчивого обнаружения.
|
||
|
||
Идём от ближнего пояса к дальнему и останавливаемся там, где вероятность
|
||
впервые опускается ниже порога: дальше объект уже теряется.
|
||
"""
|
||
best = 0.0
|
||
for c, v in sorted(zip(centres, p)):
|
||
if v < thr:
|
||
break
|
||
best = c
|
||
return best
|
||
|
||
|
||
def _funnel(rec: list[dict], obj: str = "человек_стоя") -> None:
|
||
"""Где теряется предмет: лучи → кандидат → трек → решение."""
|
||
rows = [r for r in rec if r["obj"] == obj]
|
||
if not rows:
|
||
return
|
||
d = np.concatenate([np.array(r["d"]) for r in rows])
|
||
rays = np.concatenate([np.array(r["rays"]) for r in rows])
|
||
cand = np.concatenate([np.array(r["cand"]) for r in rows])
|
||
trk = np.concatenate([np.array(r["trk"]) for r in rows])
|
||
hit = np.concatenate([np.array(r["hit"]) for r in rows])
|
||
vis = rays >= MIN_VISIBLE_RAYS
|
||
print(f"{NL}воронка потерь для «{obj}», доля от ВИДИМЫХ наблюдений:")
|
||
print(f"{'полоса':>10}{'видимых':>9}{'кандидат':>10}{'трек':>8}{'решение':>9}{'нет кандидата':>15}{'трек без решения':>18}")
|
||
for lo, hi in ((6, 30), (30, 50), (50, 70), (70, 90), (90, 120), (120, 160)):
|
||
m = vis & (d >= lo) & (d < hi)
|
||
if m.sum() < 30:
|
||
continue
|
||
c, k, h = cand[m].mean(), trk[m].mean(), hit[m].mean()
|
||
print(f"{f'{lo}-{hi} м':>10}{int(m.sum()):9d}{c:10.2f}{k:8.2f}{h:9.2f}{1 - c:15.2f}{max(k - h, 0.0):18.2f}")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
ap.add_argument("--in", dest="path", default=str(B.ARTIFACTS / "benchmark.json"))
|
||
ap.add_argument("--thr", type=float, default=0.5)
|
||
ap.add_argument("--out", default="", help="куда сохранить рисунок")
|
||
args = ap.parse_args()
|
||
|
||
rec = json.load(open(args.path, encoding="utf-8"))
|
||
by_obj: dict[str, list] = {}
|
||
for r in rec:
|
||
by_obj.setdefault(r["obj"], []).append(r)
|
||
|
||
print(f"{'предмет':16s} {'площадь':>11s} | {'алгоритм':>8s} {'как есть':>9s} | "
|
||
f"{'P@50м':>6s} {'P@100м':>7s} {'P@150м':>7s} | {'видим':>7s} | {'набл.':>6s}")
|
||
print(f"{'':16s} {'':11s} | {'рабочая дальность, м':>18s} | "
|
||
f"{'при условии видимости':>22s} |")
|
||
print("-" * 104)
|
||
|
||
fig, axes = plt.subplots(1, 3, figsize=(19, 5.2), dpi=110)
|
||
for name, rows in sorted(by_obj.items(), key=lambda kv: -_size(kv[0])):
|
||
d = np.concatenate([np.array(r["d"]) for r in rows])
|
||
hit = np.concatenate([np.array(r["hit"]) for r in rows])
|
||
rays = np.concatenate([np.array(r["rays"]) for r in rows])
|
||
|
||
# видимость: в кривом тоннеле предмет за поворотом просто не освещён
|
||
# лучами, и «непопадание» там ничего не говорит об алгоритме
|
||
vis = rays >= MIN_VISIBLE_RAYS
|
||
cv, pv, _ = curve(d, vis.astype(float))
|
||
c, p, _ = curve(d, hit)
|
||
cc, pc_, _ = curve(d[vis], hit[vis]) if vis.any() else (np.array([]),) * 3
|
||
if c.size == 0:
|
||
continue
|
||
wr_alg = working_range(cc, pc_, args.thr) if cc.size else 0.0
|
||
wr_op = working_range(c, p, args.thr)
|
||
at = {x: float(np.interp(x, cc, pc_)) if cc.size else 0.0 for x in (50, 100, 150)}
|
||
print(f"{name:16s} {_size(name):8.2f} м² | {wr_alg:8.0f} {wr_op:9.0f} | "
|
||
f"{at[50]:6.2f} {at[100]:7.2f} {at[150]:7.2f} | {vis.mean():7.1%} | {d.size:6d}")
|
||
|
||
axes[0].plot(cv, pv, marker=".", lw=1.2, label=name)
|
||
if cc.size:
|
||
axes[1].plot(cc, pc_, marker="o", ms=3.5, lw=1.4, label=name)
|
||
axes[2].plot(c, p, marker="o", ms=3.5, lw=1.4, label=name)
|
||
|
||
axes[0].set_title(f"1. Виден ли предмет вообще\n(≥{MIN_VISIBLE_RAYS} лучей попало)")
|
||
axes[0].set_ylabel("доля кадров")
|
||
axes[1].set_title("2. Обнаружение, когда предмет виден\n(качество алгоритма)")
|
||
axes[1].set_ylabel("вероятность обнаружения")
|
||
axes[2].set_title("3. Обнаружение как есть\n(эксплуатационная величина)")
|
||
axes[2].set_ylabel("вероятность обнаружения")
|
||
for ax in axes:
|
||
ax.axhline(args.thr, color="k", ls="--", lw=0.8)
|
||
ax.set_xlabel("истинная дистанция до предмета, м")
|
||
ax.set_ylim(-0.03, 1.03); ax.grid(alpha=0.25)
|
||
ax.legend(fontsize=7, ncol=2)
|
||
|
||
B.FIGURES.mkdir(parents=True, exist_ok=True)
|
||
out = Path(args.out) if args.out else B.FIGURES / "detection_range.png"
|
||
fig.tight_layout(); fig.savefig(out)
|
||
print("\nсохранено:", out)
|
||
|
||
if all("cand" in r for r in rec):
|
||
_funnel(rec)
|
||
|
||
fp = np.concatenate([np.array(r["fp"]) for r in rec])
|
||
print(f"посторонних тревог: {fp.sum():.0f} на {fp.size} наблюдений "
|
||
f"({fp.mean():.3f} на кадр)")
|
||
|
||
|
||
def _size(name: str) -> float:
|
||
from flyguard.synth import catalogue
|
||
w, h = catalogue()[name].size
|
||
return w * h
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|