p1-mp4-recording-gstreamer-10.ino
Script Python pour capturer et encoder en temps réel une vidéo H.264 dans un conteneur MP4 sur le stockage local.
Utilise les bindings Python de GStreamer (gi.repository.Gst) pour interagir avec l'encodeur matériel H.264 (omxh264enc) du V40. Nécessite les paquets gstreamer1.0-plugins-good, bad et libav. L'accès à /dev/video0 est requis. La résolution est fixée à 720p 30fps pour optimiser les performances. Exécutez le script avec les arguments de durée et de nom de fichier optionnels. Vérifiez les permissions d'écriture sur le répertoire courant.
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)