"""
|
实时AFSK解调器 - 基于Goertzel算法
|
"""
|
|
import numpy as np
|
from collections import deque
|
|
|
class TraceGoertzel:
|
"""实时Goertzel算法实现"""
|
|
def __init__(self, freq: float, n: int):
|
"""
|
初始化Goertzel算法
|
|
Args:
|
freq: 归一化频率 (目标频率/采样频率)
|
n: 窗口大小
|
"""
|
self.freq = freq
|
self.n = n
|
|
# 预计算系数 - 与参考代码一致
|
self.k = int(freq * n)
|
self.w = 2.0 * np.pi * freq
|
self.cw = np.cos(self.w)
|
self.sw = np.sin(self.w)
|
self.c = 2.0 * self.cw
|
|
# 初始化状态变量 - 使用deque存储最近两个值
|
self.zs = deque([0.0, 0.0], maxlen=2)
|
|
def reset(self):
|
"""重置算法状态"""
|
self.zs.clear()
|
self.zs.extend([0.0, 0.0])
|
|
def __call__(self, xs):
|
"""
|
处理一组采样点 - 与参考代码一致的接口
|
|
Args:
|
xs: 采样点序列
|
|
Returns:
|
计算出的振幅
|
"""
|
self.reset()
|
for x in xs:
|
z1, z2 = self.zs[-1], self.zs[-2] # Z[-1], Z[-2]
|
z0 = x + self.c * z1 - z2 # S[n] = x[n] + C * S[n-1] - S[n-2]
|
self.zs.append(float(z0)) # 更新序列
|
return self.amp
|
|
@property
|
def amp(self) -> float:
|
"""计算当前振幅 - 与参考代码一致"""
|
z1, z2 = self.zs[-1], self.zs[-2]
|
ip = self.cw * z1 - z2
|
qp = self.sw * z1
|
return np.sqrt(ip**2 + qp**2) / (self.n / 2.0)
|
|
|
class PairGoertzel:
|
"""双频Goertzel解调器"""
|
|
def __init__(self, f_sample: int, f_space: int, f_mark: int,
|
bit_rate: int, win_size: int):
|
"""
|
初始化双频解调器
|
|
Args:
|
f_sample: 采样频率
|
f_space: Space频率 (通常对应0)
|
f_mark: Mark频率 (通常对应1)
|
bit_rate: 比特率
|
win_size: Goertzel窗口大小
|
"""
|
assert f_sample % bit_rate == 0, "采样频率必须是比特率的整数倍"
|
|
self.Fs = f_sample
|
self.F0 = f_space
|
self.F1 = f_mark
|
self.bit_rate = bit_rate
|
self.n_per_bit = int(f_sample // bit_rate) # 每个比特的采样点数
|
|
# 计算归一化频率
|
f1 = f_mark / f_sample
|
f0 = f_space / f_sample
|
|
# 初始化Goertzel算法
|
self.g0 = TraceGoertzel(freq=f0, n=win_size)
|
self.g1 = TraceGoertzel(freq=f1, n=win_size)
|
|
# 输入缓冲区
|
self.in_buffer = deque(maxlen=win_size)
|
self.out_count = 0
|
|
print(f"PairGoertzel initialized: f0={f0:.6f}, f1={f1:.6f}, win_size={win_size}, n_per_bit={self.n_per_bit}")
|
|
def __call__(self, s: float):
|
"""
|
处理单个采样点 - 与参考代码一致的接口
|
|
Args:
|
s: 采样点值
|
|
Returns:
|
(amp0, amp1, p1_prob) - 空间频率振幅，标记频率振幅，标记概率
|
"""
|
self.in_buffer.append(s)
|
self.out_count += 1
|
|
amp0, amp1, p1_prob = 0, 0, None
|
|
# 每个比特周期输出一次结果
|
if self.out_count >= self.n_per_bit:
|
amp0 = self.g0(self.in_buffer) # 计算space频率振幅
|
amp1 = self.g1(self.in_buffer) # 计算mark频率振幅
|
p1_prob = amp1 / (amp0 + amp1 + 1e-8) # 计算mark概率
|
self.out_count = 0
|
|
return amp0, amp1, p1_prob
|
|
|
class RealTimeAFSKDecoder:
|
"""实时AFSK解码器 - 基于起始帧触发"""
|
|
def __init__(self, f_sample: int = 16000, mark_freq: int = 1800,
|
space_freq: int = 1500, bitrate: int = 100,
|
s_goertzel: int = 9, threshold: float = 0.5):
|
"""
|
初始化实时AFSK解码器
|
|
Args:
|
f_sample: 采样频率
|
mark_freq: Mark频率
|
space_freq: Space频率
|
bitrate: 比特率
|
s_goertzel: Goertzel窗口大小系数 (win_size = f_sample // mark_freq * s_goertzel)
|
threshold: 判决门限
|
"""
|
self.f_sample = f_sample
|
self.mark_freq = mark_freq
|
self.space_freq = space_freq
|
self.bitrate = bitrate
|
self.threshold = threshold
|
|
# 计算窗口大小 - 与参考代码一致
|
win_size = int(f_sample / mark_freq * s_goertzel)
|
|
# 初始化解调器
|
self.demodulator = PairGoertzel(f_sample, space_freq, mark_freq,
|
bitrate, win_size)
|
|
# 帧定义 - 与参考代码一致
|
self.start_bytes = b'\x01\x02'
|
self.end_bytes = b'\x03\x04'
|
self.start_bits = "".join(format(int(x), '08b') for x in self.start_bytes)
|
self.end_bits = "".join(format(int(x), '08b') for x in self.end_bytes)
|
|
# 状态机
|
self.state = "idle" # idle / entering
|
|
# 存储解调结果
|
self.buffer_prelude:deque = deque(maxlen=len(self.start_bits)) # 判断是否启动
|
self.indicators = [] # 存储概率序列
|
self.signal_bits = "" # 存储比特序列
|
self.text_cache = ""
|
|
# 解码结果
|
self.decoded_messages = []
|
self.total_bits_received = 0
|
|
print(f"Decoder initialized: win_size={win_size}")
|
print(f"Start frame: {self.start_bits} (from {self.start_bytes.hex()})")
|
print(f"End frame: {self.end_bits} (from {self.end_bytes.hex()})")
|
|
def process_audio(self, samples: np.array) -> str:
|
"""
|
处理音频数据并返回解码文本
|
|
Args:
|
audio_data: 音频字节数据 (16-bit PCM)
|
|
Returns:
|
新解码的文本
|
"""
|
new_text = ""
|
# 逐个处理采样点
|
for sample in samples:
|
amp0, amp1, p1_prob = self.demodulator(sample)
|
# 如果有概率输出，记录并判决
|
if p1_prob is not None:
|
bit = '1' if p1_prob > self.threshold else '0'
|
match self.state:
|
case "idle":
|
self.buffer_prelude.append(bit)
|
pass
|
case "entering":
|
self.buffer_prelude.append(bit)
|
self.signal_bits += bit
|
self.total_bits_received += 1
|
case _:
|
pass
|
self.indicators.append(p1_prob)
|
|
# 检查状态机
|
if self.state == "idle" and "".join(self.buffer_prelude) == self.start_bits:
|
self.state = "entering"
|
self.text_cache = ""
|
self.signal_bits = "" # 清空比特序列
|
self.buffer_prelude.clear()
|
elif self.state == "entering" and ("".join(self.buffer_prelude) == self.end_bits or len(self.signal_bits) >= 256):
|
self.state = "idle"
|
self.buffer_prelude.clear()
|
|
# 每收集一定数量的比特后尝试解码
|
if len(self.signal_bits) >= 8:
|
text = self._decode_bits_to_text(self.signal_bits)
|
if len(text) > len(self.text_cache):
|
new_text = text[len(self.text_cache) - len(text):]
|
self.text_cache = text
|
return new_text
|
|
def _decode_bits_to_text(self, bits: str) -> str:
|
"""
|
将比特串解码为文本
|
|
Args:
|
bits: 比特串
|
|
Returns:
|
解码出的文本
|
"""
|
if len(bits) < 8:
|
return ""
|
|
decoded_text = ""
|
byte_count = len(bits) // 8
|
|
for i in range(byte_count):
|
# 提取8位
|
byte_bits = bits[i*8:(i+1)*8]
|
|
# 位转字节
|
byte_val = int(byte_bits, 2)
|
|
# 尝试解码为ASCII字符
|
if 32 <= byte_val <= 126: # 可打印ASCII字符
|
decoded_text += chr(byte_val)
|
elif byte_val == 0: # NULL字符，忽略
|
continue
|
else:
|
# 非可打印字符pass，以十六进制显示
|
pass
|
# decoded_text += f"\\x{byte_val:02X}"
|
|
return decoded_text
|
|
def clear(self):
|
"""清空解码状态"""
|
self.indicators = []
|
self.signal_bits = ""
|
self.decoded_messages = []
|
self.total_bits_received = 0
|
print("解码器状态已清空")
|
|
def get_stats(self) -> dict:
|
"""获取解码统计信息"""
|
return {
|
'prelude_bits': "".join(self.buffer_prelude),
|
"state": self.state,
|
'total_chars': sum(len(msg) for msg in self.text_cache),
|
'buffer_bits': len(self.signal_bits),
|
'mark_freq': self.mark_freq,
|
'space_freq': self.space_freq,
|
'bitrate': self.bitrate,
|
'threshold': self.threshold,
|
}
|
