Initial commit

client.py

@@ -0,0 +1,90 @@
+from config import *
+from httpx_ws import aconnect_ws
+import orjson
+import asyncio
+import uuid
+
+class MMClient:
+	def __init__(self):
+		self.url = WEBSOCKET_URL
+		self.ws = None
+		self.lock = asyncio.Lock()
+		self.recv_queue = asyncio.Queue()
+		self.send_queue = asyncio.Queue()
+		self.pending_acks = {}
+		self.connected = False
+
+	async def run(self):
+		while True:
+			try:
+				async with aconnect_ws(self.url) as ws:
+					self.ws = ws
+					self.connected = True
+					send_task = asyncio.create_task(self._send_loop())
+					recv_task = asyncio.create_task(self._recv_loop())
+					await asyncio.gather(send_task, recv_task)
+			except Exception as e:
+				print(f"Got exception: {e}. Retrying in 5 seconds...")
+				self.connected = False
+				await asyncio.sleep(5)
+
+	async def _send_loop(self):
+		while True:
+			packet = await self.send_queue.get()
+
+			# Requeue packet for later if disconnected
+			if not self.connected:
+				await self.send_queue.put(packet)
+				await asyncio.sleep(1)
+				continue
+
+			print(packet)
+			packet = orjson.dumps(packet)
+			await self.ws.send_bytes(packet)
+
+	async def _recv_loop(self):
+		while True:
+			try:
+				packet = await self.ws.receive()
+				packet = orjson.loads(packet.data)
+			except Exception as e:
+				print("Something went wrong", e)
+				continue
+
+			if nonce := packet.get('nonce'):
+				if nonce and nonce in self.pending_acks:
+					fut = self.pending_acks.pop(nonce)
+					if not fut.done():
+						fut.set_result(packet)
+
+			print("RECEIVED EVENT", packet)
+
+	async def send_event(self, event_id, payload, ack=False):
+		packet = {
+			"_": event_id,
+			"payload": payload,
+			"nonce": str(uuid.uuid1())
+		}
+
+
+		print("Sending packet...")
+		if ack:
+			fut = asyncio.get_event_loop().create_future()
+			self.pending_acks[str(packet["nonce"])] = fut
+
+		await self.send_queue.put(packet)
+
+		if ack:
+			try:
+				async with asyncio.timeout(2):
+					print("WAITING FOR RETURN....")
+					result = await fut
+			except asyncio.TimeoutError:
+				print("RETURN DID NOT ARRIVE....")
+				return False
+			else:
+				print("GOT RETURN RESULT....", result)
+				return result
+
+		print("CCCCCCC")
+		return

config.py

@@ -0,0 +1,4 @@
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
+WEBSOCKET_URL = 'ws://[::1]:8000/api/event/stream'

device/serial.py

@@ -0,0 +1,82 @@
+import logging
+import asyncio
+import serial
+
+log = logging.getLogger('serial-device')
+
+def _zebra_beep(beep_id):
+	command = [0x05, 0xE6, 0x04, 0x00, beep_id]
+	checksum = (~sum(command) + 1)
+	return bytes(command) + (checksum&0xffff).to_bytes(2, 'big')
+
+def _honeywell_beep(beep_id):
+	return b"0\x1b" + f"{beep_id}".encode('ASCII') + b','
+
+_beep_map = {
+	'ack': {'zebra': _zebra_beep(5), 'honeywell': _honeywell_beep(7)},
+	'nack': {'zebra': _zebra_beep(15), 'honeywell': _honeywell_beep('b')},
+	'err': {'zebra': _zebra_beep(16), 'honeywell': _honeywell_beep(8)}
+}
+
+class SerialDevice:
+	def __init__(self, path, mm, brand):
+		self.path = path
+		self.ser = None
+		self.run = False
+		self.mm = mm
+		self.brand = None
+
+	def start(self):
+		self.run = True
+		asyncio.create_task(self.loop())
+
+	def stop(self):
+		self.run = False
+
+	async def loop(self):
+		with serial.Serial(self.path) as ser:
+
+			# Flush data that was sent during startup
+			ser.reset_input_buffer()
+
+			self.ser = ser
+			while self.run:
+				try:
+					await self.cycle(ser)
+				except Exception as e:
+					log.error(f"Halting serial loop due to exception {e}")
+					return
+
+	async def cycle(self, ser):
+
+		buf = b''
+
+		# Try to read all the remaining data
+		while ser.in_waiting:
+			buf += ser.read()
+		else:
+			if not buf:
+				# Sleep until data is available
+				while not ser.in_waiting:
+					await asyncio.sleep(0.1) # If there is no data, sleep
+				return
+
+		# We have data, let's apply some basic cleaning
+		buf = buf.strip(b"\r\n") # Remove line break
+
+		log.debug(f"RECV {buf}")
+		res = await self.mm.send_event("BarcodeScan", {'content': buf.decode('ASCII')}, ack=True)
+
+		if beep := res.get('beep'):
+			if beep not in _beep_map:
+				beep = 'nack'
+
+			if self.brand in _beep_map[beep]:
+				log.debug(f"BEEP {beep}")
+				ser.write(_beep_map[beep][brand])
+
+			# Reset the input buffer (discard the response)
+			# TODO: make this more elegant
+			async with asyncio.timeout(0.5):
+				ser.read()
+				ser.reset_input_buffer()

main.py

@@ -0,0 +1,50 @@
+from config import *
+from client import MMClient
+import asyncio
+from glob import glob
+from os.path import join
+from device.serial import SerialDevice
+
+serial_mapping = {
+	'usb-Symbol*': (SerialDevice, 'zebra'),
+	'usb-Honeywell*': (SerialDevice, 'honeywell'),
+	'usb-XinChip*': (SerialDevice, 'china')
+}
+
+class USBSerialDetect:
+	def __init__(self, mm):
+		self.devices = {}
+		self.mm = mm
+
+	async def detect(self):
+		while 1:
+			for pat, handler in serial_mapping.items():
+				for dev in glob(join('/dev/serial/by-id', pat)):
+					if self.devices.get(dev):
+						if self.devices[dev].ser.is_open:
+							continue
+						else:
+							print("Destroying device.")
+							del self.devices[dev]
+
+					print("Adding", dev, 'as', handler)
+					self.devices[dev] = handler[0](dev, self.mm, handler[1])
+					self.devices[dev].start()
+
+			await asyncio.sleep(1)
+
+async def main():
+
+	print("Connecting...")
+	mm = MMClient()
+	asyncio.create_task(mm.run())
+
+	print("Detecting barcodes...")
+	a = USBSerialDetect(mm)
+	asyncio.create_task(a.detect())
+
+	while 1:
+		await asyncio.sleep(100)
+
+if __name__ == '__main__':
+	asyncio.run(main())