acodec.py 是使用G.711编码和解码音频的示例程序。
程序分为三个主要部分:编码音频、解码音频和循环编解码(实时采集、编码、解码并播放)。
程序使用了media.pyaudio进行音频的采集和播放,使用media.g711进行G.711编解码。
三个主要函数如下:
1. 编码函数 (encode_audio)
将原始音频数据编码为 G.711 格式并保存到文件:
设置音频参数:44.1kHz 采样率,16位精度,双声道
初始化 PyAudio 和 G.711 编码器
从音频输入流读取数据并编码
将编码后的数据写入文件
2. 解码函数 (decode_audio)
从 G.711 文件读取数据并解码播放:
打开 G.711 编码的文件
初始化 PyAudio 和 G.711 解码器
读取文件数据并解码
通过音频输出流播放解码后的数据
3. 实时编解码函数 (loop_codec)
实时采集音频,编码为 G.711,然后立即解码并播放:
同时初始化编码器和解码器
从输入设备采集音频
编码为 G.711 格式
立即解码回原始格式
通过输出设备播放
注意事项
程序需要 SD 卡来存储编码后的音频文件;
可以选择三种操作模式:
仅编码:采集音频并保存为 G.711 文件;
仅解码:读取 G.711 文件并播放;
实时编解码:采集-编码-解码-播放的完整流程。
音频参数固定为 44.1kHz/16位/双声道;
程序包含完整的异常处理和资源清理机制。
实际运行时的效果(实时编解码播放):
#encode_audio('/sdcard/app/test.g711a', 15) #采集并编码g711文件
#decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频
效果:板载麦克风拾音,通过喇叭实时播放麦克风拾音的内容,运行15秒钟。
实际运行时的效果(采集并编码后,保存音频文件):
encode_audio('/sdcard/app/test.g711a', 10) #采集并编码g711文件
#decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
#loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频
效果:板载麦克风拾音15秒钟,保存为/test.g711a文件至虚拟U盘的/sdcard/app/目录下。
运行时错误1:默认运行会提示错误:OSError: [Errno 2] ENOENT ,是指定的文件或目录不存在。
处理方法:在/sdcard/目录下新建app目录。
运行时错误2:默认的采集15秒,在运行时会出现错误:MemoryError: memory allocation failed, allocating 1323001 bytes ,意思是内存分配错误。
处理方法:将参数的15秒,改为10秒,能够正常运行。
实际运行时的效果(播放已经保存的音频文件):
#encode_audio('/sdcard/app/test.g711a', 15) #采集并编码g711文件
decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
#loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频 效果:板载麦克风拾音,通过喇叭实时播放麦克风拾音的内容。
程序源代码如下:
# g711 encode/decode example
#
# Note: You will need an SD card to run this example.
#
# You can collect raw data and encode it into g711 or decode it into raw data output.
import os
from mpp.payload_struct import * #导入payload模块,用于获取音视频编解码类型
from media.media import * #导入media模块,用于初始化vb buffer
from media.pyaudio import * #导入pyaudio模块,用于采集和播放音频
import media.g711 as g711 #导入g711模块,用于g711编解码
def exit_check():
try:
os.exitpoint()
except KeyboardInterrupt as e:
import sys
sys.print_exception(e)
return True
return False
def encode_audio(filename, duration):
CHUNK = int(44100/25) #设置音频chunk值
FORMAT = paInt16 #设置采样精度
CHANNELS = 2 #设置声道数
RATE = 44100 #设置采样率
try:
p = PyAudio()
p.initialize(CHUNK) #初始化PyAudio对象
enc = g711.Encoder(K_PT_G711A,CHUNK) #创建g711编码器对象
MediaManager.init() #vb buffer初始化
enc.create() #创建编码器
#创建音频输入流
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK)
frames = []
#采集音频数据编码并存入列表
for i in range(0, int(RATE / CHUNK * duration)):
frame_data = stream.read() #从音频输入流中读取音频数据
data = enc.encode(frame_data) #编码音频数据为g711
frames.append(data) #将g711编码数据保存到列表中
if exit_check():
break
#将g711编码数据存入文件中
with open(filename,mode='w') as wf:
wf.write(b''.join(frames))
stream.stop_stream() #停止音频输入流
stream.close() #关闭音频输入流
p.terminate() #释放音频对象
enc.destroy() #销毁g711音频编码器
except BaseException as e:
import sys
sys.print_exception(e)
finally:
MediaManager.deinit() #释放vb buffer
def decode_audio(filename):
FORMAT = paInt16 #设置音频chunk值
CHANNELS = 2 #设置声道数
RATE = 44100 #设置采样率
CHUNK = int(RATE/25) #设置音频chunk值
try:
wf = open(filename,mode='rb') #打开g711文件
p = PyAudio()
p.initialize(CHUNK) #初始化PyAudio对象
dec = g711.Decoder(K_PT_G711A,CHUNK) #创建g711解码器对象
MediaManager.init() #vb buffer初始化
dec.create() #创建解码器
#创建音频输出流
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
output=True,
frames_per_buffer=CHUNK)
stream_len = CHUNK*CHANNELS*2//2 #设置每次读取的g711数据流长度
stream_data = wf.read(stream_len) #从g711文件中读取数据
#解码g711文件并播放
while stream_data:
frame_data = dec.decode(stream_data) #解码g711文件
stream.write(frame_data) #播放raw数据
stream_data = wf.read(stream_len) #从g711文件中读取数据
if exit_check():
break
stream.stop_stream() #停止音频输入流
stream.close() #关闭音频输入流
p.terminate() #释放音频对象
dec.destroy() #销毁解码器
wf.close() #关闭g711文件
except BaseException as e:
import sys
sys.print_exception(e)
finally:
MediaManager.deinit() #释放vb buffer
def loop_codec(duration):
CHUNK = int(44100/25) #设置音频chunk值
FORMAT = paInt16 #设置采样精度
CHANNELS = 2 #设置声道数
RATE = 44100 #设置采样率
try:
p = PyAudio()
p.initialize(CHUNK) #初始化PyAudio对象
dec = g711.Decoder(K_PT_G711A,CHUNK) #创建g711解码器对象
enc = g711.Encoder(K_PT_G711A,CHUNK) #创建g711编码器对象
MediaManager.init() #vb buffer初始化
dec.create() #创建g711解码器
enc.create() #创建g711编码器
#创建音频输入流
input_stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK)
#创建音频输出流
output_stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
output=True,
frames_per_buffer=CHUNK)
#从音频输入流中获取数据->编码->解码->写入到音频输出流中
for i in range(0, int(RATE / CHUNK * duration)):
frame_data = input_stream.read() #从音频输入流中获取raw音频数据
stream_data = enc.encode(frame_data) #编码音频数据为g711
frame_data = dec.decode(stream_data) #解码g711数据为raw数据
output_stream.write(frame_data) #播放raw数据
if exit_check():
break
input_stream.stop_stream() #停止音频输入流
output_stream.stop_stream() #停止音频输出流
input_stream.close() #关闭音频输入流
output_stream.close() #关闭音频输出流
p.terminate() #释放音频对象
dec.destroy() #销毁g711解码器
enc.destroy() #销毁g711编码器
except BaseException as e:
import sys
sys.print_exception(e)
finally:
MediaManager.deinit() #释放vb buffer
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
print("audio codec sample start")
#encode_audio('/sdcard/app/test.g711a', 15) #采集并编码g711文件
#decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频
print("audio codec sample done")
课后思考:为何官方例程,运行保存15秒的程序时会有错误提示?
为了排查官方15秒录音的例程为何会运行提示错误,结合前面例程
13.examples1-Micropython-Basicsdemo_view_mem.py 加强版-CSDN博客
增加内存监控后再看下录音10秒正常时的数据,以后再排查15秒内存异常的原因:
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
print("audio codec sample start")
import gc # 导入垃圾回收模块
# 查看初始内存状态
print("初始状态:")
print(f"已分配内存: {gc.mem_alloc()} Bytes")
print(f"空闲内存: {gc.mem_free()} Bytes")
# 手动触发垃圾回收
gc.collect()
print("
垃圾回收后:")
print(f"已分配内存: {gc.mem_alloc()} Bytes")
print(f"空闲内存: {gc.mem_free()} Bytes")
encode_audio('/sdcard/app/test.g711a', 10) #采集并编码g711文件
#decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
#loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频
print("
encode_audio后:")
print(f"已分配内存: {gc.mem_alloc()} Bytes")
print(f"空闲内存: {gc.mem_free()} Bytes")
print("audio codec sample done")
#-----------------------------------------------------
#实际运行输出参数如下:
audio codec sample start
初始状态:
已分配内存: 190912 Bytes
空闲内存: 3954624 Bytes
垃圾回收后:
已分配内存: 183904 Bytes
空闲内存: 3961632 Bytes
vb common pool count 4
encode_audio后:
已分配内存: 3848544 Bytes
空闲内存: 296992 Bytes
audio codec sample done
MPY: soft reboot
CanMV v1.3-132-gb19c756(based on Micropython e00a144) on 2025-08-09; k230_canmv_01studio with K230