pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[IMP] hw_drivers: Map drivers actions

Browse Source 
Add possibilities to map actions for the drivers

To avoid a sequence of "if elif else" we integrate a '_actions()' dictionary
where the key is the name of the action and the value is
the name of the function to be executed in the drivers
with the action data as a parameter

closes odoo/odoo#63375

Related: odoo/enterprise#15310
Signed-off-by: pimodoo <pimodoo@users.noreply.github.com>
This commit is contained in:
lejeune quentin
2020-12-16 15:52:34 +00:00
parent 6843d63769
commit 2be28bfbde
5 changed files with 80 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/hw_drivers/driver.py
+6 -3
View File
@@ -32,6 +32,7 @@ class Driver(Thread, metaclass=DriverMetaClass):
self.device_type = ''
self.device_manufacturer = ''
self.data = {'value': ''}
self._actions = {}
self._stopped = Event()
@classmethod
@@ -43,10 +44,12 @@ class Driver(Thread, metaclass=DriverMetaClass):
return False
def action(self, data):
"""Helper function that calls a specific action method on the device.
:param data: the `_actions` key mapped to the action method we want to call
:type data: string
"""
On specific driver override this method to make a action with device (take picture, printing,...)
"""
raise NotImplementedError()
self._actions[data['action']](data)
def disconnect(self):
self._stopped.set()
addons/hw_drivers/iot_handlers/drivers/DisplayDriver.py
+29 -16
View File
@@ -44,6 +44,14 @@ class DisplayDriver(Driver):
self._x_screen = device.get('x_screen', '0')
self.load_url()
self._actions.update({
'update_url': self._action_update_url,
'display_refresh': self._action_display_refresh,
'take_control': self._action_take_control,
'customer_facing_display': self._action_customer_facing_display,
'get_owner': self._action_get_owner,
})
@classmethod
def supported(cls, device):
return True # All devices with connection_type == 'display' are supported
@@ -53,22 +61,6 @@ class DisplayDriver(Driver):
displays = list(filter(lambda d: iot_devices[d].device_type == 'display', iot_devices))
return len(displays) and iot_devices[displays[0]]
def action(self, data):
if data.get('action') == "update_url" and self.device_identifier != 'distant_display':
self.update_url(data.get('url'))
elif data.get('action') == "display_refresh" and self.device_identifier != 'distant_display':
self.call_xdotools('F5')
elif data.get('action') == "take_control":
self.take_control(self.data['owner'], data.get('html'))
elif data.get('action') == "customer_facing_display":
self.update_customer_facing_display(self.data['owner'], data.get('html'))
elif data.get('action') == "get_owner":
self.data = {
'value': '',
'owner': self.owner,
}
event_manager.device_changed(self)
def run(self):
while self.device_identifier != 'distant_display' and not self._stopped.isSet():
time.sleep(60)
@@ -137,6 +129,27 @@ class DisplayDriver(Driver):
event_manager.device_changed(self)
self.event_data.set()
def _action_update_url(self, data):
if self.device_identifier != 'distant_display':
self.update_url(data.get('url'))
def _action_display_refresh(self, data):
if self.device_identifier != 'distant_display':
self.call_xdotools('F5')
def _action_take_control(self, data):
self.take_control(self.data.get('owner'), data.get('html'))
def _action_customer_facing_display(self, data):
self.update_customer_facing_display(self.data.get('owner'), data.get('html'))
def _action_get_owner(self, data):
self.data = {
'value': '',
'owner': self.owner,
}
event_manager.device_changed(self)
class DisplayController(http.Controller):
@http.route('/hw_proxy/display_refresh', type='json', auth='none', cors='*')
addons/hw_drivers/iot_handlers/drivers/KeyboardUSBDriver.py
+32 -28
View File
@@ -40,6 +40,12 @@ class KeyboardUSBDriver(Driver):
self.device_connection = 'direct'
self.device_name = self._set_name()
self._actions.update({
'update_layout': self._update_layout,
'update_is_scanner': self._save_is_scanner,
'': self._action_default,
})
# from https://github.com/xkbcommon/libxkbcommon/blob/master/test/evdev-scancodes.h
self._scancode_to_modifier = {
42: 'left_shift',
@@ -117,21 +123,6 @@ class KeyboardUSBDriver(Driver):
_logger.warning(e)
return _('Unknown input device')
def action(self, data):
if data.get('action', False) == 'update_layout':
layout = {
'layout': data.get('layout'),
'variant': data.get('variant'),
}
self._change_keyboard_layout(layout)
self.save_layout(layout)
elif data.get('action', False) == 'update_is_scanner':
is_scanner = {'is_scanner': data.get('is_scanner')}
self.save_is_scanner(is_scanner)
else:
self.data['value'] = ''
event_manager.device_changed(self)
def run(self):
try:
for event in self.input_device.read_loop():
@@ -198,19 +189,6 @@ class KeyboardUSBDriver(Driver):
data[self.device_identifier] = layout
helpers.write_file('odoo-keyboard-layouts.conf', json.dumps(data))
def save_is_scanner(self, is_scanner):
"""Save the type of device.
We need that in order to keep the selected type of device after a reboot.
"""
file_path = Path.home() / 'odoo-keyboard-is-scanner.conf'
if file_path.exists():
data = json.loads(file_path.read_text())
else:
data = {}
data[self.device_identifier] = is_scanner
helpers.write_file('odoo-keyboard-is-scanner.conf', json.dumps(data))
self._set_device_type('scanner') if is_scanner.get('is_scanner') else self._set_device_type()
def load_layout(self):
"""Read the layout from the saved filed and set it as current layout.
If no file or no layout is found we use 'us' by default.
@@ -223,6 +201,10 @@ class KeyboardUSBDriver(Driver):
layout = {'layout': 'us'}
self._change_keyboard_layout(layout)
def _action_default(self, data):
self.data['value'] = ''
event_manager.device_changed(self)
def _is_scanner(self):
"""Read the device type from the saved filed and set it as current type.
If no file or no device type is found we try to detect it automatically.
@@ -268,6 +250,28 @@ class KeyboardUSBDriver(Driver):
else:
self._current_barcode += self._scancode_to_char(scancode)
def _save_is_scanner(self, data):
"""Save the type of device.
We need that in order to keep the selected type of device after a reboot.
"""
is_scanner = {'is_scanner': data.get('is_scanner')}
file_path = Path.home() / 'odoo-keyboard-is-scanner.conf'
if file_path.exists():
data = json.loads(file_path.read_text())
else:
data = {}
data[self.device_identifier] = is_scanner
helpers.write_file('odoo-keyboard-is-scanner.conf', json.dumps(data))
self._set_device_type('scanner') if is_scanner.get('is_scanner') else self._set_device_type()
def _update_layout(self, data):
layout = {
'layout': data.get('layout'),
'variant': data.get('variant'),
}
self._change_keyboard_layout(layout)
self.save_layout(layout)
def _set_device_type(self, device_type='keyboard'):
"""Modify the device type between 'keyboard' and 'scanner'
addons/hw_drivers/iot_handlers/drivers/PrinterDriver.py
+11 -9
View File
@@ -80,6 +80,12 @@ class PrinterDriver(Driver):
}
self.send_status()
self._actions.update({
'cashbox': self.open_cashbox,
'print_receipt': self.print_receipt,
'': self._action_default,
})
self.receipt_protocol = 'star' if 'STR_T' in device['device-id'] else 'escpos'
if 'direct' in self.device_connection and any(cmd in device['device-id'] for cmd in ['CMD:STAR;', 'CMD:ESC/POS;']):
self.print_status()
@@ -127,14 +133,6 @@ class PrinterDriver(Driver):
status = 'connected' if any(iot_devices[d].device_type == "printer" and iot_devices[d].device_connection == 'direct' for d in iot_devices) else 'disconnected'
return {'status': status, 'messages': ''}
def action(self, data):
if data.get('action') == 'cashbox':
self.open_cashbox()
elif data.get('action') == 'print_receipt':
self.print_receipt(b64decode(data['receipt']))
else:
self.print_raw(b64decode(data['document']))
def disconnect(self):
self.update_status('disconnected', 'Printer was disconnected')
super(PrinterDriver, self).disconnect()
@@ -168,7 +166,8 @@ class PrinterDriver(Driver):
process = subprocess.Popen(["lp", "-d", self.device_identifier], stdin=subprocess.PIPE)
process.communicate(data)
def print_receipt(self, receipt):
def print_receipt(self, data):
receipt = b64decode(data['receipt'])
im = Image.open(io.BytesIO(receipt))
# Convert to greyscale then to black and white
@@ -257,6 +256,9 @@ class PrinterDriver(Driver):
for drawer in commands['drawers']:
self.print_raw(drawer)
def _action_default(self, data):
self.print_raw(b64decode(data['document']))
class PrinterController(http.Controller):
addons/hw_drivers/iot_handlers/drivers/SerialBaseDriver.py
+2 -2
View File
@@ -66,9 +66,9 @@ class SerialDriver(Driver):
"""
super(SerialDriver, self).__init__(identifier, device)
self._actions = {
self._actions.update({
'get_status': self._push_status,
}
})
self.device_connection = 'serial'
self._device_lock = Lock()
self._status = {'status': self.STATUS_CONNECTING, 'message_title': '', 'message_body': ''}