diff options
Diffstat (limited to 'test-cli/test/helpers')
| -rw-r--r-- | test-cli/test/helpers/qrreader.py | 120 | ||||
| -rw-r--r-- | test-cli/test/helpers/testsrv_db.py | 25 |
2 files changed, 143 insertions, 2 deletions
diff --git a/test-cli/test/helpers/qrreader.py b/test-cli/test/helpers/qrreader.py new file mode 100644 index 0000000..1d3768b --- /dev/null +++ b/test-cli/test/helpers/qrreader.py @@ -0,0 +1,120 @@ +import evdev +from evdev import InputDevice, categorize, ecodes +import threading +import time +import selectors +from selectors import DefaultSelector, EVENT_READ + +selector = selectors.DefaultSelector() + +qrdevice_list = [ + "Honeywell Imaging & Mobility 1900", + "Manufacturer Barcode Reader" +] + + +qrkey_list = { 'KEY_0':'0', + 'KEY_1':'1', + 'KEY_2':'2', + 'KEY_3':'3', + 'KEY_4':'4', + 'KEY_5':'5', + 'KEY_6':'6', + 'KEY_7':'7', + 'KEY_8':'8', + 'KEY_9':'9' + } + + +class QRReader: + __qrstr = "" + __myReader = {} + __numdigits = 10 + __dev = None + + def __init__(self): + self.getQRlist() + + def getQRlist(self): + devices = [evdev.InputDevice(path) for path in evdev.list_devices()] + for device in devices: + if device.name in qrdevice_list: + self.__myReader['NAME'] = device.name + print(self.__myReader['NAME']) + self.__myReader['PATH'] = device.path + print(self.__myReader['PATH']) + self.__myReader['PHYS'] = device.phys + print(self.__myReader['PHYS']) + + def IsQR (self): + return 'NAME' in self.__myReader + + def getQRNumber(self): + return self.__qrstr + + def openQR(self): + if self.IsQR(): + self.closeQR() + self.__dev = InputDevice(self.__myReader['PATH']) + return True + return False + + def closeQR(self): + if self.__dev: + del self.__dev + self.__dev = None + + def readQR(self): + """" Sync Read up to numdigits """ + count = 0 + self.__qrstr = "" + if self.__dev: + self.__dev.grab() + for event in self.__dev.read_loop(): + if event.type == evdev.ecodes.EV_KEY: + c_event = categorize(event) + if c_event.keycode in qrkey_list: + if c_event.keystate == 0: + self.__qrstr += qrkey_list[c_event.keycode] + count += 1 + if count == self.__numdigits: + break + self.__dev.ungrab() + return True + return False + + def wait_event (self, timeout): + selector.register(self.__dev, selectors.EVENT_READ) + while True: + events = selector.select(timeout); + if not events: + return False + else: + return True + + def readQRasync(self, timeout): + count = 0 + r = False + self.__qrstr = "" + if self.__dev: + self.__dev.grab() + if self.wait_event(timeout): + for event in self.__dev.read_loop(): + if event.type == evdev.ecodes.EV_KEY: + c_event = categorize(event) + if c_event.keycode in qrkey_list: + if c_event.keystate == 0: + self.__qrstr += qrkey_list[c_event.keycode] + count += 1 + if count == self.__numdigits: + r = True + break + self.__dev.ungrab() + return r + return False + + +#qr = QRReader() +#if qr.openQR(): +# print(qr.readQRasync(2)) +#qr.closeQR()
\ No newline at end of file diff --git a/test-cli/test/helpers/testsrv_db.py b/test-cli/test/helpers/testsrv_db.py index d563e74..dc4fb55 100644 --- a/test-cli/test/helpers/testsrv_db.py +++ b/test-cli/test/helpers/testsrv_db.py @@ -139,12 +139,12 @@ class TestSrv_Database(object): def create_task_result(self, taskid_ctl, name, newstatus, newinfo=None): sql = "SELECT isee.f_create_task_result({},'{}','{}','{}')".format(taskid_ctl, name, newstatus, newinfo) - print('>>>' + sql) + # print('>>>' + sql) try: self.__sqlObject.db_execute_query(sql) except Exception as err: r = find_between(str(err), '#', '#') - print(r) + # print(r) return None def update_taskctl_status(self, taskid_ctl, newstatus): @@ -156,3 +156,24 @@ class TestSrv_Database(object): r = find_between(str(err), '#', '#') # print(r) return None + + def set_factorycode(self, uuid, factorycode): + sql = "SELECT isee.f_set_factorycode('{}','{}')".format(uuid, factorycode) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def bond_to_station(self, uuid, station): + sql = "SELECT station.bond_to_station('{}','{}')".format(uuid, station) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + |