enforces request/response communication with Orbiter

Author: MAKOMO

src/artisanlib/async_comm.py

@@ -251,7 +251,7 @@ async def create_serial_connection(
 
 class AsyncComm:
 
-    __slots__ = [ '_asyncLoopThread', '_write_queue', '_running', '_host', '_port', '_serial', '_connected_handler', '_disconnected_handler',
+    __slots__ = [ '_asyncLoopThread', '_write_queue', '_running', '_serialize_write_lock', '_host', '_port', '_serial', '_connected_handler', '_disconnected_handler',
                     '_verify_crc', '_logging', '_send_timeout' ]
 
     def __init__(self, host:str = '127.0.0.1', port:int = 8080, serial:'SerialSettings|None' = None,
@@ -262,6 +262,9 @@ def __init__(self, host:str = '127.0.0.1', port:int = 8080, serial:'SerialSettin
         self._write_queue:  asyncio.Queue[bytes]|None     = None # noqa: UP037 # quotes for Python3.8 # the write_queue
         self._running:bool                                = False              # while true we keep running the thread
 
+        # lock to serialize write_await calls to realize request/response patterns
+        self._serialize_write_lock:asyncio.Lock = asyncio.Lock()
+
         # connection
         self._host:str = host
         self._port:int = port
@@ -355,12 +358,12 @@ async def handle_writes(self, writer: asyncio.StreamWriter, queue: 'asyncio.Queu
         try:
             with suppress(asyncio.CancelledError):
 # assignments in while are only only available from Python 3.8
-#                while (message := await queue.get()) != b'':
-#                    await self.write(writer, message)
-                message = await queue.get()
-                while message != b'':
+                while (message := await queue.get()) != b'':
                     await self.write(writer, message)
-                    message = await queue.get()
+#                message = await queue.get()
+#                while message != b'':
+#                    await self.write(writer, message)
+#                    message = await queue.get()
                 # on empty messages we close the connection
                 writer.close()
         except Exception as e: # pylint: disable=broad-except
@@ -434,36 +437,60 @@ def send(self, message:bytes) -> None:
         if self.async_loop_thread is not None and self._write_queue is not None:
             asyncio.run_coroutine_threadsafe(self._write_queue.put(message), self.async_loop_thread.loop)
 
-
-    # adds message to write queue and awaits new data
-    async def write_await(self, message:bytes, event:asyncio.Event, send_timeout:float) -> None:
+    # adds message to write queue and awaits new data which is assumed to event.set() once received()
+    # if serialize is set, writes are serialized such that at any momemnt only one response is awaited using the given events
+    # ensuring minimum delay of 'delay' between writes (in seconds)
+    # returns True on success and False on timeout
+    # on return the event is always cleared
+    async def write_await(self, message:bytes, event:asyncio.Event, send_timeout:float, serialize:bool, delay:float) -> bool:
         if self._write_queue is None:
-            return
-        await self._write_queue.put(message)
-        # await a response containing a new value for var with timeout
+            return False
         try:
-            await asyncio.wait_for(event.wait(), send_timeout)
-        except TimeoutError:
-            if self._logging:
-                _log.info('write_await (msg=%s, send_timeout:%s)', message.strip(), send_timeout)
+            if serialize:
+                was_writing = self._serialize_write_lock.locked() # remember if writing was in progress
+                try:
+                    await asyncio.wait_for(self._serialize_write_lock.acquire(), send_timeout)
+                except TimeoutError:
+                    pass
+                if was_writing and delay > 0: # only if writing was in progress we add a delay before writing
+                    await asyncio.sleep(delay)
+            # write out the message
+            await self._write_queue.put(message)
+            # await a response with timeout indicated by the event being set
+            try:
+                await asyncio.wait_for(event.wait(), send_timeout)
+                return True
+            except TimeoutError:
+                if self._logging:
+                    _log.info('write_await (msg=%s, send_timeout:%s)', message.strip(), send_timeout)
+            return False
+        finally:
+            # in any case, clear the event belonging to this message
+            event.clear()
+            if serialize:
+                # release the serializing write lock
+                self._serialize_write_lock.release()
+
 
-    def send_await(self, message:bytes, event:asyncio.Event, timeout:float|None = None) -> None:
+    # returns True if message was sent successfully
+    def send_await(self, message:bytes, event:asyncio.Event, timeout:float|None = None, serialize:bool = False, delay:float = 0) -> bool:
         if self.async_loop_thread is not None and self._write_queue is not None:
             send_timeout:float = self._send_timeout
             if timeout is not None:
                 send_timeout = timeout
-            task = self.write_await(message, event, send_timeout)
+            task = self.write_await(message, event, send_timeout, serialize, delay)
             if self._asyncLoopThread is not None:
                 future = asyncio.run_coroutine_threadsafe(task, self._asyncLoopThread.loop)
                 try:
-                    future.result()
+                    return future.result()
                 except TimeoutError:
                     # the coroutine took too long, cancelling the task...
                     if self._logging:
                         _log.info('send_request timeout (msg=%s, timeout:%s, send_timeout:%s)',message,timeout,send_timeout)
                     future.cancel()
                 except Exception as ex: # pylint: disable=broad-except
                     _log.error(ex)
+        return False
 
 
 

src/artisanlib/main.py

@@ -17967,7 +17967,7 @@ def kaleidoSendMessage(self, target:str, value:str) -> None:
     @pyqtSlot(bytes,bytes,bytes,int)
     def orbiterSendMessage(self, cmd:bytes, data:bytes, param:bytes, time:int) -> None:
         if self.orbiter is not None:
-            self.orbiter.send_msg(cmd, data, param, time)
+            self.orbiter.send_msg_await(cmd, data, param, time)
 
     # if record is True, an event is added during recording, otherwise only the slider is moved
     # if fire_slider_action is True, the slider action is fired

src/artisanlib/orbiter.py

@@ -63,10 +63,9 @@ class Orbiter(AsyncComm):
 
     HEADER:Final[bytes] = b'\xFF\xFF'
     EVENT:Final[bytes] = b'\x00'
-    CMD_INIT:Final[bytes] = b'\x01'
     CMD_SYNC:Final[bytes] = b'\x00'
 
-    __slots__ = [ 'send_timeout', '_connected', '_new_readings_available', '_BT', '_ET', '_IT', '_DT', '_air', '_drum', '_damper',
+    __slots__ = [ 'send_timeout', '_connected', '_ACK_received', '_BT', '_ET', '_IT', '_DT', '_air', '_drum', '_damper',
             '_heater', '_sound', '_RoR', '_master_control', '_SERIAL',  '_FW_VERSION', '_PCB_VERSION', '_DASHBOARD_STATUS', '_MODEL', '_MODEL_NUM',
             'isRoaster_Roasting' ]
 
@@ -82,7 +81,7 @@ def __init__(self, serial:'SerialSettings',
 
         # current readings
         self._connected:int = 0  # connection status (0:disconnected, 1:connected)
-        self._new_readings_available:asyncio.Event = asyncio.Event()
+        self._ACK_received:asyncio.Event = asyncio.Event()
         #-
         self._BT:float = -1      # bean temperature
         self._ET:float = -1      # environmental temperature
@@ -116,9 +115,8 @@ def setLogging(self, b:bool) -> None:
     # getBT triggers fetching a complete set of new readings
     # time is the preheat/roasting/cooling time in seconds send along the sync command to the machine
     def getBT(self, time:int = 0) -> float:
-        if not self._new_readings_available.is_set():
+        if not self._ACK_received.is_set(): # only send if no sync command is currently send
             self.send_sync_await(time)
-        self._new_readings_available.clear()
         return self._BT
     def getET(self) -> float:
         return self._ET
@@ -178,7 +176,7 @@ async def read_msg(self, stream: asyncio.StreamReader|IteratorReader) -> None:
         # check for the second header byte
         if await stream.readexactly(1) == self.HEADER[1:2]:
             cmd = await stream.readexactly(1)
-            if cmd[0] == 0: # sync data (total 28 bytes)
+            if cmd[0] == 0: # sync data ACK (total 28 bytes)
                 if self._logging:
                     _log.debug('Orbiter CMD sync data')
                 data = await stream.readexactly(25)
@@ -207,7 +205,7 @@ async def read_msg(self, stream: asyncio.StreamReader|IteratorReader) -> None:
                         self._damper = data[21]
                         self._sound = data[22]
                         self._master_control = data[23]
-                        self._new_readings_available.set()
+                        self._ACK_received.set()
                     else:
                         _log.debug('Orbiter CRC failed: %s != %s', compute_crc(cmd[0:1] + data[:24]), data[24])
                 except Exception as e: # pylint: disable=broad-except
@@ -237,6 +235,7 @@ async def read_msg(self, stream: asyncio.StreamReader|IteratorReader) -> None:
                         _log.debug('Orbiter MODEL_NUM: %s', self._MODEL_NUM)
                         _log.debug('Orbiter _sound: %s', self._sound)
                         _log.debug('Orbiter _master_control: %s', self._master_control)
+                        self._ACK_received.set()
                     else:
                         _log.debug('Orbiter CRC failed: %s != %s', compute_crc(cmd[0:1] + data[:24]), data[24])
                 except Exception as e: # pylint: disable=broad-except
@@ -257,24 +256,15 @@ def create_msg(self, cmd:bytes, data:bytes, param:bytes, time:int) -> bytes:
         return self.HEADER + payload + crc.to_bytes(1, 'little')
 
     # data byte order: LSB last (little-endian); eg. data=b'\x07\x00' equals 7
-    def send_msg(self, cmd:bytes, data:bytes = b'\x00\x00', param:bytes = b'\x00', time:int = 0) -> None:
-        # send via socket
-        self.send(self.create_msg(cmd, data, param, time))
-
-    def send_msg_await(self, event:asyncio.Event, timeout:float, cmd:bytes, data:bytes = b'\x00\x00', param:bytes = b'\x00', time:int = 0) -> None:
-        # send via socket
-        self.send_await(self.create_msg(cmd, data, param, time), event, timeout)
+    def send_msg_await(self, cmd:bytes, data:bytes = b'\x00\x00', param:bytes = b'\x00', time:int = 0) -> None:
+        # send via socket using a request/response pattern (serialize=True) awaiting a response that sets the _ACK_received event
+        # ensuring a 100ms delay between those request/response pairs
+        self.send_await(self.create_msg(cmd, data, param, time), self._ACK_received, self.send_timeout, serialize=True, delay=0.1)
 
     #
 
-    def send_init(self) -> None:
-        self.send_msg(self.CMD_INIT)
-
-    def send_sync(self) -> None:
-        self.send_msg(self.CMD_SYNC)
-
     def send_sync_await(self, time:int) -> None:
-        self.send_msg_await(self._new_readings_available, self.send_timeout, self.CMD_SYNC, time=time)
+        self.send_msg_await(self.CMD_SYNC, time=time)
 
     #
 
@@ -672,7 +662,7 @@ def main() -> None:
     for _ in range(4):
         print('>>> hallo')
         val:int = 7
-        orbiter.send_msg(b'\x0D', val.to_bytes(2, 'little')) # set power to 7
+        orbiter.send_msg_await(b'\x0D', val.to_bytes(2, 'little')) # set power to 7
         time.sleep(1)
         print('BT', orbiter.getBT())
         time.sleep(1)

src/includes/Machines/Orbiter/OB-1.aset

@@ -101,7 +101,7 @@ bytesize=8
 comport=COM3
 parity=N
 stopbits=1
-timeout=1
+timeout=0.4
 
 [Sliders]
 ModeTempSliders=C