python-gui-barcode-reader/barcode_manager.py at main · owenlyn0123/python-gui-barcode-reader · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

from dbr import *
import cv2
from multiprocessing import Process, Queue
import time
import numpy as np


def process_barcode_frame(license, frameQueue, resultQueue, template=None, types=0, types2=0):
    # Create Dynamsoft Barcode Reader
    reader = BarcodeReader()
    # Apply for a trial license: https://www.dynamsoft.com/customer/license/trialLicense
    reader.init_license(license)
    if template is not None and template is not '':
        error = reader.init_runtime_settings_with_string(template)
        if error[0] != EnumErrorCode.DBR_OK:
            print(error[1])

    if types != 0:
        settings = reader.get_runtime_settings()
        settings.barcode_format_ids = types
        ret = reader.update_runtime_settings(settings)

    if types2 != 0:
        settings = reader.get_runtime_settings()
        settings.barcode_format_ids_2 = types2
        ret = reader.update_runtime_settings(settings)

    settings = reader.get_runtime_settings()
    settings.max_algorithm_thread_count = 1
    reader.update_runtime_settings(settings)

    while True:
        results = None

        try:
            frame = frameQueue.get(False, 10)
            if type(frame) is str:
                break
        except:
            time.sleep(0.01)
            continue

        start, end = 0, 0
        try:
            frameHeight, frameWidth, channel = frame.shape[:3]
            start = time.time()
            # results = reader.decode_buffer(frame)
            results = reader.decode_buffer_manually(np.array(frame).tobytes(), frameWidth, frameHeight, frame.strides[0], EnumImagePixelFormat.IPF_RGB_888)
            end = time.time()
        except BarcodeReaderError as error:
            print(error)

        try:
            resultQueue.put([results, (end - start) * 1000], False, 10)
        except:
            pass


class BarcodeManager():
    def __init__(self, license):
        self._reader = BarcodeReader()
        self._reader.init_license(license)
        settings = self._reader.get_runtime_settings()
        settings.max_algorithm_thread_count = 1
        self._reader.update_runtime_settings(settings)
        self._template = None
        self._types = 0
        self._types2 = 0
        self._license = license
        self.frameQueue, self.resultQueue, self.barcodeScanning = None, None, None

    def __decode_buffer(self, frame):
        if frame is None:
            return None, None

        if self._template is not None and self._template is not '':
            error = self._reader.init_runtime_settings_with_string(self._template)
            if error[0] != EnumErrorCode.DBR_OK:
                print(error[1])

        if self._types != 0:
            settings = self._reader.get_runtime_settings()
            settings.barcode_format_ids = self._types
            ret = self._reader.update_runtime_settings(settings)

        if self._types2 != 0:
            settings = self._reader.get_runtime_settings()
            settings.barcode_format_ids_2 = self._types2
            ret = self._reader.update_runtime_settings(settings)

        start = time.time()
        results = self._reader.decode_buffer(frame)
        end = time.time()
        return frame, [results, (end - start) * 1000]

    def decode_frame(self, frame):
        return self.__decode_buffer(frame)

    def decode_file(self, filename):
        frame = cv2.imread(filename)
        return self.__decode_buffer(frame)

    def set_template(self, template):
        self._template = template

    def set_barcode_types(self, types):
        self._types = types

    def set_barcode_types_2(self, types):
        self._types2 = types

    def create_barcode_process(self):
        self.destroy_barcode_process()

        size = 1
        self.frameQueue = Queue(size)
        self.resultQueue = Queue(size)
        self.barcodeScanning = Process(target=process_barcode_frame, args=(self._license, self.frameQueue, self.resultQueue, self._template, self._types, self._types2))
        self.barcodeScanning.start()

    def destroy_barcode_process(self):
        if self.frameQueue is not None:
            self.frameQueue.put("")
            self.frameQueue = None

        if self.barcodeScanning is not None:
            self.barcodeScanning.join()
            self.barcodeScanning = None

        if self.frameQueue is not None:
            self.frameQueue.close()
            self.frameQueue = None

        if self.resultQueue is not None:
            self.resultQueue.close()
            self.resultQueue = None

    def append_frame(self, frame):
        try:
            self.frameQueue.put(frame.copy(), False, 10)
        except:
            pass

    def peek_results(self):
        try:
            return self.resultQueue.get(False, 10)
        except:
            return None