嘉楠科K230AI开发板测评8--音频采集、播放、编码与解码、视频采集、播放与编码
本帖最后由 dfjs 于 2024-11-16 22:32 编辑<p style="text-align: center;"><b>嘉楠科K230AI开发板</b><b>测评8</b></p>
<p><strong>1、音频采集与播放</strong></p>
<p> K230自带采集与播放接口，接口位置如下左图所示，音频输入为：麦克风咪头，音频输出：3.5mm音频口（双声道），查看音频采集与播放的原理图，可能是考虑到功率问题，没有功率放大电路，而采用了3.5mm的音频口输出而不是直接喇叭外放。</p>
<p style="text-align: center;"> </p>
<p style="text-align: center;"> </p>
<p> </p>
<p align="justify"> 具体代码思路如下：</p>
<p align="justify"><b> 初始化模块：</b></p>
<p align="justify"> 导入必要的模块，包括 os、media、pyaudio 和 wave。</p>
<p align="justify"> 初始化媒体管理和音频对象。</p>
<p align="justify"><b> 异常处理：</b></p>
<p align="justify"> 定义 exit_check 函数，用于捕获键盘中断（Ctrl+C）并优雅地退出程序。</p>
<p align="justify"><b> 音频录制：</b></p>
<p align="justify"> 定义 record_audio 函数，用于录制音频并保存为 WAV 文件。</p>
<p align="justify"> 设置音频参数，如采样率、采样精度、声道数和 chunk 大小。</p>
<p align="justify"> 打开音频输入流，读取音频数据并存储到列表中。</p>
<p align="justify"> 将列表中的音频数据保存到 WAV 文件中。</p>
<p align="justify"><b> 音频播放：</b></p>
<p align="justify"> 定义 play_audio 函数，用于播放 WAV 文件。</p>
<p align="justify"> 打开 WAV 文件，读取音频参数。</p>
<p align="justify"> 打开音频输出流，读取 WAV 文件中的音频数据并写入输出流。</p>
<p align="justify"><b> 实时回放：</b></p>
<p align="justify"> 定义 loop_audio 函数，用于实时采集音频并立即播放。</p>
<p align="justify"> 设置音频参数，打开音频输入流和输出流。</p>
<p align="justify"> 从输入流中读取音频数据并立即写入输出流。</p>
<p align="justify"><b> 主程序：</b></p>
<p align="justify"> 在 __main__ 块中，启用退出点，启动音频示例。</p>
<p> 调用 play_audio、record_audio 或 loop_audio 函数来执行相应的音频操作。</p>
<p align="justify"><b> 参考代码如下：</b></p>
<pre>
<code class="language-python"># audio input and output example
#
# Note: You will need an SD card to run this example.
#
# You can play wav files or capture audio to save as wav
import os
from media.media import * #导入media模块，用于初始化vb buffer
from media.pyaudio import * #导入pyaudio模块，用于采集和播放音频
import media.wave as wave #导入wav模块，用于保存和加载wav音频文件
def exit_check():
try:
os.exitpoint()
except KeyboardInterrupt as e:
print("user stop: ", e)
return True
return False
def record_audio(filename, duration):
CHUNK = int(44100/25)#设置音频chunk值
FORMAT = paInt16 #设置采样精度
CHANNELS = 2 #设置声道数
RATE = 44100 #设置采样率
try:
p = PyAudio()
p.initialize(CHUNK) #初始化PyAudio对象
MediaManager.init() #vb buffer初始化
#创建音频输入流
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)):
data = stream.read()
frames.append(data)
if exit_check():
break
#将列表中的数据保存到wav文件中
wf = wave.open(filename, 'wb') #创建wav 文件
wf.set_channels(CHANNELS) #设置wav 声道数
wf.set_sampwidth(p.get_sample_size(FORMAT))#设置wav 采样精度
wf.set_framerate(RATE)#设置wav 采样率
wf.write_frames(b''.join(frames)) #存储wav音频数据
wf.close() #关闭wav文件
except BaseException as e:
print(f"Exception {e}")
finally:
stream.stop_stream() #停止采集音频数据
stream.close()#关闭音频输入流
p.terminate()#释放音频对象
MediaManager.deinit() #释放vb buffer
def play_audio(filename):
try:
wf = wave.open(filename, 'rb')#打开wav文件
CHUNK = int(wf.get_framerate()/25)#设置音频chunk值
p = PyAudio()
p.initialize(CHUNK) #初始化PyAudio对象
MediaManager.init() #vb buffer初始化
#创建音频输出流，设置的音频参数均为wave中获取到的参数
stream = p.open(format=p.get_format_from_width(wf.get_sampwidth()),
channels=wf.get_channels(),
rate=wf.get_framerate(),
output=True,frames_per_buffer=CHUNK)
data = wf.read_frames(CHUNK)#从wav文件中读取数一帧数据
while data:
stream.write(data)#将帧数据写入到音频输出流中
data = wf.read_frames(CHUNK) #从wav文件中读取数一帧数据
if exit_check():
break
except BaseException as e:
print(f"Exception {e}")
finally:
stream.stop_stream() #停止音频输出流
stream.close()#关闭音频输出流
p.terminate()#释放音频对象
wf.close()#关闭wav文件
MediaManager.deinit() #释放vb buffer
def loop_audio(duration):
CHUNK = int(44100/25)#设置音频chunck
FORMAT = paInt16 #设置音频采样精度
CHANNELS = 2 #设置音频声道数
RATE = 44100 #设置音频采样率
try:
p = PyAudio()
p.initialize(CHUNK)#初始化PyAudio对象
MediaManager.init() #vb buffer初始化
#创建音频输入流
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)):
output_stream.write(input_stream.read())
if exit_check():
break
except BaseException as e:
print(f"Exception {e}")
finally:
input_stream.stop_stream()#停止音频输入流
output_stream.stop_stream()#停止音频输出流
input_stream.close() #关闭音频输入流
output_stream.close() #关闭音频输出流
p.terminate() #释放音频对象
MediaManager.deinit() #释放vb buffer
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
print("audio sample start")
play_audio('/sdcard/app/output.wav') #播放wav文件
#record_audio('/sdcard/app/output.wav', 15)#录制wav文件
loop_audio(15) #采集音频并输出
print("audio sample done")
</code></pre>
<p><strong> 实验结果：</strong></p>
<p align="justify"> 注意：需要用一个有线耳机插入到K230的3.5mm音频口，才可听到采集到的音频。</p>
<p> 下图1为执行音频采集代码之前“CanMV\sdcard\app”目录的文件情况，图2为执行代码之后的目录文件，可以看到成功采集到“output.wav”的音频文件。</p>
<p style="text-align: center;"> </p>
<p style="text-align: center;"> </p>
<p align="justify"> 接着注释掉音频录制代码，执行音频播放代码，如下图，可以从有线耳机听到刚刚采集到的音频。</p>
<p align="justify" style="text-align: center;"> </p>
<p><strong>2、音频编码与解码</strong></p>
<p align="justify"> 为了高效存储和传输音频数据，通过压缩技术减少文件大小和带宽需求，同时保持或优化音频质量，确保在不同设备和平台上的兼容性和一致性，因此需要音频编码与解码。G.711 是一种常用的音频编码标准，主要用于电话通信系统中。它通过脉冲编码调制（PCM）技术将模拟音频信号转换为数字信号，并进行量化和编码。G.711 编码的目的是在保持较高音频质量的同时，减少数据传输所需的带宽。K230的音频编码与解码采用G.711标准。</p>
<p align="justify"> 具体代码思路如下：</p>
<p align="justify"><b> 初始化模块：</b></p>
<p align="justify"> 导入必要的模块，包括 os、media、pyaudio 和 g711。</p>
<p align="justify"> 初始化媒体管理和音频对象。</p>
<p align="justify"><b> 异常处理：</b></p>
<p align="justify"> 定义 exit_check 函数，用于捕获键盘中断（Ctrl+C）并优雅地退出程序。</p>
<p align="justify"><b> 音频采集与编码：</b></p>
<p align="justify"> 定义 encode_audio 函数，用于采集音频数据并将其编码为 G.711 格式，然后保存到文件中。</p>
<p align="justify"> 设置音频参数，如采样率、采样精度、声道数和 chunk 大小。</p>
<p align="justify"> 打开音频输入流，读取音频数据并进行 G.711 编码，将编码后的数据保存到文件中。</p>
<p align="justify"><b> 音频解码与播放：</b></p>
<p align="justify"> 定义 decode_audio 函数，用于从文件中读取 G.711 编码的音频数据，解码为原始音频数据并播放。</p>
<p align="justify"> 打开 G.711 文件，读取音频参数。</p>
<p align="justify"> 打开音频输出流，读取 G.711 文件中的数据并解码为原始音频数据，然后写入输出流。</p>
<p align="justify"><b> 实时编码与解码回放：</b></p>
<p align="justify"> 定义 loop_codec 函数，用于实时采集音频数据，进行 G.711 编码和解码，然后立即播放。</p>
<p align="justify"> 设置音频参数，打开音频输入流和输出流。</p>
<p align="justify"> 从输入流中读取音频数据，进行 G.711 编码和解码，然后写入输出流。</p>
<p align="justify"><b> 主程序：</b></p>
<p align="justify"> 在 __main__ 块中，启用退出点，启动音频示例。</p>
<p align="justify"> 调用 encode_audio、decode_audio 或 loop_codec 函数来执行相应的音频操作。</p>
<p align="justify"><b> 参考代码如下：</b></p>
<pre>
<code class="language-python"># 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:
print("user stop: ", 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:
print(f"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:
print(f"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:
print(f"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', 5) #采集并编码g711文件
#decode_audio('/sdcard/app/test.g711a') #解码g711文件并输出
loop_codec(15) #采集音频数据->编码g711->解码g711->播放音频
print("audio codec sample done")
</code></pre>
<p align="justify"><b> 实验结果：</b></p>
<p align="justify"> 音频采集编码结果如下，生成“test.g711a”编码文件。</p>
<p align="justify" style="text-align: center;"> </p>
<p align="justify"> 解码并播放代码如下，可以从耳机听到解码后的音频。</p>
<p align="justify" style="text-align: center;"> </p>
<p><strong>3、视频采集</strong></p>
<p align="justify"> K230自带三个摄像头接口CSI0、CSI1、CSI2，如下图，可以用来图像识别，拍照，录像等功能，下面用K230录取一段视频保存在内存卡中。</p>
<p style="text-align: center;"> </p>
<p align="justify"> 具体代码思路如下：</p>
<p align="justify"><b> 初始化模块：</b></p>
<p align="justify"> 导入必要的模块，包括 media.mp4format 和 os。</p>
<p align="justify"> 初始化 MP4 容器和配置对象。</p>
<p align="justify"><b> 配置 MP4 容器：</b></p>
<p align="justify"> 设置 MP4 文件的路径、视频编码格式、分辨率和音频编码格式。</p>
<p align="justify"> 创建和启动 MP4 复用器（muxer）。</p>
<p align="justify"><b> 处理音视频数据：</b></p>
<p align="justify"> 在一个循环中，调用 MP4 复用器的 Process 方法，将音视频数据写入 MP4 文件。</p>
<p align="justify"> 控制循环次数，达到一定帧数后停止。</p>
<p align="justify"> 停止和销毁 MP4 复用器：</p>
<p align="justify"> 停止 MP4 复用器。</p>
<p align="justify"> 销毁 MP4 复用器，释放资源。</p>
<p align="justify"><b> 异常处理</b>：</p>
<p align="justify"> 捕获并处理可能发生的异常。</p>
<p><b> 参考代码如下：</b></p>
<pre>
<code class="language-python"># Save MP4 file example
#
# Note: You will need an SD card to run this example.
#
# You can capture audio and video and save them as MP4.The current version only supports MP4 format, video supports 264/265, and audio supports g711a/g711u.
from media.mp4format import *
import os
def mp4_muxer_test():
print("mp4_muxer_test start")
width = 1280
height = 720
# 实例化mp4 container
mp4_muxer = Mp4Container()
mp4_cfg = Mp4CfgStr(mp4_muxer.MP4_CONFIG_TYPE_MUXER)
if mp4_cfg.type == mp4_muxer.MP4_CONFIG_TYPE_MUXER:
file_name = "/sdcard/app/tests/test.mp4"
mp4_cfg.SetMuxerCfg(file_name, mp4_muxer.MP4_CODEC_ID_H265, width, height, mp4_muxer.MP4_CODEC_ID_G711U)
# 创建mp4 muxer
mp4_muxer.Create(mp4_cfg)
# 启动mp4 muxer
mp4_muxer.Start()
frame_count = 0
try:
while True:
os.exitpoint()
# 处理音视频数据，按MP4格式写入文件
mp4_muxer.Process()
frame_count += 1
print("frame_count = ", frame_count)
if frame_count >= 200:
break
except BaseException as e:
print(e)
# 停止mp4 muxer
mp4_muxer.Stop()
# 销毁mp4 muxer
mp4_muxer.Destroy()
print("mp4_muxer_test stop")
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
mp4_muxer_test()
</code></pre>
<p align="justify"><b> 实验结果：</b></p>
<p align="justify"> 可以看到在“\CanMV\sdcard\app\tests”目录下成功生成一个名为“test.mp4”的文件，可以用电脑带的播放器点开播放，视频带有声音。</p>
<p style="text-align: center;"> </p>
<p><strong>4、视频播放</strong></p>
<p align="justify"> 视频播放的思路如下：</p>
<p align="justify"><b> 初始化模块：</b></p>
<p align="justify"> 导入必要的模块，包括 media.player 和 os。</p>
<p align="justify"> 定义全局变量 start_play 用于控制播放状态。</p>
<p align="justify"><b> 定义播放器事件回调函数：</b></p>
<p align="justify"> 定义 player_event 函数，用于处理播放器事件，特别是播放结束事件。</p>
<p align="justify"><b> 加载和播放 MP4 文件：</b></p>
<p align="justify"> 创建播放器对象。</p>
<p align="justify"> 加载 MP4 文件。</p>
<p align="justify"> 设置播放器事件回调函数。</p>
<p align="justify"> 开始播放 MP4 文件。</p>
<p align="justify"> 等待播放结束。</p>
<p align="justify"><b> 停止播放：</b></p>
<p align="justify"> 捕获并处理可能发生的异常。</p>
<p align="justify"> 停止播放器，释放资源。</p>
<p align="justify"> <strong> 参考代码如下：</strong></p>
<pre>
<code class="language-python"># play mp4 file example
#
# Note: You will need an SD card to run this example.
#
# You can load local files to play. The current version only supports MP4 format, video supports 264/265, and audio supports g711a/g711u.
from media.player import * #导入播放器模块，用于播放mp4文件
import os
start_play = False #播放结束flag
def player_event(event,data):
global start_play
if(event == K_PLAYER_EVENT_EOF): #播放结束标识
start_play = False #设置播放结束标识
def play_mp4_test(filename):
global start_play
player=Player() #创建播放器对象
player.load(filename) #加载mp4文件
player.set_event_callback(player_event) #设置播放器事件回调
player.start() #开始播放
start_play = True
#等待播放结束
try:
while(start_play):
time.sleep(0.1)
os.exitpoint()
except KeyboardInterrupt as e:
print("user stop: ", e)
except BaseException as e:
sys.print_exception(e)
player.stop() #停止播放
print("play over")
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
play_mp4_test("/sdcard/app/tests/test.mp4")#播放mp4文件
</code></pre>
<p align="justify"><strong> 实验结果：</strong></p>
<p style="text-align: center;"> </p>
<p><strong>5、视频编码</strong></p>
<p align="justify"> 视频编码与解码是数字视频处理中不可或缺的步骤，通过压缩技术减少视频文件的大小和传输带宽需求，同时保持或优化视频质量，确保高效存储、快速传输和跨平台兼容性。编码将原始视频数据转换为压缩格式，解码则将压缩数据还原为可播放的视频，二者共同保证了视频内容的高效分发和流畅播放。</p>
<p align="justify"> 嘉楠K230内置多个高清视频图像输入处理和智能硬件处理单元，兼顾高性能、低功耗（ 采用大小核设计兼顾性能与功耗，提供百毫秒级快速启动软件SDK支持，适合电池类产品开发）和高安全性特点。</p>
<p align="justify"> 使用 Python 和 media 模块来捕获视频数据并将其编码为 H.264 或 H.265 格式的文件。</p>
<p align="justify"> 视频编码的思路如下：</p>
<p align="justify"><b> 初始化模块：</b></p>
<p align="justify"> 导入必要的模块，包括 media.vencoder、media.sensor、media.media 和 os。</p>
<p align="justify"> 初始化传感器（相机）和视频编码器。</p>
<p align="justify"><b> 配置传感器：</b></p>
<p align="justify"> 重置传感器。</p>
<p align="justify"> 设置传感器的输出分辨率和格式。</p>
<p align="justify"><b> 实例化和配置视频编码器：</b></p>
<p align="justify"> 创建视频编码器对象。</p>
<p align="justify"> 设置视频编码器的输出缓冲区。</p>
<p align="justify"> 绑定传感器和视频编码器。</p>
<p align="justify"><b> 启动传感器和编码器：</b></p>
<p align="justify"> 初始化媒体管理器。</p>
<p align="justify"> 创建和启动视频编码器。</p>
<p align="justify"> 启动传感器。</p>
<p align="justify"><b> 处理视频数据：</b></p>
<p align="justify"> 在一个循环中，从视频编码器获取编码后的码流数据，并将其写入文件。</p>
<p align="justify"> 控制循环次数，达到一定帧数后停止。</p>
<p align="justify"><b> 停止和销毁编码器：</b></p>
<p align="justify"> 停止传感器和编码器。</p>
<p align="justify"> 销毁传感器和编码器的绑定。</p>
<p align="justify"> 停止和销毁编码器，释放资源。</p>
<p align="justify"><b> 异常处理：</b></p>
<p align="justify"> 捕获并处理可能发生的异常。</p>
<p align="justify"> <strong>参考代码如下：</strong></p>
<pre>
<code class="language-python"># Video encode example
#
# Note: You will need an SD card to run this example.
#
# You can capture videos and encode them into 264 files
from media.vencoder import *
from media.sensor import *
from media.media import *
import time, os
# NOT WORK NOW!!!
def venc_test():
print("venc_test start")
width = 1280
height = 720
venc_chn = VENC_CHN_ID_0
width = ALIGN_UP(width, 16)
# 初始化sensor
sensor = Sensor()
sensor.reset()
# 设置camera 输出buffer
# set chn0 output size
sensor.set_framesize(width = width, height = height, alignment=12)
# set chn0 output format
sensor.set_pixformat(Sensor.YUV420SP)
# 实例化video encoder
encoder = Encoder()
# 设置video encoder 输出buffer
encoder.SetOutBufs(venc_chn, 15, width, height)
# 绑定camera和venc
link = MediaManager.link(sensor.bind_info()['src'], (VIDEO_ENCODE_MOD_ID, VENC_DEV_ID, venc_chn))
# init media manager
MediaManager.init()
chnAttr = ChnAttrStr(encoder.PAYLOAD_TYPE_H265, encoder.H265_PROFILE_MAIN, width, height)
streamData = StreamData()
# 创建编码器
encoder.Create(venc_chn, chnAttr)
# 开始编码
encoder.Start(venc_chn)
# 启动camera
sensor.run()
frame_count = 0
if chnAttr.payload_type == encoder.PAYLOAD_TYPE_H265:
suffix = "265"
elif chnAttr.payload_type == encoder.PAYLOAD_TYPE_H264:
suffix = "264"
else:
suffix = "unkown"
print("cam_venc_test, venc payload_type unsupport")
out_file = f"/sdcard/app/tests/venc_chn_{venc_chn:02d}.{suffix}"
print("save stream to file: ", out_file)
with open(out_file, "wb") as fo:
try:
while True:
os.exitpoint()
encoder.GetStream(venc_chn, streamData) # 获取一帧码流
for pack_idx in range(0, streamData.pack_cnt):
stream_data = uctypes.bytearray_at(streamData.data, streamData.data_size)
fo.write(stream_data) # 码流写文件
print("stream size: ", streamData.data_size, "stream type: ", streamData.stream_type)
encoder.ReleaseStream(venc_chn, streamData) # 释放一帧码流
frame_count += 1
if frame_count >= 100:
break
except KeyboardInterrupt as e:
print("user stop: ", e)
except BaseException as e:
sys.print_exception(e)
# 停止camera
sensor.stop()
# 销毁camera和venc的绑定
del link
# 停止编码
encoder.Stop(venc_chn)
# 销毁编码器
encoder.Destroy(venc_chn)
# 清理buffer
MediaManager.deinit()
print("venc_test stop")
if __name__ == "__main__":
os.exitpoint(os.EXITPOINT_ENABLE)
venc_test()
</code></pre>
<p align="justify"> 实验结果：</p>
<p align="justify"> 在”This PC\CanMV\sdcard\app\tests”目录下生成名为“venc_chnn_00.265”的编码文件。</p>
<p style="text-align: center;"> </p>
页:
[1]