pympta & webcam dev tester

pipeline_webcam.py

@@ -0,0 +1,52 @@
+import argparse
+import os
+import cv2
+import time
+import logging
+
+from siwatsystem.pympta import load_pipeline_from_zip, run_pipeline
+
+logging.basicConfig(level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s")
+
+def main(mpta_url: str, video_source: str):
+    extraction_dir = os.path.join("models", "webcam_pipeline")
+    logging.info(f"Loading pipeline from {mpta_url}")
+    model_tree = load_pipeline_from_zip(mpta_url, extraction_dir)
+    if model_tree is None:
+        logging.error("Failed to load pipeline.")
+        return
+
+    cap = cv2.VideoCapture(video_source)
+    if not cap.isOpened():
+        logging.error(f"Cannot open video source {video_source}")
+        return
+
+    logging.info("Press 'q' to exit.")
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            logging.error("Failed to capture frame.")
+            break
+
+        detection, bbox = run_pipeline(frame, model_tree, return_bbox=True)
+        if bbox:
+            x1, y1, x2, y2 = bbox
+            cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
+            label = detection["class"] if detection else "Detection"
+            cv2.putText(frame, label, (x1, y1 - 10),
+                        cv2.FONT_HERSHEY_SIMPLEX, 0.9, (36, 255, 12), 2)
+
+        cv2.imshow("Pipeline Webcam", frame)
+        if cv2.waitKey(1) & 0xFF == ord('q'):
+            break
+
+    cap.release()
+    cv2.destroyAllWindows()
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Run pipeline webcam utility.")
+    parser.add_argument("--mpta-url", type=str, required=True, help="URL to the pipeline mpta (ZIP) file.")
+    parser.add_argument("--video", type=str, default="0", help="Video source (default webcam index 0).")
+    args = parser.parse_args()
+    video_source = int(args.video) if args.video.isdigit() else args.video
+    main(args.mpta_url, video_source)