refactoring

app/src/async_ifc.py

@@ -1,11 +1,111 @@
-if __name__ == "app.src.async_ifc":
-    from app.src.byte_fifo import ByteFifo
-else:  # pragma: no cover
-    from byte_fifo import ByteFifo
+from abc import ABC, abstractmethod
 
 
-class AsyncIfc():
-    def __init__(self):
-        self.read = ByteFifo()
-        self.write = ByteFifo()
-        self.forward = ByteFifo()
+class AsyncIfc(ABC):
+    @abstractmethod
+    def get_conn_no(self):
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def set_node_id(self, value: str):
+        pass  # pragma: no cover
+
+    #
+    #  TX - QUEUE
+    #
+    @abstractmethod
+    def tx_add(self, data: bytearray):
+        ''' add data to transmit queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_flush(self):
+        ''' send transmit queue and clears it'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_get(self, size: int = None) -> bytearray:
+        '''removes size numbers of bytes and return them'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_peek(self, size: int = None) -> bytearray:
+        '''returns size numbers of byte without removing them'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_log(self, level, info):
+        ''' log the transmit queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_clear(self):
+        ''' clear transmit queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def tx_len(self):
+        ''' get numner of bytes in the transmit queue'''
+        pass  # pragma: no cover
+
+    #
+    #  FORWARD - QUEUE
+    #
+    @abstractmethod
+    def fwd_add(self, data: bytearray):
+        ''' add data to forward queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def fwd_flush(self):
+        ''' send forward queue and clears it'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def fwd_log(self, level, info):
+        ''' log the forward queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def fwd_clear(self):
+        ''' clear forward queue'''
+        pass  # pragma: no cover
+
+    #
+    #  RX - QUEUE
+    #
+    @abstractmethod
+    def rx_get(self, size: int = None) -> bytearray:
+        '''removes size numbers of bytes and return them'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def rx_peek(self, size: int = None) -> bytearray:
+        '''returns size numbers of byte without removing them'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def rx_log(self, level, info):
+        ''' logs the receive queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def rx_clear(self):
+        ''' clear receive queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def rx_len(self):
+        ''' get numner of bytes in the receive queue'''
+        pass  # pragma: no cover
+
+    @abstractmethod
+    def rx_set_cb(self, callback):
+        pass  # pragma: no cover
+
+    #
+    #  Protocol Callbacks
+    #
+    @abstractmethod
+    def prot_set_timeout_cb(self, callback):
+        pass  # pragma: no cover

app/src/async_stream.py

@@ -7,19 +7,114 @@ from typing import Self
 from itertools import count
 
 if __name__ == "app.src.async_stream":
+    from app.src.byte_fifo import ByteFifo
     from app.src.async_ifc import AsyncIfc
-    from app.src.messages import State
+    from app.src.infos import Infos
 else:  # pragma: no cover
+    from byte_fifo import ByteFifo
     from async_ifc import AsyncIfc
-    from messages import State
+    from infos import Infos
 
 
 import gc
 logger = logging.getLogger('conn')
 
 
-class AsyncStream():
+class AsyncIfcImpl(AsyncIfc):
     _ids = count(0)
+
+    def __init__(self) -> None:
+        logger.debug('AsyncIfcImpl.__init__')
+        self.fwd_fifo = ByteFifo()
+        self.tx_fifo = ByteFifo()
+        self.rx_fifo = ByteFifo()
+        self.conn_no = next(self._ids)
+        self.node_id = ''
+        self.timeout_cb = None
+
+    def set_node_id(self, value: str):
+        self.node_id = value
+
+    def get_conn_no(self):
+        return self.conn_no
+
+    def tx_add(self, data: bytearray):
+        ''' add data to transmit queue'''
+        self.tx_fifo += data
+
+    def tx_flush(self):
+        ''' send transmit queue and clears it'''
+        self.tx_fifo()
+
+    def tx_get(self, size: int = None) -> bytearray:
+        '''removes size numbers of bytes and return them'''
+        return self.tx_fifo.get(size)
+
+    def tx_peek(self, size: int = None) -> bytearray:
+        '''returns size numbers of byte without removing them'''
+        return self.tx_fifo.peek(size)
+
+    def tx_log(self, level, info):
+        ''' log the transmit queue'''
+        self.tx_fifo.logging(level, info)
+
+    def tx_clear(self):
+        ''' clear transmit queue'''
+        self.tx_fifo.clear()
+
+    def tx_len(self):
+        ''' get numner of bytes in the transmit queue'''
+        return len(self.tx_fifo)
+
+    def fwd_add(self, data: bytearray):
+        ''' add data to forward queue'''
+        self.fwd_fifo += data
+
+    def fwd_flush(self):
+        ''' send forward queue and clears it'''
+        self.fwd_fifo()
+
+    def fwd_log(self, level, info):
+        ''' log the forward queue'''
+        self.fwd_fifo.logging(level, info)
+
+    def fwd_clear(self):
+        ''' clear forward queue'''
+        self.fwd_fifo.clear()
+
+    def rx_get(self, size: int = None) -> bytearray:
+        '''removes size numbers of bytes and return them'''
+        return self.rx_fifo.get(size)
+
+    def rx_peek(self, size: int = None) -> bytearray:
+        '''returns size numbers of byte without removing them'''
+        return self.rx_fifo.peek(size)
+
+    def rx_log(self, level, info):
+        ''' logs the receive queue'''
+        self.rx_fifo.logging(level, info)
+
+    def rx_clear(self):
+        ''' clear receive queue'''
+        self.rx_fifo.clear()
+
+    def rx_len(self):
+        ''' get numner of bytes in the receive queue'''
+        return len(self.rx_fifo)
+
+    def rx_set_cb(self, callback):
+        self.rx_fifo.reg_trigger(callback)
+
+    def prot_set_timeout_cb(self, callback):
+        self.timeout_cb = callback
+
+
+class StreamPtr():
+    def __init__(self, stream):
+        self.stream = stream
+
+
+class AsyncStream(AsyncIfcImpl):
     MAX_PROC_TIME = 2
     '''maximum processing time for a received msg in sec'''
     MAX_START_TIME = 400
@@ -30,31 +125,31 @@ class AsyncStream():
     '''maximum default time without a received msg in sec'''
 
     def __init__(self, reader: StreamReader, writer: StreamWriter,
-                 addr, ifc: "AsyncIfc") -> None:
+                 addr, async_publ_mqtt, async_create_remote,
+                 rstream: "StreamPtr") -> None:
+        AsyncIfcImpl.__init__(self)
+
         logger.debug('AsyncStream.__init__')
-        ifc.write.reg_trigger(self.__write_cb)
-        self.ifc = ifc
+
+        self.remote = rstream
+        self.tx_fifo.reg_trigger(self.__write_cb)
+        self.async_create_remote = async_create_remote
         self._reader = reader
         self._writer = writer
         self.addr = addr
         self.r_addr = ''
         self.l_addr = ''
-        self.conn_no = next(self._ids)
         self.proc_start = None  # start processing start timestamp
         self.proc_max = 0
+        self.async_publ_mqtt = async_publ_mqtt
 
     def __write_cb(self):
-        self._writer.write(self.ifc.write.get())
+        self._writer.write(self.tx_fifo.get())
 
     def __timeout(self) -> int:
-        if self.state == State.init or self.state == State.received:
-            to = self.MAX_START_TIME
-        elif self.state == State.up and \
-                self.server_side and self.modbus_polling:
-            to = self.MAX_INV_IDLE_TIME
-        else:
-            to = self.MAX_DEF_IDLE_TIME
-        return to
+        if self.timeout_cb is callable:
+            return self.timeout_cb
+        return 360
 
     async def publish_outstanding_mqtt(self):
         '''Publish all outstanding MQTT topics'''
@@ -69,25 +164,25 @@ class AsyncStream():
         '''Loop for receiving messages from the inverter (server-side)'''
         logger.info(f'[{self.node_id}:{self.conn_no}] '
                     f'Accept connection from {addr}')
-        self.inc_counter('Inverter_Cnt')
+        Infos.inc_counter('Inverter_Cnt')
         await self.publish_outstanding_mqtt()
         await self.loop()
-        self.dec_counter('Inverter_Cnt')
+        Infos.dec_counter('Inverter_Cnt')
         await self.publish_outstanding_mqtt()
         logger.info(f'[{self.node_id}:{self.conn_no}] Server loop stopped for'
                     f' r{self.r_addr}')
 
         # if the server connection closes, we also have to disconnect
         # the connection to te TSUN cloud
-        if self.remote_stream:
+        if self.remote.stream:
             logger.info(f'[{self.node_id}:{self.conn_no}] disc client '
-                        f'connection: [{self.remote_stream.node_id}:'
-                        f'{self.remote_stream.conn_no}]')
-            await self.remote_stream.disc()
+                        f'connection: [{self.remote.stream.node_id}:'
+                        f'{self.remote.stream.conn_no}]')
+            await self.remote.stream._ifc.disc()
 
     async def client_loop(self, _: str) -> None:
         '''Loop for receiving messages from the TSUN cloud (client-side)'''
-        client_stream = await self.remote_stream.loop()
+        client_stream = await self.remote.stream._ifc.loop()
         logger.info(f'[{client_stream.node_id}:{client_stream.conn_no}] '
                     'Client loop stopped for'
                     f' l{client_stream.l_addr}')
@@ -98,13 +193,13 @@ class AsyncStream():
         # establish a new connection to the TSUN cloud
 
         # erase backlink to inverter
-        client_stream.remote_stream = None
+        client_stream.remote.stream = None
 
-        if self.remote_stream == client_stream:
+        if self.remote.stream == client_stream:
             # logging.debug(f'Client l{client_stream.l_addr} refs:'
             #               f' {gc.get_referrers(client_stream)}')
             # than erase client connection
-            self.remote_stream = None
+            self.remote.stream = None
 
     async def loop(self) -> Self:
         """Async loop handler for precessing all received messages"""
@@ -121,10 +216,10 @@ class AsyncStream():
                 await asyncio.wait_for(self.__async_read(),
                                        dead_conn_to)
 
-                if self.unique_id:
-                    await self.__async_write()
-                    await self.__async_forward()
-                    await self.async_publ_mqtt()
+                # if self.unique_id:
+                await self.__async_write()
+                await self.__async_forward()
+                await self.async_publ_mqtt()
 
             except asyncio.TimeoutError:
                 logger.warning(f'[{self.node_id}:{self.conn_no}] Dead '
@@ -150,7 +245,7 @@ class AsyncStream():
                 return self
 
             except Exception:
-                self.inc_counter('SW_Exception')
+                Infos.inc_counter('SW_Exception')
                 logger.error(
                     f"Exception for {self.addr}:\n"
                     f"{traceback.format_exc()}")
@@ -158,9 +253,9 @@ class AsyncStream():
 
     async def __async_write(self, headline: str = 'Transmit to ') -> None:
         """Async write handler to transmit the send_buffer"""
-        if len(self.ifc.write) > 0:
-            self.ifc.write.logging(logging.INFO, f'{headline}{self.addr}:')
-            self._writer.write(self.ifc.write.get())
+        if len(self.tx_fifo) > 0:
+            self.tx_fifo.logging(logging.INFO, f'{headline}{self.addr}:')
+            self._writer.write(self.tx_fifo.get())
             await self._writer.drain()
 
     async def disc(self) -> None:
@@ -176,18 +271,19 @@ class AsyncStream():
 
            hint: must be called before releasing the connection instance
         """
+        self.tx_fifo.reg_trigger(None)
+        self.async_create_remote = None
         self._reader.feed_eof()          # abort awaited read
         if self._writer.is_closing():
             return
         logger.debug(f'AsyncStream.close() l{self.l_addr} | r{self.r_addr}')
-        self.ifc.write.reg_trigger(None)
         self._writer.close()
 
     def healthy(self) -> bool:
         elapsed = 0
         if self.proc_start is not None:
             elapsed = time.time() - self.proc_start
-        if self.state == State.closed or elapsed > self.MAX_PROC_TIME:
+        if elapsed > self.MAX_PROC_TIME:
             logging.debug(f'[{self.node_id}:{self.conn_no}:'
                           f'{type(self).__name__}]'
                           f' act:{round(1000*elapsed)}ms'
@@ -203,8 +299,8 @@ class AsyncStream():
         data = await self._reader.read(4096)
         if data:
             self.proc_start = time.time()
-            self.ifc.read += data
-            wait = self.ifc.read()                # call read in parent class
+            self.rx_fifo += data
+            wait = self.rx_fifo()                # call read in parent class
             if wait > 0:
                 await asyncio.sleep(wait)
         else:
@@ -212,42 +308,42 @@ class AsyncStream():
 
     async def __async_forward(self) -> None:
         """forward handler transmits data over the remote connection"""
-        if len(self.ifc.forward) == 0:
+        if len(self.fwd_fifo) == 0:
             return
         try:
-            if not self.remote_stream:
+            if not self.remote.stream:
                 await self.async_create_remote()
-                if self.remote_stream:
-                    if self.remote_stream._init_new_client_conn():
-                        await self.remote_stream.__async_write()
+                if self.remote.stream:
+                    if self.remote.stream._init_new_client_conn():
+                        await self.remote.stream._ifc.__async_write()
 
-            if self.remote_stream:
-                self.remote_stream._update_header(self.ifc.forward.peek())
-                self.ifc.forward.logging(logging.INFO, 'Forward to '
-                                         f'{self.remote_stream.addr}:')
-                self.remote_stream._writer.write(self.ifc.forward.get())
-                await self.remote_stream._writer.drain()
+            if self.remote.stream:
+                self.remote.stream._update_header(self.fwd_fifo.peek())
+                self.fwd_fifo.logging(logging.INFO, 'Forward to '
+                                      f'{self.remote.stream.addr}:')
+                self.remote.stream._ifc._writer.write(self.fwd_fifo.get())
+                await self.remote.stream._ifc._writer.drain()
 
         except OSError as error:
-            if self.remote_stream:
-                rmt = self.remote_stream
-                self.remote_stream = None
+            if self.remote.stream:
+                rmt = self.remote.stream
+                self.remote.stream = None
                 logger.error(f'[{rmt.node_id}:{rmt.conn_no}] Fwd: {error} for '
-                             f'l{rmt.l_addr} | r{rmt.r_addr}')
-                await rmt.disc()
-                rmt.close()
+                             f'l{rmt._ifc.l_addr} | r{rmt._ifc.r_addr}')
+                await rmt._ifc.disc()
+                rmt._ifc.close()
 
         except RuntimeError as error:
-            if self.remote_stream:
-                rmt = self.remote_stream
-                self.remote_stream = None
+            if self.remote.stream:
+                rmt = self.remote.stream
+                self.remote.stream = None
                 logger.info(f'[{rmt.node_id}:{rmt.conn_no}] '
-                            f'Fwd: {error} for {rmt.l_addr}')
-                await rmt.disc()
-                rmt.close()
+                            f'Fwd: {error} for {rmt._ifc.l_addr}')
+                await rmt._ifc.disc()
+                rmt._ifc.close()
 
         except Exception:
-            self.inc_counter('SW_Exception')
+            Infos.inc_counter('SW_Exception')
             logger.error(
                 f"Fwd Exception for {self.addr}:\n"
                 f"{traceback.format_exc()}")

app/src/gen3/connection_g3.py

@@ -2,33 +2,35 @@ import logging
 from asyncio import StreamReader, StreamWriter
 
 if __name__ == "app.src.gen3.connection_g3":
-    from app.src.async_ifc import AsyncIfc
-    from app.src.async_stream import AsyncStream
+    from app.src.async_stream import AsyncStream, StreamPtr
     from app.src.gen3.talent import Talent
 else:  # pragma: no cover
-    from async_ifc import AsyncIfc
-    from async_stream import AsyncStream
+    from async_stream import AsyncStream, StreamPtr
     from gen3.talent import Talent
 
 logger = logging.getLogger('conn')
 
 
-class ConnectionG3(AsyncStream, Talent):
+class ConnectionG3(Talent):
 
     def __init__(self, reader: StreamReader, writer: StreamWriter,
-                 addr, remote_stream: 'ConnectionG3', server_side: bool,
+                 addr, rstream: 'ConnectionG3', server_side: bool,
                  id_str=b'') -> None:
-        self._ifc = AsyncIfc()
-        AsyncStream.__init__(self, reader, writer, addr, self._ifc)
+        self.remote = StreamPtr(rstream)
+        self._ifc = AsyncStream(reader, writer, addr,
+                                self.async_publ_mqtt,
+                                self.async_create_remote,
+                                self.remote)
         Talent.__init__(self, server_side, self._ifc, id_str)
 
-        self.remote_stream: 'ConnectionG3' = remote_stream
+        self.conn_no = self._ifc.get_conn_no()
+        self.addr = addr
 
     '''
     Our puplic methods
     '''
     def close(self):
-        AsyncStream.close(self)
+        self._ifc.close()
         Talent.close(self)
         # logger.info(f'AsyncStream refs: {gc.get_referrers(self)}')
 
@@ -40,10 +42,4 @@ class ConnectionG3(AsyncStream, Talent):
 
     def healthy(self) -> bool:
         logger.debug('ConnectionG3 healthy()')
-        return AsyncStream.healthy(self)
-
-    '''
-    Our private methods
-    '''
-    def __del__(self):
-        super().__del__()
+        return self._ifc.healthy()

app/src/gen3/inverter_g3.py

@@ -53,6 +53,7 @@ class InverterG3(Inverter, ConnectionG3):
     def __init__(self, reader: StreamReader, writer: StreamWriter, addr):
         super().__init__(reader, writer, addr, None, True)
         self.__ha_restarts = -1
+        self.addr = addr
 
     async def async_create_remote(self) -> None:
         '''Establish a client connection to the TSUN cloud'''
@@ -65,12 +66,12 @@ class InverterG3(Inverter, ConnectionG3):
             logging.info(f'[{self.node_id}] Connect to {addr}')
             connect = asyncio.open_connection(host, port)
             reader, writer = await connect
-            self.remote_stream = ConnectionG3(reader, writer, addr, self,
+            self.remote.stream = ConnectionG3(reader, writer, addr, self,
                                               False, self.id_str)
-            logging.info(f'[{self.remote_stream.node_id}:'
-                         f'{self.remote_stream.conn_no}] '
+            logging.info(f'[{self.remote.stream.node_id}:'
+                         f'{self.remote.stream.conn_no}] '
                          f'Connected to {addr}')
-            asyncio.create_task(self.client_loop(addr))
+            asyncio.create_task(self._ifc.client_loop(addr))
 
         except (ConnectionRefusedError, TimeoutError) as error:
             logging.info(f'{error}')
@@ -128,10 +129,6 @@ class InverterG3(Inverter, ConnectionG3):
         self.db.reg_clr_at_midnight(f'{self.entity_prfx}{self.node_id}')
 
     def close(self) -> None:
-        logging.debug(f'InverterG3.close() l{self.l_addr} | r{self.r_addr}')
+        logging.debug(f'InverterG3.close() {self.addr}')
         super().close()         # call close handler in the parent class
 #         logging.info(f'Inverter refs: {gc.get_referrers(self)}')
-
-    def __del__(self):
-        logging.debug("InverterG3.__del__")
-        super().__del__()

app/src/gen3/talent.py

@@ -48,7 +48,8 @@ class Talent(Message):
 
     def __init__(self, server_side: bool, ifc: "AsyncIfc", id_str=b''):
         super().__init__(server_side, self.send_modbus_cb, mb_timeout=15)
-        ifc.read.reg_trigger(self.read)
+        ifc.rx_set_cb(self.read)
+        ifc.prot_set_timeout_cb(self._timeout)
         self.ifc = ifc
         self.await_conn_resp_cnt = 0
         self.id_str = id_str
@@ -107,7 +108,8 @@ class Talent(Message):
         self.log_lvl.clear()
         self.state = State.closed
         self.mb_timer.close()
-        self.ifc.read.reg_trigger(None)
+        self.ifc.rx_set_cb(None)
+        self.ifc.prot_set_timeout_cb(None)
         super().close()
 
     def __set_serial_no(self, serial_no: str):
@@ -143,10 +145,10 @@ class Talent(Message):
         self._read()
         while True:
             if not self.header_valid:
-                self.__parse_header(self.ifc.read.peek(), len(self.ifc.read))
+                self.__parse_header(self.ifc.rx_peek(), self.ifc.rx_len())
 
             if self.header_valid and \
-               len(self.ifc.read) >= (self.header_len + self.data_len):
+               self.ifc.rx_len() >= (self.header_len + self.data_len):
                 if self.state == State.init:
                     self.state = State.received     # received 1st package
 
@@ -154,10 +156,10 @@ class Talent(Message):
                 if callable(log_lvl):
                     log_lvl = log_lvl()
 
-                self.ifc.read.logging(log_lvl, f'Received from {self.addr}:'
-                                      f' BufLen: {len(self.ifc.read)}'
-                                      f' HdrLen: {self.header_len}'
-                                      f' DtaLen: {self.data_len}')
+                self.ifc.rx_log(log_lvl, f'Received from {self.addr}:'
+                                f' BufLen: {self.ifc.rx_len()}'
+                                f' HdrLen: {self.header_len}'
+                                f' DtaLen: {self.data_len}')
 
                 self.__set_serial_no(self.id_str.decode("utf-8"))
                 self.__dispatch_msg()
@@ -170,9 +172,9 @@ class Talent(Message):
         tsun = Config.get('tsun')
         if tsun['enabled']:
             buflen = self.header_len+self.data_len
-            buffer = self.ifc.read.peek(buflen)
-            self.ifc.forward += buffer
-            self.ifc.forward.logging(logging.DEBUG, 'Store for forwarding:')
+            buffer = self.ifc.rx_peek(buflen)
+            self.ifc.fwd_add(buffer)
+            self.ifc.fwd_log(logging.DEBUG, 'Store for forwarding:')
 
             fnc = self.switch.get(self.msg_id, self.msg_unknown)
             logger.info(self.__flow_str(self.server_side, 'forwrd') +
@@ -181,18 +183,18 @@ class Talent(Message):
     def forward_snd(self) -> None:
         '''add the build send msg to the forwarding queue'''
         tsun = Config.get('tsun')
-        rest = self.ifc.write.get(self.send_msg_ofs)
-        buffer = self.ifc.write.get(len(self.ifc.write))
+        rest = self.ifc.tx_get(self.send_msg_ofs)
+        buffer = self.ifc.tx_get()
         if tsun['enabled']:
             _len = len(buffer)
             struct.pack_into('!l', buffer, 0, _len-4)
-            self.ifc.forward += buffer
-            self.ifc.forward.logging(logging.INFO, 'Store for forwarding:')
+            self.ifc.fwd_add(buffer)
+            self.ifc.fwd_log(logging.INFO, 'Store for forwarding:')
 
             fnc = self.switch.get(self.msg_id, self.msg_unknown)
             logger.info(self.__flow_str(self.server_side, 'forwrd') +
                         f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
-        self.ifc.write += rest
+        self.ifc.tx_add(rest)
 
     def send_modbus_cb(self, modbus_pdu: bytearray, log_lvl: int, state: str):
         if self.state != State.up:
@@ -201,13 +203,13 @@ class Talent(Message):
             return
 
         self.__build_header(0x70, 0x77)
-        self.ifc.write += b'\x00\x01\xa3\x28'   # magic ?
-        self.ifc.write += struct.pack('!B', len(modbus_pdu))
-        self.ifc.write += modbus_pdu
+        self.ifc.tx_add(b'\x00\x01\xa3\x28')   # magic ?
+        self.ifc.tx_add(struct.pack('!B', len(modbus_pdu)))
+        self.ifc.tx_add(modbus_pdu)
         self.__finish_send_msg()
 
-        self.ifc.write.logging(log_lvl, f'Send Modbus {state}:{self.addr}:')
-        self.ifc.write()
+        self.ifc.tx_log(log_lvl, f'Send Modbus {state}:{self.addr}:')
+        self.ifc.tx_flush()
 
     def _send_modbus_cmd(self, func, addr, val, log_lvl) -> None:
         if self.state != State.up:
@@ -235,9 +237,9 @@ class Talent(Message):
         self.msg_id = 0
         self.await_conn_resp_cnt += 1
         self.__build_header(0x91)
-        self.ifc.write += struct.pack(f'!{len(contact_name)+1}p'
-                                      f'{len(contact_mail)+1}p',
-                                      contact_name, contact_mail)
+        self.ifc.tx_add(struct.pack(f'!{len(contact_name)+1}p'
+                                    f'{len(contact_mail)+1}p',
+                                    contact_name, contact_mail))
 
         self.__finish_send_msg()
         return True
@@ -321,7 +323,7 @@ class Talent(Message):
             self.inc_counter('Invalid_Msg_Format')
 
             # erase broken recv buffer
-            self.ifc.read.clear()
+            self.ifc.rx_clear()
             return
 
         hdr_len = 5+id_len+2
@@ -342,16 +344,16 @@ class Talent(Message):
     def __build_header(self, ctrl, msg_id=None) -> None:
         if not msg_id:
             msg_id = self.msg_id
-        self.send_msg_ofs = len(self.ifc.write)
-        self.ifc.write += struct.pack(f'!l{len(self.id_str)+1}pBB',
-                                      0, self.id_str, ctrl, msg_id)
+        self.send_msg_ofs = self.ifc.tx_len()
+        self.ifc.tx_add(struct.pack(f'!l{len(self.id_str)+1}pBB',
+                                    0, self.id_str, ctrl, msg_id))
         fnc = self.switch.get(msg_id, self.msg_unknown)
         logger.info(self.__flow_str(self.server_side, 'tx') +
                     f' Ctl: {int(ctrl):#02x} Msg: {fnc.__name__!r}')
 
     def __finish_send_msg(self) -> None:
-        _len = len(self.ifc.write) - self.send_msg_ofs
-        struct.pack_into('!l', self.ifc.write.peek(), self.send_msg_ofs,
+        _len = self.ifc.tx_len() - self.send_msg_ofs
+        struct.pack_into('!l', self.ifc.tx_peek(), self.send_msg_ofs,
                          _len-4)
 
     def __dispatch_msg(self) -> None:
@@ -366,7 +368,7 @@ class Talent(Message):
                         f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
 
     def __flush_recv_msg(self) -> None:
-        self.ifc.read.get(self.header_len+self.data_len)
+        self.ifc.rx_get(self.header_len+self.data_len)
         self.header_valid = False
 
     '''
@@ -376,7 +378,7 @@ class Talent(Message):
         if self.ctrl.is_ind():
             if self.server_side and self.__process_contact_info():
                 self.__build_header(0x91)
-                self.ifc.write += b'\x01'
+                self.ifc.tx_add(b'\x01')
                 self.__finish_send_msg()
             # don't forward this contact info here, we will build one
             # when the remote connection is established
@@ -390,7 +392,7 @@ class Talent(Message):
             self.forward()
 
     def __process_contact_info(self) -> bool:
-        buf = self.ifc.read.peek()
+        buf = self.ifc.rx_peek()
         result = struct.unpack_from('!B', buf, self.header_len)
         name_len = result[0]
         if self.data_len == 1:  # this is a response withone status byte
@@ -417,16 +419,16 @@ class Talent(Message):
                 ts = self._timestamp()
                 logger.debug(f'time: {ts:08x}')
                 self.__build_header(0x91)
-                self.ifc.write += struct.pack('!q', ts)
+                self.ifc.tx_add(struct.pack('!q', ts))
                 self.__finish_send_msg()
 
             elif self.data_len >= 8:
                 ts = self._timestamp()
-                result = struct.unpack_from('!q', self.ifc.read.peek(),
+                result = struct.unpack_from('!q', self.ifc.rx_peek(),
                                             self.header_len)
                 self.ts_offset = result[0]-ts
-                if self.remote_stream:
-                    self.remote_stream.ts_offset = self.ts_offset
+                if self.remote.stream:
+                    self.remote.stream.ts_offset = self.ts_offset
                 logger.debug(f'tsun-time: {int(result[0]):08x}'
                              f'  proxy-time: {ts:08x}'
                              f'  offset: {self.ts_offset}')
@@ -446,10 +448,10 @@ class Talent(Message):
                     self.db.set_db_def_value(Register.POLLING_INTERVAL,
                                              self.mb_timeout)
                 self.__build_header(0x99)
-                self.ifc.write += b'\x02'
+                self.ifc.tx_add(b'\x02')
                 self.__finish_send_msg()
 
-                result = struct.unpack_from('!Bq', self.ifc.read.peek(),
+                result = struct.unpack_from('!Bq', self.ifc.rx_peek(),
                                             self.header_len)
                 resp_code = result[0]
                 ts = result[1]+self.ts_offset
@@ -457,11 +459,11 @@ class Talent(Message):
                              f'  tsun-time: {ts:08x}'
                              f'  offset: {self.ts_offset}')
                 self.__build_header(0x91)
-                self.ifc.write += struct.pack('!Bq', resp_code, ts)
+                self.ifc.tx_add(struct.pack('!Bq', resp_code, ts))
                 self.forward_snd()
                 return
         elif self.ctrl.is_resp():
-            result = struct.unpack_from('!B', self.ifc.read.peek(),
+            result = struct.unpack_from('!B', self.ifc.rx_peek(),
                                         self.header_len)
             resp_code = result[0]
             logging.debug(f'TimeActRespCode: {resp_code}')
@@ -473,7 +475,7 @@ class Talent(Message):
         self.forward()
 
     def parse_msg_header(self):
-        result = struct.unpack_from('!lB', self.ifc.read.peek(),
+        result = struct.unpack_from('!lB', self.ifc.rx_peek(),
                                     self.header_len)
 
         data_id = result[0]    # len of complete message
@@ -482,7 +484,7 @@ class Talent(Message):
 
         msg_hdr_len = 5+id_len+9
 
-        result = struct.unpack_from(f'!{id_len+1}pBq', self.ifc.read.peek(),
+        result = struct.unpack_from(f'!{id_len+1}pBq', self.ifc.rx_peek(),
                                     self.header_len + 4)
 
         timestamp = result[2]
@@ -495,7 +497,7 @@ class Talent(Message):
     def msg_collector_data(self):
         if self.ctrl.is_ind():
             self.__build_header(0x99)
-            self.ifc.write += b'\x01'
+            self.ifc.tx_add(b'\x01')
             self.__finish_send_msg()
             self.__process_data()
 
@@ -510,7 +512,7 @@ class Talent(Message):
     def msg_inverter_data(self):
         if self.ctrl.is_ind():
             self.__build_header(0x99)
-            self.ifc.write += b'\x01'
+            self.ifc.tx_add(b'\x01')
             self.__finish_send_msg()
             self.__process_data()
             self.state = State.up  # allow MODBUS cmds
@@ -530,7 +532,7 @@ class Talent(Message):
     def __process_data(self):
         msg_hdr_len, ts = self.parse_msg_header()
 
-        for key, update in self.db.parse(self.ifc.read.peek(), self.header_len
+        for key, update in self.db.parse(self.ifc.rx_peek(), self.header_len
                                          + msg_hdr_len, self.node_id):
             if update:
                 self._set_mqtt_timestamp(key, self._utcfromts(ts))
@@ -550,7 +552,7 @@ class Talent(Message):
 
         msg_hdr_len = 5
 
-        result = struct.unpack_from('!lBB', self.ifc.read.peek(),
+        result = struct.unpack_from('!lBB', self.ifc.rx_peek(),
                                     self.header_len)
         modbus_len = result[1]
         return msg_hdr_len, modbus_len
@@ -559,7 +561,7 @@ class Talent(Message):
 
         msg_hdr_len = 6
 
-        result = struct.unpack_from('!lBBB', self.ifc.read.peek(),
+        result = struct.unpack_from('!lBBB', self.ifc.rx_peek(),
                                     self.header_len)
         modbus_len = result[2]
         return msg_hdr_len, modbus_len
@@ -580,12 +582,12 @@ class Talent(Message):
         self.__msg_modbus(hdr_len)
 
     def __msg_modbus(self, hdr_len):
-        data = self.ifc.read.peek()[self.header_len:
-                                    self.header_len+self.data_len]
+        data = self.ifc.rx_peek()[self.header_len:
+                                  self.header_len+self.data_len]
 
         if self.ctrl.is_req():
-            if self.remote_stream.mb.recv_req(data[hdr_len:],
-                                              self.remote_stream.
+            if self.remote.stream.mb.recv_req(data[hdr_len:],
+                                              self.remote.stream.
                                               msg_forward):
                 self.inc_counter('Modbus_Command')
             else:

app/src/gen3plus/connection_g3p.py

@@ -2,34 +2,37 @@ import logging
 from asyncio import StreamReader, StreamWriter
 
 if __name__ == "app.src.gen3plus.connection_g3p":
-    from app.src.async_ifc import AsyncIfc
-    from app.src.async_stream import AsyncStream
+    from app.src.async_stream import AsyncStream, StreamPtr
     from app.src.gen3plus.solarman_v5 import SolarmanV5
 else:  # pragma: no cover
-    from async_ifc import AsyncIfc
-    from async_stream import AsyncStream
+    from async_stream import AsyncStream, StreamPtr
     from gen3plus.solarman_v5 import SolarmanV5
 
 logger = logging.getLogger('conn')
 
 
-class ConnectionG3P(AsyncStream, SolarmanV5):
+class ConnectionG3P(SolarmanV5):
 
     def __init__(self, reader: StreamReader, writer: StreamWriter,
-                 addr, remote_stream: 'ConnectionG3P',
+                 addr, rstream: 'ConnectionG3P',
                  server_side: bool,
                  client_mode: bool) -> None:
-        self._ifc = AsyncIfc()
-        AsyncStream.__init__(self, reader, writer, addr, self._ifc)
+
+        self.remote = StreamPtr(rstream)
+        self._ifc = AsyncStream(reader, writer, addr,
+                                self.async_publ_mqtt,
+                                self.async_create_remote,
+                                self.remote)
         SolarmanV5.__init__(self, server_side, client_mode, self._ifc)
 
-        self.remote_stream: 'ConnectionG3P' = remote_stream
+        self.conn_no = self._ifc.get_conn_no()
+        self.addr = addr
 
     '''
     Our puplic methods
     '''
     def close(self):
-        AsyncStream.close(self)
+        self._ifc.close()
         SolarmanV5.close(self)
         #  logger.info(f'AsyncStream refs: {gc.get_referrers(self)}')
 
@@ -41,10 +44,4 @@ class ConnectionG3P(AsyncStream, SolarmanV5):
 
     def healthy(self) -> bool:
         logger.debug('ConnectionG3P healthy()')
-        return AsyncStream.healthy(self)
-
-    '''
-    Our private methods
-    '''
-    def __del__(self):
-        super().__del__()
+        return self._ifc.healthy()

app/src/gen3plus/inverter_g3p.py

@@ -55,6 +55,7 @@ class InverterG3P(Inverter, ConnectionG3P):
         super().__init__(reader, writer, addr, None,
                          server_side=True, client_mode=client_mode)
         self.__ha_restarts = -1
+        self.addr = addr
 
     async def async_create_remote(self) -> None:
         '''Establish a client connection to the TSUN cloud'''
@@ -67,13 +68,13 @@ class InverterG3P(Inverter, ConnectionG3P):
             logging.info(f'[{self.node_id}] Connect to {addr}')
             connect = asyncio.open_connection(host, port)
             reader, writer = await connect
-            self.remote_stream = ConnectionG3P(reader, writer, addr, self,
+            self.remote.stream = ConnectionG3P(reader, writer, addr, self,
                                                server_side=False,
                                                client_mode=False)
-            logging.info(f'[{self.remote_stream.node_id}:'
-                         f'{self.remote_stream.conn_no}] '
+            logging.info(f'[{self.remote.stream.node_id}:'
+                         f'{self.remote.stream.conn_no}] '
                          f'Connected to {addr}')
-            asyncio.create_task(self.client_loop(addr))
+            asyncio.create_task(self._ifc.client_loop(addr))
 
         except (ConnectionRefusedError, TimeoutError) as error:
             logging.info(f'{error}')
@@ -131,10 +132,6 @@ class InverterG3P(Inverter, ConnectionG3P):
         self.db.reg_clr_at_midnight(f'{self.entity_prfx}{self.node_id}')
 
     def close(self) -> None:
-        logging.debug(f'InverterG3P.close() l{self.l_addr} | r{self.r_addr}')
+        logging.debug(f'InverterG3P.close() {self.addr}')
         super().close()         # call close handler in the parent class
 #        logger.debug (f'Inverter refs: {gc.get_referrers(self)}')
-
-    def __del__(self):
-        logging.debug("InverterG3P.__del__")
-        super().__del__()

app/src/gen3plus/solarman_v5.py

@@ -64,7 +64,8 @@ class SolarmanV5(Message):
 
     def __init__(self, server_side: bool, client_mode: bool, ifc: "AsyncIfc"):
         super().__init__(server_side, self.send_modbus_cb, mb_timeout=8)
-        ifc.read.reg_trigger(self.read)
+        ifc.rx_set_cb(self.read)
+        ifc.prot_set_timeout_cb(self._timeout)
         self.ifc = ifc
         self.header_len = 11  # overwrite construcor in class Message
         self.control = 0
@@ -163,7 +164,8 @@ class SolarmanV5(Message):
         self.log_lvl.clear()
         self.state = State.closed
         self.mb_timer.close()
-        self.ifc.read.reg_trigger(None)
+        self.ifc.rx_set_cb(None)
+        self.ifc.prot_set_timeout_cb(None)
         super().close()
 
     async def send_start_cmd(self, snr: int, host: str,
@@ -234,10 +236,10 @@ class SolarmanV5(Message):
         self._read()
         while True:
             if not self.header_valid:
-                self.__parse_header(self.ifc.read.peek(),
-                                    len(self.ifc.read))
+                self.__parse_header(self.ifc.rx_peek(),
+                                    self.ifc.rx_len())
 
-            if self.header_valid and len(self.ifc.read) >= \
+            if self.header_valid and self.ifc.rx_len() >= \
                (self.header_len + self.data_len+2):
                 self.__process_complete_received_msg()
                 self.__flush_recv_msg()
@@ -248,10 +250,10 @@ class SolarmanV5(Message):
         log_lvl = self.log_lvl.get(self.control, logging.WARNING)
         if callable(log_lvl):
             log_lvl = log_lvl()
-        self.ifc.read.logging(log_lvl, f'Received from {self.addr}:')
+        self.ifc.rx_log(log_lvl, f'Received from {self.addr}:')
         # self._recv_buffer, self.header_len +
         # self.data_len+2)
-        if self.__trailer_is_ok(self.ifc.read.peek(), self.header_len
+        if self.__trailer_is_ok(self.ifc.rx_peek(), self.header_len
                                 + self.data_len + 2):
             if self.state == State.init:
                 self.state = State.received
@@ -264,8 +266,8 @@ class SolarmanV5(Message):
             return
         tsun = Config.get('solarman')
         if tsun['enabled']:
-            self.ifc.forward += buffer[:buflen]
-            self.ifc.forward.logging(logging.DEBUG, 'Store for forwarding:')
+            self.ifc.fwd_add(buffer[:buflen])
+            self.ifc.fwd_log(logging.DEBUG, 'Store for forwarding:')
 
             fnc = self.switch.get(self.control, self.msg_unknown)
             logger.info(self.__flow_str(self.server_side, 'forwrd') +
@@ -321,7 +323,7 @@ class SolarmanV5(Message):
 
             self.inc_counter('Invalid_Msg_Format')
             # erase broken recv buffer
-            self.ifc.read.clear()
+            self.ifc.rx_clear()
             return
         self.header_valid = True
 
@@ -333,11 +335,11 @@ class SolarmanV5(Message):
                             'Drop packet w invalid stop byte from '
                             f'{self.addr}:', buf, buf_len)
             self.inc_counter('Invalid_Msg_Format')
-            if len(self.ifc.read) > (self.data_len+13):
+            if self.ifc.rx_len() > (self.data_len+13):
                 next_start = buf[self.data_len+13]
                 if next_start != 0xa5:
                     # erase broken recv buffer
-                    self.ifc.read.clear()
+                    self.ifc.rx_clear()
 
             return False
 
@@ -353,22 +355,22 @@ class SolarmanV5(Message):
 
     def __build_header(self, ctrl) -> None:
         '''build header for new transmit message'''
-        self.send_msg_ofs = len(self.ifc.write)
+        self.send_msg_ofs = self.ifc.tx_len()
 
-        self.ifc.write += struct.pack(
-            '<BHHHL', 0xA5, 0, ctrl, self.seq.get_send(), self.snr)
+        self.ifc.tx_add(struct.pack(
+            '<BHHHL', 0xA5, 0, ctrl, self.seq.get_send(), self.snr))
         fnc = self.switch.get(ctrl, self.msg_unknown)
         logger.info(self.__flow_str(self.server_side, 'tx') +
                     f' Ctl: {int(ctrl):#04x} Msg: {fnc.__name__!r}')
 
     def __finish_send_msg(self) -> None:
         '''finish the transmit message, set lenght and checksum'''
-        _len = len(self.ifc.write) - self.send_msg_ofs
-        struct.pack_into('<H', self.ifc.write.peek(), self.send_msg_ofs+1,
+        _len = self.ifc.tx_len() - self.send_msg_ofs
+        struct.pack_into('<H', self.ifc.tx_peek(), self.send_msg_ofs+1,
                          _len-11)
-        check = sum(self.ifc.write.peek()[
+        check = sum(self.ifc.tx_peek()[
             self.send_msg_ofs+1:self.send_msg_ofs + _len]) & 0xff
-        self.ifc.write += struct.pack('<BB', check, 0x15)    # crc & stop
+        self.ifc.tx_add(struct.pack('<BB', check, 0x15))    # crc & stop
 
     def _update_header(self, _forward_buffer):
         '''update header for message before forwarding,
@@ -399,14 +401,14 @@ class SolarmanV5(Message):
                         f' Msg: {fnc.__name__!r}')
 
     def __flush_recv_msg(self) -> None:
-        self.ifc.read.get(self.header_len + self.data_len+2)
+        self.ifc.rx_get(self.header_len + self.data_len+2)
         self.header_valid = False
 
     def __send_ack_rsp(self, msgtype, ftype, ack=1):
         self.__build_header(msgtype)
-        self.ifc.write += struct.pack('<BBLL', ftype, ack,
-                                      self._timestamp(),
-                                      self._heartbeat())
+        self.ifc.tx_add(struct.pack('<BBLL', ftype, ack,
+                                    self._timestamp(),
+                                    self._heartbeat()))
         self.__finish_send_msg()
 
     def send_modbus_cb(self, pdu: bytearray, log_lvl: int, state: str):
@@ -415,12 +417,12 @@ class SolarmanV5(Message):
                            ' cause the state is not UP anymore')
             return
         self.__build_header(0x4510)
-        self.ifc.write += struct.pack('<BHLLL', self.MB_RTU_CMD,
-                                      self.sensor_list, 0, 0, 0)
-        self.ifc.write += pdu
+        self.ifc.tx_add(struct.pack('<BHLLL', self.MB_RTU_CMD,
+                                    self.sensor_list, 0, 0, 0))
+        self.ifc.tx_add(pdu)
         self.__finish_send_msg()
-        self.ifc.write.logging(log_lvl, f'Send Modbus {state}:{self.addr}:')
-        self.ifc.write()
+        self.ifc.tx_log(log_lvl, f'Send Modbus {state}:{self.addr}:')
+        self.ifc.tx_flush()
 
     def _send_modbus_cmd(self, func, addr, val, log_lvl) -> None:
         if self.state != State.up:
@@ -462,18 +464,18 @@ class SolarmanV5(Message):
 
         self.forward_at_cmd_resp = False
         self.__build_header(0x4510)
-        self.ifc.write += struct.pack(f'<BHLLL{len(at_cmd)}sc', self.AT_CMD,
-                                      0x0002, 0, 0, 0,
-                                      at_cmd.encode('utf-8'), b'\r')
+        self.ifc.tx_add(struct.pack(f'<BHLLL{len(at_cmd)}sc', self.AT_CMD,
+                                    0x0002, 0, 0, 0,
+                                    at_cmd.encode('utf-8'), b'\r'))
         self.__finish_send_msg()
-        self.ifc.write.logging(logging.INFO, 'Send AT Command:')
+        self.ifc.tx_log(logging.INFO, 'Send AT Command:')
         try:
-            self.ifc.write()
+            self.ifc.tx_flush()
         except Exception:
-            self.ifc.write.clear()
+            self.ifc.tx_clear()
 
     def __forward_msg(self):
-        self.forward(self.ifc.read.peek(), self.header_len+self.data_len+2)
+        self.forward(self.ifc.rx_peek(), self.header_len+self.data_len+2)
 
     def __build_model_name(self):
         db = self.db
@@ -494,7 +496,7 @@ class SolarmanV5(Message):
     def __process_data(self, ftype, ts):
         inv_update = False
         msg_type = self.control >> 8
-        for key, update in self.db.parse(self.ifc.read.peek(), msg_type, ftype,
+        for key, update in self.db.parse(self.ifc.rx_peek(), msg_type, ftype,
                                          self.node_id):
             if update:
                 if key == 'inverter':
@@ -513,7 +515,7 @@ class SolarmanV5(Message):
         self.__forward_msg()
 
     def msg_dev_ind(self):
-        data = self.ifc.read.peek()[self.header_len:]
+        data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from(self.HDR_FMT, data, 0)
         ftype = result[0]  # always 2
         total = result[1]
@@ -534,7 +536,7 @@ class SolarmanV5(Message):
         self.__send_ack_rsp(0x1110, ftype)
 
     def msg_data_ind(self):
-        data = self.ifc.read.peek()
+        data = self.ifc.rx_peek()
         result = struct.unpack_from('<BHLLLHL', data, self.header_len)
         ftype = result[0]  # 1 or 0x81
         sensor = result[1]
@@ -562,7 +564,7 @@ class SolarmanV5(Message):
         self.new_state_up()
 
     def msg_sync_start(self):
-        data = self.ifc.read.peek()[self.header_len:]
+        data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from(self.HDR_FMT, data, 0)
         ftype = result[0]
         total = result[1]
@@ -575,8 +577,8 @@ class SolarmanV5(Message):
         self.__send_ack_rsp(0x1310, ftype)
 
     def msg_command_req(self):
-        data = self.ifc.read.peek()[self.header_len:
-                                    self.header_len+self.data_len]
+        data = self.ifc.rx_peek()[self.header_len:
+                                  self.header_len+self.data_len]
         result = struct.unpack_from('<B', data, 0)
         ftype = result[0]
         if ftype == self.AT_CMD:
@@ -588,8 +590,8 @@ class SolarmanV5(Message):
             self.forward_at_cmd_resp = True
 
         elif ftype == self.MB_RTU_CMD:
-            if self.remote_stream.mb.recv_req(data[15:],
-                                              self.remote_stream.
+            if self.remote.stream.mb.recv_req(data[15:],
+                                              self.remote.stream.
                                               __forward_msg):
                 self.inc_counter('Modbus_Command')
             else:
@@ -604,7 +606,7 @@ class SolarmanV5(Message):
             self.mqtt.publish(key, data))
 
     def get_cmd_rsp_log_lvl(self) -> int:
-        ftype = self.ifc.read.peek()[self.header_len]
+        ftype = self.ifc.rx_peek()[self.header_len]
         if ftype == self.AT_CMD:
             if self.forward_at_cmd_resp:
                 return logging.INFO
@@ -616,8 +618,8 @@ class SolarmanV5(Message):
         return logging.WARNING
 
     def msg_command_rsp(self):
-        data = self.ifc.read.peek()[self.header_len:
-                                    self.header_len+self.data_len]
+        data = self.ifc.rx_peek()[self.header_len:
+                                  self.header_len+self.data_len]
         ftype = data[0]
         if ftype == self.AT_CMD:
             if not self.forward_at_cmd_resp:
@@ -653,7 +655,7 @@ class SolarmanV5(Message):
                 self.__build_model_name()
 
     def msg_hbeat_ind(self):
-        data = self.ifc.read.peek()[self.header_len:]
+        data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from('<B', data, 0)
         ftype = result[0]
 
@@ -662,7 +664,7 @@ class SolarmanV5(Message):
         self.new_state_up()
 
     def msg_sync_end(self):
-        data = self.ifc.read.peek()[self.header_len:]
+        data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from(self.HDR_FMT, data, 0)
         ftype = result[0]
         total = result[1]
@@ -675,7 +677,7 @@ class SolarmanV5(Message):
         self.__send_ack_rsp(0x1810, ftype)
 
     def msg_response(self):
-        data = self.ifc.read.peek()[self.header_len:]
+        data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from('<BBLL', data, 0)
         ftype = result[0]  # always 2
         valid = result[1] == 1  # status

app/src/infos.py

@@ -383,12 +383,14 @@ class Infos:
         '''inc proxy statistic counter'''
         db_dict = cls.stat['proxy']
         db_dict[counter] += 1
+        cls.new_stat_data['proxy'] = True
 
     @classmethod
     def dec_counter(cls, counter: str) -> None:
         '''dec proxy statistic counter'''
         db_dict = cls.stat['proxy']
         db_dict[counter] -= 1
+        cls.new_stat_data['proxy'] = True
 
     def ha_proxy_confs(self, ha_prfx: str, node_id: str, snr: str) \
             -> Generator[tuple[str, str, str, str], None, None]:

app/src/messages.py

@@ -90,6 +90,12 @@ class State(Enum):
 
 class Message(metaclass=IterRegistry):
     _registry = []
+    MAX_START_TIME = 400
+    '''maximum time without a received msg in sec'''
+    MAX_INV_IDLE_TIME = 120
+    '''maximum time without a received msg from the inverter in sec'''
+    MAX_DEF_IDLE_TIME = 360
+    '''maximum default time without a received msg in sec'''
 
     def __init__(self, server_side: bool, send_modbus_cb:
                  Callable[[bytes, int, str], None], mb_timeout: int):
@@ -105,12 +111,21 @@ class Message(metaclass=IterRegistry):
         self.header_len = 0
         self.data_len = 0
         self.unique_id = 0
-        self.node_id = ''  # will be overwritten in the child class's __init__
+        self._node_id = ''
         self.sug_area = ''
         self.new_data = {}
         self.state = State.init
         self.shutdown_started = False
 
+    @property
+    def node_id(self):
+        return self._node_id
+
+    @node_id.setter
+    def node_id(self, value):
+        self._node_id = value
+        self.ifc.set_node_id(value)
+
     '''
     Empty methods, that have to be implemented in any child class which
     don't use asyncio
@@ -138,6 +153,16 @@ class Message(metaclass=IterRegistry):
             # logger.info(f'update: key: {key} ts:{tstr}'
             self.db.set_db_def_value(info_id, round(ts))
 
+    def _timeout(self) -> int:
+        if self.state == State.init or self.state == State.received:
+            to = self.MAX_START_TIME
+        elif self.state == State.up and \
+                self.server_side and self.modbus_polling:
+            to = self.MAX_INV_IDLE_TIME
+        else:
+            to = self.MAX_DEF_IDLE_TIME
+        return to
+
     '''
     Our puplic methods
     '''

app/src/modbus_tcp.py

@@ -5,9 +5,11 @@ import asyncio
 if __name__ == "app.src.modbus_tcp":
     from app.src.config import Config
     from app.src.gen3plus.inverter_g3p import InverterG3P
+    from app.src.infos import Infos
 else:  # pragma: no cover
     from config import Config
     from gen3plus.inverter_g3p import InverterG3P
+    from infos import Infos
 
 logger = logging.getLogger('conn')
 
@@ -27,13 +29,14 @@ class ModbusConn():
                                   client_mode=True)
         logging.info(f'[{self.stream.node_id}:{self.stream.conn_no}] '
                      f'Connected to {self.addr}')
-        self.stream.inc_counter('Inverter_Cnt')
-        await self.stream.publish_outstanding_mqtt()
+        Infos.inc_counter('Inverter_Cnt')
+        await self.stream._ifc.publish_outstanding_mqtt()
         return self.stream
 
     async def __aexit__(self, exc_type, exc, tb):
-        self.stream.dec_counter('Inverter_Cnt')
-        await self.stream.publish_outstanding_mqtt()
+        Infos.dec_counter('Inverter_Cnt')
+        await self.stream._ifc.publish_outstanding_mqtt()
+        self.stream.close()
 
 
 class ModbusTcp():
@@ -60,7 +63,7 @@ class ModbusTcp():
             try:
                 async with ModbusConn(host, port) as stream:
                     await stream.send_start_cmd(snr, host)
-                    await stream.loop()
+                    await stream._ifc.loop()
                     logger.info(f'[{stream.node_id}:{stream.conn_no}] '
                                 f'Connection closed - Shutdown: '
                                 f'{stream.shutdown_started}')

app/src/server.py

@@ -74,14 +74,14 @@ async def handle_client(reader: StreamReader, writer: StreamWriter):
     '''Handles a new incoming connection and starts an async loop'''
 
     addr = writer.get_extra_info('peername')
-    await InverterG3(reader, writer, addr).server_loop(addr)
+    await InverterG3(reader, writer, addr)._ifc.server_loop(addr)
 
 
 async def handle_client_v2(reader: StreamReader, writer: StreamWriter):
     '''Handles a new incoming connection and starts an async loop'''
 
     addr = writer.get_extra_info('peername')
-    await InverterG3P(reader, writer, addr).server_loop(addr)
+    await InverterG3P(reader, writer, addr)._ifc.server_loop(addr)
 
 
 async def handle_shutdown(web_task):