diff --git a/gst/linescan/rollover_example.py b/gst/linescan/rollover_example.py deleted file mode 100644 index 14973af..0000000 --- a/gst/linescan/rollover_example.py +++ /dev/null @@ -1,198 +0,0 @@ -#!/usr/bin/env python3 -# /// script -# requires-python = ">=3.10" -# dependencies = ["numpy", "pillow"] -# /// -# -# Simple Rollover Example - Linescan Buffer Capture Demo -# -# This is a minimal example showing how to capture linescan images using -# the rollover signal with a videotestsrc. Files are saved via Python/PIL -# in the rollover callback, demonstrating the correct approach. -# -# Prerequisites: -# 1. Build the GStreamer plugins: .\build.ps1 -# 2. Ensure GST_PLUGIN_PATH includes the build directory -# -# Usage: -# uv run .\gst\linescan\rollover_example.py # Run with defaults -# python gst/linescan/rollover_example.py # Alternative without uv -# -# Features: -# - Uses videotestsrc (ball pattern) for testing -# - Saves images on rollover signal (not every frame) -# - Live preview window showing linescan accumulation -# - JPEG output to current directory -# -# Pipeline: -# videotestsrc -> linescan -> videoconvert -> autovideosink -# ↓ -# (rollover signal) -# ↓ -# Python callback saves file -# -# See launch_with_signal.py for a full-featured version with camera support. - -import os -import sys - -# Check for required environment variable -gst_root = os.environ.get("GSTREAMER_1_0_ROOT_MSVC_X86_64") -if not gst_root: - print("ERROR: GSTREAMER_1_0_ROOT_MSVC_X86_64 environment variable is not set") - print("Expected: C:\\bin\\gstreamer\\1.0\\msvc_x86_64\\") - sys.exit(1) -else: - # Remove trailing backslash if present - gst_root = gst_root.rstrip("\\") - - gst_site_packages = os.path.join(gst_root, "lib", "site-packages") - sys.path.insert(0, gst_site_packages) - - # Add GI typelibs - os.environ["GI_TYPELIB_PATH"] = os.path.join(gst_root, "lib", "girepository-1.0") - - # Add GStreamer DLL bin directory - os.environ["PATH"] = os.path.join(gst_root, "bin") + ";" + os.environ["PATH"] - -import gi -gi.require_version('Gst', '1.0') -from gi.repository import Gst, GLib -import numpy as np -from PIL import Image - -frame_counter = 0 - -def on_rollover(linescan, buffer): - """Callback when rollover signal is emitted - save buffer to JPEG""" - global frame_counter - - filename = f"linescan_frame_{frame_counter:04d}.jpeg" - - # Map the buffer to get image data - success, mapinfo = buffer.map(Gst.MapFlags.READ) - if not success: - print(f"[ERROR] Failed to map buffer for frame {frame_counter}") - return - - try: - # Get caps to determine image dimensions - caps = linescan.get_static_pad("src").get_current_caps() - structure = caps.get_structure(0) - width = structure.get_value('width') - height = structure.get_value('height') - format_str = structure.get_value('format') - - # Convert buffer data to numpy array - data = np.frombuffer(mapinfo.data, dtype=np.uint8) - - # Handle different formats - if format_str in ['RGB', 'BGR']: - channels = 3 - data = data.reshape((height, width, channels)) - if format_str == 'BGR': - data = data[:, :, ::-1] # Convert BGR to RGB - elif format_str == 'GRAY8': - channels = 1 - data = data.reshape((height, width)) - else: - data = data.reshape((height, width, -1)) - - # Create PIL Image and save - if channels == 1: - img = Image.fromarray(data, mode='L') - else: - img = Image.fromarray(data, mode='RGB') - - img.save(filename, 'JPEG', quality=95) - print(f"[ROLLOVER] Frame {frame_counter} saved to: {filename}") - - except Exception as e: - print(f"[ERROR] Failed to save frame {frame_counter}: {e}") - finally: - buffer.unmap(mapinfo) - frame_counter += 1 - -def main(): - Gst.init(None) - - # Create pipeline - simplified without file branch - # videotestsrc -> linescan -> videoconvert -> autovideosink - # ↓ - # (rollover callback saves files) - - pipeline = Gst.Pipeline.new("rollover-pipeline") - - # Create elements - source = Gst.ElementFactory.make("videotestsrc", "source") - linescan = Gst.ElementFactory.make("linescan", "linescan") - convert = Gst.ElementFactory.make("videoconvert", "convert") - videosink = Gst.ElementFactory.make("autovideosink", "videosink") - - if not all([pipeline, source, linescan, convert, videosink]): - print("ERROR: Not all elements could be created") - print("Make sure linescan plugin is built and in GST_PLUGIN_PATH") - return -1 - - # Configure elements - source.set_property("pattern", 18) # ball pattern - linescan.set_property("direction", 0) # horizontal - linescan.set_property("line-index", 100) - linescan.set_property("output-size", 400) - - # Connect to rollover signal - files saved in callback - linescan.connect("rollover", on_rollover) - - # Add elements to pipeline - pipeline.add(source) - pipeline.add(linescan) - pipeline.add(convert) - pipeline.add(videosink) - - # Link elements - simple chain - if not source.link(linescan): - print("ERROR: Could not link source to linescan") - return -1 - if not linescan.link(convert): - print("ERROR: Could not link linescan to convert") - return -1 - if not convert.link(videosink): - print("ERROR: Could not link convert to videosink") - return -1 - - # Start playing - pipeline.set_state(Gst.State.PLAYING) - - # Create main loop - loop = GLib.MainLoop() - - # Handle bus messages - bus = pipeline.get_bus() - bus.add_signal_watch() - - def on_message(bus, message): - t = message.type - if t == Gst.MessageType.EOS: - print("\nEnd-of-stream") - loop.quit() - elif t == Gst.MessageType.ERROR: - err, debug = message.parse_error() - print(f"\nError: {err}, {debug}") - loop.quit() - - bus.connect("message", on_message) - - print("Pipeline running. Files saved on rollover to current directory.") - print("Press Ctrl+C to stop.\n") - try: - loop.run() - except KeyboardInterrupt: - print("\nStopping...") - - # Cleanup - pipeline.set_state(Gst.State.NULL) - print(f"Captured {frame_counter} rollover frames") - return 0 - -if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file