test python script

src/detectionSoftware/run.py

@@ -0,0 +1,177 @@
+import sys
+import threading
+import math
+import gi
+import logging
+from flask import Flask, Response, render_template_string
+from pypylon import pylon
+
+# GStreamer dependencies
+gi.require_version('Gst', '1.0')
+from gi.repository import Gst, GLib
+
+# --- Configuration ---
+STREAM_WIDTH = 1920
+STREAM_HEIGHT = 1080
+# The final output resolution of the tiled web stream
+WEB_OUTPUT_WIDTH = 1920
+WEB_OUTPUT_HEIGHT = 1080
+
+# --- Flask Setup ---
+app = Flask(__name__)
+frame_buffer = None
+buffer_lock = threading.Lock()
+
+def discover_cameras():
+    """
+    Uses pypylon to find all connected Basler cameras.
+    Returns a list of Serial Numbers.
+    """
+    tl_factory = pylon.TlFactory.GetInstance()
+    devices = tl_factory.EnumerateDevices()
+    
+    serials = []
+    for dev in devices:
+        serials.append(dev.GetSerialNumber())
+    
+    if not serials:
+        print("CRITICAL ERROR: No Basler cameras detected via Pylon.")
+        sys.exit(1)
+        
+    print(f"Discovered {len(serials)} cameras: {serials}")
+    return serials
+
+class GStreamerPipeline(threading.Thread):
+    def __init__(self, camera_serials):
+        super().__init__()
+        self.camera_serials = camera_serials
+        self.loop = GLib.MainLoop()
+        self.pipeline = None
+
+    def calculate_grid(self, num_cams):
+        """
+        Calculates the most square-like grid (rows, cols) for N cameras.
+        """
+        rows = int(math.ceil(math.sqrt(num_cams)))
+        cols = int(math.ceil(num_cams / rows))
+        return rows, cols
+
+    def run(self):
+        Gst.init(None)
+        self.build_dynamic_pipeline()
+        self.pipeline.set_state(Gst.State.PLAYING)
+        try:
+            self.loop.run()
+        except Exception as e:
+            print(f"Error in GStreamer loop: {e}")
+        finally:
+            self.pipeline.set_state(Gst.State.NULL)
+
+    def on_new_sample(self, sink):
+        """
+        Callback: grabs the already-encoded JPEG from the pipeline.
+        """
+        sample = sink.emit("pull-sample")
+        if not sample:
+            return Gst.FlowReturn.ERROR
+
+        buffer = sample.get_buffer()
+        success, map_info = buffer.map(Gst.MapFlags.READ)
+        
+        if not success:
+            return Gst.FlowReturn.ERROR
+
+        global frame_buffer
+        with buffer_lock:
+            frame_buffer = bytes(map_info.data)
+
+        buffer.unmap(map_info)
+        return Gst.FlowReturn.OK
+
+    def build_dynamic_pipeline(self):
+        num_cams = len(self.camera_serials)
+        rows, cols = self.calculate_grid(num_cams)
+        
+        print(f"Building pipeline for {num_cams} cameras (Grid: {cols}x{rows})")
+
+        # 1. Construct Sources
+        # We need to build N pylonsrc elements, each linking to a specific pad on the muxer.
+        sources_str = ""
+        for i, serial in enumerate(self.camera_serials):
+            # We explicitly link to muxer sink pad: m.sink_0, m.sink_1, etc.
+            sources_str += (
+                f"pylonsrc camera-device-serial-number={serial} ! "
+                f"videoconvert ! nvvideoconvert ! m.sink_{i} "
+            )
+
+        # 2. Configure Muxer
+        # batch-size must match number of cameras
+        muxer_str = (
+            f"nvstreammux name=m batch-size={num_cams} "
+            f"width={STREAM_WIDTH} height={STREAM_HEIGHT} live-source=1 "
+        )
+
+        # 3. Configure Tiler
+        # This combines the batch into one 2D image
+        tiler_str = (
+            f"nvmultistreamtiler width={WEB_OUTPUT_WIDTH} height={WEB_OUTPUT_HEIGHT} "
+            f"rows={rows} columns={cols} "
+        )
+
+        # 4. Final Processing (Convert -> JPEG -> AppSink)
+        output_str = (
+            "nvvideoconvert ! video/x-raw, format=I420 ! "
+            "jpegenc quality=85 ! "
+            "appsink name=sink emit-signals=True sync=False max-buffers=1 drop=True"
+        )
+
+        # Combine all parts
+        full_pipeline_str = f"{sources_str} {muxer_str} ! {tiler_str} ! {output_str}"
+        
+        print(f"Pipeline String:\n{full_pipeline_str}")
+        
+        self.pipeline = Gst.parse_launch(full_pipeline_str)
+        
+        # Link callback
+        appsink = self.pipeline.get_by_name("sink")
+        appsink.connect("new-sample", self.on_new_sample)
+
+# --- Flask Routes ---
+
+@app.route('/')
+def index():
+    return render_template_string('''
+        <html>
+        <body style="background:#111; color:white; text-align:center;">
+            <h1>Basler Auto-Discovery Feed</h1>
+            <img src="{{ url_for('video_feed') }}" style="border: 2px solid green; max-width:95%;">
+        </body>
+        </html>
+    ''')
+
+@app.route('/video_feed')
+def video_feed():
+    def generate():
+        while True:
+            with buffer_lock:
+                if frame_buffer:
+                    yield (b'--frame\r\n'
+                           b'Content-Type: image/jpeg\r\n\r\n' + frame_buffer + b'\r\n')
+            GLib.usleep(15000) # ~60fps poll cap
+
+    return Response(generate(), mimetype='multipart/x-mixed-replace; boundary=frame')
+
+# --- Main ---
+
+if __name__ == "__main__":
+    # 1. Discover Cameras
+    found_serials = discover_cameras()
+
+    # 2. Start Pipeline Thread
+    gst_thread = GStreamerPipeline(found_serials)
+    gst_thread.daemon = True
+    gst_thread.start()
+
+    # 3. Start Web Server
+    print(f"Stream available at http://0.0.0.0:5000")
+    app.run(host='0.0.0.0', port=5000, debug=False, threaded=True)