viktorиClaude Opus 4.8
cb4e8a9bb1
Панель: • шесть страниц с маршрутизацией по хэшу: Обзор, Last Heard, Хотспоты, Группы, Доступ, Настройки; • журнал приходит живым потоком (SSE) вместо опроса раз в две секунды; • Last Heard: кто, куда, когда, сколько, через какой хотспот; • в таблице хотспотов — ВСЁ, что устройство сообщает в RPTC. Контроль доступа (два независимых уровня, их легко перепутать): • хотспот пускается по паролю — у каждого может быть свой, плюс режим «только из списка» и отключение конкретного устройства; • абонент (поле src, владелец рации) допускается белым/чёрным списком с поддержкой диапазонов. Встроено в штатный хук guard ядра. Два бага по дороге: • поля RPTC читались как colorcode/latitude/longitude, а разборщик отдаёт color_code/lat/lon — половина таблицы молча пустовала; • столбец «описание» на деле содержит СТРАНУ: именно туда Pi-Star кладёт country. Переименован. tests/test_access.py 20/20, tests/test_panel.py 31/31 (включая наполнение Last Heard настоящим рукопожатием и голосовым вызовом по UDP). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XvX94Vwy7vXWEuiFm1g9Av
81 строка
3.6 KiB
Python
81 строка
3.6 KiB
Python
"""Last Heard — кто и когда выходил в эфир.
|
|
|
|
Строится по потоку кадров, а не по хуку вызовов ядра: тот не отдаёт номер
|
|
хотспота, а без него не ответить на вопрос «через какое устройство слышали».
|
|
|
|
Начало передачи определяется по смене stream_id (заголовки теряются в эфире),
|
|
конец — по терминатору ИЛИ по тишине: терминаторы теряются так же часто.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from .analyzer import CALL_GROUP
|
|
from .state import Call
|
|
|
|
# 360 мс = 6 пропущенных голосовых кадров. Столько же ждёт HBlink, прежде чем
|
|
# считать передачу законченной. Берём с запасом, чтобы не рвать вызов на джиттере.
|
|
SILENCE_TIMEOUT = 2.0
|
|
|
|
|
|
class LastHeard:
|
|
def __init__(self, state) -> None:
|
|
self.state = state
|
|
self._open: dict[tuple[int, int], Call] = {} # (peer_id, slot) -> вызов
|
|
self._sid: dict[tuple[int, int], bytes] = {} # (peer_id, slot) -> stream_id
|
|
self._seen: dict[tuple[int, int], float] = {} # (peer_id, slot) -> когда был кадр
|
|
|
|
def feed(self, peer, dmr) -> None:
|
|
key = (int(dmr.peer_id), int(dmr.slot))
|
|
sid = dmr.stream_id
|
|
now = time.time()
|
|
|
|
if self._sid.get(key) != sid: # новая передача
|
|
self._close(key, "новый поток")
|
|
call = Call(
|
|
stream_id=sid if isinstance(sid, int) else int.from_bytes(bytes(sid or b"\0"), "big"),
|
|
src_id=int(dmr.src or 0),
|
|
dst_id=int(dmr.dst or 0),
|
|
slot=int(dmr.slot),
|
|
group=(dmr.call_type == CALL_GROUP),
|
|
peer_id=int(dmr.peer_id),
|
|
)
|
|
self._open[key] = call
|
|
self._sid[key] = sid
|
|
with self.state._lock:
|
|
self.state.calls.append(call)
|
|
|
|
call = self._open.get(key)
|
|
if call is not None:
|
|
call.frames += 1
|
|
self._seen[key] = now
|
|
|
|
if getattr(dmr, "is_terminator", False):
|
|
self._close(key, "терминатор")
|
|
|
|
def sweep(self) -> None:
|
|
"""Закрыть передачи, по которым перестали приходить кадры.
|
|
|
|
Без этого зависший вызов вечно висел бы «активным»: терминатор теряется
|
|
в эфире регулярно, полагаться только на него нельзя.
|
|
"""
|
|
now = time.time()
|
|
for key, last in list(self._seen.items()):
|
|
if now - last > SILENCE_TIMEOUT:
|
|
self._close(key, "тишина")
|
|
|
|
def _close(self, key, why: str) -> None:
|
|
call = self._open.pop(key, None)
|
|
self._sid.pop(key, None)
|
|
self._seen.pop(key, None)
|
|
if call is None or call.ended:
|
|
return
|
|
call.ended = time.time()
|
|
if call.frames > 1: # одиночный кадр — не передача, а шум
|
|
kind = "группа TG %s" % call.dst_id if call.group else "приват %s" % call.dst_id
|
|
self.state.event(
|
|
"call",
|
|
f"эфир: {call.src_id} → {kind} · TS{call.slot} · "
|
|
f"{round(call.ended - call.started, 1)} с · хотспот {call.peer_id}",
|
|
)
|
© 2026 StackWorks