嘉楠科K230AI开发板测评8--音频采集、播放、编码与解码、视频采集、播放与编码

dfjs

嘉楠科K230AI开发板测评8--音频采集、播放、编码与解码、视频采集、播放与编码 [复制链接]

 

嘉楠科K230AI开发板测评8

1、音频采集与播放

    K230自带采集与播放接口，接口位置如下左图所示，音频输入为：麦克风咪头，音频输出：3.5mm音频口（双声道），查看音频采集与播放的原理图，可能是考虑到功率问题，没有功率放大电路，而采用了3.5mm的音频口输出而不是直接喇叭外放。

 

 

 

    具体代码思路如下：

    初始化模块：

        导入必要的模块，包括 os、media、pyaudio 和 wave。

        初始化媒体管理和音频对象。

    异常处理：

        定义 exit_check 函数，用于捕获键盘中断（Ctrl+C）并优雅地退出程序。

    音频录制：

        定义 record_audio 函数，用于录制音频并保存为 WAV 文件。

        设置音频参数，如采样率、采样精度、声道数和 chunk 大小。

        打开音频输入流，读取音频数据并存储到列表中。

        将列表中的音频数据保存到 WAV 文件中。

    音频播放：

        定义 play_audio 函数，用于播放 WAV 文件。

        打开 WAV 文件，读取音频参数。

        打开音频输出流，读取 WAV 文件中的音频数据并写入输出流。

    实时回放：

        定义 loop_audio 函数，用于实时采集音频并立即播放。

        设置音频参数，打开音频输入流和输出流。

        从输入流中读取音频数据并立即写入输出流。

    主程序：

        在 __main__ 块中，启用退出点，启动音频示例。

        调用 play_audio、record_audio 或 loop_audio 函数来执行相应的音频操作。

    参考代码如下：

# 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")

    实验结果：

    注意：需要用一个有线耳机插入到K230的3.5mm音频口，才可听到采集到的音频。

    下图1为执行音频采集代码之前“CanMV\sdcard\app”目录的文件情况，图2为执行代码之后的目录文件，可以看到成功采集到“output.wav”的音频文件。

 

 

    接着注释掉音频录制代码，执行音频播放代码，如下图，可以从有线耳机听到刚刚采集到的音频。

 

2、音频编码与解码

    为了高效存储和传输音频数据，通过压缩技术减少文件大小和带宽需求，同时保持或优化音频质量，确保在不同设备和平台上的兼容性和一致性，因此需要音频编码与解码。G.711 是一种常用的音频编码标准，主要用于电话通信系统中。它通过脉冲编码调制（PCM）技术将模拟音频信号转换为数字信号，并进行量化和编码。G.711 编码的目的是在保持较高音频质量的同时，减少数据传输所需的带宽。K230的音频编码与解码采用G.711标准。

    具体代码思路如下：

    初始化模块：

        导入必要的模块，包括 os、media、pyaudio 和 g711。

        初始化媒体管理和音频对象。

    异常处理：

        定义 exit_check 函数，用于捕获键盘中断（Ctrl+C）并优雅地退出程序。

    音频采集与编码：

        定义 encode_audio 函数，用于采集音频数据并将其编码为 G.711 格式，然后保存到文件中。

        设置音频参数，如采样率、采样精度、声道数和 chunk 大小。

        打开音频输入流，读取音频数据并进行 G.711 编码，将编码后的数据保存到文件中。

    音频解码与播放：

        定义 decode_audio 函数，用于从文件中读取 G.711 编码的音频数据，解码为原始音频数据并播放。

        打开 G.711 文件，读取音频参数。

        打开音频输出流，读取 G.711 文件中的数据并解码为原始音频数据，然后写入输出流。

    实时编码与解码回放：

        定义 loop_codec 函数，用于实时采集音频数据，进行 G.711 编码和解码，然后立即播放。

        设置音频参数，打开音频输入流和输出流。

        从输入流中读取音频数据，进行 G.711 编码和解码，然后写入输出流。

    主程序：

        在 __main__ 块中，启用退出点，启动音频示例。

        调用 encode_audio、decode_audio 或 loop_codec 函数来执行相应的音频操作。

    参考代码如下：

# 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")

    实验结果：

    音频采集编码结果如下，生成“test.g711a”编码文件。

 

    解码并播放代码如下，可以从耳机听到解码后的音频。

 

3、视频采集

    K230自带三个摄像头接口CSI0、CSI1、CSI2，如下图，可以用来图像识别，拍照，录像等功能，下面用K230录取一段视频保存在内存卡中。

 

    具体代码思路如下：

    初始化模块：

        导入必要的模块，包括 media.mp4format 和 os。

        初始化 MP4 容器和配置对象。

    配置 MP4 容器：

        设置 MP4 文件的路径、视频编码格式、分辨率和音频编码格式。

        创建和启动 MP4 复用器（muxer）。

    处理音视频数据：

        在一个循环中，调用 MP4 复用器的 Process 方法，将音视频数据写入 MP4 文件。

        控制循环次数，达到一定帧数后停止。

        停止和销毁 MP4 复用器：

        停止 MP4 复用器。

        销毁 MP4 复用器，释放资源。

    异常处理：

        捕获并处理可能发生的异常。

    参考代码如下：

# 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()

    实验结果：

    可以看到在“\CanMV\sdcard\app\tests”目录下成功生成一个名为“test.mp4”的文件，可以用电脑带的播放器点开播放，视频带有声音。

 

4、视频播放

    视频播放的思路如下：

    初始化模块：

        导入必要的模块，包括 media.player 和 os。

        定义全局变量 start_play 用于控制播放状态。

    定义播放器事件回调函数：

        定义 player_event 函数，用于处理播放器事件，特别是播放结束事件。

    加载和播放 MP4 文件：

        创建播放器对象。

        加载 MP4 文件。

        设置播放器事件回调函数。

        开始播放 MP4 文件。

        等待播放结束。

    停止播放：

        捕获并处理可能发生的异常。

        停止播放器，释放资源。

    参考代码如下：

# 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文件

    实验结果：

 

5、视频编码

    视频编码与解码是数字视频处理中不可或缺的步骤，通过压缩技术减少视频文件的大小和传输带宽需求，同时保持或优化视频质量，确保高效存储、快速传输和跨平台兼容性。编码将原始视频数据转换为压缩格式，解码则将压缩数据还原为可播放的视频，二者共同保证了视频内容的高效分发和流畅播放。

    嘉楠K230内置多个高清视频图像输入处理和智能硬件处理单元，兼顾高性能、低功耗（ 采用大小核设计兼顾性能与功耗，提供百毫秒级快速启动软件SDK支持，适合电池类产品开发）和高安全性特点。

    使用 Python 和 media 模块来捕获视频数据并将其编码为 H.264 或 H.265 格式的文件。

    视频编码的思路如下：

    初始化模块：

        导入必要的模块，包括 media.vencoder、media.sensor、media.media 和 os。

        初始化传感器（相机）和视频编码器。

    配置传感器：

        重置传感器。

        设置传感器的输出分辨率和格式。

    实例化和配置视频编码器：

        创建视频编码器对象。

        设置视频编码器的输出缓冲区。

        绑定传感器和视频编码器。

    启动传感器和编码器：

        初始化媒体管理器。

        创建和启动视频编码器。

        启动传感器。

    处理视频数据：

        在一个循环中，从视频编码器获取编码后的码流数据，并将其写入文件。

        控制循环次数，达到一定帧数后停止。

    停止和销毁编码器：

        停止传感器和编码器。

        销毁传感器和编码器的绑定。

        停止和销毁编码器，释放资源。

    异常处理：

        捕获并处理可能发生的异常。

    参考代码如下：

# 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[pack_idx], streamData.data_size[pack_idx])
                    fo.write(stream_data) # 码流写文件
                    print("stream size: ", streamData.data_size[pack_idx], "stream type: ", streamData.stream_type[pack_idx])

                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()

    实验结果：

    在”This PC\CanMV\sdcard\app\tests”目录下生成名为“venc_chnn_00.265”的编码文件。