"""DESCENDING NEURONS — решение. Весь разбор сцены у мухи сходится на нескольких десятках нисходящих нейронов, идущих в грудной ганглий. Два из них работают как раз по надвигающемуся объекту: * **Giant Fiber (DNp01)** — один толстый аксон с высоким порогом. Срабатывает только на близкое и быстрое надвигание и запускает немедленный аварийный взлёт, жертвуя устойчивостью ради скорости. * **DNp02/DNp11** — порог ниже, реакция раньше и мягче: муха успевает подготовиться, не срываясь в паническое движение. Поезду нужна ровно такая же пара уровней: заблаговременное предупреждение с запасом по дальности и экстренное торможение по надёжному близкому объекту. Гистерезис здесь — не украшение: без него трек на пороге даёт дребезг, а дребезжащая команда торможения хуже её отсутствия. """ from __future__ import annotations from dataclasses import dataclass, field from .central_complex import CentralComplex, Track BRAKING_DECEL = 1.0 # м/с², служебное торможение метропоезда (оценка) REACTION_TIME = 1.5 # с, задержка канала «решение → тормоз» @dataclass class DetectedObject: distance: float lateral: float height: float width: float size_v: float confidence: float novelty: float n_rays: int track_id: int ttc: float # Смещение в системе лидара: `lateral` отсчитан от оси пути, и в кривой # рамка по нему стоит в стороне от точек. NaN — неизвестно. sensor_x: float = float("nan") @dataclass class Decision: """Выход системы за один кадр.""" detected: bool = False emergency: bool = False distance: float = float("inf") ttc: float = float("inf") confidence: float = 0.0 stopping_distance: float = 0.0 objects: list[DetectedObject] = field(default_factory=list) speed: float = 0.0 @property def clear(self) -> bool: return not self.detected class DescendingNeurons: """Два порога с гистерезисом поверх подтверждённых треков.""" def __init__(self, *, warn_evidence: float = 0.5, clear_evidence: float = 0.3, emergency_evidence: float = 0.75, min_hits: int = 2, novelty_floor: float = 0.10, max_range: float = 200.0, warn_far: float | None = None, warn_far_from: float = 90.0, track_blend: float = 0.0, track_score: str = "w_mean", track_gate: float = 0.0, track_readout=None, far_channel: bool = False, far_confirm: int = 0, far_confirm_from: float = 100.0): self.warn_evidence = warn_evidence self.clear_evidence = clear_evidence self.emergency_evidence = emergency_evidence self.min_hits = min_hits self.novelty_floor = novelty_floor self.max_range = max_range # Порог предупреждения, ослабевающий с дальностью. Пара нисходящих # нейронов у мухи не равноправна: DNp02/DNp11 срабатывают рано и # мягко, Giant Fiber — поздно и жёстко. Здесь то же самое по # дальности: далёкая улика физически слабее (4…8 лучей вместо # сотни), а цена ошибки ниже — до предмета ещё сотня метров и # десятки кадров на подтверждение. Экстренное торможение своего # порога не снижает. None — послабления нет. self.warn_far = warn_evidence if warn_far is None else warn_far self.warn_far_from = warn_far_from # Считывание по ТРЕКУ (см. track_readout.py). Улика — это сумма # покадровых весов, и она ничего не знает ни о том, держится ли # место в тоннеле, ни о том, растёт ли число лучей по закону # решётки. Модель знает. Смешивание геометрическое, как и для # покадрового считывания: 0 — решает только улика, ровно как # раньше, 1 — только модель. Так размен меряется, а не объявляется. self.track_blend = track_blend # Жёсткий порог по тому же отсчёту — вместо смешивания или вместе с # ним. Разница существенная: смешивание опускает отсчёт И предмету # тоже (обе величины меньше единицы), а порог ничего не отнимает у # того, кто его прошёл. На реальном объекте средний вес 0.998 у # предмета против 0.269 у ложных треков — разделение, под которое # порог и просится. 0 — не проверять. self.track_gate = track_gate # Дальний канал предупреждения (Zhirik1337, 22.09): далёкий трек с # хорошими наблюдениями поднимает тревогу в обход порога, даже если # улика просела на пропусках. Та же задача, что у `warn_far`, другим # способом; включать вместе с ним — ослабить дальний край дважды. # Замер — EXPERIMENTS п. 16. self.far_channel = far_channel # Дальний трек поднимает тревогу, только прожив `far_confirm` кадров. # Вдали предмет — три-четыре луча, и по одному кадру он неотличим от # вспышки на кромке или стыке; фантомы на синтетике и в записях жили # 0.5–1 с и пропадали при подъезде. Торможению дальний трек не нужен # срочно: до него сотня метров. 0 — выключено. EXPERIMENTS п. 19. self.far_confirm = far_confirm self.far_confirm_from = far_confirm_from self.track_score = track_score self.track_readout = track_readout self._latched: set[int] = set() @property def _hyst(self) -> float: return self.clear_evidence / max(self.warn_evidence, 1e-6) def _warn_at(self, d: float) -> float: """Порог тревоги на дальности d: линейно от warn до warn_far.""" span = self.max_range - self.warn_far_from if span <= 0 or d <= self.warn_far_from: return self.warn_evidence f = min((d - self.warn_far_from) / span, 1.0) return self.warn_evidence + (self.warn_far - self.warn_evidence) * f def decide(self, cx: CentralComplex, speed: float) -> Decision: out = Decision(speed=speed) stop = speed * REACTION_TIME + speed * speed / (2 * BRAKING_DECEL) out.stopping_distance = stop # Дешёвые проверки сначала: модель считается только по тем трекам, # которые их прошли, — обычно это единицы, а не полсотни. ready: list[tuple[Track, float]] = [] for t in cx.tracks: d = t.distance(cx.s_world) far_floor = 0.05 if self.far_channel else self.novelty_floor nov_ok = (t.novelty >= self.novelty_floor if d <= self.warn_far_from else t.novelty >= far_floor) if (not (0.0 < d <= self.max_range) or t.hits < self.min_hits or not nov_ok): self._latched.discard(t.id) continue if (self.far_confirm > 0 and d > self.far_confirm_from and t.age < self.far_confirm and t.id not in self._latched): continue ready.append((t, d)) use_track = ((self.track_blend > 0.0 or self.track_gate > 0.0) and bool(ready)) if use_track and self.track_score == "model": if self.track_readout is None: use_track = False else: self.track_readout.annotate_tracks([t for t, _ in ready], cx.s_world) live: list[tuple[Track, float]] = [] for t, d in ready: warn = self._warn_at(d) # гистерезис: попавший в тревогу трек держится до нижнего порога on = warn if t.id not in self._latched else warn * self._hyst score = t.evidence if use_track: q = (t.p_track if self.track_score == "model" else t.p_mean if self.track_score == "p_mean" else t.w_ema if self.track_score == "w_ema" else t.w_mean) if self.track_gate > 0.0 and q < self.track_gate: self._latched.discard(t.id) continue if self.track_blend > 0.0: b = min(max(self.track_blend, 0.0), 1.0) score = (max(q, 1e-4) ** b * max(t.evidence, 1e-4) ** (1.0 - b)) # Дальний канал раннего предупреждения (DNp02/DNp11): # На дальности d > warn_far_from при наличии подтверждений (hits >= min_hits) # трек с устойчивыми признаками (p_mean, w_mean, w_ema или p_track) # квалифицируется как предупреждение, даже если улика просела из-за пропусков. is_far = (self.far_channel and d > self.warn_far_from and t.hits >= self.min_hits) quality = max(t.p_mean, t.w_mean, t.w_ema) far_qualified = is_far and ( (quality >= 0.40 and t.evidence >= 0.15) or (getattr(t, "p_track", 0.0) >= 0.65) ) if score < on and not far_qualified: self._latched.discard(t.id) continue self._latched.add(t.id) live.append((t, d)) if not live: return out live.sort(key=lambda p: p[1]) for t, d in live: ttc = d / speed if speed > 0.5 else float("inf") out.objects.append(DetectedObject( distance=d, lateral=t.u, height=t.h, width=t.width, size_v=t.height, confidence=min(1.0, t.evidence * t.novelty + 0.0), novelty=t.novelty, n_rays=t.last_n_rays, track_id=t.id, ttc=ttc, sensor_x=t.u_raw)) nearest, d0 = live[0] out.detected = True out.distance = d0 out.ttc = d0 / speed if speed > 0.5 else float("inf") out.confidence = max(o.confidence for o in out.objects) out.emergency = any( t.evidence >= self.emergency_evidence and (d <= max(stop, 25.0)) for t, d in live) return out