p1-mp4-recording-gstreamer-10.ino

AIWorld P1 AI Module for BPI-M2 Ultra and Berry · File : p1-mp4-recording-gstreamer-10.ino · Language : Python · Use case : Enregistrer une vidéo en MP4 (H.264) à partir de la caméra OV5640 en utilisant GStreamer via Python pour une sauvegarde locale.

Python script to capture and encode real-time H.264 video into an MP4 container on local storage.

p1-mp4-recording-gstreamer-10.ino

#!/usr/bin/env python3
"""
AIWorld P1: Enregistrement vidéo H.264 via GStreamer Python bindings
Dépendances: python3-gi, gstreamer1.0-plugins-good, gstreamer1.0-plugins-bad, gstreamer1.0-libav
Exécution: python3 p1_gst_record.py
"""
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import sys
import time

# Initialisation de GStreamer
Gst.init(None)

def record_video(duration=5, filename="output.mp4"):
    pipeline_str = (
        "v4l2src device=/dev/video0 ! "
        "video/x-raw, width=1280, height=720, framerate=30/1 ! "
        "videoconvert ! "
        "queue ! "
        "omxh264enc target-bitrate=2000000 ! "
        "h264parse ! "
        "mp4mux ! "
        "filesink location={filename}"
    ).format(filename=filename)

    try:
        pipeline = Gst.parse_launch(pipeline_str)
        bus = pipeline.get_bus()
        
        # Passer à l'état PLAYING
        pipeline.set_state(Gst.State.PLAYING)
        print(f"Enregistrement de {filename} pour {duration} secondes...")
        
        # Définir un timer pour arrêter après 'duration' secondes
        # Note: Dans un script simple, on utilise time.sleep, mais un bus de messages est préférable pour la robustesse.
        start_time = time.time()
        while time.time() - start_time < duration:
            time.sleep(0.1)
            # Vérifier les erreurs ou EOS
            msg = bus.poll(Gst.MessageType.ERROR | Gst.MessageType.EOS, 100)
            if msg:
                if msg.type == Gst.MessageType.ERROR:
                    desc = msg.parse_error()[1]
                    print(f"Erreur GStreamer: {desc}")
                    break
                elif msg.type == Gst.MessageType.EOS:
                    print("Fin du flux (EOS).")
                    break
        
        # Arrêter proprement
        print("Arrêt de l'enregistrement...")
        pipeline.send_event(Gst.Event.new_eos())
        pipeline.set_state(Gst.State.NULL)
        print(f"Fichier {filename} créé.")

    except Exception as e:
        print(f"Exception: {e}")
        pipeline.set_state(Gst.State.NULL)

if __name__ == "__main__":
    if len(sys.argv) > 2:
        duration = int(sys.argv[1])
        filename = sys.argv[2]
    else:
        duration = 5
        filename = "aiworld_recording.mp4"
    
    record_video(duration, filename)

← Back to the board