mirror of
https://git.vladimir.cc/vladimir/ewsdr.git
synced 2026-08-25 20:37:33 +00:00
281 lines
12 KiB
Python
Executable File
281 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Стенд совместимости с TCI-клиентом MSHV.
|
|
|
|
Зачем: MSHV — самый распространённый TCI-клиент для цифры, и ведёт он себя не
|
|
так, как наш собственный стенд. Две его особенности определяют всё:
|
|
|
|
1. Входящие строки фильтруются по номеру приёмника, поэтому клиенту в режиме
|
|
RX2 (tci_trx = "1") строка `trx:0,false`
|
|
не существует в природе.
|
|
|
|
2. Инициализация — это ожидание ОДНОГО ответа. Он шлёт `vfo:<rx>,0;` раз в
|
|
500 мс, максимум 5 попыток, и ждёт `vfo:<rx>,0,<частота>`. Не дождался —
|
|
сообщает об ошибке инициализации.
|
|
|
|
Соответствие настроек: «TCI Client rx1» → приёмник 0, «TCI Client rx2» →
|
|
приёмник 1.
|
|
|
|
Стенд по умолчанию БЕЗОПАСЕН на приём: ни TRX, ни TUNE, ни START/STOP, ни
|
|
TX-аудио не отправляются. Передающую часть включает явный --tx.
|
|
|
|
Запуск (приложение уже работает, TCI-сервер поднят):
|
|
test/tci/mshv_sim.py --rx 1
|
|
test/tci/mshv_sim.py --rx 0 --audio
|
|
Только стандартная библиотека; транспорт переиспользуется из live_rx_test.py.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from live_rx_test import WS, split_commands, parse_command # noqa: E402
|
|
|
|
HDR = struct.Struct("<16I")
|
|
|
|
# Параметры совместимого клиента: 2 канала, 2048 сэмплов, float32,
|
|
# частота 48 кГц и буфер TX 24 мс.
|
|
MSHV_CHANNELS = "2"
|
|
MSHV_SAMPLES = "2048"
|
|
MSHV_TYPE = "float32"
|
|
MSHV_RATE = "48000"
|
|
MSHV_TXBUF = "24"
|
|
|
|
|
|
class Verdict:
|
|
def __init__(self):
|
|
self.ok_count = 0
|
|
self.fail_count = 0
|
|
|
|
def ok(self, label, detail=""):
|
|
self.ok_count += 1
|
|
print(f" ok {label}" + (f" — {detail}" if detail else ""))
|
|
|
|
def fail(self, label, detail=""):
|
|
self.fail_count += 1
|
|
print(f" FAIL {label}" + (f" — {detail}" if detail else ""))
|
|
|
|
def note(self, label, detail=""):
|
|
print(f" .. {label}" + (f" — {detail}" if detail else ""))
|
|
|
|
|
|
class MshvClient:
|
|
"""Минимальная модель поведения совместимого TCI-клиента."""
|
|
|
|
def __init__(self, ws, rx, verdict):
|
|
self.ws = ws
|
|
self.rx = str(rx)
|
|
self.v = verdict
|
|
# Состояние, которое MSHV ведёт по ФИЛЬТРОВАННОМУ потоку строк.
|
|
self.tx_enable = False
|
|
self.rx_enable = False
|
|
self.rx_mute = False
|
|
self.split_enable = False
|
|
self.drive = 0
|
|
self.is_trx = False
|
|
self.vfo_hz = None
|
|
self.mode = None
|
|
self.device = "NONE"
|
|
self.vendor = "ExpertSDR3"
|
|
self.proto_id = 0 # 0, 1 = 1.5.0 и выше, 2 = 1.9.0 и выше
|
|
self.audio_blocks = 0
|
|
self.audio_foreign = 0 # блоки чужих приёмников (MSHV их выбрасывает)
|
|
self.seen_for_me = [] # что вообще дошло до логики MSHV
|
|
self.seen_all = []
|
|
|
|
# ── приём ────────────────────────────────────────────────────────────
|
|
def pump(self, seconds):
|
|
deadline = time.monotonic() + seconds
|
|
while time.monotonic() < deadline:
|
|
frame = self.ws.recv_frame(min(0.2, deadline - time.monotonic()))
|
|
if frame is None:
|
|
continue
|
|
opcode, payload = frame
|
|
if opcode == 1:
|
|
for command in split_commands(payload.decode("utf-8", "replace")):
|
|
self.on_text(command)
|
|
elif opcode == 2:
|
|
self.on_binary(payload)
|
|
|
|
def on_text(self, command):
|
|
name, args = parse_command(command)
|
|
self.seen_all.append(command)
|
|
args = args + ["", "", ""]
|
|
|
|
# protocol разбирается ДО фильтра по приёмнику (так же у MSHV).
|
|
if name == "protocol":
|
|
self.vendor = args[0]
|
|
parts = (args[1].split(".") + ["0", "0", "0"])[:3]
|
|
v0, v1, v2 = (int(p) if p.isdigit() else 0 for p in parts)
|
|
self.proto_id = 2
|
|
if v0 < 2 and v1 < 9 and v2 < 1000:
|
|
self.proto_id = 1
|
|
if v0 < 2 and v1 < 5 and v2 < 1000:
|
|
self.proto_id = 0
|
|
return
|
|
if name == "device":
|
|
self.device = args[0]
|
|
return
|
|
|
|
# ★Фильтр MSHV: всё, адресованное не нашему приёмнику, не существует.
|
|
if args[0] != self.rx:
|
|
return
|
|
self.seen_for_me.append(command)
|
|
|
|
if name == "vfo" and args[1] == "0":
|
|
self.vfo_hz = args[2]
|
|
elif name == "modulation":
|
|
self.mode = args[1]
|
|
elif name == "trx":
|
|
self.is_trx = args[1] == "true"
|
|
elif name == "tx_enable":
|
|
self.tx_enable = args[1] == "true"
|
|
elif name == "rx_enable":
|
|
self.rx_enable = args[1] == "true"
|
|
elif name == "rx_mute":
|
|
self.rx_mute = args[1] == "true"
|
|
elif name == "split_enable":
|
|
self.split_enable = args[1] == "true"
|
|
elif name == "drive":
|
|
self.drive = int(args[1]) if args[1].lstrip("-").isdigit() else 0
|
|
|
|
def on_binary(self, payload):
|
|
if len(payload) < HDR.size:
|
|
return
|
|
head = HDR.unpack(payload[:HDR.size])
|
|
# Блоки других приёмников клиент игнорирует.
|
|
if head[0] != int(self.rx):
|
|
self.audio_foreign += 1
|
|
return
|
|
self.audio_blocks += 1
|
|
|
|
# ── инициализация: единственная причина «Error To Initialize» ────────
|
|
def initialize(self):
|
|
for attempt in range(1, 6):
|
|
self.ws.send_text(f"vfo:{self.rx},0;")
|
|
self.pump(0.5)
|
|
if self.vfo_hz is not None:
|
|
self.v.ok(f"инициализация прошла с попытки {attempt}",
|
|
f"vfo:{self.rx},0,{self.vfo_hz}")
|
|
return True
|
|
self.v.note(f"попытка {attempt}: ответа vfo:{self.rx},0,… нет")
|
|
self.v.fail("Error To Initialize TCI Server",
|
|
f"за 5 попыток не пришло vfo:{self.rx},0,<частота> — "
|
|
f"приёмника {self.rx} для сервера не существует")
|
|
return False
|
|
|
|
# ── то, что MSHV делает сразу после успешной инициализации ───────────
|
|
def after_connect(self):
|
|
if not self.tx_enable:
|
|
self.ws.send_text(f"tx_enable:{self.rx},true;")
|
|
if not self.rx_enable:
|
|
self.ws.send_text(f"rx_enable:{self.rx},true;")
|
|
if self.rx_mute:
|
|
self.ws.send_text(f"rx_mute:{self.rx},false;")
|
|
if self.split_enable:
|
|
self.ws.send_text(f"split_enable:{self.rx},false;")
|
|
if self.drive <= 0:
|
|
self.ws.send_text(f"drive:{self.rx},25;")
|
|
self.pump(1.0)
|
|
|
|
def start_audio(self):
|
|
self.ws.send_text(f"audio_samplerate:{MSHV_RATE};")
|
|
if self.proto_id > 1:
|
|
self.ws.send_text(f"audio_stream_sample_type:{MSHV_TYPE};")
|
|
self.ws.send_text(f"audio_stream_channels:{MSHV_CHANNELS};")
|
|
self.ws.send_text(f"audio_stream_samples:{MSHV_SAMPLES};")
|
|
self.ws.send_text(f"tx_stream_audio_buffering:{MSHV_TXBUF};")
|
|
self.ws.send_text(f"audio_start:{self.rx};")
|
|
|
|
def stop_audio(self):
|
|
self.ws.send_text(f"audio_stop:{self.rx};")
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--host", default="127.0.0.1")
|
|
ap.add_argument("--port", type=int, default=40001)
|
|
ap.add_argument("--rx", type=int, default=0,
|
|
help="приёмник TCI: 0 = «TCI Client rx1» в MSHV, 1 = «rx2»")
|
|
ap.add_argument("--audio", action="store_true",
|
|
help="запросить аудиопоток, как это делает MSHV")
|
|
ap.add_argument("--seconds", type=float, default=3.0,
|
|
help="сколько слушать поток")
|
|
ap.add_argument("--tx", action="store_true",
|
|
help="★ВКЛЮЧАЕТ ПЕРЕДАЧУ: короткий trx:<rx>,true,tci")
|
|
args = ap.parse_args()
|
|
|
|
v = Verdict()
|
|
print(f"Стенд MSHV: приёмник {args.rx} "
|
|
f"(в MSHV это «TCI Client rx{args.rx + 1}»)")
|
|
try:
|
|
ws = WS(args.host, args.port)
|
|
except Exception as exc:
|
|
v.fail("подключение к TCI-серверу", str(exc))
|
|
print(f"\nИтого: {v.ok_count} ok, {v.fail_count} FAIL")
|
|
return 1
|
|
|
|
client = MshvClient(ws, args.rx, v)
|
|
try:
|
|
# Пачка инициализации приходит сама; MSHV её тоже читает.
|
|
client.pump(1.5)
|
|
v.ok("пачка инициализации получена",
|
|
f"{client.vendor} {client.device}, protocol id {client.proto_id}")
|
|
|
|
if not client.initialize():
|
|
return 1
|
|
|
|
client.after_connect()
|
|
|
|
# Дальше — то, без чего MSHV работает молча неправильно.
|
|
if client.tx_enable:
|
|
v.ok(f"tx_enable для приёмника {args.rx} получен")
|
|
else:
|
|
v.fail(f"tx_enable для приёмника {args.rx} НЕ получен",
|
|
"MSHV начинает set_ptt со строки `if (!tci_tx_enable) return;` "
|
|
"— PTT будет молча мёртв")
|
|
if client.mode:
|
|
v.ok("модуляция известна", client.mode)
|
|
else:
|
|
v.fail("modulation для этого приёмника не приходила",
|
|
"MSHV не узнает вид связи")
|
|
|
|
if args.audio:
|
|
client.start_audio()
|
|
client.pump(args.seconds)
|
|
if client.audio_blocks > 0:
|
|
v.ok("аудиопоток идёт", f"{client.audio_blocks} блоков")
|
|
else:
|
|
v.fail("блоков аудио нет",
|
|
f"чужих (не нашего приёмника) отброшено: {client.audio_foreign}")
|
|
client.stop_audio()
|
|
client.pump(0.3)
|
|
|
|
if args.tx:
|
|
print(" .. ★передача: trx:%d,true,tci" % args.rx)
|
|
client.ws.send_text(f"trx:{args.rx},true,tci;")
|
|
client.pump(0.7)
|
|
if client.is_trx:
|
|
v.ok("MSHV увидел бы себя в эфире", f"trx:{args.rx},true")
|
|
else:
|
|
v.fail("подтверждения передачи для своего приёмника нет",
|
|
"MSHV фильтрует ответ по номеру приёмника: "
|
|
"строку с чужим номером он не увидит")
|
|
client.ws.send_text(f"trx:{args.rx},false;")
|
|
client.pump(0.5)
|
|
|
|
print(f"\nСтрок, дошедших до логики MSHV: {len(client.seen_for_me)} "
|
|
f"из {len(client.seen_all)} присланных")
|
|
finally:
|
|
ws.close()
|
|
|
|
print(f"\nИтого: {v.ok_count} ok, {v.fail_count} FAIL")
|
|
return 1 if v.fail_count else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|