[REF] hw_{driver, scanner}: Barcode scanners as IoT Devices

Barcode scanners are now part of KeyboardUSBDriver. We also add
the possibility to change the keyboard/scanner layout from an
action. The list of available layout can be retrieved with a new
route.

`hw_scanner` has been removed. Backward compatibility is ensured
because the name of the route that is now used in community is the
same that was used previously.

TaskID: 1939455
This commit is contained in:
Antoine Prieels
2019-08-13 15:35:38 +02:00
parent 9790c81d5b
commit 4e649a8ca9
9 changed files with 210 additions and 316 deletions
+1 -1
View File
@@ -166,7 +166,7 @@ class Driver(Thread, metaclass=DriverMetaClass):
"""
On specific driver override this method to give type of device
return string
possible value : printer - camera - device
possible value : printer - camera - keyboard - scanner - device
"""
return self._device_type
+208 -17
View File
@@ -1,25 +1,51 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import ctypes
import evdev
import logging
from lxml import etree
import os
import subprocess
import time
from usb import util
try:
from queue import Queue, Empty
except ImportError:
from Queue import Queue, Empty # pylint: disable=deprecated-module
from odoo import _
from odoo.addons.hw_drivers.controllers.driver import event_manager, Driver
from odoo import http, _
from odoo.addons.hw_proxy.controllers.main import drivers as old_drivers
from odoo.addons.hw_drivers.controllers.driver import event_manager, Driver, iot_devices
_logger = logging.getLogger(__name__)
xlib = ctypes.cdll.LoadLibrary('libX11.so.6')
class KeyboardUSBDriver(Driver):
connection_type = 'usb'
keyboard_layout_groups = []
def __init__(self, device):
if not hasattr(KeyboardUSBDriver, 'display'):
os.environ['XAUTHORITY'] = "/run/lightdm/pi/xauthority"
KeyboardUSBDriver.display = xlib.XOpenDisplay(bytes(":0.0", "utf-8"))
super(KeyboardUSBDriver, self).__init__(device)
self._device_type = 'device'
self._device_type = 'keyboard'
self._device_connection = 'direct'
self._device_name = self._get_name()
for device in [evdev.InputDevice(path) for path in evdev.list_devices()]:
if (self.dev.idVendor == device.info.vendor) and (self.dev.idProduct == device.info.product):
self.input_device = device
if 'barcode' in self._device_name.lower() or 'scanner' in self._device_name.lower():
self._device_type = 'scanner'
self._barcodes = Queue()
self._current_barcode = ''
self.input_device.grab()
@classmethod
def supported(cls, device):
for cfg in device:
@@ -28,32 +54,197 @@ class KeyboardUSBDriver(Driver):
return True
return False
def _get_name(self):
@classmethod
def get_status(self):
"""Allows `hw_proxy.Proxy` to retrieve the status of the scanners"""
status = 'connected' if any(iot_devices[d].device_type == "scanner" for d in iot_devices) else 'disconnected'
return {'status': status, 'messages': ''}
def _set_name(self):
try:
manufacturer = util.get_string(self.dev, 256, self.dev.iManufacturer)
product = util.get_string(self.dev, 256, self.dev.iProduct)
return ("%s - %s") % (manufacturer, product)
except ValueError as e:
_logger.warning(e)
return _('Unknow keyboard')
return _('Unknow keyboard or scanner')
def action(self, data):
self.data['value'] = ''
event_manager.device_changed(self)
if data.get('action', False) == 'update_layout':
layout = {
'layout': data.get('layout'),
'variant': data.get('variant'),
}
self._change_keyboard_layout(layout)
else:
self.data['value'] = ''
event_manager.device_changed(self)
def run(self):
devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
for device in devices:
if (self.dev.idVendor == device.info.vendor) and (self.dev.idProduct == device.info.product):
path = device.path
device = evdev.InputDevice(path)
key_input = self._barcode_scanner_input if self._device_type == "scanner" else self._keyboard_input
try:
for event in device.read_loop():
for event in self.input_device.read_loop():
if event.type == evdev.ecodes.EV_KEY:
data = evdev.categorize(event)
if data.keystate:
self.data['value'] = data.keycode.replace('KEY_','')
event_manager.device_changed(self)
if data.keystate == 1:
key_input(data.scancode)
except Exception as err:
_logger.warning(err)
def _change_keyboard_layout(self, new_layout):
"""Change the layout of the current device to what is specified in
new_layout.
Args:
new_layout (dict): A dict containing two keys:
- layout (str): The layout code
- variant (str): An optional key to represent the variant of the
selected layout
"""
if hasattr(self, 'keyboard_layout'):
KeyboardUSBDriver.keyboard_layout_groups.remove(self.keyboard_layout)
if new_layout:
self.keyboard_layout = new_layout.get('layout') or 'us'
if new_layout.get('variant'):
self.keyboard_layout += "(%s)" % new_layout['variant']
else:
self.keyboard_layout = 'us'
KeyboardUSBDriver.keyboard_layout_groups.append(self.keyboard_layout)
subprocess.call(["setxkbmap", "-display", ":0.0", ",".join(KeyboardUSBDriver.keyboard_layout_groups)])
# Close then re-open display to refresh the mapping
xlib.XCloseDisplay(KeyboardUSBDriver.display)
KeyboardUSBDriver.display = xlib.XOpenDisplay(bytes(":0.0", "utf-8"))
def _keyboard_input(self, scancode):
"""Deal with a keyboard input. Send the character corresponding to the
pressed key represented by its scancode to the connected Odoo instance.
Args:
scancode (int): The scancode of the pressed key.
"""
self.data['value'] = self._scancode_to_char(scancode)
if self.data['value']:
event_manager.device_changed(self)
def _barcode_scanner_input(self, scancode):
"""Deal with a barcode scanner input. Add the new character scanned to
the current barcode or complete the barcode if "Return" is pressed.
When a barcode is completed, two tasks are performed:
- Send a device_changed update to the event manager to notify the
listeners that the value has changed (used in Enterprise).
- Add the barcode to the list barcodes that are being queried in
Community.
Args:
scancode (int): The scancode of the pressed key.
"""
if scancode == 28: # Return
self.data['value'] = self._current_barcode
event_manager.device_changed(self)
self._barcodes.put((time.time(), self._current_barcode))
self._current_barcode = ''
else:
self._current_barcode += self._scancode_to_char(scancode)
def _scancode_to_char(self, scancode):
"""Translate a received scancode to a character depending on the
selected keyboard layout and the current state of the keyboard's
modifiers.
Args:
scancode (int): The scancode of the pressed key, to be translated to
a character
Returns:
str: The translated scancode.
"""
# Scancode -> Keysym : Depends on the keyboard layout
group = KeyboardUSBDriver.keyboard_layout_groups.index(self.keyboard_layout)
modifiers = self._get_active_modifiers(scancode)
keysym = ctypes.c_int(xlib.XkbKeycodeToKeysym(KeyboardUSBDriver.display, scancode + 8, group, modifiers))
# Translate Keysym to a character
key_pressed = ctypes.create_string_buffer(5)
xlib.XkbTranslateKeySym(KeyboardUSBDriver.display, ctypes.byref(keysym), 0, ctypes.byref(key_pressed), 5, ctypes.byref(ctypes.c_int()))
return key_pressed.value.decode('latin1') if key_pressed.value else ''
def _get_active_modifiers(self, scancode):
"""Get the state of currently active modifiers.
Args:
scancode (int): The scancode of the key being translated
Returns:
int: The current state of the modifiers:
0 -- Lowercase
1 -- Highercase or (NumLock + key pressed on keypad)
2 -- AltGr
3 -- Highercase + AltGr
"""
keyboard_leds = self.input_device.leds()
num_lock = 0 in keyboard_leds # Led Num 0
caps_lock = 1 in keyboard_leds # Led Num 1
keyboard_pressed = self.input_device.active_keys()
shift = bool([x for x in [42, 54] if x in keyboard_pressed]) # 42 = Left Shift, 54 = Right Shift
alt_gr = 100 in keyboard_pressed
modifiers = 0
if (scancode in [71, 72, 73, 75, 76, 77, 79, 80, 81, 82, 83] and num_lock) or (shift ^ caps_lock):
modifiers += 1
if alt_gr:
modifiers += 2
return modifiers
def read_next_barcode(self):
"""Get the value of the last barcode that was scanned but not sent yet
and not older than 5 seconds. This function is used in Community, when
we don't have access to the IoTLongpolling.
Returns:
str: The next barcode to be read or an empty string.
"""
try:
timestamp, barcode = self._barcodes.get(True, 55)
if timestamp > time.time() - 5:
return barcode
except Empty:
return ''
old_drivers['scanner'] = KeyboardUSBDriver
class KeyboardUSBController(http.Controller):
@http.route('/hw_proxy/scanner', type='json', auth='none', cors='*')
def get_barcode(self):
scanners = [iot_devices[d] for d in iot_devices if iot_devices[d].device_type == "scanner"]
if scanners:
return scanners[0].read_next_barcode()
time.sleep(5)
return None
@http.route('/hw_proxy/load_keyboard_layouts', type='json', auth='none', cors='*')
def load_keyboard_layouts(self):
available_layouts = []
tree = etree.parse("/usr/share/X11/xkb/rules/base.xml", etree.XMLParser(ns_clean=True, recover=True))
layouts = tree.xpath("//layout")
for layout in layouts:
layout_name = layout.xpath("./configItem/name")[0].text
layout_description = layout.xpath("./configItem/description")[0].text
available_layouts.append({
'name': layout_description,
'layout': layout_name,
})
for variant in layout.xpath("./variantList/variant"):
variant_name = variant.xpath("./configItem/name")[0].text
variant_description = variant.xpath("./configItem/description")[0].text
available_layouts.append({
'name': variant_description,
'layout': layout_name,
'variant': variant_name,
})
return available_layouts
-11
View File
@@ -209,17 +209,6 @@ class Proxy(http.Controller):
def print_receipt(self, receipt):
print('print_receipt %s', receipt)
@http.route('/hw_proxy/is_scanner_connected', type='json', auth='none', cors='*')
def is_scanner_connected(self, receipt):
print('is_scanner_connected?')
return False
@http.route('/hw_proxy/scanner', type='json', auth='none', cors='*')
def scanner(self, receipt):
print('scanner')
time.sleep(10)
return ''
@http.route('/hw_proxy/log', type='json', auth='none', cors='*')
def log(self, arguments):
_logger.info(' '.join(str(v) for v in arguments))
-4
View File
@@ -1,4 +0,0 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import controllers
-22
View File
@@ -1,22 +0,0 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
{
'name': 'Barcode Scanner Hardware Driver',
'category': 'Point of Sale',
'sequence': 6,
'summary': 'Hardware Driver for Barcode Scanners',
'website': 'https://www.odoo.com/page/point-of-sale-hardware',
'description': """
Barcode Scanner Hardware Driver
================================
This module allows the web client to access a remotely installed barcode
scanner, and is used by the IoTBox to provide barcode scanner support to the
point of sale module.
""",
'depends': ['hw_proxy'],
'external_dependencies': {'python': ['evdev']},
'installable': False,
}
@@ -1,4 +0,0 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import main
-240
View File
@@ -1,240 +0,0 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
import time
from os import listdir
from os.path import join, isdir
try:
from queue import Queue, Empty
except ImportError:
from Queue import Queue, Empty # pylint: disable=deprecated-module
from select import select
from threading import Thread, Lock
from odoo import http
from odoo.addons.hw_proxy.controllers import main as hw_proxy
_logger = logging.getLogger(__name__)
try:
import evdev
except ImportError:
_logger.error('Odoo module hw_scanner depends on the evdev python module')
evdev = None
class ScannerDevice():
def __init__(self, path):
self.evdev = evdev.InputDevice(path)
self.evdev.grab()
self.barcode = []
self.shift = False
class Scanner(Thread):
def __init__(self):
Thread.__init__(self)
self.lock = Lock()
self.status = {'status':'connecting', 'messages':[]}
self.input_dir = '/dev/input/by-id/'
self.open_devices = []
self.barcodes = Queue()
self.keymap = {
2: ("1","!"),
3: ("2","@"),
4: ("3","#"),
5: ("4","$"),
6: ("5","%"),
7: ("6","^"),
8: ("7","&"),
9: ("8","*"),
10:("9","("),
11:("0",")"),
12:("-","_"),
13:("=","+"),
# 14 BACKSPACE
# 15 TAB
16:("q","Q"),
17:("w","W"),
18:("e","E"),
19:("r","R"),
20:("t","T"),
21:("y","Y"),
22:("u","U"),
23:("i","I"),
24:("o","O"),
25:("p","P"),
26:("[","{"),
27:("]","}"),
# 28 ENTER
# 29 LEFT_CTRL
30:("a","A"),
31:("s","S"),
32:("d","D"),
33:("f","F"),
34:("g","G"),
35:("h","H"),
36:("j","J"),
37:("k","K"),
38:("l","L"),
39:(";",":"),
40:("'","\""),
41:("`","~"),
# 42 LEFT SHIFT
43:("\\","|"),
44:("z","Z"),
45:("x","X"),
46:("c","C"),
47:("v","V"),
48:("b","B"),
49:("n","N"),
50:("m","M"),
51:(",","<"),
52:(".",">"),
53:("/","?"),
# 54 RIGHT SHIFT
57:(" "," "),
}
def lockedstart(self):
with self.lock:
if not self.isAlive():
self.daemon = True
self.start()
def set_status(self, status, message = None):
if status == self.status['status']:
if message != None and message != self.status['messages'][-1]:
self.status['messages'].append(message)
else:
self.status['status'] = status
if message:
self.status['messages'] = [message]
else:
self.status['messages'] = []
if status == 'error' and message:
_logger.error('Barcode Scanner Error: '+message)
elif status == 'disconnected' and message:
_logger.info('Disconnected Barcode Scanner: %s', message)
def get_devices(self):
try:
if not evdev:
return []
if not isdir(self.input_dir):
return []
new_devices = [device for device in listdir(self.input_dir)
if join(self.input_dir, device) not in [dev.evdev.fn for dev in self.open_devices]]
scanners = [device for device in new_devices
if (('kbd' in device) and ('keyboard' not in device.lower()))
or ('barcode' in device.lower()) or ('scanner' in device.lower())]
for device in scanners:
_logger.debug('opening device %s', join(self.input_dir,device))
self.open_devices.append(ScannerDevice(join(self.input_dir,device)))
if self.open_devices:
self.set_status('connected','Connected to '+ str([dev.evdev.name for dev in self.open_devices]))
else:
self.set_status('disconnected','Barcode Scanner Not Found')
return self.open_devices
except Exception as e:
self.set_status('error',str(e))
return []
def release_device(self, dev):
self.open_devices.remove(dev)
def get_barcode(self):
""" Returns a scanned barcode. Will wait at most 5 seconds to get a barcode, and will
return barcode scanned in the past if they are not older than 5 seconds and have not
been returned before. This is necessary to catch barcodes scanned while the POS is
busy reading another barcode
"""
self.lockedstart()
while True:
try:
timestamp, barcode = self.barcodes.get(True, 5)
if timestamp > time.time() - 5:
return barcode
except Empty:
return ''
def get_status(self):
self.lockedstart()
return self.status
def _get_open_device_by_fd(self, fd):
for dev in self.open_devices:
if dev.evdev.fd == fd:
return dev
def run(self):
""" This will start a loop that catches all keyboard events, parse barcode
sequences and put them on a timestamped queue that can be consumed by
the point of sale's requests for barcode events
"""
self.barcodes = Queue()
barcode = []
shift = False
devices = None
while True: # barcodes loop
devices = self.get_devices()
try:
while True: # keycode loop
r,w,x = select({dev.fd: dev for dev in [d.evdev for d in devices]},[],[],5)
if len(r) == 0: # timeout
break
for fd in r:
device = self._get_open_device_by_fd(fd)
if not evdev.util.is_device(device.evdev.fn):
_logger.info('%s disconnected', str(device.evdev))
self.release_device(device)
break
events = device.evdev.read()
for event in events:
if event.type == evdev.ecodes.EV_KEY:
# _logger.debug('Evdev Keyboard event %s',evdev.categorize(event))
if event.value == 1: # keydown events
if event.code in self.keymap:
if device.shift:
device.barcode.append(self.keymap[event.code][1])
else:
device.barcode.append(self.keymap[event.code][0])
elif event.code == 42 or event.code == 54: # SHIFT
device.shift = True
elif event.code == 28: # ENTER, end of barcode
_logger.debug('pushing barcode %s from %s', ''.join(device.barcode), str(device.evdev))
self.barcodes.put( (time.time(),''.join(device.barcode)) )
device.barcode = []
elif event.value == 0: #keyup events
if event.code == 42 or event.code == 54: # LEFT SHIFT
device.shift = False
except Exception as e:
self.set_status('error',str(e))
scanner_thread = None
if evdev:
scanner_thread = Scanner()
hw_proxy.drivers['scanner'] = scanner_thread
class ScannerDriver(hw_proxy.Proxy):
@http.route('/hw_proxy/scanner', type='json', auth='none', cors='*')
def scanner(self):
return scanner_thread.get_barcode() if scanner_thread else None
-16
View File
@@ -1,16 +0,0 @@
# Translation of Odoo Server.
# This file contains the translation of the following modules:
#
msgid ""
msgstr ""
"Project-Id-Version: Odoo Server 9.0\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2015-09-07 14:39+0000\n"
"PO-Revision-Date: 2015-09-07 14:39+0000\n"
"Last-Translator: <>\n"
"Language-Team: \n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: \n"
"Plural-Forms: \n"
@@ -26,7 +26,7 @@ test -x $DAEMON || exit 0
set -e
function _start() {
start-stop-daemon --start --quiet --pidfile $PIDFILE --chuid $USER:$USER --background --make-pidfile --exec $DAEMON -- --config $CONFIG --logfile $LOGFILE --load=web,hw_proxy,hw_posbox_homepage,hw_posbox_upgrade,hw_scanner,hw_escpos,hw_blackbox_be,hw_screen,hw_drivers
start-stop-daemon --start --quiet --pidfile $PIDFILE --chuid $USER:$USER --background --make-pidfile --exec $DAEMON -- --config $CONFIG --logfile $LOGFILE --load=web,hw_proxy,hw_posbox_homepage,hw_posbox_upgrade,hw_escpos,hw_blackbox_be,hw_screen,hw_drivers
}
function _stop() {