p1-mp4-recording-gstreamer-10.ino
Script Python para capturar y codificar video H.264 en tiempo real en un contenedor MP4 en el almacenamiento local.
Utiliza los enlaces Python de GStreamer (gi.repository.Gst) para interactuar con el codificador de hardware H.264 (omxh264enc) de V40. Requiere los paquetes gstreamer1.0-plugins-good, bad y libav. Se requiere acceso a /dev/video0. La resolución se establece en 720p 30fps para optimizar el rendimiento. Ejecute el script con argumentos opcionales de duración y nombre de archivo. Verifique los permisos de escritura en el directorio actual.
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)