mp4demuxer.py 程序是一个MP4解复用(demux)的例子,和前面一篇文章《02-Media-4-mp4muxer.py 录制视频并保存为MP4文件的示例》配套使用,用于从MP4文件中提取视频和音频流。它支持H.264和H.265视频编解码器,以及G.711A和G.711U音频编解码器。

程序主要流程如下:

  1. 导入必要的模块,包括媒体处理相关的模块、MP4格式解析模块、音频处理模块、显示模块等。

  2. 定义了一个demuxer_mp4函数,该函数接收一个MP4文件名作为参数,传入的地址为上篇文章中设定的MP4文件保存路径。

  3. demuxer_mp4函数中,首先配置MP4解复用器,设置文件名称和标志。

  4. 创建MP4解复用器句柄,如果创建失败则抛出异常。

  5. 获取MP4文件信息,包括轨道数量和持续时间。

  6. 遍历每个轨道,获取轨道信息。如果是视频轨道(H.264或H.265),则记录视频信息;如果是音频轨道(G.711A或G.711U),则记录音频信息。

  7. 如果没有找到视频轨道,则抛出异常。

  8. 然后进入一个循环,从MP4文件中读取每一帧数据。如果是视频帧,则根据时间戳进行同步(通过延时来匹配视频播放的时间);如果是音频帧,则同样处理,示例中没有对音频做同步处理,只是打印了信息。

  9. 当读取到文件结束帧时,退出循环。

  10. 最后销毁MP4解复用器句柄。

  11. 在主函数中,设置退出点并调用demuxer_mp4函数,指定要解复用的MP4文件路径。

需要注意的是,程序中对视频帧进行了简单的同步处理,通过比较系统时间和视频时间戳来调整播放速度,但音频帧并没有进行同步播放,只是打印了信息。在项目应用中,可能需要同时处理音频和视频的同步播放。

程序运行输出效果如下:


=====file_info: track_num: 2 duration: 17520
    codec_id:  1
    track_id:  1
    width:  1280
    height:  720
    codec_id:  3
    track_id:  2
    channels:  1
    sample_rate:  8000
    bit_per_sample:  16
video frame_data.codec_id: 1 data_length: 4871 timestamp: 0
audio frame_data.codec_id: 3 data_length: 320 timestamp: 0
video frame_data.codec_id: 1 data_length: 3234 timestamp: 17
audio frame_data.codec_id: 3 data_length: 320 timestamp: 40
video frame_data.codec_id: 1 data_length: 1328 timestamp: 35
audio frame_data.codec_id: 3 data_length: 320 timestamp: 80
video frame_data.codec_id: 1 data_length: 25850 timestamp: 52
audio frame_data.codec_id: 3 data_length: 320 timestamp: 120
video frame_data.codec_id: 1 data_length: 3080 timestamp: 70
audio frame_data.codec_id: 3 data_length: 320 timestamp: 160
...省略类似输出...
video frame_data.codec_id: 1 data_length: 18196 timestamp: 17351
video frame_data.codec_id: 1 data_length: 87 timestamp: 17352
MPY: soft reboot
CanMV v1.3-147-g46fd58c(based on Micropython e00a144) on 2025-09-02; k230_canmv_01studio with K230

源程序如下:

# MP4 Demuxer Example
#
# This script demuxes an MP4 file, extracting video and audio streams.
# Supported video codecs: H.264, H.265
# Supported audio codecs: G.711A, G.711U

from media.media import *
from mpp.mp4_format import *
from mpp.mp4_format_struct import *
from media.pyaudio import *
import media.g711 as g711
from mpp.payload_struct import *
import media.vdecoder as vdecoder
from media.display import *
import uctypes
import time
import _thread
import os

def demuxer_mp4(filename):
    mp4_cfg = k_mp4_config_s()
    video_info = k_mp4_video_info_s()
    video_track = False
    audio_info = k_mp4_audio_info_s()
    audio_track = False
    mp4_handle = k_u64_ptr()

    mp4_cfg.config_type = K_MP4_CONFIG_DEMUXER
    mp4_cfg.muxer_config.file_name[:] = bytes(filename, 'utf-8')
    mp4_cfg.muxer_config.fmp4_flag = 0

    ret = kd_mp4_create(mp4_handle, mp4_cfg)
    if ret:
        raise OSError("kd_mp4_create failed:",filename)

    file_info = k_mp4_file_info_s()
    kd_mp4_get_file_info(mp4_handle.value, file_info)
    print("=====file_info: track_num:",file_info.track_num,"duration:",file_info.duration)

    for i in range(file_info.track_num):
        track_info = k_mp4_track_info_s()
        ret = kd_mp4_get_track_by_index(mp4_handle.value, i, track_info)
        if (ret < 0):
            raise ValueError("kd_mp4_get_track_by_index failed")

        if (track_info.track_type == K_MP4_STREAM_VIDEO):
            if (track_info.video_info.codec_id == K_MP4_CODEC_ID_H264 or track_info.video_info.codec_id == K_MP4_CODEC_ID_H265):
                video_track = True
                video_info = track_info.video_info
                print("    codec_id: ", video_info.codec_id)
                print("    track_id: ", video_info.track_id)
                print("    width: ", video_info.width)
                print("    height: ", video_info.height)
            else:
                print("video not support codecid:",track_info.video_info.codec_id)
        elif (track_info.track_type == K_MP4_STREAM_AUDIO):
            if (track_info.audio_info.codec_id == K_MP4_CODEC_ID_G711A or track_info.audio_info.codec_id == K_MP4_CODEC_ID_G711U):
                audio_track = True
                audio_info = track_info.audio_info
                print("    codec_id: ", audio_info.codec_id)
                print("    track_id: ", audio_info.track_id)
                print("    channels: ", audio_info.channels)
                print("    sample_rate: ", audio_info.sample_rate)
                print("    bit_per_sample: ", audio_info.bit_per_sample)
                #audio_info.channels = 2
            else:
                print("audio not support codecid:",track_info.audio_info.codec_id)

    if (video_track == False):
        raise ValueError("video track not found")

    # 记录初始系统时间
    start_system_time = time.ticks_ms()
    # 记录初始视频时间戳
    start_video_timestamp = 0

    while (True):
        frame_data =  k_mp4_frame_data_s()
        ret = kd_mp4_get_frame(mp4_handle.value, frame_data)
        if (ret < 0):
            raise OSError("get frame data failed")

        if (frame_data.eof):
            break

        if (frame_data.codec_id == K_MP4_CODEC_ID_H264 or frame_data.codec_id == K_MP4_CODEC_ID_H265):
            data = uctypes.bytes_at(frame_data.data,frame_data.data_length)

            print("video frame_data.codec_id:",frame_data.codec_id,"data_length:",frame_data.data_length,"timestamp:",frame_data.time_stamp)

            # 计算视频时间戳经历的时长
            video_timestamp_elapsed = frame_data.time_stamp - start_video_timestamp
            # 计算系统时间戳经历的时长
            current_system_time = time.ticks_ms()
            system_time_elapsed = current_system_time - start_system_time

            # 如果系统时间戳经历的时长小于视频时间戳经历的时长,进行延时
            if system_time_elapsed < video_timestamp_elapsed:
                time.sleep_ms(video_timestamp_elapsed - system_time_elapsed)

        elif(frame_data.codec_id == K_MP4_CODEC_ID_G711A or frame_data.codec_id == K_MP4_CODEC_ID_G711U):
            data = uctypes.bytes_at(frame_data.data,frame_data.data_length)
            print("audio frame_data.codec_id:",frame_data.codec_id,"data_length:",frame_data.data_length,"timestamp:",frame_data.time_stamp)

    kd_mp4_destroy(mp4_handle.value)

if __name__ == "__main__":
    os.exitpoint(os.EXITPOINT_ENABLE)
    demuxer_mp4("/sdcard/examples/test.mp4")

给自己留个问题,在程序中:

if (frame_data.codec_id == K_MP4_CODEC_ID_H264 or frame_data.codec_id == K_MP4_CODEC_ID_H265):
            data = uctypes.bytes_at(frame_data.data,frame_data.data_length)

此次将MP4的视频数据进行了解码,获得了画面的字节数据,但是没有显示到液晶屏或HDMI接口输出,以后有机会在此处增加显示功能,将解码的MP4画面通过HDMI显示出来。

Logo

智能硬件社区聚焦AI智能硬件技术生态,汇聚嵌入式AI、物联网硬件开发者,打造交流分享平台,同步全国赛事资讯、开展 OPC 核心人才招募,助力技术落地与开发者成长。

更多推荐