nikit3s
c1ac8d59ca
Разбор кадров и запись .raw/.bin не трогали. Заменена только шкала: у микрофона нет опорного напряжения, уровни считаются от полной шкалы. Бит1 STATUS теперь показывается как насыщение дециматора.
331 lines
13 KiB
Python
331 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Приём кадров цифрового микрофона по UART.
|
||
|
||
Пишет два файла:
|
||
<out>.raw все байты из порта, до разбора -- отладка канала
|
||
<out>.bin только отсчёты, 3 байта BE -- читается read_24bit_bin()
|
||
из F:\\PyPrj\\Signal_analayzer
|
||
|
||
Кадр с несошедшейся CRC отбрасывается, пропуск восстанавливается по FRAME_CNT
|
||
и добивается нулями, чтобы не сдвинулась ось времени.
|
||
Формат кадра -- README.md.
|
||
"""
|
||
|
||
import argparse
|
||
import math
|
||
import sys
|
||
import time
|
||
|
||
import serial
|
||
from serial.tools import list_ports
|
||
|
||
SYNC = b"\xA5\x5A"
|
||
SAMPLES_PER_FRAME = 50
|
||
PAYLOAD_LEN = 2 + 1 + SAMPLES_PER_FRAME * 3 # FRAME_CNT + STATUS + DATA
|
||
FRAME_LEN = len(SYNC) + PAYLOAD_LEN + 2 # + CRC16
|
||
FS_DEFAULT = 12500.0
|
||
FS_CODE = 2 ** 23 # полная шкала: постоянная '1' на PDM
|
||
|
||
|
||
def dbfs(code) -> float:
|
||
"""Код -> dBFS. Шкалу задаёт микрофон, вольт в тракте нет."""
|
||
return 20.0 * math.log10(max(abs(code), 1e-9) / FS_CODE)
|
||
|
||
|
||
def _make_crc_table():
|
||
tbl = []
|
||
for b in range(256):
|
||
c = b << 8
|
||
for _ in range(8):
|
||
c = ((c << 1) ^ 0x1021) & 0xFFFF if c & 0x8000 else (c << 1) & 0xFFFF
|
||
tbl.append(c)
|
||
return tbl
|
||
|
||
|
||
_CRC_TBL = _make_crc_table()
|
||
|
||
|
||
def crc16_ccitt(data: bytes) -> int:
|
||
"""CRC-16/CCITT, poly 0x1021, init 0xFFFF, MSB-first -- как в ПЛИС."""
|
||
crc = 0xFFFF
|
||
for byte in data:
|
||
crc = ((crc << 8) & 0xFFFF) ^ _CRC_TBL[(crc >> 8) ^ byte]
|
||
return crc
|
||
|
||
|
||
class Stats:
|
||
def __init__(self):
|
||
self.frames_ok = 0
|
||
self.crc_errors = 0
|
||
self.resyncs = 0
|
||
self.padded_frames = 0
|
||
self.status_flags = 0
|
||
self.samples = 0
|
||
self.code_min = None
|
||
self.code_max = None
|
||
self.code_sum = 0
|
||
self.code_sqsum = 0
|
||
self.bits_or = 0 # какие биты хоть раз были 1
|
||
self.bits_and = 0xFFFFFF # какие биты всегда были 1
|
||
|
||
|
||
def decode_samples(data: bytes):
|
||
"""150 байт -> 50 знаковых отсчётов."""
|
||
out = []
|
||
for i in range(0, len(data), 3):
|
||
v = (data[i] << 16) | (data[i + 1] << 8) | data[i + 2]
|
||
if v & 0x800000:
|
||
v -= 0x1000000
|
||
out.append(v)
|
||
return out
|
||
|
||
|
||
def open_port(port, baud):
|
||
ser = serial.Serial(port, baud, timeout=0.1)
|
||
try:
|
||
# Буфер COM-порта в Windows по умолчанию около 4 кБ -- это всего 100 мс
|
||
# потока, любая заминка в разборе стоит пачки байт
|
||
ser.set_buffer_size(rx_size=1 << 20, tx_size=1 << 12)
|
||
except Exception:
|
||
pass
|
||
ser.reset_input_buffer()
|
||
return ser
|
||
|
||
|
||
def run_dump(port, baud, duration, out_base):
|
||
"""Только приём в файл, без разбора. Показывает, теряет ли байты сам канал."""
|
||
raw_path = out_base + ".raw"
|
||
total = 0
|
||
slowest = 0.0
|
||
with open_port(port, baud) as ser, open(raw_path, "wb") as fraw:
|
||
t0 = time.time()
|
||
t_end = t0 + duration
|
||
while time.time() < t_end:
|
||
t1 = time.perf_counter()
|
||
chunk = ser.read(4096)
|
||
slowest = max(slowest, time.perf_counter() - t1)
|
||
if chunk:
|
||
fraw.write(chunk)
|
||
total += len(chunk)
|
||
elapsed = time.time() - t0
|
||
|
||
rate = total / elapsed
|
||
expect = FRAME_LEN * FS_DEFAULT / SAMPLES_PER_FRAME
|
||
print(f"принято : {total} байт за {elapsed:.2f} с")
|
||
print(f"скорость : {rate / 1000:.1f} кБ/с")
|
||
print(f"ожидалось : {expect / 1000:.1f} кБ/с ({FRAME_LEN} байт/кадр, "
|
||
f"{FS_DEFAULT / SAMPLES_PER_FRAME:.0f} кадров/с)")
|
||
print(f"дошло : {100.0 * rate / expect:.1f}%")
|
||
print(f"дольше всего ждали чтение: {slowest * 1000:.0f} мс")
|
||
if rate < expect * 0.95:
|
||
print("\nБайты теряются в канале, разбор тут ни при чём: "
|
||
"мост, драйвер или скорость порта.")
|
||
else:
|
||
print("\nКанал отдаёт весь поток. Значит потери возникают при разборе.")
|
||
return 0
|
||
|
||
|
||
class FrameParser:
|
||
"""Разбор потока кадров. Единственное место, где описан формат, --
|
||
им пользуются и запись, и живой просмотр.
|
||
|
||
feed() возвращает список (frame_cnt, status, data, gap): data -- 150 байт
|
||
отсчётов, gap -- сколько кадров потерялось перед этим.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.buf = bytearray()
|
||
self.expected_cnt = None
|
||
self.frames_ok = 0
|
||
self.crc_errors = 0
|
||
self.resyncs = 0
|
||
self.padded_frames = 0
|
||
self.status_flags = 0
|
||
|
||
def feed(self, chunk):
|
||
out = []
|
||
self.buf += chunk
|
||
while True:
|
||
idx = self.buf.find(SYNC)
|
||
if idx < 0:
|
||
# Оставляем хвост, в котором SYNC мог разорваться между чтениями
|
||
if len(self.buf) > 1:
|
||
del self.buf[:-1]
|
||
break
|
||
if idx > 0:
|
||
self.resyncs += 1
|
||
del self.buf[:idx]
|
||
if len(self.buf) < FRAME_LEN:
|
||
break
|
||
|
||
frame = bytes(self.buf[:FRAME_LEN])
|
||
payload = frame[2:2 + PAYLOAD_LEN]
|
||
crc_rx = (frame[-2] << 8) | frame[-1]
|
||
|
||
if crc16_ccitt(payload) != crc_rx:
|
||
self.crc_errors += 1
|
||
# SYNC мог оказаться внутри данных: сдвигаемся и ищем дальше
|
||
del self.buf[:2]
|
||
continue
|
||
|
||
del self.buf[:FRAME_LEN]
|
||
|
||
cnt = (payload[0] << 8) | payload[1]
|
||
status = payload[2]
|
||
self.status_flags |= status
|
||
|
||
gap = 0
|
||
if self.expected_cnt is not None and cnt != self.expected_cnt:
|
||
gap = (cnt - self.expected_cnt) & 0xFFFF
|
||
self.padded_frames += gap
|
||
self.expected_cnt = (cnt + 1) & 0xFFFF
|
||
self.frames_ok += 1
|
||
out.append((cnt, status, payload[3:], gap))
|
||
return out
|
||
|
||
|
||
def run(port, baud, duration, out_base, quiet):
|
||
raw_path = out_base + ".raw"
|
||
bin_path = out_base + ".bin"
|
||
log_path = out_base + ".log"
|
||
|
||
st = Stats()
|
||
log_lines = []
|
||
|
||
def log(msg):
|
||
log_lines.append(msg)
|
||
if not quiet:
|
||
print(msg)
|
||
|
||
log(f"порт {port}, {baud} бод, длительность {duration} с")
|
||
|
||
parser = FrameParser()
|
||
t_end = time.time() + duration
|
||
gaps_printed = 0
|
||
|
||
with open_port(port, baud) as ser, \
|
||
open(raw_path, "wb") as fraw, open(bin_path, "wb") as fbin:
|
||
while time.time() < t_end:
|
||
chunk = ser.read(4096)
|
||
if not chunk:
|
||
continue
|
||
fraw.write(chunk)
|
||
|
||
for cnt, status, data, gap in parser.feed(chunk):
|
||
if gap:
|
||
fbin.write(b"\x00" * (gap * SAMPLES_PER_FRAME * 3))
|
||
st.samples += gap * SAMPLES_PER_FRAME
|
||
# Печатаем только первые сообщения: вывод в консоль медленный
|
||
# и на длинной записи сам становится причиной потерь
|
||
msg = (f"пропуск {gap} кадр(ов) перед FRAME_CNT={cnt}, "
|
||
f"добито {gap * SAMPLES_PER_FRAME} нулей")
|
||
log_lines.append(msg)
|
||
if gaps_printed < 10 and not quiet:
|
||
print(msg)
|
||
gaps_printed += 1
|
||
|
||
fbin.write(data)
|
||
st.samples += SAMPLES_PER_FRAME
|
||
|
||
for v in decode_samples(data):
|
||
st.code_sum += v
|
||
st.code_sqsum += v * v
|
||
st.bits_or |= v & 0xFFFFFF
|
||
st.bits_and &= v & 0xFFFFFF
|
||
if st.code_min is None or v < st.code_min:
|
||
st.code_min = v
|
||
if st.code_max is None or v > st.code_max:
|
||
st.code_max = v
|
||
|
||
st.frames_ok = parser.frames_ok
|
||
st.crc_errors = parser.crc_errors
|
||
st.resyncs = parser.resyncs
|
||
st.padded_frames = parser.padded_frames
|
||
st.status_flags = parser.status_flags
|
||
|
||
log("")
|
||
log(f"кадров принято : {st.frames_ok}")
|
||
log(f"ошибок CRC : {st.crc_errors}")
|
||
log(f"ресинхронизаций : {st.resyncs}")
|
||
log(f"кадров добито нулями: {st.padded_frames}")
|
||
log(f"отсчётов в .bin : {st.samples} ({st.samples / FS_DEFAULT:.3f} с при {FS_DEFAULT:.0f} Гц)")
|
||
|
||
if st.status_flags & 0x01:
|
||
log("STATUS: было переполнение FIFO в ПЛИС")
|
||
if st.status_flags & 0x02:
|
||
log("STATUS: выход дециматора упирался в шкалу")
|
||
if st.status_flags == 0 and st.frames_ok:
|
||
log("STATUS: чисто")
|
||
|
||
if st.frames_ok:
|
||
n = st.frames_ok * SAMPLES_PER_FRAME
|
||
mean = st.code_sum / n
|
||
var = max(st.code_sqsum / n - mean * mean, 0.0)
|
||
sigma = var ** 0.5
|
||
peak = max(abs(st.code_min), abs(st.code_max))
|
||
log("")
|
||
log(f"код min/max/среднее : {st.code_min} / {st.code_max} / {mean:.1f}")
|
||
log(f"пик : {peak} код ({dbfs(peak):+.1f} dBFS)")
|
||
log(f"смещение нуля : {mean:.1f} код ({dbfs(mean):+.1f} dBFS)")
|
||
log(f"СКО : {sigma:.1f} код ({dbfs(sigma):+.1f} dBFS)")
|
||
|
||
# Постоянные старшие биты -- норма: при малом размахе сигнала меняются
|
||
# только младшие. Тревожно обратное: залипшие младшие биты, которые
|
||
# обязан расшевелить хотя бы собственный шум микрофона.
|
||
frozen = ((~st.bits_or) | st.bits_and) & 0xFFFFFF
|
||
low_frozen = frozen & 0x0F
|
||
if frozen == 0xFFFFFF:
|
||
log("биты : НИ ОДИН из 24 не менялся — тракт отдаёт константу")
|
||
elif low_frozen:
|
||
log("биты : ВНИМАНИЕ, не менялись младшие " + ", ".join(
|
||
str(i) for i in range(3, -1, -1) if low_frozen >> i & 1) +
|
||
" — похоже на сбой чтения, а не на сигнал")
|
||
else:
|
||
top = 0
|
||
for i in range(23, -1, -1):
|
||
if frozen >> i & 1:
|
||
top += 1
|
||
else:
|
||
break
|
||
if top:
|
||
log(f"биты : младшие {24 - top} переключались, "
|
||
f"старшие {top} постоянны — норма при малом размахе")
|
||
else:
|
||
log("биты : все 24 переключались")
|
||
|
||
with open(log_path, "w", encoding="utf-8") as flog:
|
||
flog.write("\n".join(log_lines) + "\n")
|
||
|
||
return 1 if (st.crc_errors or st.status_flags or not st.frames_ok) else 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("--port", help="например COM5")
|
||
ap.add_argument("--baud", type=int, default=921600)
|
||
ap.add_argument("--seconds", type=float, default=10.0)
|
||
ap.add_argument("--out", default="capture", help="базовое имя файлов без расширения")
|
||
ap.add_argument("--quiet", action="store_true")
|
||
ap.add_argument("--list", action="store_true", help="показать доступные порты")
|
||
ap.add_argument("--dump", action="store_true",
|
||
help="только приём в .raw без разбора: проверка канала")
|
||
args = ap.parse_args()
|
||
|
||
if args.list:
|
||
for p in list_ports.comports():
|
||
print(f"{p.device} {p.description}")
|
||
return 0
|
||
|
||
if not args.port:
|
||
ap.error("нужен --port (список портов: --list)")
|
||
|
||
if args.dump:
|
||
return run_dump(args.port, args.baud, args.seconds, args.out)
|
||
|
||
return run(args.port, args.baud, args.seconds, args.out, args.quiet)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|