ウェブカメラ、OpenCVとPythonを使用してスキャンバーコードとQRコード

pip install opencv-python

pip install dbr

import numpy as np
import cv2 as cv

from multiprocessing.pool import ThreadPool
from collections import deque

import dbr
from dbr import *

BarcodeReader.init_license("LICENSE-KEY")
reader = BarcodeReader()

threadn = 1 # cv.getNumberOfCPUs()
pool = ThreadPool(processes = threadn)
barcodeTasks = deque()

def process_frame(frame):
    results = None
    try:
        results = reader.decode_buffer(frame)
    except BarcodeReaderError as bre:
        print(bre)

    return results

while True:
    ret, frame = cap.read()
    while len(barcodeTasks) > 0 and barcodeTasks[0].ready():
        results = barcodeTasks.popleft().get()
        if results != None:
            for result in results:
                points = result.localization_result.localization_points
                cv.line(frame, points[0], points[1], (0,255,0), 2)
                cv.line(frame, points[1], points[2], (0,255,0), 2)
                cv.line(frame, points[2], points[3], (0,255,0), 2)
                cv.line(frame, points[3], points[0], (0,255,0), 2)
                cv.putText(frame, result.barcode_text, points[0], cv.FONT_HERSHEY_SIMPLEX, 0.5, (0,0,255))

    if len(barcodeTasks) < threadn:
        task = pool.apply_async(process_frame, (frame.copy(), ))
        barcodeTasks.append(task)

    cv.imshow('Barcode & QR Code Scanner', frame)
    ch = cv.waitKey(1)
    if ch == 27:
        break

modes = (cv.Stitcher_PANORAMA, cv.Stitcher_SCANS)
stitcher = cv.Stitcher.create(modes[1])
stitcher.setPanoConfidenceThresh(0.5)

panoramaPool = ThreadPool(processes = threadn)
panoramaTask = deque()

def stitch_frame(self, frame):
    try:
        results = self.reader.decode_buffer(frame)
        if results != None:
            for result in results:
                points = result.localization_result.localization_points
                cv.line(frame, points[0], points[1], (0,255,0), 2)
                cv.line(frame, points[1], points[2], (0,255,0), 2)
                cv.line(frame, points[2], points[3], (0,255,0), 2)
                cv.line(frame, points[3], points[0], (0,255,0), 2)
                cv.putText(frame, result.barcode_text, points[0], cv.FONT_HERSHEY_SIMPLEX, 0.5, (0,0,255))

            self.panorama.append((frame, len(results)))
            print('Stitching .............')
            try:
                all_images = [frame for frame, count in self.panorama]
                status, image = self.stitcher.stitch(all_images)

                if status != cv.Stitcher_OK:
                    print("Can't stitch images, error code = %d" % status)
                    return self.panorama[0][0]
                else:
                    # Stop stitching if the output image is out of control
                    if image.shape[0] >= frame.shape[0] * 1.5:
                        self.isPanoramaDone = True
                        self.save_frame(all_images[0])
                        print('Stitching is done.............')
                        return None

                    # Drop the stitched image if its quality is not good enough
                    total = 0
                    for frame, count in self.panorama:
                        total += count

                    count_stitch = self.count_barcodes(image)
                    if count_stitch > total or count_stitch < self.panorama[0][1]:
                        return self.panorama[0][0]

                    # Wait for the next stitching and return the current stitched image
                    self.panorama = [(image, count_stitch)]
                    return image
            except Exception as e:
                print(e)
                return None

    except BarcodeReaderError as e:
        print(e)
        return None

    return None

while len(panoramaTask) > 0 and panoramaTask[0].ready():
    image = panoramaTask.popleft().get()
    if image is not None:
        cv.imshow('panorama', image)

if len(panoramaTask) < threadn:
    task = panoramaPool.apply_async(self.stitch_frame, (frame_cp, ))
    panoramaTask.append(task)