# -*- coding: utf-8 -*- # Part of Odoo, Flectra. See LICENSE file for full copyright and licensing details. from __future__ import print_function import logging import subprocess import time import subprocess from threading import Lock from flectra import http from flectra.http import request _logger = logging.getLogger(__name__) # Those are the builtin raspberry pi USB modules, they should # not appear in the list of connected devices. BANNED_DEVICES = { "0424:9514", # Standard Microsystem Corp. Builtin Ethernet module "1d6b:0002", # Linux Foundation 2.0 root hub "0424:ec00", # Standard Microsystem Corp. Other Builtin Ethernet module "0424:2514", # Standard Microsystems Corp. USB 2.0 Hub (rpi3b+) "0424:7800", # Standard Microsystems Corp. (rpi3b+) } # drivers modules must add to drivers an object with a get_status() method # so that 'status' can return the status of all active drivers drivers = {} # keep a list of RS-232 devices that have been recognized by a driver, # so other drivers can skip them during probes rs232_devices = {} # {'/path/to/device': 'driver'} rs232_lock = Lock() # must be held to update `rs232_devices` class Proxy(http.Controller): def get_status(self): statuses = {} for driver in drivers: statuses[driver] = drivers[driver].get_status() return statuses @http.route('/hw_proxy/hello', type='http', auth='none', cors='*') def hello(self): return "ping" @http.route('/hw_proxy/handshake', type='json', auth='none', cors='*') def handshake(self): return True @http.route('/hw_proxy/status', type='http', auth='none', cors='*') def status_http(self, debug=None, **kwargs): resp = """ Flectra's PosBox

Hardware Status

The list of enabled drivers and their status

""" statuses = self.get_status() for driver in statuses: status = statuses[driver] if status['status'] == 'connecting': color = 'black' elif status['status'] == 'connected': color = 'green' else: color = 'red' resp += "

"+driver+' : '+status['status']+"

\n" resp += "\n" resp += """

Connected Devices

The list of connected USB devices as seen by the posbox

""" if debug is None: resp += """(debug version)""" devices = subprocess.check_output("lsusb").decode('utf-8').split('\n') count = 0 resp += "
\n" for device in devices: device_name = device[device.find('ID')+2:] if device_name: # to avoid last empty line device_id = device_name.split()[0] if not (device_id in BANNED_DEVICES): resp += "
"+device_name+"
\n" count += 1 if count == 0: resp += "
No USB Device Found
" resp += "
\n\n\n\n" if debug is not None: resp += """

Debug version

lsusb -v output:

                %s
""" % subprocess.check_output('lsusb -v', shell=True).decode('utf-8') return request.make_response(resp,{ 'Cache-Control': 'no-cache', 'Content-Type': 'text/html; charset=utf-8', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'GET', }) @http.route('/hw_proxy/status_json', type='json', auth='none', cors='*') def status_json(self): return self.get_status() @http.route('/hw_proxy/scan_item_success', type='json', auth='none', cors='*') def scan_item_success(self, ean): """ A product has been scanned with success """ print('scan_item_success: %s', ean) @http.route('/hw_proxy/scan_item_error_unrecognized', type='json', auth='none', cors='*') def scan_item_error_unrecognized(self, ean): """ A product has been scanned without success """ print('scan_item_error_unrecognized: %s', ean) @http.route('/hw_proxy/help_needed', type='json', auth='none', cors='*') def help_needed(self): """ The user wants an help (ex: light is on) """ print("help_needed") @http.route('/hw_proxy/help_canceled', type='json', auth='none', cors='*') def help_canceled(self): """ The user stops the help request """ print("help_canceled") @http.route('/hw_proxy/payment_request', type='json', auth='none', cors='*') def payment_request(self, price): """ The PoS will activate the method payment """ print("payment_request: price:%s" % price) return 'ok' @http.route('/hw_proxy/payment_status', type='json', auth='none', cors='*') def payment_status(self): print("payment_status") return { 'status':'waiting' } @http.route('/hw_proxy/payment_cancel', type='json', auth='none', cors='*') def payment_cancel(self): print("payment_cancel") @http.route('/hw_proxy/transaction_start', type='json', auth='none', cors='*') def transaction_start(self): print('transaction_start') @http.route('/hw_proxy/transaction_end', type='json', auth='none', cors='*') def transaction_end(self): print('transaction_end') @http.route('/hw_proxy/cashier_mode_activated', type='json', auth='none', cors='*') def cashier_mode_activated(self): print('cashier_mode_activated') @http.route('/hw_proxy/cashier_mode_deactivated', type='json', auth='none', cors='*') def cashier_mode_deactivated(self): print('cashier_mode_deactivated') @http.route('/hw_proxy/open_cashbox', type='json', auth='none', cors='*') def open_cashbox(self): print('open_cashbox') @http.route('/hw_proxy/print_receipt', type='json', auth='none', cors='*') 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)) @http.route('/hw_proxy/print_pdf_invoice', type='json', auth='none', cors='*') def print_pdf_invoice(self, pdfinvoice): print('print_pdf_invoice %s' % pdfinvoice)