dmr-server/dmrsrv/parrot.py

"""DMRhub v2 · Паррот (эхотест) на TG 9990.

Записывает групповой вызов на 9990 и после отпускания PTT проигрывает его ОБРАТНО —
абонент слышит себя («слышно ли в эфире»).

Ключ: проигрываем ЗАПИСАННЫЕ кадры ДОСЛОВНО (payload [20:53] как есть), только с новым
stream_id. У кадров рации уже корректные sync/EMB/slot-type/AMBE — ничего не пересобираем
(реконструкция фрейминга давала «секундное молчание»: рация видела вызов, но не декодировала
голос). Это бит-в-бит повтор принятого — гарантированно декодируется.

Доставка — НАПРЯМУЮ в хотспот вызвавшего (master._send на его addr), минуя роутер/гвард:
  • не идёт через _route_dmr → не попадает обратно в feed (петли исключены by design);
  • не портит last-heard рации и не триггерит анти-эхо мостов;
  • слышит только тестирующий. Темп 60мс/кадр (как реальный слот TDMA).
"""
from __future__ import annotations

import asyncio
import os
import random

from .analyzer import DMR, FrameType, CALL_GROUP, CALL_UNIT, build_dmrd
from .server import PeerState
from .log import get_logger
from . import dmr_voice

log = get_logger(__name__)

# ДИАГНОСТИКА (env DMRHUB_PARROT_RECON): изоляция причины «тишины» v1. Дефолт пусто —
# дословный повтор (рабочий). Режимы:
#   "voice"  — пересобрать payload голос-бёрстов (voice_burst), заголовок/терм дословно;
#   "header" — пересобрать заголовок/терминатор (voice_lc_header/terminator), голос дословно.
RECON = os.getenv("DMRHUB_PARROT_RECON", "")
RECON_VOICE = RECON in ("1", "voice")
RECON_HEADER = RECON == "header"

PARROT_TG = 9990
PARROT_SLOT = 2                    # слот эхотеста (переопределяется из конфига)
PARROT_PEER = 0x0F990000          # peer_id в собранных DMRD (для direct-send не критичен)
FRAME_S = 0.06                     # 60мс на кадр (шаг слота TDMA)
GAP_S = 1.0                        # пауза после отпускания PTT перед проигрыванием
MAX_FRAMES = 4000                  # предохранитель (~4 мин) от бесконечной записи


class _Rec:
    __slots__ = ("caller_peer", "src", "frames", "group")

    def __init__(self, caller_peer: int, src: int, group: bool = True):
        self.caller_peer = caller_peer
        self.src = src
        self.group = group                 # True=групповой эхотест на 9990, False=приватный
        self.frames: list[tuple] = []      # (frame_type, data_type, payload33) в порядке приёма


class Parrot:
    """Эхотест TG9990. `feed(dmr)` цепляется на on_dmrd мастера (как P25-мост)."""

    def __init__(self, master):
        self.master = master
        self._rec: dict[tuple, _Rec] = {}      # (peer_id, stream_id) → запись

    def feed(self, dmr) -> None:
        if dmr.peer_id == PARROT_PEER:                 # свой replay не писать (страховка)
            return
        if dmr.dst != PARROT_TG or dmr.slot != PARROT_SLOT:
            return
        if dmr.call_type not in (CALL_GROUP, CALL_UNIT):   # групповой И приватный эхотест
            return
        is_group = dmr.call_type == CALL_GROUP
        key = (dmr.peer_id, dmr.stream_id)
        rec = self._rec.get(key)
        if rec is None:
            if dmr.is_terminator:                      # хвост без записи — игнор
                return
            rec = self._rec[key] = _Rec(dmr.peer_id, dmr.src, is_group)

        payload = bytes(dmr.raw[20:53])
        if len(payload) >= 33 and len(rec.frames) < MAX_FRAMES:
            rec.frames.append((dmr.frame_type, int(dmr.data_type), payload[:33]))

        if dmr.is_terminator:
            self._rec.pop(key, None)
            if any(ft in (FrameType.VOICE, FrameType.VOICE_SYNC) for ft, _, _ in rec.frames):
                self._schedule(rec)                    # проигрываем, только если был голос


    def _schedule(self, rec: _Rec) -> None:
        try:
            asyncio.ensure_future(self._replay(rec))
        except RuntimeError:
            log.warning("паррот: нет event-loop для проигрывания")

    # ── сборка кадров эха из записанных payload (тестируемо отдельно от пейсинга) ──
    def build_frames(self, rec: _Rec, sid: int) -> list[DMR]:
        """Пересобрать DMR-кадры эха. Группа — дословно (тот же payload, новый stream_id).
        Приват — обратным адресом (src=9990 → dst=вызывающий, unit): дословный повтор не
        годится, там dst=9990, рация не примет как «мне»."""
        from analyzer import DataType
        out: list[DMR] = []
        if not rec.group:                              # ── приватный эхотест ──
            dst, src = rec.src, PARROT_TG              # эхо приватом ОБРАТНО вызывающему
            for i, (ft, dt, payload33) in enumerate(rec.frames):
                if ft in (FrameType.VOICE, FrameType.VOICE_SYNC):
                    pl = dmr_voice.voice_burst(dmr_voice.ambe_from_burst(payload33),
                                               dt, dst, src, unit=True)
                elif ft == FrameType.DATA_SYNC and dt == DataType.VOICE_LC_HEADER:
                    pl = dmr_voice.voice_lc_header(dst, src, unit=True)
                elif ft == FrameType.DATA_SYNC and dt == DataType.TERMINATOR_LC:
                    pl = dmr_voice.terminator(dst, src, unit=True)
                else:
                    pl = payload33
                d = DMR(raw=b"", seq=i & 0xFF, src=src, dst=dst, peer_id=PARROT_PEER,
                        slot=PARROT_SLOT, call_type=CALL_UNIT, frame_type=ft,
                        data_type=dt, stream_id=sid)
                d.raw = build_dmrd(d, pl)
                out.append(d)
            return out
        for i, (ft, dt, payload33) in enumerate(rec.frames):
            pl = payload33
            is_voice = ft in (FrameType.VOICE, FrameType.VOICE_SYNC)
            if RECON_VOICE and is_voice:                # пересобрать голос-бёрст (dt = vseq 0..5)
                pl = dmr_voice.voice_burst(dmr_voice.ambe_from_burst(payload33),
                                           dt, PARROT_TG, rec.src)
            elif RECON_HEADER and ft == FrameType.DATA_SYNC and dt == DataType.VOICE_LC_HEADER:
                pl = dmr_voice.voice_lc_header(PARROT_TG, rec.src)
            elif RECON_HEADER and ft == FrameType.DATA_SYNC and dt == DataType.TERMINATOR_LC:
                pl = dmr_voice.terminator(PARROT_TG, rec.src)
            d = DMR(raw=b"", seq=i & 0xFF, src=rec.src, dst=PARROT_TG, peer_id=PARROT_PEER,
                    slot=PARROT_SLOT, call_type=CALL_GROUP, frame_type=ft,
                    data_type=dt, stream_id=sid)
            d.raw = build_dmrd(d, pl)
            out.append(d)
        return out

    async def _replay(self, rec: _Rec) -> None:
        peer = self.master.peers.get(rec.caller_peer)
        if peer is None or peer.state != PeerState.CONNECTED or not self.master._send:
            return
        await asyncio.sleep(GAP_S)                      # дать рации перейти на приём
        frames = self.build_frames(rec, random.getrandbits(32))
        log.info("паррот: эхо src=%s → хотспот %s, %d кадров",
                 rec.src, rec.caller_peer, len(frames))
        for i, d in enumerate(frames):
            try:
                self.master._send(d.raw, peer.addr)
            except Exception:
                log.exception("паррот: ошибка отправки эха")
                return
            if i < len(frames) - 1:
                await asyncio.sleep(FRAME_S)            # шаг слота 60мс между кадрами