pip install paho-mqtt
发送json，含有音频文件；
import json, base64 import paho.mqtt.client as mqtt import time def on_connect(client, userdata, flags, rc): if rc == 0: print("[A] Connected to broker") client.subscribe(TOPIC_SUB) print(f"[A] Subscribed to {TOPIC_SUB}") else: print(f"[A] Connection failed: {rc}") def on_message(client, userdata, msg): print(f"[A] Received message on '{msg.topic}': {msg.payload.decode()}") if __name__ == '__main__': #install "pip install paho-mqtt" BROKER = "192.168.8.127" BROKER = "localhost" PORT = 1883 TOPIC_SUB = "ue/state" # 订阅状态 TOPIC_PUB = "ue/command" # 发布命令 client = mqtt.Client(client_id="ClientA") client.on_connect = on_connect client.on_message = on_message client.connect(BROKER, PORT, keepalive=30) # non-blocking loop client.loop_start() wav_path = r"D:\data\audios\post_2.wav" # 发送一些命令 for i in range(5000): msg = f"command {i}" with open(wav_path, "rb") as f: audio_bytes = f.read() audio_b64 = base64.b64encode(audio_bytes).decode("utf-8") msg_audio = json.dumps({ "type": "audio", "data": audio_b64, }) client.publish("ue/messages", msg_audio) # ===== 发送命令 ===== msg_cmd = json.dumps({ "type": "command", "data": "pause", }) # client.publish("ue/messages", msg_cmd) print(f"[A] Sent: {msg}") time.sleep(3) client.loop_stop() client.disconnect()
接收json，含有音频文件
import json import base64 import paho.mqtt.client as mqtt import os # ====== 配置 ====== BROKER = "localhost" # MQTT Broker PORT = 1883 TOPIC_SUB = "ue/messages" # 订阅 Topic SAVE_DIR = r"D:\data\audios_received" os.makedirs(SAVE_DIR, exist_ok=True) file_index = 0 # ====== MQTT 回调 ====== def on_connect(client, userdata, flags, rc): if rc == 0: print("[B] Connected to broker") client.subscribe(TOPIC_SUB) print(f"[B] Subscribed to {TOPIC_SUB}") else: print(f"[B] Connection failed: {rc}") def on_message(client, userdata, msg): global file_index try: payload_str = msg.payload.decode('utf-8') data = json.loads(payload_str) if data.get("type") == "audio": audio_b64 = data.get("data", "") if not audio_b64: return audio_bytes = base64.b64decode(audio_b64) file_path = os.path.join(SAVE_DIR, f"received_{file_index}.wav") file_index += 1 # 自动判断是否是完整 WAV if is_wav_file(audio_bytes): # 已经是完整 WAV，直接写入文件 with open(file_path, 'wb') as f: f.write(audio_bytes) print(f"[B] Saved complete WAV to {file_path}") else: # 裸 PCM，需要自己加 WAV 头 # 默认假设 16-bit PCM，单声道，采样率 44100 save_pcm_as_wav(file_path, audio_bytes, sample_rate=44100, num_channels=1) print(f"[B] Saved PCM as WAV to {file_path}") elif data.get("type") == "command": cmd = data.get("data") print(f"[B] Received command: {cmd}") except Exception as e: print(f"[B] Error processing message: {e}") # ====== 判断是否完整 WAV 文件 ====== def is_wav_file(data_bytes): # WAV 文件头前4字节是 "RIFF" return data_bytes[:4] == b'RIFF' # ====== 将 PCM 数据写入 WAV 文件 ====== def save_pcm_as_wav(file_path, audio_bytes, sample_rate=44100, num_channels=1): import wave with wave.open(file_path, 'wb') as wf: wf.setnchannels(num_channels) wf.setsampwidth(2) # 16-bit wf.setframerate(sample_rate) wf.writeframes(audio_bytes) # ====== 主程序 ====== if __name__ == "__main__": client = mqtt.Client(client_id="ClientB") client.on_connect = on_connect client.on_message = on_message client.connect(BROKER, PORT, keepalive=30) client.loop_forever()
