Barcode detector and web reporter update

Luxonis Apps/QR Code Reader/app.py

@@ -68,6 +68,8 @@ def manage_device(self, device: dai.Device):
                                                  input_names=["high_res_rgb", "qr_bboxes", "h264_frame"],
                                                  output_message_obj=messages.FramesWithDetections)
         qr_code_decoder = host_node.QrCodeDecoder(input_node=qr_boxes_and_frame_sync, qr_crop_queue=qr_crops_queue)
+        if rh.CONFIGURATION["enable_web_reporter"]:
+            host_node.WebReporter(input_node=qr_code_decoder)
         host_node.ResultsReporter(input_node=qr_code_decoder)
         host_node.Monitor(input_node=qr_code_decoder, name="qr_boxes_and_frame_sync")
 

Luxonis Apps/QR Code Reader/app_PoC.py

@@ -11,10 +11,14 @@
 OUTPUT_TO_FILE = False
 OUTPUT_FILENAME = "results.txt"
 
-NN_INPUT_SIZE_W = 512
-NN_INPUT_SIZE_H = 288
+# NN_INPUT_SIZE_W = 512
+# NN_INPUT_SIZE_H = 288
 
-CONFIDENCE_THRESHOLD = 0.2
+NN_INPUT_SIZE_W = 416
+NN_INPUT_SIZE_H = 416
+
+
+CONFIDENCE_THRESHOLD = 0.75
 
 NUMBER_OF_CROPPED_IMAGES = 9
 crop_vals = [(0.0, 0.0), (0.0, 0.3), (0.0, 0.6), (0.3, 0.0), (0.3, 0.3), (0.3, 0.6), (0.6, 0.0), (0.6, 0.3), (0.6, 0.6)]
@@ -34,7 +38,7 @@
 cam.setResolution(depthai.ColorCameraProperties.SensorResolution.THE_1080_P)
 cam.setColorOrder(depthai.ColorCameraProperties.ColorOrder.BGR)
 cam.setInterleaved(False)
-cam.setPreviewSize(1920,1080)
+cam.setPreviewSize(1920, 1080)
 
 # Script
 # https://docs.luxonis.com/projects/api/en/latest/components/nodes/script/
@@ -98,7 +102,8 @@ def create_image_manip(crop):
 # YoloDetectionNetwork
 # https://docs.luxonis.com/projects/api/en/latest/components/nodes/yolo_detection_network/
 nn_yolo = pipeline.create(depthai.node.YoloDetectionNetwork)
-nn_yolo.setBlobPath(str((Path(__file__).parent / Path('qr_model_512x288_rvc2_openvino_2022.1_6shave.blob')).resolve().absolute()))
+# nn_yolo.setBlobPath(str((Path('models\\qr_model_512x288_rvc2_openvino_2022.1_6shave.blob')).resolve().absolute()))
+nn_yolo.setBlobPath(str((Path('models/barcode-det_best-416x416_openvino_2022.1_6shave.blob')).resolve().absolute()))
 nn_yolo.setConfidenceThreshold(CONFIDENCE_THRESHOLD)
 nn_yolo.setNumClasses(1)
 nn_yolo.setCoordinateSize(4)

Luxonis Apps/QR Code Reader/app_pipeline/host_node/__init__.py

@@ -7,3 +7,4 @@
 from .results_reporter import *
 from .sync import *
 from .video_reporter import *
+from .web_reporter import *

Luxonis Apps/QR Code Reader/app_pipeline/host_node/qr_code_decoder.py

@@ -69,6 +69,7 @@ def __callback(self, frames_and_detections: messages.FramesWithDetections):
                         log.warning(f"More than one QR code detected in crop {i}")
                     decoded_code = decoded_codes[0]
                     bbox.set_label(label=decoded_code.text)
+                    bbox.set_format(code_format=str(decoded_code.format))
         # cv2.imshow("4k", high_res_frame)
         if len(qr_bboxes.bounding_boxes) > 0:
             if qr_bboxes.bounding_boxes[0].crop.getSequenceNum() != qr_bboxes.bounding_boxes[-1].crop.getSequenceNum():

Luxonis Apps/QR Code Reader/app_pipeline/host_node/web_reporter.py

@@ -0,0 +1,75 @@
+import logging as log
+from collections import deque
+
+import robothub as rh
+from datetime import datetime
+import requests
+
+from app_pipeline import host_node, messages
+
+__all__ = ["WebReporter"]
+
+
+class WebReporter(host_node.BaseNode):
+    NOT_SEEN_THRESHOLD = 10
+    ELAPSED_TIME_THRESHOLD = 10
+    MAX_REPORT_BUFFER_LEN = 4
+
+    def __init__(self, input_node: host_node.BaseNode):
+        super().__init__()
+        input_node.set_callback(callback=self.__callback)
+
+        self._web_report_buffer = deque(maxlen=self.MAX_REPORT_BUFFER_LEN)
+        self._qr_code_memory = {}  # label -> not seen for x frames
+
+    def __callback(self, frames_and_detections: messages.FramesWithDetections):
+        qr_detections = frames_and_detections.qr_bboxes.bounding_boxes
+        new_qr_codes = {}
+        existing_qr_codes = {}
+        for qr_code in qr_detections:
+            if qr_code.label and qr_code.label not in self._qr_code_memory:
+                self._qr_code_memory[qr_code.label] = 0
+                new_qr_codes[qr_code.label] = qr_code
+            else:
+                existing_qr_codes[qr_code.label] = qr_code
+
+        if new_qr_codes:
+            log.info(f"[WebReporter] New QR codes found: {new_qr_codes.keys()}")
+            qr_boxes = messages.QrBoundingBoxes(bounding_boxes=list(new_qr_codes.values()),
+                                                sequence_number=frames_and_detections.getSequenceNum())
+            for bbox in qr_boxes.bounding_boxes:
+                web_report = messages.WebReport(crop_image=bbox.crop.getCvFrame(), label=bbox.label,
+                                                code_format=bbox.code_format, timestamp=datetime.now(),
+                                                sequence_number=frames_and_detections.getSequenceNum())
+                if len(self._web_report_buffer) < self._web_report_buffer.maxlen:
+                    self._web_report_buffer.append(web_report)
+                else:
+                    log.warning(f"[WebReporter] Too many reports in buffer, dropping {web_report.getSequenceNum()}")
+
+        if len(self._web_report_buffer) > 0:
+            web_report: messages.WebReport = self._web_report_buffer.popleft()
+            log.info(f"[WebReporter] Sending QR code report with label {web_report.label}")
+            self.__send_report(web_report)
+
+        for qr_code_label in list(self._qr_code_memory.keys()):
+            # when seen, reset counter to zero, because it means for how long ar label was not spotted
+            if qr_code_label in existing_qr_codes:
+                self._qr_code_memory[qr_code_label] = 0
+            # not in new and not in existing, increment counter
+            elif qr_code_label not in new_qr_codes:
+                self._qr_code_memory[qr_code_label] += 1
+                if self._qr_code_memory[qr_code_label] >= self.NOT_SEEN_THRESHOLD:
+                    log.info(f"[WebReporter] QR code {qr_code_label} not seen for {self._qr_code_memory[qr_code_label]} frames, removing from memory.")
+                    self._qr_code_memory.pop(qr_code_label)
+
+    @staticmethod
+    def __send_report(report: messages.WebReport) -> None:
+        """Send the report on the customer URL. URL is from robotapp.toml configuration file."""
+        try:
+            response = requests.post(rh.CONFIGURATION["url"], json=report.to_dict())
+            if response.status_code == 200:
+                log.info("[WebReporter] Data was successfully received!")
+            else:
+                log.error(f"[WebReporter] {response.status_code} - {response.text}")
+        except Exception as e:
+            log.error(f"[WebReporter] {e}")

Luxonis Apps/QR Code Reader/app_pipeline/messages.py

@@ -2,6 +2,9 @@
 
 import depthai as dai
 import numpy as np
+from datetime import datetime
+import cv2
+import base64
 
 from node_helpers import BoundingBox
 
@@ -37,3 +40,25 @@ class FramesWithDetections(Message):
 class RhReport(Message):
     context_image: np.ndarray
     qr_bboxes: QrBoundingBoxes
+
+
+@dataclass(slots=True, kw_only=True)
+class WebReport(Message):
+    crop_image: np.ndarray
+    label: str
+    code_format: str
+    timestamp: datetime
+
+    def to_dict(self) -> dict:
+        return {
+            "label": self.label,
+            "code_format": self.code_format,
+            "timestamp": str(self.timestamp),
+            "crop_image": self.__encode_image_to_base64()
+        }
+
+    def __encode_image_to_base64(self) -> str:
+        """Convert np.ndarray to string."""
+        _, buffer = cv2.imencode('.jpg', self.crop_image)
+        encoded_string = base64.b64encode(buffer).decode('utf-8')
+        return encoded_string

Luxonis Apps/QR Code Reader/app_pipeline/oak_pipeline.py

@@ -18,6 +18,7 @@ def create_pipeline(pipeline: dai.Pipeline) -> None:
 
     script_node = create_script_node(pipeline=pipeline, script_name="app_pipeline/script_node.py")
     script_node_qr_crops = create_script_node(pipeline=pipeline, script_name="app_pipeline/script_node_qr_crops.py")
+    # camera (isp) > script
     rgb_sensor.isp.link(script_node.inputs["rgb_frame"])
     rgb_sensor.isp.link(script_node_qr_crops.inputs["rgb_frame"])
     script_node.inputs["rgb_frame"].setBlocking(True)
@@ -39,16 +40,23 @@ def create_pipeline(pipeline: dai.Pipeline) -> None:
     h264_encoder.input.setQueueSize(2)
     script_node.outputs["image_manip_1to1_crop_cfg"].link(image_manip_1to1_crop.inputConfig)
 
-    nn_input_width = 512
-    nn_input_height = 512 if rh.CONFIGURATION["resolution"] == "5312x6000" else 288
-    if rh.CONFIGURATION["resolution"] == "5312x6000":
-        nn_model_path = "models/qrdet-512x512_openvino_2022.1_3shave.blob"
-    elif rh.CONFIGURATION["resolution"] == "4k":
-        nn_model_path = "models/qrdet-512x288_openvino_2022.1_5shave.blob"
-    elif rh.CONFIGURATION["resolution"] == "1080p":
-        nn_model_path = "models/qr_model_512x288_rvc2_openvino_2022.1_6shave.blob"
+    if rh.CONFIGURATION["barcode_mode"]:
+        nn_input_width = 416
+        nn_input_height = 416
+        nn_model_path = "models\\barcode-det_best-416x416_openvino_2022.1_6shave.blob"
+        nn_confidence_threshold = 0.75
     else:
-        raise ValueError(f"Unknown resolution: {rh.CONFIGURATION['resolution']}")
+        nn_input_width = 512
+        nn_input_height = 512 if rh.CONFIGURATION["resolution"] == "5312x6000" else 288
+        nn_confidence_threshold = 0.5
+        if rh.CONFIGURATION["resolution"] == "5312x6000":
+            nn_model_path = "models/qrdet-512x512_openvino_2022.1_3shave.blob"
+        elif rh.CONFIGURATION["resolution"] == "4k":
+            nn_model_path = "models/qrdet-512x288_openvino_2022.1_5shave.blob"
+        elif rh.CONFIGURATION["resolution"] == "1080p":
+            nn_model_path = "models/qr_model_512x288_rvc2_openvino_2022.1_6shave.blob"
+        else:
+            raise ValueError(f"Unknown resolution: {rh.CONFIGURATION['resolution']}")
 
     image_manip_nn_input_crop = create_image_manip(pipeline=pipeline, source=image_manip_1to1_crop.out,
                                                    resize=(nn_input_width, nn_input_height), frames_pool=9, blocking_input_queue=True,
@@ -61,7 +69,7 @@ def create_pipeline(pipeline: dai.Pipeline) -> None:
 
     qr_detection_nn = create_yolo_nn(pipeline=pipeline, source=image_manip_nn_input_crop.out,
                                      model_path=nn_model_path,
-                                     confidence_threshold=0.5)
+                                     confidence_threshold=nn_confidence_threshold)
     qr_detection_nn.setNumPoolFrames(10)
     qr_detection_nn.input.setBlocking(True)
     qr_detection_nn.input.setQueueSize(9)

Luxonis Apps/QR Code Reader/node_helpers/bounding_box.py

@@ -11,6 +11,7 @@ def __init__(self, xmin: float, xmax: float, ymin: float, ymax: float, confidenc
                  sequence_number: int = 0):
         self.frame_sequence_number: int = sequence_number
         self.label = ""
+        self.code_format = ""
         self.crop = None
         self.counter = 0
 
@@ -49,6 +50,9 @@ def __repr__(self):
     def set_label(self, label: str):
         self.label = label
 
+    def set_format(self, code_format: str):
+        self.code_format = code_format.replace('BarcodeFormat.', '')
+
     def set_crop(self, crop):
         self.crop = crop
 

Luxonis Apps/QR Code Reader/requirements.txt

@@ -1,4 +1,9 @@
-depthai==2.25.0.0
-opencv-python
+depthai==2.27.0.0
+opencv-python~=4.10.0.84
 robothub==2.6.0
 zxing-cpp==2.2.0
+toml
+depthai-sdk==1.10.0
+av
+numpy~=1.26.4
+requests==2.31.0

Luxonis Apps/QR Code Reader/robotapp.toml

@@ -1,3 +1,4 @@
+# https://docs.luxonis.com/cloud/perception-apps/configuration/
 config_version = "2.0"
 
 [info]
@@ -73,3 +74,23 @@ step = 1
 min = 0
 max = 255
 initial_value = 0
+
+[[configuration]]
+key = "enable_web_reporter"
+label = "Enable sending reports to customer URL."
+field = "boolean"
+initial_value = false
+
+[[configuration]]
+field = "text"
+key = "url"
+label = "Customer URL to which the application sends reports."
+prefix = ""
+initial_value = "http://127.0.0.1:5000/webhook"
+
+[[configuration]]
+key = "barcode_mode"
+label = "Enable barcode recognition and decoding (barcodes only)."
+field = "boolean"
+initial_value = false
+

Luxonis Apps/QR Code Reader/server.py

@@ -0,0 +1,39 @@
+"""
+    Example server for testing WebReporter node
+"""
+from flask import Flask, request, jsonify
+import base64
+import numpy as np
+import cv2
+
+app = Flask(__name__)
+
+
+# Function to decode base64 to a NumPy array image
+def decode_image_from_base64(base64_string):
+    img_data = base64.b64decode(base64_string)
+    np_arr = np.frombuffer(img_data, np.uint8)
+    image = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)
+    return image
+
+
+@app.route('/webhook', methods=['POST'])
+def webhook():
+    data = request.json
+    # Extract the base64-encoded image from the JSON payload
+    base64_image = data.get('crop_image')
+
+    if base64_image is None:
+        return jsonify({'error': 'No image provided'}), 400
+
+    # Decode the image
+    image = decode_image_from_base64(base64_image)
+    # Save image
+    cv2.imwrite('received_image.jpg', image)
+    # Show received data
+    print(f"Received data:\nlabel: {data.get('label')}\ncode_format: {data.get('code_format')}\ntimestamp: {data.get('timestamp')}\n")
+    return jsonify({"status": "success", "data_received": data}), 200
+
+
+if __name__ == "__main__":
+    app.run(port=5000)