"""CENTRAL COMPLEX — накопление улик и треки в мировой системе координат. Эллипсоидное тело мухи держит **кольцевой аттрактор**: клетки EPG образуют один «бугор» активности, кодирующий текущий курс, клетки PEN сдвигают этот бугор по сигналам собственного вращения, а взаимное торможение не даёт возникнуть второму бугру. Так муха помнит направление, даже когда ориентир пропал из виду. Здесь тот же механизм решает другую задачу. Кандидат на дальности 150 м — это пять-десять лучей, и по одному кадру он неотличим от шума. Но поезд едет, и объект, если он настоящий, остаётся **на одном и том же месте в тоннеле**, а не в поле зрения. Поэтому треки живут в координате «расстояние по пути от точки старта», сдвигаемой оценкой собственного движения (роль PEN), каждое совпадение подкачивает улику (локальное возбуждение), несовпадение — утечка, а конкуренция за одно и то же место не даёт плодить дубликаты (глобальное торможение APL/Δ7). Это и есть подтверждение по нескольким кадрам, которого требует ТЗ, — но не как эвристический фильтр, а как накопитель, который вытягивает слабый сигнал из шума. """ from __future__ import annotations import itertools from dataclasses import dataclass, field import numpy as np from .lobula import Candidate @dataclass class Track: """Подтверждаемая гипотеза о препятствии.""" id: int s_world: float # положение вдоль пути от начала записи, м u: float # смещение от оси пути, м h: float # высота над рельсом, м width: float height: float evidence: float = 0.0 # накопленная улика, 0…1 hits: int = 0 misses: int = 0 age: int = 0 first_d: float = 0.0 # на какой дальности впервые замечен last_d: float = 0.0 last_n_rays: int = 0 novelty: float = 1.0 speed_lat: float = 0.0 # поперечная скорость, м/с p_track: float = 0.0 # отсчёт считывания по треку, если оно подключено # То же смещение в системе лидара, без поправки на кривую. В решении не # участвует: по нему рамка рисуется у своих точек, а не у оси пути. u_raw: float = 0.0 history: list = field(default_factory=list) # Сводка по всем наблюдениям трека, накапливаемая на лету. Считать её из # `history` нельзя: описание трека нужно каждый кадр для всех живых # треков, а разворачивать ради этого полсотни списков в массивы — заметная # доля бюджета кадра. Все обновления здесь O(1). n_obs: int = 0 s_sum: float = 0.0 # мировая привязка, подразумеваемая наблюдением s_sq: float = 0.0 u_sum: float = 0.0 u_sq: float = 0.0 w_sum: float = 0.0 # вес наблюдения — ровно то, что идёт в улику w_max: float = 0.0 p_sum: float = 0.0 # отсчёт считывания по кандидату, если оно есть p_max: float = 0.0 p_n: int = 0 # сколько раз он вообще был w_ema: float = 0.0 # он же, но со скользящим забыванием r_sum: float = 0.0 # лучи, нормированные на ожидаемые для дальности r_last: float = 0.0 ri_sum: float = 0.0 # суммы для наклона нормированных лучей по времени i_sum: float = 0.0 i_sq: float = 0.0 @property def confirmed(self) -> bool: return self.evidence >= 0.5 @property def w_mean(self) -> float: """Средний вес наблюдения за жизнь трека. Улика — это тот же вес, но просуммированный и обрезанный единицей, и обрезание её убивает: на реальном объекте медиана улики 1.000 и у предмета, и у ложных треков (AUC 0.624), а у среднего веса 0.998 против 0.269 (AUC 1.000). Сумма отвечает «давно ли я на это смотрю», среднее — «на что именно я смотрю». EXPERIMENTS п. 15. """ return self.w_sum / max(self.n_obs, 1) @property def p_mean(self) -> float: """Средний покадровый отсчёт считывания MBON за жизнь трека. Без переданной конвейеру модели отсчёта нет вовсе, и возвращать тут ноль значило бы тихо погасить все треки разом. Поэтому в таком случае отдаётся средний вес наблюдения — он считается всегда. """ return self.p_sum / self.p_n if self.p_n else self.w_mean def distance(self, s_now: float) -> float: return self.s_world - s_now def observe(self, c: Candidate, s_obs: float, w: float) -> None: """Записать наблюдение в сводку. Полезное здесь — `w_sum` и `p_sum`: улика суммируется и упирается в потолок, а среднее того же веса различает предмет и обстановку и после насыщения (EXPERIMENTS п. 15.1). `s_obs` — место в тоннеле, которое подразумевает наблюдение: путь плюс дальность. Копится ТОЛЬКО ради замеров: на полигоне его разброс выглядит сильным признаком, а на настоящем объекте он перевёрнут, потому что вставка стоит на дальности, посчитанной из той же оценки движения (п. 15.4). В решении не участвует и участвовать не должен. """ i = float(self.n_obs) self.n_obs += 1 self.s_sum += s_obs self.s_sq += s_obs * s_obs self.u_sum += c.u self.u_sq += c.u * c.u self.w_sum += w self.w_max = max(self.w_max, w) # Среднее за всю жизнь отстаёт у СБЛИЖАЮЩЕГОСЯ предмета: он # появляется далеко и слабым, и первые десятки наблюдений тянут # среднее вниз ещё долго после того, как он стал отчётливым. # Скользящее забывание (полупериод около трёх наблюдений) отвечает # на вопрос «каков он сейчас», а не «каким был с самого начала». self.w_ema = w if self.n_obs == 1 else 0.8 * self.w_ema + 0.2 * w p = c.extra.get("mbon") if c.extra else None if p is not None: self.p_n += 1 self.p_sum += float(p) self.p_max = max(self.p_max, float(p)) r = c.n_rays / expected_rays(c.d) self.r_sum += r self.r_last = r self.ri_sum += i * r self.i_sum += i self.i_sq += i * i class CentralComplex: """Накопитель улик и менеджер треков.""" def __init__(self, *, gate_d: float = 4.0, gate_u: float = 1.2, gain: float = 0.34, leak: float = 0.12, leak_far: float | None = None, leak_far_from: float = 90.0, inhibition: float = 0.05, max_misses: int = 12, max_tracks: int = 48, use_shape: bool = True, mbon_power: float = 1.0, mbon_blend: float = 1.0, nov_fade_from: float = 0.0, nov_fade_to: float = 160.0, lying_exempt: bool = False, top_from: float = float("inf"), flat: tuple[float, float, float, float] = (0.0, 0.0, 0.0, 1.0), hover: tuple[float, ...] = (0.0, 0.6, 0.9, 1.2, 80.0, 1.5, 0.0)): self.use_shape = use_shape self.lying_exempt = lying_exempt self.top_from = top_from # низ выше этого — предмет в верхней секции габарита self.flat = flat # (высота, верх, дальность, множитель), см. _quality self.hover = hover # (пол, низ от, доля в габарите, |u| до, дальность до, протяжённость до, ширина от), см. _quality self.mbon_power = mbon_power self.mbon_blend = mbon_blend self.nov_fade_from = nov_fade_from self.nov_fade_to = nov_fade_to self.gate_d = gate_d self.gate_u = gate_u self.gain = gain self.leak = leak # Утечка улики за промах, слабеющая с дальностью. Промах на 150 м # — не свидетельство отсутствия, а свойство решётки: предмет там # даёт единицы лучей и виден через кадр. Ровно по этой причине # число лучей уже нормируется на ожидаемое (см. _quality), и # промах нормировать надо так же. # # Считать легко: при доле попаданий p улика не убывает, только # если p·gain·w > (1-p)·leak. На 120…160 м кандидат появляется в # 62 % наблюдений, и при gain 0.34 и leak 0.12 нужно качество # выше 0.22 — столько далёкое наблюдение не даёт никогда. Поэтому # улика у далёкого трека не «чуть ниже порога», а около нуля, и # снижением порога это не лечится (замер: 0.5 → 0.3 не изменил # ничего). None — зависимости нет, как было. self.leak_far = leak if leak_far is None else leak_far self.leak_far_from = leak_far_from self.inhibition = inhibition self.max_misses = max_misses self.max_tracks = max_tracks self.tracks: list[Track] = [] self.s_world = 0.0 self._ids = itertools.count(1) # ------------------------------------------------------------------ обновление def _leak_at(self, d: float) -> float: """Утечка на дальности d: линейно от leak до leak_far к 200 м.""" span = 200.0 - self.leak_far_from if span <= 0 or d <= self.leak_far_from: return self.leak f = min((d - self.leak_far_from) / span, 1.0) return self.leak + (self.leak_far - self.leak) * f def update(self, candidates: list[Candidate], ds: float, dt: float) -> list[Track]: """Сдвинуть мир на `ds`, сопоставить кандидатов, обновить улики.""" self.s_world += ds # допуск по дальности растёт с расстоянием: там и разрешение грубее, # и ошибка оценки собственного движения успевает накопиться for t in self.tracks: t.age += 1 used = set() for t in self.tracks: d_pred = t.distance(self.s_world) best, best_cost = None, None for k, c in enumerate(candidates): if k in used: continue gd = self.gate_d + 0.05 * max(c.d, 0.0) dd = abs(c.d - d_pred) du = abs(c.u - t.u) if dd > gd or du > self.gate_u + 0.4: continue cost = dd / gd + du / (self.gate_u + 0.4) if best_cost is None or cost < best_cost: best, best_cost = k, cost if best is None: t.misses += 1 t.evidence = max(0.0, t.evidence - self._leak_at(t.distance(self.s_world))) continue c = candidates[best] used.add(best) t.hits += 1 t.misses = 0 w = _quality(c, self.use_shape, self.mbon_power, self.mbon_blend, self.nov_fade_from, self.nov_fade_to, self.lying_exempt, self.top_from, self.flat, self.hover) t.observe(c, self.s_world + c.d, w) t.evidence = min(1.0, t.evidence + self.gain * w) if dt > 1e-3: t.speed_lat = 0.6 * t.speed_lat + 0.4 * (c.u - t.u) / dt # положение сглаживается: ближние наблюдения точнее дальних a = float(np.clip(0.5 * (60.0 / max(c.d, 20.0)), 0.15, 0.6)) t.s_world = (1 - a) * t.s_world + a * (self.s_world + c.d) t.u = (1 - a) * t.u + a * c.u t.u_raw = (1 - a) * t.u_raw + a * float(c.extra.get("u_raw", c.u)) t.h = (1 - a) * t.h + a * c.h t.width = max(t.width * 0.7, c.width) t.height = max(t.height * 0.7, c.height) t.last_d = c.d t.last_n_rays = c.n_rays t.novelty = 0.7 * t.novelty + 0.3 * c.novelty t.history.append((c.d, c.u, c.n_rays)) if len(t.history) > 64: del t.history[:-64] # новые гипотезы из несопоставленных кандидатов for k, c in enumerate(candidates): if k in used or len(self.tracks) >= self.max_tracks: continue t = Track(id=next(self._ids), s_world=self.s_world + c.d, u=c.u, h=c.h, width=c.width, height=c.height, first_d=c.d, last_d=c.d, last_n_rays=c.n_rays, novelty=c.novelty, u_raw=float(c.extra.get("u_raw", c.u))) w = _quality(c, self.use_shape, self.mbon_power, self.mbon_blend, self.nov_fade_from, self.nov_fade_to, self.lying_exempt, self.top_from, self.flat, self.hover) t.observe(c, self.s_world + c.d, w) t.evidence = self.gain * w t.hits = 1 self.tracks.append(t) # глобальное торможение: сильный трек подавляет соседей по месту self._inhibit() self.tracks = [t for t in self.tracks if t.misses <= self.max_misses and t.evidence > 0.02 and t.distance(self.s_world) > -5.0] self.tracks.sort(key=lambda t: -t.evidence) del self.tracks[self.max_tracks:] return self.tracks def _inhibit(self) -> None: if len(self.tracks) < 2 or self.inhibition <= 0: return order = sorted(self.tracks, key=lambda t: -t.evidence) for i, strong in enumerate(order): for weak in order[i + 1:]: if (abs(strong.s_world - weak.s_world) < self.gate_d and abs(strong.u - weak.u) < self.gate_u): weak.evidence = max(0.0, weak.evidence - self.inhibition) def confirmed(self) -> list[Track]: out = [t for t in self.tracks if t.confirmed and t.distance(self.s_world) > 0] out.sort(key=lambda t: t.distance(self.s_world)) return out def expected_rays(d: float) -> float: """Сколько лучей предмет даёт на дальности d просто по устройству решётки. Нужна в двух местах — в весе наблюдения и в описании трека, — и оба раза за одним и тем же: дальний предмет даёт мало лучей не потому, что он сомнительный, а потому, что так устроена развёртка. """ return max(3.0, 2500.0 / max(d, 5.0) ** 1.4) def _quality(c: Candidate, use_shape: bool = True, mbon_power: float = 1.0, mbon_blend: float = 1.0, nov_fade_from: float = 0.0, nov_fade_to: float = 160.0, lying_exempt: bool = False, top_from: float = float("inf"), flat: tuple[float, float, float, float] = (0.0, 0.0, 0.0, 1.0), hover: tuple[float, ...] = (0.0, 0.6, 0.9, 1.2, 80.0, 1.5, 0.0)) -> float: """Вес одного наблюдения: сколько улики оно добавляет. Дальний объект даёт мало лучей не потому, что он сомнительный, а потому что так устроена решётка, — поэтому число лучей нормируется на ожидаемое для этой дальности. Остальные множители отделяют предмет от конструкции тоннеля: предмет целиком помещается в габарит и компактен вдоль пути, а лоток или стена тянутся дальше и в стороны. Новизна из грибовидного тела входит сюда же множителем. """ # Висящее в верхней секции (`top_from`) — отдельная статья. Обе обученные # части, считывание MBON и память тоннеля, собирались до того, как у # габарита появилась верхняя секция, и таких форм не видели: их ответ там # не значит ничего, и оба канала здесь выключены. Нормировка числа лучей # рассчитана на предмет с человека, а свисающий кабель или стержень тонок # по природе: у стержня 5 см опора упиралась в нижний край 0.25 на любой # дальности. Пояс над вагоном в исправном тоннеле пуст, поэтому нижний # край для висящего поднят до 0.5. EXPERIMENTS п. 19. top = c.h_min >= top_from if c.extra and c.extra.get("small"): return _small_quality(c, nov_fade_from, nov_fade_to) support = float(np.clip(c.n_rays / expected_rays(c.d), 0.5 if top else 0.25, 1.0)) # Контраст к фону — главная улика вблизи и недоступная вдали. На 170 м # кольцо окружения ламины упирается в стену тоннеля, стоящую на той же # дальности, и `gap` структурно равен нулю: предмет не «ближе окружения», # он «на той же дальности, что окружение». Замер это подтверждает — # у вставленного человека на 140…185 м gap = 0.0 во ВСЕХ кадрах. # # Там, где контраст не измеряется, его заменяет опора веерного тела: # возвращались ли лучи из этой точки пути кадр за кадром. Берётся # максимум — вблизи решает контраст, вдали накопление. contrast = float(np.clip(c.gap / 3.0, 0.2, 1.0)) acc = float(c.extra.get("acc_support", 0.0)) if c.extra else 0.0 if acc > 0.0: contrast = max(contrast, float(np.clip(acc, 0.2, 1.0))) if use_shape: whole = float(np.clip((c.containment - 0.25) / 0.45, 0.05, 1.0)) span = max(c.width, c.height, 0.2) # Предмет или человек, лежащий в колее вдоль пути (depth ~ 1.8 м при малой высоте), # лежит прямо на полотне перед поездом — не штрафуем его как конструкцию тоннеля: is_lying_on_track = (lying_exempt and abs(c.u) <= 0.85 and c.h <= 0.40 and c.depth <= 2.2) if is_lying_on_track: compact = 1.0 else: compact = float(np.clip(1.5 - c.depth / (3.0 * span), 0.1, 1.0)) # опора снизу: упавший предмет, человек, камень стоят на полотне, а знак, # лоток или кронштейн висят на стене, и под ними пусто. Кроме верхней # секции габарита (`top_from`): у оси под сводом стен нет, и висящее # там — это то, что свисает в путь вагона. grounded = 1.0 if top else float(np.clip(1.25 - c.h_min / 1.2, 0.15, 1.0)) else: whole = compact = grounded = 1.0 # Знакомость на большой дальности не просто бесполезна, а вредна. Замер: # на 120…185 м у вставленного человека новизна 0.150, а у окружающей # обстановки 0.199, разделяющая способность AUC 0.293 — то есть # признак перевёрнут. Причина понятна: на шести лучах дескриптор # вырождается, и память узнаёт в предмете любую далёкую конструкцию. # Поэтому вклад новизны гасится с дальностью, а вблизи, где она и # работает, остаётся как был. 0 — не гасить (EXPERIMENTS п. 12.3). novel = 1.0 if top else novelty_gain(c.novelty) if not top and nov_fade_from > 0.0 and c.d > nov_fade_from: span = max(nov_fade_to - nov_fade_from, 1e-3) w = 1.0 - min((c.d - nov_fade_from) / span, 1.0) novel = novel ** w hand = support * contrast * whole * compact * grounded # Обученное считывание MBON, если оно есть. Модель видит те же признаки, # что и шесть множителей выше, плюс тень и интенсивность, которых в ручной # формуле нет вовсе. Знакомость в неё НЕ входит и остаётся отдельным # каналом: физику решает модель, конкретный тоннель — память. # # `mbon_blend` — геометрическое смешивание с ручной формулой: 1 — только # модель, 0 — только руками, между ними всё промежуточное. Нужно затем, # чтобы размен «модель против ручной формулы» мерился, а не объявлялся. p = c.extra.get("mbon") if c.extra else None # висящее в верхней секции модель не видела ни разу (обучена на стоящих # предметах) и гасит его — здесь решает ручная формула if p is not None and mbon_blend > 0.0 and not top: # Висящее посреди габарита модель тоже не видела: её учили на стоящих # предметах, и низ в метре над рельсом для неё — примета конструкции # на стене. Кубы 0.3 м синтетики организаторов на высоте 1.2 м она # гасит до 0.02–0.13 уже с 40–50 м, хотя кандидат целиком в габарите. # Пол вероятности для такого кандидата; 0 — выключено. EXPERIMENTS п. 24. # Только ближе `d_max`: вдали ось пути неточна, и в габарит заходят # предметы на стенах той же высоты — с правилом без предела они давали # ложные треки на 90–180 м. И только компактное вдоль пути: труба или # поручень у платформы на той же высоте тянется на метры (6.4 м), а у # висящего предмета протяжённость — десятки сантиметров. floor, h_from, cont_min, u_max, d_max, depth_max, width_min = hover if (floor > 0.0 and c.h_min >= h_from and c.containment >= cont_min and abs(c.u) <= u_max and c.d <= d_max and c.depth <= depth_max and c.width >= width_min): p = max(float(p), floor) pm = float(np.clip(p, 1e-4, 1.0)) ** max(mbon_power, 1e-3) b = float(np.clip(mbon_blend, 0.0, 1.0)) hand = pm ** b * max(hand, 1e-4) ** (1.0 - b) # Плоское у самого пола: высотой в сантиметры, целиком ниже `flat[1]` над # рельсом. Это пластины и кромки на полотне, края жёлоба, порог # гермозатвора — по ответу организаторов то, что в жёлобе, препятствием не # считается. Модель таким ставит 0.8–0.99: в её обучении был кабель на # путях. А ложных тревог это давало больше всего: порог гермозатвора на # 65→31 м и два фантома на синтетике. Настоящий низкий предмет (0.3 м на # рельсе, 2×0.2 м поперёк путей) вблизи даёт 0.1–0.4 м по высоте и верх # выше 0.4 м. Дальше `flat[2]` решётка слишком редкая, чтобы судить о # высоте, и там правило молчит. Множитель, а не запрет: если предмет хоть в # одном кадре покажет высоту, он своё доберёт. EXPERIMENTS п. 19. fh, ftop, fd, fw = flat if fh > 0.0 and c.d <= fd and c.height < fh and c.h_min + c.height < ftop: hand *= fw return float(np.clip(hand * novel, 0.0, 1.0)) def _small_quality(c: Candidate, nov_fade_from: float, nov_fade_to: float) -> float: """Вес наблюдения из канала малых целей (`lobula._small_targets`). Лобула уже потребовала от такого кандидата всё, что отличает висящий в пустоте предмет от конструкции: крошечная компонента, низ над полотном, ближе к оси, далёкий фон за ним. Поэтому из ручной формулы остаются опора (с тем же нижним краем 0.5, что у висящего сверху: предмет в два луча — не сомнительный, а маленький), контраст и целостность, а опора снизу и компактность — нет: первая наказала бы как раз висящее, вторая у двух лучей всегда единица. Обученное считывание MBON такой цели не судит: его выборка собрана при обычном пороге, и кандидатов из двух лучей в ней не было ни одного. Знакомость остаётся — память тоннеля по-прежнему гасит то, что на этом месте видела. """ support = float(np.clip(c.n_rays / expected_rays(c.d), 0.5, 1.0)) contrast = float(np.clip(c.gap / 3.0, 0.2, 1.0)) whole = float(np.clip((c.containment - 0.25) / 0.45, 0.05, 1.0)) novel = novelty_gain(c.novelty) if nov_fade_from > 0.0 and c.d > nov_fade_from: span = max(nov_fade_to - nov_fade_from, 1e-3) novel = novel ** (1.0 - min((c.d - nov_fade_from) / span, 1.0)) return float(np.clip(support * contrast * whole * novel, 0.0, 1.0)) def novelty_gain(novelty: float, lo: float = 0.15, hi: float = 0.50, floor: float = 0.05) -> float: """Ответ MBON → множитель улики. Пороги взяты по измеренному разделению (`tools/tune_memory.py`): знакомая обстановка даёт новизну около 0.14, реальный объект — около 0.50. Отображение делает разницу резкой, но оставляет ненулевой пол: даже похожий на привычную конструкцию предмет должен накапливать улику, просто медленнее. """ return float(np.clip((novelty - lo) / (hi - lo), floor, 1.0))