How to Improve Python Barcode QR Code Scanning Performance on Raspberry Pi

Raspberry Pi is a cheap single-board computer, which is often adopted by developers as an economical IoT solution, such as scanning barcode and Qr code. One thing you have to know is the CPU clock speed affects the decoding speed and accuracy. Dynamsoft Barcode Reader SDK supports Raspberry Pi. Its leading strengths are multiple and low-quality barcode decoding capabilities, which heavily rely on the CPU clock rate. Although the latest Raspberry Pi model has a 1.5GHz quad-core CPU, it is still a big challenge to trade off the speed and accuracy by customizing the algorithm parameters. In this article, I will show you how to scan barcode and Qr code in Python asynchronously on Raspberry Pi, as well as how to breakthrough the CPU bottleneck using server-side decoding via socket.

Prerequisites

• Python 3.6 or above
• Dynamsoft Barcode Reader v9.4
  • License Key
  • Python Package
    • pip install dbr: the official Python package of Dynamsoft Barcode Reader, which provides full API and relevant documentation.
    • pip install barcode-qr-code-sdk: a community version based on Dynamsoft C/C++ Barcode SDK, providing async decoding API for easy usage.

Building Python Barcode Scanner on Raspberry Pi

Implementing a barcode scanner in Python involves the following steps:

1. Use OpenCV to capture the video stream from the camera.
2. Use Dynamsoft Barcode Reader to decode the barcode from the image.

Barcode reading is a CPU-intensive task. When running synchronous API on a high clock rate CPU, we may not be aware of the latency on desktop computers. However, the CPU clock rate of Raspberry Pi is much lower. To avoid FPS (frames per second) dropping, it is necessary to carry out barcode detection algorithm in a separate thread.

Python's GIL (Global Interpreter Lock) limits the thread concurrency performance, especially for CPU-intensive tasks. The multiprocessing module is a better choice for the barcode scanning scenario. However, it is not easy to share the memory between processes. The Python package released by Dynamsoft Barcode Reader SDK provides three asynchronous decoding methods start_video_mode(), append_video_frame and stop_video_mode to overcome the GIL limitation. They maintains a C/C++ thread pool and a buffer queue.

To simplify the native-threaded API, the community version adds an alternative method called decodeMatAsync(), which decodes the latest image buffer and send decoding results via a registered callback function.

import barcodeQrSDK
import numpy as np
import cv2
import json

g_results = None

def callback(results, elapsed_time):
    global g_results
    g_results = (results, elapsed_time)

def run():
    # set license
    barcodeQrSDK.initLicense("DLS2eyJoYW5kc2hha2VDb2RlIjoiMjAwMDAxLTE2NDk4Mjk3OTI2MzUiLCJvcmdhbml6YXRpb25JRCI6IjIwMDAwMSIsInNlc3Npb25QYXNzd29yZCI6IndTcGR6Vm05WDJrcEQ5YUoifQ==")

    # initialize barcode scanner
    scanner = barcodeQrSDK.createInstance()
    params = scanner.getParameters()

    # register callback function to native thread
    scanner.addAsyncListener(callback)

    cap = cv2.VideoCapture(0)
    while True:
        ret, image = cap.read()
        if image is not None:
            scanner.decodeMatAsync(image)

        if g_results != None:
            print('Elapsed time: ' + str(g_results[1]) + 'ms')
            cv2.putText(image, 'Elapsed time: ' + str(g_results[1]) + 'ms', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
            for result in g_results[0]:
                x1 = result.x1
                y1 = result.y1
                x2 = result.x2
                y2 = result.y2
                x3 = result.x3
                y3 = result.y3
                x4 = result.x4
                y4 = result.y4

                cv2.drawContours(image, [np.int0([(x1, y1), (x2, y2), (x3, y3), (x4, y4)])], 0, (0, 255, 0), 2)
                cv2.putText(image, result.text, (x1, y1), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

        cv2.imshow('Barcode QR Code Scanner', image)
        ch = cv2.waitKey(1)
        if ch == 27:
            break

    scanner.clearAsyncListener()

if __name__ == '__main__':
    run()

Moving Barcode Detection Work to Server Side

The performance of running above Python barcode scanner on Raspberry Pi 4 looks not bad. However, if you want to run the program on some older Raspberry Pi models or other cheaper single-board computers with lower CPU clock rate, the performance will be much worse. To relieve the CPU burden, we can move the heavy computation to a powerful server. Here we use Python socket programming. You can get started with the article - Socket Programming in Python.

How to Compress Camera Frames for Socket Transmission

WebP is a modern image format that provides superior lossless and lossy compression for images. It is supported by OpenCV. The following code shows how to encode and decode an image with WebP using OpenCV API:

import cv2 as cv
import numpy as np

cap = cv.VideoCapture(0)
rval, frame = cap.read()
# Send
webp = cv.imencode('.webp', frame, [cv.IMWRITE_WEBP_QUALITY, 90])[1]
bytes_sent = webp.tobytes()

# Receive
frame = cv.imdecode(np.frombuffer(bytes_sent, np.uint8), cv.IMREAD_COLOR)

A Simple Socket Class for Sending and Receiving Data

We create a SimpleSocket class with socket and selector modules. The selector is used to implement non-blocking I/O.

import socket
import selectors

class SimpleSocket():
    def __init__(self) -> None:
        self.sel = selectors.DefaultSelector()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.callback = None

    def registerEventCb(self, callback):
        self.callback = callback

    def startClient(self, address, port):
        self.sock.setblocking(False)
        self.sock.connect_ex((address, port))
        events = selectors.EVENT_READ | selectors.EVENT_WRITE
        self.sel.register(self.sock, events, data=self.callback)

    def startServer(self, port, number_of_clients=1):
        self.sock.bind(('', port))
        self.sock.listen(number_of_clients)
        print('waiting for a connection at port %s' % port)
        self.sock.setblocking(False)
        self.sel.register(self.sock, selectors.EVENT_READ, data=None)

The callback parameter is a tuple of read and write callback functions.

In an infinite loop, we call select() to wait for I/O events. It monitors connection, read, and write events.

def monitorEvents(self):
    events = self.sel.select(timeout=None)
    for key, mask in events:
        if key.data is None:
            self.acceptConn(key.fileobj, self.callback)
        else:
            self.serveConn(key, mask)

def acceptConn(self, sock, callback):
    connection, addr = sock.accept()
    print('Connected to %s' % addr[0])
    connection.setblocking(False)
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    self.sel.register(connection, events, data=callback)

def serveConn(self, key, mask):
    sock = key.fileobj
    callback = key.data
    if callback != None and len(callback) == 2:
        if mask & selectors.EVENT_READ:
            data_type, data = self.receiveData(sock)
            callback[0](data_type, data)
        if mask & selectors.EVENT_WRITE:
            data_type, data = callback[1]()
            if data_type != None and data != None:
                self.sendData(sock, data_type, data)

For C/S communication, we need to define a simple protocol.

'''
+-+-+-+-+-------+-+-------------+-------------------------------+
|Type (1 byte)                  |   Payload length (4 bytes)    |
|0: text, 1: json 2: webp       |                               |
+-------------------------------+-------------------------------+
|                           Payload Data                        |
+-------------------------------- - - - - - - - - - - - - - - - +
:                     Payload Data continued ...                :
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|                     Payload Data continued ...                |
+---------------------------------------------------------------+
'''

The first byte is the data type that represents text, json, or webp. The next 4 bytes are the payload length. The rest of the data is the payload data.

According to the protocol, we can implement the receiveData() and sendData() functions.

def sendData(self, sock, data_type, data):
    msg = data_type + len(data).to_bytes(4, 'big') + data
    return self.send(sock, msg)

def receiveData(self, sock):
    data_type = self.receive(sock, 1)
    if data_type == b'':
        return b'', b''
    data_length = self.receive(sock, 4)
    if data_length == b'':
        return b'', b''
    data = self.receive(sock, int.from_bytes(data_length, 'big'))
    return data_type, data

def send(self, sock, msg):
    try:
        totalsent = 0
        while totalsent < len(msg):
            sent = sock.send(msg[totalsent:])
            if sent == 0:
                # connection closed
                return False

            totalsent = totalsent + sent
    except Exception as e:
        print(e)
        return False

    return True

def receive(self, sock, size):
    try:
        chunks = []
        bytes_recd = 0
        while bytes_recd < size:
            chunk = sock.recv(min(size, 1024))
            if chunk == b'':
                # connection closed
                return b''

            chunks.append(chunk)
            bytes_recd = bytes_recd + len(chunk)

    except Exception as e:
        print(e)
        return b''

    return b''.join(chunks)

Implementing Server-side Barcode Scanning Solution

The steps to implement the client.

1. Create a client.py file.

2. Set camera resolution to 640x480 and create a loop to capture frames.
   

   import cv2 as cv
   from simplesocket import SimpleSocket, DataType
   import json
   import numpy as np
   
   g_local_results = None
   g_remote_results = None
   isDisconnected = False
   msgQueue = []
   isReady = True
   
   cap = cv.VideoCapture(0)
   if cap.isOpened() == False:
       print("Unable to read camera feed")
       exit()
   
   cap.set(cv.CAP_PROP_FRAME_WIDTH, 640)
   cap.set(cv.CAP_PROP_FRAME_HEIGHT, 480)
   
   def run():
       while True:
           rval, frame = cap.read()
           cv.imshow('client', frame)
           if cv.waitKey(10) == 27:
               break
   
   if __name__ == '__main__':
       run()  
   

3. Initialize the socket client and register the callback functions.
   

   def callback(results, elapsed_time):
       global g_local_results
       print("Local decoding time: " + str(elapsed_time) + " ms")
       g_local_results = (results, elapsed_time)
   
   def readCb(data_type, data):
       global isDisconnected, g_remote_results, isReady
       if data == b'':
           isDisconnected = True
   
       if data_type == DataType.TEXT:
           text = data.decode('utf-8')
           print(text)
   
       if data_type == DataType.JSON:
           obj = json.loads(data)
           g_remote_results = (obj['results'], obj['time'])
           isReady = True
   
   # Data for sending
   def writeCb():
       if len(msgQueue) > 0:
           data_type, data =  msgQueue.pop(0)
           return data_type, data
   
       return None, None
   
   def run():
       global isDisconnected, g_local_results, g_remote_results, isReady
   
       client = SimpleSocket()
       client.registerEventCb((readCb, writeCb))
       client.startClient('192.168.8.72', 8080)
   

4. Keep reading frames from the camera and send them to the server.
   

   while True:
       client.monitorEvents()
       if (isDisconnected):
               break
   
       rval, frame = cap.read()
   
       # Send data to server
       if isReady:
           isReady = False
           webp = cv.imencode('.webp', frame, [cv.IMWRITE_WEBP_QUALITY, 90])[1]
           msgQueue.append((DataType.WEBP, webp.tobytes()))
   
       cv.imshow('client', frame)
       if cv.waitKey(10) == 27:
           break
   
   

5. Display the results when the results are returned from the server.
   

   if g_remote_results != None:
       print("Remote decoding time: " + str(int(g_remote_results[1])) + " ms")
       for result in g_remote_results[0]:
           text = result['text']
           x1 = result['x1']
           y1 = result['y1']
           x2 = result['x2']
           y2 = result['y2']
           x3 = result['x3']
           y3 = result['y3']
           x4 = result['x4']
           y4 = result['y4']
           cv.putText(frame, text, (x1, y1), cv.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2)
           cv.drawContours(frame, [np.int0([(x1, y1), (x2, y2), (x3, y3), (x4, y4)])], 0, (0, 255, 0), 2)
   
       cv.putText(frame, "Remote decoding time: " + str(int(g_remote_results[1])) + " ms", (10, 60), cv.FONT_HERSHEY_SIMPLEX, 0.8, (255, 0, 0), 2)
   

Considering the CPU performance is good and there is no UI required on the server side, we can use synchronous API to recognize barcode and Qr code after receiving the frame from the client. The returned results are encoded as JSON string.

import cv2 as cv
import numpy as np
from simplesocket import SimpleSocket, DataType
import json
import barcodeQrSDK

g_results = None
isDisconnected = False
msgQueue = []

# Initialize Dynamsoft Barcode Reader
barcodeQrSDK.initLicense("DLS2eyJoYW5kc2hha2VDb2RlIjoiMjAwMDAxLTE2NDk4Mjk3OTI2MzUiLCJvcmdhbml6YXRpb25JRCI6IjIwMDAwMSIsInNlc3Npb25QYXNzd29yZCI6IndTcGR6Vm05WDJrcEQ5YUoifQ==")
reader = barcodeQrSDK.createInstance()

# Process received data     
def readCb(data_type, data):
    global isDisconnected, g_results, msgQueue

    if data == b'':
        isDisconnected = True

    if data_type == DataType.TEXT:
        text = data.decode('utf-8')
        print(text)

    if data_type == DataType.JSON:
        obj = json.loads(data)
        print(obj)


    if data_type == DataType.WEBP:
        try:
            frame = cv.imdecode(np.frombuffer(data, np.uint8), cv.IMREAD_COLOR)

            if frame is not None:
                results, elpased_time = reader.decodeMat(frame)
                g_results = (results, elpased_time)

            if g_results != None:
                jsonData = {'results': [], 'time': g_results[1]}
                for result in g_results[0]:
                    format = result.format
                    text = result.text
                    x1 = result.x1
                    y1 = result.y1
                    x2 = result.x2
                    y2 = result.y2
                    x3 = result.x3
                    y3 = result.y3
                    x4 = result.x4
                    y4 = result.y4
                    data = {'format': format, 'text': text, 'x1': x1, 'y1': y1, 'x2': x2, 'y2': y2, 'x3': x3, 'y3': y3, 'x4': x4, 'y4': y4}
                    jsonData['results'].append(data)

                msgQueue.append((DataType.JSON, json.dumps(jsonData).encode('utf-8')))

        except Exception as e:
            isDisconnected = True

def writeCb():
    if len(msgQueue) > 0:
        data_type, data =  msgQueue.pop(0)
        return data_type, data

    return None, None

def run():
    global isDisconnected

    server = SimpleSocket()
    server.registerEventCb((readCb, writeCb))
    server.startServer(8080, 1)

    try:
        while True:
            server.monitorEvents()
            if (isDisconnected):
                break

    except KeyboardInterrupt:
            print("Caught keyboard interrupt, exiting")
    finally:
        server.shutdown()

if __name__ == '__main__':
    run()

Testing the Server-side Barcode Scanning Solution

When running the client.py and server.py, you need to change the IP address and port. If the network transmission is stable, the server-side barcode scanning solution can achieve an ideal performance.

Source Code

https://github.com/yushulx/python-barcode-qrcode-sdk/tree/main/examples/socket