add fifo and timeout handler for modbus

app/src/gen3/talent.py

@@ -36,7 +36,7 @@ class Control:
 
 class Talent(Message):
     def __init__(self, server_side: bool, id_str=b''):
-        super().__init__(server_side)
+        super().__init__(server_side, self.send_modbus_cb, 15)
         self.await_conn_resp_cnt = 0
         self.id_str = id_str
         self.contact_name = b''
@@ -65,6 +65,7 @@ class Talent(Message):
         # deallocated by the garbage collector ==> we get a memory leak
         self.switch.clear()
         self.state = self.STATE_CLOSED
+        super().close()
 
     def __set_serial_no(self, serial_no: str):
 
@@ -122,20 +123,22 @@ class Talent(Message):
                         f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
         return
 
-    async def send_modbus_cmd(self, func, addr, val) -> None:
+    def send_modbus_cb(self, modbus_pdu: bytearray):
-        if self.state != self.STATE_UP:
-            return
         self.forward_modbus_resp = False
         self.__build_header(0x70, 0x77)
         self._send_buffer += b'\x00\x01\xa3\x28'   # fixme
-        modbus_msg = self.mb.build_msg(Modbus.INV_ADDR, func, addr, val)
+        self._send_buffer += struct.pack('!B', len(modbus_pdu))
-        self._send_buffer += struct.pack('!B', len(modbus_msg))
+        self._send_buffer += modbus_pdu
-        self._send_buffer += modbus_msg
         self.__finish_send_msg()
-        try:
+        hex_dump_memory(logging.INFO, f'Send Modbus Command:{self.addr}:',
-            await self.async_write('Send Modbus Command:')
+                        self._send_buffer, len(self._send_buffer))
-        except Exception:
+        self.writer.write(self._send_buffer)
-            self._send_buffer = bytearray(0)
+        self._send_buffer = bytearray(0)  # self._send_buffer[sent:]
+
+    async def send_modbus_cmd(self, func, addr, val) -> None:
+        if self.state != self.STATE_UP:
+            return
+        self.mb.build_msg(Modbus.INV_ADDR, func, addr, val)
 
     def _init_new_client_conn(self) -> bool:
         contact_name = self.contact_name

app/src/gen3plus/solarman_v5.py

@@ -52,7 +52,7 @@ class SolarmanV5(Message):
     MB_RTU_CMD = 2
 
     def __init__(self, server_side: bool):
-        super().__init__(server_side)
+        super().__init__(server_side, self.send_modbus_cb, 5)
 
         self.header_len = 11  # overwrite construcor in class Message
         self.control = 0
@@ -104,6 +104,7 @@ class SolarmanV5(Message):
         # deallocated by the garbage collector ==> we get a memory leak
         self.switch.clear()
         self.state = self.STATE_CLOSED
+        super().close()
 
     def __set_serial_no(self, snr: int):
         serial_no = str(snr)
@@ -301,20 +302,22 @@ class SolarmanV5(Message):
                                          self._heartbeat())
         self.__finish_send_msg()
 
-    async def send_modbus_cmd(self, func, addr, val) -> None:
+    def send_modbus_cb(self, pdu: bytearray):
-        if self.state != self.STATE_UP:
-            return
         self.forward_modbus_resp = False
         self.__build_header(0x4510)
         self._send_buffer += struct.pack('<BHLLL', self.MB_RTU_CMD,
                                          0x2b0, 0, 0, 0)
-        self._send_buffer += self.mb.build_msg(Modbus.INV_ADDR,
+        self._send_buffer += pdu
-                                               func, addr, val)
         self.__finish_send_msg()
-        try:
+        hex_dump_memory(logging.INFO, f'Send Modbus Command:{self.addr}:',
-            await self.async_write('Send Modbus Command:')
+                        self._send_buffer, len(self._send_buffer))
-        except Exception:
+        self.writer.write(self._send_buffer)
-            self._send_buffer = bytearray(0)
+        self._send_buffer = bytearray(0)  # self._send_buffer[sent:]
+
+    async def send_modbus_cmd(self, func, addr, val) -> None:
+        if self.state != self.STATE_UP:
+            return
+        self.mb.build_msg(Modbus.INV_ADDR, func, addr, val)
 
     async def send_at_cmd(self, AT_cmd: str) -> None:
         if self.state != self.STATE_UP:

app/src/messages.py

@@ -56,12 +56,14 @@ class Message(metaclass=IterRegistry):
     STATE_UP = 2
     STATE_CLOSED = 3
 
-    def __init__(self, server_side: bool):
+    def __init__(self, server_side: bool, send_modbus_cb, mb_timeout):
         self._registry.append(weakref.ref(self))
 
         self.server_side = server_side
         if server_side:
-            self.mb = Modbus()
+            self.mb = Modbus(send_modbus_cb, mb_timeout)
+        else:
+            self.mb = None
 
         self.header_valid = False
         self.header_len = 0
@@ -91,6 +93,9 @@ class Message(metaclass=IterRegistry):
     Our puplic methods
     '''
     def close(self) -> None:
+        if self.mb:
+            del self.mb
+            self.mb = None
         pass  # pragma: no cover
 
     def inc_counter(self, counter: str) -> None:

app/src/modbus.py

@@ -1,5 +1,6 @@
 import struct
 import logging
+import asyncio
 from typing import Generator
 
 if __name__ == "app.src.modbus":
@@ -65,85 +66,133 @@ class Modbus():
         0x3029: {'reg': Register.PV4_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01},  # noqa: E501
     }
 
-    def __init__(self):
+    def __init__(self, snd_handler, timeout: int = 1):
         if not len(self.__crc_tab):
             self.__build_crc_tab(CRC_POLY)
+        self.que = asyncio.Queue(100)
+        self.snd_handler = snd_handler
+        self.timeout = timeout
+        self.last_addr = 0
         self.last_fcode = 0
         self.last_len = 0
         self.last_reg = 0
         self.err = 0
+        self.loop = asyncio.get_event_loop()
+        self.req_pend = False
+        self.tim = None
 
-    def build_msg(self, addr, func, reg, val):
+    def start_timer(self):
+        if self.req_pend:
+            return
+        self.req_pend = True
+        self.tim = self.loop.call_later(self.timeout, self.timeout_cb)
+        # logging.debug(f'Modbus start timer {self}')
+
+    def stop_timer(self):
+        self.req_pend = False
+        # logging.debug(f'Modbus stop timer {self}')
+        if self.tim:
+            self.tim.cancel()
+        self.get_next_req()
+
+    def timeout_cb(self):
+        self.req_pend = False
+        logging.info(f'Modbus timeout {self}')
+        self.get_next_req()
+
+    def get_next_req(self) -> None:
+        if self.req_pend:
+            return
+        try:
+            item = self.que.get_nowait()
+            req = item['req']
+            self.rsp_handler = item['rsp_hdl']
+            self.last_addr = req[0]
+            self.last_fcode = req[1]
+
+            res = struct.unpack_from('>HH', req, 2)
+            self.last_reg = res[0]
+            self.last_len = res[1]
+            self.start_timer()
+            self.snd_handler(req)
+        except asyncio.QueueEmpty:
+            pass
+
+    def build_msg(self, addr, func, reg, val) -> None:
         msg = struct.pack('>BBHH', addr, func, reg, val)
         msg += struct.pack('<H', self.__calc_crc(msg))
-        self.last_fcode = func
+        self.que.put_nowait({'req': msg,
-        self.last_reg = reg
+                             'rsp_hdl': None})
-        self.last_len = val
+        if self.que.qsize() == 1:
-        self.err = 0
+            self.get_next_req()
-        return msg
 
-    def recv_req(self, buf: bytearray) -> bool:
+    def recv_req(self, buf: bytearray, rsp_handler=None) -> bool:
         # logging.info(f'recv_req: first byte modbus:{buf[0]} len:{len(buf)}')
         if not self.check_crc(buf):
             self.err = 1
             logging.error('Modbus recv: CRC error')
             return False
-        if buf[0] != self.INV_ADDR:
+        self.que.put_nowait({'req': buf,
-            self.err = 2
+                             'rsp_hdl': rsp_handler})
-            logging.info(f'Modbus recv: Wrong addr{buf[0]}')
+        if self.que.qsize() == 1:
-            return False
+            self.get_next_req()
-        res = struct.unpack_from('>BHH', buf, 1)
+
-        self.last_fcode = res[0]
-        self.last_reg = res[1]
-        self.last_len = res[2]
-        self.err = 0
         return True
 
     def recv_resp(self, info_db, buf: bytearray, node_id: str) -> \
             Generator[tuple[str, bool], None, None]:
         # logging.info(f'recv_resp: first byte modbus:{buf[0]} len:{len(buf)}')
+        if not self.req_pend:
+            self.err = 5
+            return
         if not self.check_crc(buf):
             logging.error('Modbus resp: CRC error')
             self.err = 1
             return
-        if buf[0] != self.INV_ADDR:
+        if buf[0] != self.last_addr:
             logging.info(f'Modbus resp: Wrong addr {buf[0]}')
             self.err = 2
             return
-        if buf[1] != self.last_fcode:
+        fcode = buf[1]
-            logging.info(f'Modbus: Wrong fcode {buf[1]} != {self.last_fcode}')
+        if fcode != self.last_fcode:
+            logging.info(f'Modbus: Wrong fcode {fcode} != {self.last_fcode}')
             self.err = 3
             return
-        elmlen = buf[2] >> 1
+        if self.last_addr == self.INV_ADDR and \
-        if elmlen != self.last_len:
+                (fcode == 3 or fcode == 4):
-            logging.info(f'Modbus: len error {elmlen} != {self.last_len}')
+            elmlen = buf[2] >> 1
-            self.err = 4
+            if elmlen != self.last_len:
-            return
+                logging.info(f'Modbus: len error {elmlen} != {self.last_len}')
-        self.err = 0
+                self.err = 4
+                return
+            self.stop_timer()
 
-        for i in range(0, elmlen):
+            for i in range(0, elmlen):
-            addr = self.last_reg+i
+                addr = self.last_reg+i
-            if addr in self.map:
+                if addr in self.map:
-                row = self.map[addr]
+                    row = self.map[addr]
-                info_id = row['reg']
+                    info_id = row['reg']
-                fmt = row['fmt']
+                    fmt = row['fmt']
-                val = struct.unpack_from(fmt, buf, 3+2*i)
+                    val = struct.unpack_from(fmt, buf, 3+2*i)
-                result = val[0]
+                    result = val[0]
 
-                if 'eval' in row:
+                    if 'eval' in row:
-                    result = eval(row['eval'])
+                        result = eval(row['eval'])
-                if 'ratio' in row:
+                    if 'ratio' in row:
-                    result = round(result * row['ratio'], 2)
+                        result = round(result * row['ratio'], 2)
 
-                keys, level, unit, must_incr = info_db._key_obj(info_id)
+                    keys, level, unit, must_incr = info_db._key_obj(info_id)
 
-                if keys:
+                    if keys:
-                    name, update = info_db.update_db(keys, must_incr, result)
+                        name, update = info_db.update_db(keys, must_incr,
-                    yield keys[0], update, result
+                                                         result)
-                    if update:
+                        yield keys[0], update, result
-                        info_db.tracer.log(level,
+                        if update:
-                                           f'[\'{node_id}\']MODBUS: {name}'
+                            info_db.tracer.log(level,
-                                           f' : {result}{unit}')
+                                               f'[\'{node_id}\']MODBUS: {name}'
+                                               f' : {result}{unit}')
+        else:
+            self.stop_timer()
 
     def check_crc(self, msg) -> bool:
         return 0 == self.__calc_crc(msg)