S allius/issue205 (#207)

* Add SolarmanEmu class

* Forward a device ind to establish the EMU connection

* Move SolarmanEmu class into a dedicated file

* Add cloud connection counter

* Send inverter data in emulator mode

* Improve emulator mode

- parse more values from MQTT register
- differ between inverter and logger serial no

* Add some unit tests for SolarmanEmu class

* Send seconds since last sync in data packets

* Increase test coverage

app/src/async_stream.py

@@ -305,6 +305,14 @@ class AsyncStream(AsyncIfcImpl):
                 f"Fwd Exception for {self.r_addr}:\n"
                 f"{traceback.format_exc()}")
 
+    async def publish_outstanding_mqtt(self):
+        '''Publish all outstanding MQTT topics'''
+        try:
+            await self.async_publ_mqtt()
+            await Proxy._async_publ_mqtt_proxy_stat('proxy')
+        except Exception:
+            pass
+
 
 class AsyncStreamServer(AsyncStream):
     def __init__(self, reader: StreamReader, writer: StreamWriter,
@@ -354,14 +362,6 @@ class AsyncStreamServer(AsyncStream):
             self.remote.ifc._writer.write(self.fwd_fifo.get())
             await self.remote.ifc._writer.drain()
 
-    async def publish_outstanding_mqtt(self):
-        '''Publish all outstanding MQTT topics'''
-        try:
-            await self.async_publ_mqtt()
-            await Proxy._async_publ_mqtt_proxy_stat('proxy')
-        except Exception:
-            pass
-
 
 class AsyncStreamClient(AsyncStream):
     def __init__(self, reader: StreamReader, writer: StreamWriter,
@@ -381,7 +381,11 @@ class AsyncStreamClient(AsyncStream):
 
     async def client_loop(self, _: str) -> None:
         '''Loop for receiving messages from the TSUN cloud (client-side)'''
+        Infos.inc_counter('Cloud_Conn_Cnt')
+        await self.publish_outstanding_mqtt()
         await self.loop()
+        Infos.dec_counter('Cloud_Conn_Cnt')
+        await self.publish_outstanding_mqtt()
         logger.info(f'[{self.node_id}:{self.conn_no}] '
                     'Client loop stopped for'
                     f' l{self.l_addr}')

app/src/config.py

@@ -57,7 +57,8 @@ class Config():
                 Optional('client_mode'): {
                     'host': Use(str),
                     Optional('port', default=8899):
-                        And(Use(int), lambda n: 1024 <= n <= 65535)
+                        And(Use(int), lambda n: 1024 <= n <= 65535),
+                    Optional('forward', default=False): Use(bool),
                 },
                 Optional('modbus_polling', default=True): Use(bool),
                 Optional('suggested_area', default=""): Use(str),

app/src/gen3/infos_g3.py

@@ -75,7 +75,15 @@ class RegisterMap:
         0x00000258: {'reg': Register.EVENT_BF1},
         0x000002bc: {'reg': Register.EVENT_BF2},
         0x00000064: {'reg': Register.INVERTER_STATUS},
+
+        0x00000fa0: {'reg': Register.BOOT_STATUS},
+        0x00001004: {'reg': Register.DSP_STATUS},
+        0x000010cc: {'reg': Register.WORK_MODE},
+        0x000011f8: {'reg': Register.OUTPUT_SHUTDOWN},
         0x0000125c: {'reg': Register.MAX_DESIGNED_POWER},
+        0x000012c0: {'reg': Register.RATED_LEVEL},
+        0x00001324: {'reg': Register.INPUT_COEFFICIENT, 'ratio':  100/1024},
+        0x00001388: {'reg': Register.GRID_VOLT_CAL_COEF},
         0x00003200: {'reg': Register.OUTPUT_COEFFICIENT, 'ratio':  100/1024},
     }
 

app/src/gen3plus/infos_g3p.py

@@ -17,17 +17,17 @@ class RegisterMap:
 
     map = {
         # 0x41020007: {'reg': Register.DEVICE_SNR,           'fmt': '<L'},                 # noqa: E501
-        0x41020018: {'reg': Register.DATA_UP_INTERVAL,     'fmt': '<B', 'ratio':   60, 'dep': ProxyMode.SERVER},  # noqa: E501
+        0x41020018: {'reg': Register.DATA_UP_INTERVAL,     'fmt': '<B', 'ratio':    60, 'dep': ProxyMode.SERVER},  # noqa: E501
         0x41020019: {'reg': Register.COLLECT_INTERVAL,     'fmt': '<B', 'quotient': 60, 'dep': ProxyMode.SERVER},  # noqa: E501
         0x4102001a: {'reg': Register.HEARTBEAT_INTERVAL,   'fmt': '<B', 'ratio':    1},  # noqa: E501
-        0x4102001b: {'reg': None,                          'fmt': '<B', 'const':    1},  # noqa: E501
+        0x4102001b: {'reg': None,                          'fmt': '<B', 'const':    1},  # noqa: E501 Max No Of Connected Devices
         0x4102001c: {'reg': Register.SIGNAL_STRENGTH,      'fmt': '<B', 'ratio':    1, 'dep': ProxyMode.SERVER},  # noqa: E501
         0x4102001d: {'reg': None,                          'fmt': '<B', 'const':    1},  # noqa: E501
         0x4102001e: {'reg': Register.CHIP_MODEL,           'fmt': '!40s'},               # noqa: E501
         0x41020046: {'reg': Register.MAC_ADDR,             'fmt': '!6B', 'func': Fmt.mac},  # noqa: E501
         0x4102004c: {'reg': Register.IP_ADDRESS,           'fmt': '!16s'},               # noqa: E501
         0x4102005c: {'reg': None,                          'fmt': '<B', 'const':   15},  # noqa: E501
-        0x4102005e: {'reg': None,                          'fmt': '<B', 'const':    1},  # noqa: E501
+        0x4102005e: {'reg': None,                          'fmt': '<B', 'const':    1},  # noqa: E501 No Of Sensors (ListLen)
         0x4102005f: {'reg': Register.SENSOR_LIST,          'fmt': '<H', 'func': Fmt.hex4},   # noqa: E501
         0x41020061: {'reg': None,                          'fmt': '<BBB', 'const':  (15, 0, 255)},  # noqa: E501
         0x41020064: {'reg': Register.COLLECTOR_FW_VERSION, 'fmt': '!40s'},               # noqa: E501
@@ -80,13 +80,16 @@ class RegisterMap:
         0x42010116: {'reg': Register.INV_UNKNOWN_1,        'fmt': '!H'},                 # noqa: E501
 
         # Start MODBUS Block: 0x2000 (R/W Config Paramaneters)
-        0x42010118: {'reg': Register.BOOT_STATUS,          'fmt': '!H'},                 # noqa: E501
+        0x42010118: {'reg': Register.BOOT_STATUS,          'fmt': '!H'},
-        0x4201011a: {'reg': Register.DSP_STATUS,           'fmt': '!H'},                 # noqa: E501
+        0x4201011a: {'reg': Register.DSP_STATUS,           'fmt': '!H'},
-        0x4201011c: {'reg': None,                          'fmt': FMT_2_16BIT_VAL, 'const':   (1, 0)},  # noqa: E501
+        0x4201011c: {'reg': None,                          'fmt': '!H', 'const':    1},  # noqa: E501
-        0x42010124: {'reg': None,                          'fmt': '!H', 'const':    0xffff},  # noqa: E501
+        0x4201011e: {'reg': Register.WORK_MODE,            'fmt': '!H'},
-        0x42010126: {'reg': Register.MAX_DESIGNED_POWER,   'fmt': '!H', 'ratio':    1},  # noqa: E501
+        0x42010124: {'reg': Register.OUTPUT_SHUTDOWN,      'fmt': '!H'},
-        0x42010128: {'reg': None,                          'fmt': '!H', 'const':    3},  # noqa: E501
+        0x42010126: {'reg': Register.MAX_DESIGNED_POWER,   'fmt': '!H'},
-        0x4201012a: {'reg': None,                          'fmt': FMT_3_16BIT_VAL, 'const':  (1024, 1024, 1024)},  # noqa: E501
+        0x42010128: {'reg': Register.RATED_LEVEL,          'fmt': '!H'},
+        0x4201012a: {'reg': Register.INPUT_COEFFICIENT,    'fmt': '!H', 'ratio':  100/1024},  # noqa: E501
+        0x4201012c: {'reg': Register.GRID_VOLT_CAL_COEF,   'fmt': '!H'},
+        0x4201012e: {'reg': None,                          'fmt': '!H', 'const':   1024},  # noqa: E501
         0x42010130: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (1024, 1, 0xffff, 1)},  # noqa: E501
         0x42010138: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (6, 0x68, 0x68, 0x500)},  # noqa: E501
         0x42010140: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (0x9cd, 0x7b6, 0x139c, 0x1324)},  # noqa: E501
@@ -101,7 +104,8 @@ class RegisterMap:
         0x42010180: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (0x9ce, 0x7a8, 0x139c, 0x1326)},  # noqa: E501
         0x42010188: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (0x0, 0x0, 0x0, 0)},  # noqa: E501
         0x42010190: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (0x0, 0x0, 1024, 1024)},  # noqa: E501
-        0x4201019a: {'reg': None,                          'fmt': FMT_2_16BIT_VAL, 'const':   (0, 0xffff)},  # noqa: E501
+        0x42010198: {'reg': None,                          'fmt': FMT_4_16BIT_VAL, 'const': (0, 0, 0xffff, 0)},  # noqa: E501
+        0x420101a0: {'reg': None,                          'fmt': FMT_2_16BIT_VAL, 'const': (0x0, 0x0)},  # noqa: E501
 
         0xffffff02: {'reg': Register.POLLING_INTERVAL},
         # 0x4281001c: {'reg': Register.POWER_ON_TIME,        'fmt': '<H', 'ratio':    1},  # noqa: E501

app/src/gen3plus/inverter_g3p.py

@@ -3,13 +3,18 @@ from asyncio import StreamReader, StreamWriter
 if __name__ == "app.src.gen3plus.inverter_g3p":
     from app.src.inverter_base import InverterBase
     from app.src.gen3plus.solarman_v5 import SolarmanV5
+    from app.src.gen3plus.solarman_emu import SolarmanEmu
 else:  # pragma: no cover
     from inverter_base import InverterBase
     from gen3plus.solarman_v5 import SolarmanV5
+    from gen3plus.solarman_emu import SolarmanEmu
 
 
 class InverterG3P(InverterBase):
     def __init__(self, reader: StreamReader, writer: StreamWriter,
                  client_mode: bool = False):
+        remote_prot = None
+        if client_mode:
+            remote_prot = SolarmanEmu
         super().__init__(reader, writer, 'solarman',
-                         SolarmanV5, client_mode)
+                         SolarmanV5, client_mode, remote_prot)

app/src/gen3plus/solarman_emu.py

@@ -0,0 +1,144 @@
+import logging
+import struct
+
+if __name__ == "app.src.gen3plus.solarman_emu":
+    from app.src.async_ifc import AsyncIfc
+    from app.src.gen3plus.solarman_v5 import SolarmanBase
+    from app.src.my_timer import Timer
+    from app.src.infos import Register
+else:  # pragma: no cover
+    from async_ifc import AsyncIfc
+    from gen3plus.solarman_v5 import SolarmanBase
+    from my_timer import Timer
+    from infos import Register
+
+logger = logging.getLogger('msg')
+
+
+class SolarmanEmu(SolarmanBase):
+    def __init__(self, addr, ifc: "AsyncIfc",
+                 server_side: bool, client_mode: bool):
+        super().__init__(addr, ifc, server_side=False,
+                         _send_modbus_cb=None,
+                         mb_timeout=8)
+        logging.debug('SolarmanEmu.init()')
+        self.db = ifc.remote.stream.db
+        self.snr = ifc.remote.stream.snr
+        self.hb_timeout = 60
+        '''actual heatbeat timeout from the last response message'''
+        self.data_up_inv = self.db.get_db_value(Register.DATA_UP_INTERVAL)
+        '''time interval for getting new MQTT data messages'''
+        self.hb_timer = Timer(self.send_heartbeat_cb, self.node_id)
+        self.data_timer = Timer(self.send_data_cb, self.node_id)
+        self.last_sync = self._emu_timestamp()
+        '''timestamp when we send the last sync message (4110)'''
+        self.pkt_cnt = 0
+        '''last sent packet number'''
+
+        self.switch = {
+
+            0x4210: 'msg_data_ind',      # real time data
+            0x1210: self.msg_response,   # at least every 5 minutes
+
+            0x4710: 'msg_hbeat_ind',     # heatbeat
+            0x1710: self.msg_response,   # every 2 minutes
+
+            0x4110: 'msg_dev_ind',        # device data, sync start
+            0x1110: self.msg_response,    # every 3 hours
+
+        }
+
+        self.log_lvl = {
+
+            0x4110: logging.INFO,   # device data, sync start
+            0x1110: logging.INFO,   # every 3 hours
+
+            0x4210: logging.INFO,   # real time data
+            0x1210: logging.INFO,   # at least every 5 minutes
+
+            0x4710: logging.DEBUG,  # heatbeat
+            0x1710: logging.DEBUG,  # every 2 minutes
+
+        }
+
+    '''
+    Our puplic methods
+    '''
+    def close(self) -> None:
+        logging.info('SolarmanEmu.close()')
+        # we have references to methods of this class in self.switch
+        # so we have to erase self.switch, otherwise this instance can't be
+        # deallocated by the garbage collector ==> we get a memory leak
+        self.switch.clear()
+        self.log_lvl.clear()
+        self.hb_timer.close()
+        self.data_timer.close()
+        self.db = None
+        super().close()
+
+    def _set_serial_no(self, snr: int):
+        logging.debug(f'SolarmanEmu._set_serial_no, snr: {snr}')
+        self.unique_id = str(snr)
+
+    def _init_new_client_conn(self) -> bool:
+        logging.debug('SolarmanEmu.init_new()')
+        self.data_timer.start(self.data_up_inv)
+        return False
+
+    def next_pkt_cnt(self):
+        '''get the next packet number'''
+        self.pkt_cnt = (self.pkt_cnt + 1) & 0xffffffff
+        return self.pkt_cnt
+
+    def seconds_since_last_sync(self):
+        '''get seconds since last 0x4110 message was sent'''
+        return self._emu_timestamp() - self.last_sync
+
+    def send_heartbeat_cb(self, exp_cnt):
+        '''send a heartbeat to the TSUN cloud'''
+        self._build_header(0x4710)
+        self.ifc.tx_add(struct.pack('<B', 0))
+        self._finish_send_msg()
+        log_lvl = self.log_lvl.get(0x4710, logging.WARNING)
+        self.ifc.tx_log(log_lvl, 'Send heartbeat:')
+        self.ifc.tx_flush()
+
+    def send_data_cb(self, exp_cnt):
+        '''send a inverter data message to the TSUN cloud'''
+        self.hb_timer.start(self.hb_timeout)
+        self.data_timer.start(self.data_up_inv)
+        _len = 420
+        ftype = 1
+        build_msg = self.db.build(_len, 0x42, ftype)
+
+        self._build_header(0x4210)
+        self.ifc.tx_add(
+            struct.pack(
+                '<BHLLLHL', ftype, 0x02b0,
+                self._emu_timestamp(),
+                self.seconds_since_last_sync(),
+                self.time_ofs,
+                1,  # offset 0x1a
+                self.next_pkt_cnt()))
+        self.ifc.tx_add(build_msg[0x20:])
+        self._finish_send_msg()
+        log_lvl = self.log_lvl.get(0x4210, logging.WARNING)
+        self.ifc.tx_log(log_lvl, 'Send inv-data:')
+        self.ifc.tx_flush()
+
+    '''
+    Message handler methods
+    '''
+    def msg_response(self):
+        '''handle a received response from the TSUN cloud'''
+        logger.debug("EMU received rsp:")
+        _, _, ts, hb = super().msg_response()
+        logger.debug(f"EMU ts:{ts} hb:{hb}")
+        self.hb_timeout = hb
+        self.time_ofs = ts - self._emu_timestamp()
+        self.hb_timer.start(self.hb_timeout)
+
+    def msg_unknown(self):
+        '''counts a unknown or unexpected message from the TSUN cloud'''
+        logger.warning(f"EMU Unknow Msg: ID:{int(self.control):#04x}")
+        self.inc_counter('Unknown_Msg')

app/src/gen3plus/solarman_v5.py

@@ -10,14 +10,14 @@ if __name__ == "app.src.gen3plus.solarman_v5":
     from app.src.modbus import Modbus
     from app.src.config import Config
     from app.src.gen3plus.infos_g3p import InfosG3P
-    from app.src.infos import Register
+    from app.src.infos import Register, Fmt
 else:  # pragma: no cover
     from async_ifc import AsyncIfc
     from messages import hex_dump_memory, Message, State
     from config import Config
     from modbus import Modbus
     from gen3plus.infos_g3p import InfosG3P
-    from infos import Register
+    from infos import Register, Fmt
 
 logger = logging.getLogger('msg')
 
@@ -48,7 +48,209 @@ class Sequence():
         return f'{self.rcv_idx:02x}:{self.snd_idx:02x}'
 
 
-class SolarmanV5(Message):
+class SolarmanBase(Message):
+    def __init__(self, addr, ifc: "AsyncIfc", server_side: bool,
+                 _send_modbus_cb, mb_timeout: int):
+        super().__init__('G3P', ifc, server_side, _send_modbus_cb,
+                         mb_timeout)
+        ifc.rx_set_cb(self.read)
+        ifc.prot_set_timeout_cb(self._timeout)
+        ifc.prot_set_init_new_client_conn_cb(self._init_new_client_conn)
+        ifc.prot_set_update_header_cb(self.__update_header)
+        self.addr = addr
+        self.conn_no = ifc.get_conn_no()
+        self.header_len = 11  # overwrite construcor in class Message
+        self.control = 0
+        self.seq = Sequence(server_side)
+        self.snr = 0
+        self.time_ofs = 0
+
+    def read(self) -> float:
+        '''process all received messages in the _recv_buffer'''
+        self._read()
+        while True:
+            if not self.header_valid:
+                self.__parse_header(self.ifc.rx_peek(),
+                                    self.ifc.rx_len())
+
+            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()
+            else:
+                return 0  # wait 0s before sending a response
+    '''
+    Our public methods
+    '''
+    def _flow_str(self, server_side: bool, type: str):  # noqa: F821
+        switch = {
+            'rx':      '  <',
+            'tx':      '  >',
+            'forwrd':  '<< ',
+            'drop':    ' xx',
+            'rxS':     '>  ',
+            'txS':     '<  ',
+            'forwrdS': ' >>',
+            'dropS':   'xx ',
+        }
+        if server_side:
+            type += 'S'
+        return switch.get(type, '???')
+
+    def get_fnc_handler(self, ctrl):
+        fnc = self.switch.get(ctrl, self.msg_unknown)
+        if callable(fnc):
+            return fnc, repr(fnc.__name__)
+        else:
+            return self.msg_unknown, repr(fnc)
+
+    def _build_header(self, ctrl) -> None:
+        '''build header for new transmit message'''
+        self.send_msg_ofs = self.ifc.tx_len()
+
+        self.ifc.tx_add(struct.pack(
+            '<BHHHL', 0xA5, 0, ctrl, self.seq.get_send(), self.snr))
+        _fnc, _str = self.get_fnc_handler(ctrl)
+        logger.info(self._flow_str(self.server_side, 'tx') +
+                    f' Ctl: {int(ctrl):#04x} Msg: {_str}')
+
+    def _finish_send_msg(self) -> None:
+        '''finish the transmit message, set lenght and checksum'''
+        _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.tx_peek()[
+            self.send_msg_ofs+1:self.send_msg_ofs + _len]) & 0xff
+        self.ifc.tx_add(struct.pack('<BB', check, 0x15))    # crc & stop
+
+    def _timestamp(self):
+        # utc as epoche
+        return int(time.time())    # pragma: no cover
+
+    def _emu_timestamp(self):
+        '''timestamp for an emulated inverter (realtime - 1 day)'''
+        one_day = 24*60*60
+        return self._timestamp()-one_day
+
+    '''
+    Our private methods
+    '''
+    def __update_header(self, _forward_buffer):
+        '''update header for message before forwarding,
+        set sequence and checksum'''
+        _len = len(_forward_buffer)
+        ofs = 0
+        while ofs < _len:
+            result = struct.unpack_from('<BH', _forward_buffer, ofs)
+            data_len = result[1]    # len of variable id string
+
+            struct.pack_into('<H', _forward_buffer, ofs+5,
+                             self.seq.get_send())
+
+            check = sum(_forward_buffer[ofs+1:ofs+data_len+11]) & 0xff
+            struct.pack_into('<B', _forward_buffer, ofs+data_len+11, check)
+            ofs += (13 + data_len)
+
+    def __process_complete_received_msg(self):
+        log_lvl = self.log_lvl.get(self.control, logging.WARNING)
+        if callable(log_lvl):
+            log_lvl = log_lvl()
+        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.rx_peek(), self.header_len
+                                + self.data_len + 2):
+            if self.state == State.init:
+                self.state = State.received
+            self._set_serial_no(self.snr)
+            self.__dispatch_msg()
+
+    def __parse_header(self, buf: bytes, buf_len: int) -> None:
+
+        if (buf_len < self.header_len):  # enough bytes for complete header?
+            return
+
+        result = struct.unpack_from('<BHHHL', buf, 0)
+
+        # store parsed header values in the class
+        start = result[0]            # start byte
+        self.data_len = result[1]    # len of variable id string
+        self.control = result[2]
+        self.seq.set_recv(result[3])
+        self.snr = result[4]
+
+        if start != 0xA5:
+            hex_dump_memory(logging.ERROR,
+                            'Drop packet w invalid start byte from'
+                            f' {self.addr}:', buf, buf_len)
+
+            self.inc_counter('Invalid_Msg_Format')
+            # erase broken recv buffer
+            self.ifc.rx_clear()
+            return
+        self.header_valid = True
+
+    def __trailer_is_ok(self, buf: bytes, buf_len: int) -> bool:
+        crc = buf[self.data_len+11]
+        stop = buf[self.data_len+12]
+        if stop != 0x15:
+            hex_dump_memory(logging.ERROR,
+                            'Drop packet w invalid stop byte from '
+                            f'{self.addr}:', buf, buf_len)
+            self.inc_counter('Invalid_Msg_Format')
+            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.rx_clear()
+
+            return False
+
+        check = sum(buf[1:buf_len-2]) & 0xff
+        if check != crc:
+            self.inc_counter('Invalid_Msg_Format')
+            logger.debug(f'CRC {int(crc):#02x} {int(check):#08x}'
+                         f' Stop:{int(stop):#02x}')
+            # start & stop byte are valid, discard only this message
+            return False
+
+        return True
+
+    def __flush_recv_msg(self) -> None:
+        self.ifc.rx_get(self.header_len + self.data_len+2)
+        self.header_valid = False
+
+    def __dispatch_msg(self) -> None:
+        _fnc, _str = self.get_fnc_handler(self.control)
+        if self.unique_id:
+            logger.info(self._flow_str(self.server_side, 'rx') +
+                        f' Ctl: {int(self.control):#04x}' +
+                        f' Msg: {_str}')
+            _fnc()
+        else:
+            logger.info(self._flow_str(self.server_side, 'drop') +
+                        f' Ctl: {int(self.control):#04x}' +
+                        f' Msg: {_str}')
+
+    '''
+    Message handler methods
+    '''
+    def msg_response(self):
+        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
+        ts = result[2]
+        set_hb = result[3]  # always 60 or 120
+        logger.debug(f'ftype:{ftype} accepted:{valid}'
+                     f' ts:{ts:08x}  nextHeartbeat: {set_hb}s')
+
+        dt = datetime.fromtimestamp(ts)
+        logger.debug(f'ts: {dt.strftime("%Y-%m-%d %H:%M:%S")}')
+        return ftype, valid, ts, set_hb
+
+
+class SolarmanV5(SolarmanBase):
     AT_CMD = 1
     MB_RTU_CMD = 2
     MB_CLIENT_DATA_UP = 30
@@ -58,24 +260,15 @@ class SolarmanV5(Message):
 
     def __init__(self, addr, ifc: "AsyncIfc",
                  server_side: bool, client_mode: bool):
-        super().__init__('G3P', ifc, server_side, self.send_modbus_cb,
+        super().__init__(addr, ifc, server_side, self.send_modbus_cb,
                          mb_timeout=8)
-        ifc.rx_set_cb(self.read)
-        ifc.prot_set_timeout_cb(self._timeout)
-        ifc.prot_set_init_new_client_conn_cb(self._init_new_client_conn)
-        ifc.prot_set_update_header_cb(self._update_header)
 
-        self.addr = addr
-        self.conn_no = ifc.get_conn_no()
-        self.header_len = 11  # overwrite construcor in class Message
-        self.control = 0
-        self.seq = Sequence(server_side)
-        self.snr = 0
         self.db = InfosG3P(client_mode)
-        self.time_ofs = 0
         self.forward_at_cmd_resp = False
         self.no_forwarding = False
         '''not allowed to connect to TSUN cloud by connection type'''
+        self.establish_inv_emu = False
+        '''create an Solarman EMU instance to send data to the TSUN cloud'''
         self.switch = {
 
             0x4210: self.msg_data_ind,   # real time data
@@ -135,7 +328,7 @@ class SolarmanV5(Message):
         if 'at_acl' in g3p_cnf:  # pragma: no cover
             self.at_acl = g3p_cnf['at_acl']
 
-        self.sensor_list = 0x0000
+        self.sensor_list = 0
 
     '''
     Our puplic methods
@@ -150,16 +343,24 @@ class SolarmanV5(Message):
         super().close()
 
     async def send_start_cmd(self, snr: int, host: str,
+                             forward: bool,
                              start_timeout=MB_CLIENT_DATA_UP):
         self.no_forwarding = True
+        self.establish_inv_emu = forward
         self.snr = snr
-        self.__set_serial_no(snr)
+        self._set_serial_no(snr)
         self.mb_timeout = start_timeout
         self.db.set_db_def_value(Register.IP_ADDRESS, host)
         self.db.set_db_def_value(Register.POLLING_INTERVAL,
                                  self.mb_timeout)
+        self.db.set_db_def_value(Register.DATA_UP_INTERVAL,
+                                 300)
+        self.db.set_db_def_value(Register.COLLECT_INTERVAL,
+                                 1)
         self.db.set_db_def_value(Register.HEARTBEAT_INTERVAL,
                                  120)
+        self.db.set_db_def_value(Register.SENSOR_LIST,
+                                 Fmt.hex4((self.sensor_list, )))
         self.new_data['controller'] = True
 
         self.state = State.up
@@ -174,6 +375,15 @@ class SolarmanV5(Message):
                 self.db.set_db_def_value(Register.POLLING_INTERVAL,
                                          self.mb_timeout)
 
+    def establish_emu(self):
+        _len = 223
+        build_msg = self.db.build(_len, 0x41, 2)
+        struct.pack_into(
+            '<BHHHLBL', build_msg, 0, 0xA5, _len-11, 0x4110,
+            0, self.snr, 2, self._emu_timestamp())
+        self.ifc.fwd_add(build_msg)
+        self.ifc.fwd_add(struct.pack('<BB', 0, 0x15))    # crc & stop
+
     def __set_config_parms(self, inv: dict):
         '''init connection with params from the configuration'''
         self.node_id = inv['node_id']
@@ -183,7 +393,7 @@ class SolarmanV5(Message):
         if self.mb:
             self.mb.set_node_id(self.node_id)
 
-    def __set_serial_no(self, snr: int):
+    def _set_serial_no(self, snr: int):
         '''check the serial number and configure the inverter connection'''
         serial_no = str(snr)
         if self.unique_id == serial_no:
@@ -200,7 +410,8 @@ class SolarmanV5(Message):
                     self.db.set_pv_module_details(inv)
                     logger.debug(f'SerialNo {serial_no} allowed! area:{self.sug_area}')  # noqa: E501
 
-                    self.db.set_db_def_value(Register.COLLECTOR_SNR, key)
+                    self.db.set_db_def_value(Register.COLLECTOR_SNR, snr)
+                    self.db.set_db_def_value(Register.SERIAL_NUMBER, key)
                     break
             else:
                 self.node_id = ''
@@ -214,35 +425,6 @@ class SolarmanV5(Message):
 
             self.unique_id = serial_no
 
-    def read(self) -> float:
-        '''process all received messages in the _recv_buffer'''
-        self._read()
-        while True:
-            if not self.header_valid:
-                self.__parse_header(self.ifc.rx_peek(),
-                                    self.ifc.rx_len())
-
-            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()
-            else:
-                return 0  # wait 0s before sending a response
-
-    def __process_complete_received_msg(self):
-        log_lvl = self.log_lvl.get(self.control, logging.WARNING)
-        if callable(log_lvl):
-            log_lvl = log_lvl()
-        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.rx_peek(), self.header_len
-                                + self.data_len + 2):
-            if self.state == State.init:
-                self.state = State.received
-            self.__set_serial_no(self.snr)
-            self.__dispatch_msg()
-
     def forward(self, buffer, buflen) -> None:
         '''add the actual receive msg to the forwarding queue'''
         if self.no_forwarding:
@@ -252,158 +434,34 @@ class SolarmanV5(Message):
             self.ifc.fwd_add(buffer[:buflen])
             self.ifc.fwd_log(logging.DEBUG, 'Store for forwarding:')
 
-            fnc = self.switch.get(self.control, self.msg_unknown)
+            _, _str = self.get_fnc_handler(self.control)
-            logger.info(self.__flow_str(self.server_side, 'forwrd') +
+            logger.info(self._flow_str(self.server_side, 'forwrd') +
                         f' Ctl: {int(self.control):#04x}'
-                        f' Msg: {fnc.__name__!r}')
+                        f' Msg: {_str}')
 
     def _init_new_client_conn(self) -> bool:
         return False
 
-    '''
-    Our private methods
-    '''
-    def __flow_str(self, server_side: bool, type: str):  # noqa: F821
-        switch = {
-            'rx':      '  <',
-            'tx':      '  >',
-            'forwrd':  '<< ',
-            'drop':    ' xx',
-            'rxS':     '>  ',
-            'txS':     '<  ',
-            'forwrdS': ' >>',
-            'dropS':   'xx ',
-        }
-        if server_side:
-            type += 'S'
-        return switch.get(type, '???')
-
-    def _timestamp(self):
-        # utc as epoche
-        return int(time.time())    # pragma: no cover
-
     def _heartbeat(self) -> int:
         return 60                  # pragma: no cover
 
-    def __parse_header(self, buf: bytes, buf_len: int) -> None:
-
-        if (buf_len < self.header_len):  # enough bytes for complete header?
-            return
-
-        result = struct.unpack_from('<BHHHL', buf, 0)
-
-        # store parsed header values in the class
-        start = result[0]            # start byte
-        self.data_len = result[1]    # len of variable id string
-        self.control = result[2]
-        self.seq.set_recv(result[3])
-        self.snr = result[4]
-
-        if start != 0xA5:
-            hex_dump_memory(logging.ERROR,
-                            'Drop packet w invalid start byte from'
-                            f' {self.addr}:', buf, buf_len)
-
-            self.inc_counter('Invalid_Msg_Format')
-            # erase broken recv buffer
-            self.ifc.rx_clear()
-            return
-        self.header_valid = True
-
-    def __trailer_is_ok(self, buf: bytes, buf_len: int) -> bool:
-        crc = buf[self.data_len+11]
-        stop = buf[self.data_len+12]
-        if stop != 0x15:
-            hex_dump_memory(logging.ERROR,
-                            'Drop packet w invalid stop byte from '
-                            f'{self.addr}:', buf, buf_len)
-            self.inc_counter('Invalid_Msg_Format')
-            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.rx_clear()
-
-            return False
-
-        check = sum(buf[1:buf_len-2]) & 0xff
-        if check != crc:
-            self.inc_counter('Invalid_Msg_Format')
-            logger.debug(f'CRC {int(crc):#02x} {int(check):#08x}'
-                         f' Stop:{int(stop):#02x}')
-            # start & stop byte are valid, discard only this message
-            return False
-
-        return True
-
-    def __build_header(self, ctrl) -> None:
-        '''build header for new transmit message'''
-        self.send_msg_ofs = self.ifc.tx_len()
-
-        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 = 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.tx_peek()[
-            self.send_msg_ofs+1:self.send_msg_ofs + _len]) & 0xff
-        self.ifc.tx_add(struct.pack('<BB', check, 0x15))    # crc & stop
-
-    def _update_header(self, _forward_buffer):
-        '''update header for message before forwarding,
-        set sequence and checksum'''
-        _len = len(_forward_buffer)
-        ofs = 0
-        while ofs < _len:
-            result = struct.unpack_from('<BH', _forward_buffer, ofs)
-            data_len = result[1]    # len of variable id string
-
-            struct.pack_into('<H', _forward_buffer, ofs+5,
-                             self.seq.get_send())
-
-            check = sum(_forward_buffer[ofs+1:ofs+data_len+11]) & 0xff
-            struct.pack_into('<B', _forward_buffer, ofs+data_len+11, check)
-            ofs += (13 + data_len)
-
-    def __dispatch_msg(self) -> None:
-        fnc = self.switch.get(self.control, self.msg_unknown)
-        if self.unique_id:
-            logger.info(self.__flow_str(self.server_side, 'rx') +
-                        f' Ctl: {int(self.control):#04x}' +
-                        f' Msg: {fnc.__name__!r}')
-            fnc()
-        else:
-            logger.info(self.__flow_str(self.server_side, 'drop') +
-                        f' Ctl: {int(self.control):#04x}' +
-                        f' Msg: {fnc.__name__!r}')
-
-    def __flush_recv_msg(self) -> None:
-        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._build_header(msgtype)
         self.ifc.tx_add(struct.pack('<BBLL', ftype, ack,
                                     self._timestamp(),
                                     self._heartbeat()))
-        self.__finish_send_msg()
+        self._finish_send_msg()
 
     def send_modbus_cb(self, pdu: bytearray, log_lvl: int, state: str):
         if self.state != State.up:
             logger.warning(f'[{self.node_id}] ignore MODBUS cmd,'
                            ' cause the state is not UP anymore')
             return
-        self.__build_header(0x4510)
+        self._build_header(0x4510)
         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._finish_send_msg()
         self.ifc.tx_log(log_lvl, f'Send Modbus {state}:{self.addr}:')
         self.ifc.tx_flush()
 
@@ -436,11 +494,11 @@ class SolarmanV5(Message):
             return
 
         self.forward_at_cmd_resp = False
-        self.__build_header(0x4510)
+        self._build_header(0x4510)
         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._finish_send_msg()
         self.ifc.tx_log(logging.INFO, 'Send AT Command:')
         try:
             self.ifc.tx_flush()
@@ -607,6 +665,18 @@ class SolarmanV5(Message):
             return
         self.__forward_msg()
 
+    def __parse_modbus_rsp(self, data):
+        inv_update = False
+        self.modbus_elms = 0
+        for key, update, _ in self.mb.recv_resp(self.db, data[14:]):
+            self.modbus_elms += 1
+            if update:
+                if key == 'inverter':
+                    inv_update = True
+                self._set_mqtt_timestamp(key, self._timestamp())
+                self.new_data[key] = True
+        return inv_update
+
     def __modbus_command_rsp(self, data):
         '''precess MODBUS RTU response'''
         valid = data[1]
@@ -614,18 +684,13 @@ class SolarmanV5(Message):
         # logger.debug(f'modbus_len:{modbus_msg_len} accepted:{valid}')
         if valid == 1 and modbus_msg_len > 4:
             # logger.info(f'first byte modbus:{data[14]}')
-            inv_update = False
+            inv_update = self.__parse_modbus_rsp(data)
-            self.modbus_elms = 0
-            for key, update, _ in self.mb.recv_resp(self.db, data[14:]):
-                self.modbus_elms += 1
-                if update:
-                    if key == 'inverter':
-                        inv_update = True
-                    self._set_mqtt_timestamp(key, self._timestamp())
-                    self.new_data[key] = True
             if inv_update:
                 self.__build_model_name()
 
+            if self.establish_inv_emu and not self.ifc.remote.stream:
+                self.establish_emu()
+
     def msg_hbeat_ind(self):
         data = self.ifc.rx_peek()[self.header_len:]
         result = struct.unpack_from('<B', data, 0)
@@ -647,16 +712,3 @@ class SolarmanV5(Message):
 
         self.__forward_msg()
         self.__send_ack_rsp(0x1810, ftype)
-
-    def msg_response(self):
-        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
-        ts = result[2]
-        set_hb = result[3]  # always 60 or 120
-        logger.debug(f'ftype:{ftype} accepted:{valid}'
-                     f' ts:{ts:08x}  nextHeartbeat: {set_hb}s')
-
-        dt = datetime.fromtimestamp(ts)
-        logger.debug(f'ts: {dt.strftime("%Y-%m-%d %H:%M:%S")}')

app/src/infos.py

@@ -26,7 +26,10 @@ class Register(Enum):
     EQUIPMENT_MODEL = 24
     NO_INPUTS = 25
     MAX_DESIGNED_POWER = 26
-    OUTPUT_COEFFICIENT = 27
+    RATED_LEVEL = 27
+    INPUT_COEFFICIENT = 28
+    GRID_VOLT_CAL_COEF = 29
+    OUTPUT_COEFFICIENT = 30
     INVERTER_CNT = 50
     UNKNOWN_SNR = 51
     UNKNOWN_MSG = 52
@@ -39,6 +42,7 @@ class Register(Enum):
     AT_COMMAND = 59
     MODBUS_COMMAND = 60
     AT_COMMAND_BLOCKED = 61
+    CLOUD_CONN_CNT = 62
     OUTPUT_POWER = 83
     RATED_POWER = 84
     INVERTER_TEMP = 85
@@ -90,6 +94,8 @@ class Register(Enum):
     INV_UNKNOWN_1 = 252
     BOOT_STATUS = 253
     DSP_STATUS = 254
+    WORK_MODE = 255
+    OUTPUT_SHUTDOWN = 256
 
     GRID_VOLTAGE = 300
     GRID_CURRENT = 301
@@ -269,6 +275,7 @@ class Infos:
     }
 
     __comm_type_val_tpl = "{%set com_types = ['n/a','Wi-Fi', 'G4', 'G5', 'GPRS'] %}{{com_types[value_json['Communication_Type']|int(0)]|default(value_json['Communication_Type'])}}"    # noqa: E501
+    __work_mode_val_tpl = "{%set mode = ['Normal-Mode', 'Aging-Mode', 'ATE-Mode', 'Shielding GFDI', 'DTU-Mode'] %}{{mode[value_json['Work_Mode']|int(0)]|default(value_json['Work_Mode'])}}"    # noqa: E501
     __status_type_val_tpl = "{%set inv_status = ['Off-line', 'On-grid', 'Off-grid'] %}{{inv_status[value_json['Inverter_Status']|int(0)]|default(value_json['Inverter_Status'])}}"    # noqa: E501
     __rated_power_val_tpl = "{% if 'Rated_Power' in value_json and value_json['Rated_Power'] != None %}{{value_json['Rated_Power']|string() +' W'}}{% else %}{{ this.state }}{% endif %}"  # noqa: E501
     __designed_power_val_tpl = '''
@@ -376,6 +383,7 @@ class Infos:
 {% endif %}
 '''
 
+    __input_coef_val_tpl = "{% if 'Output_Coefficient' in value_json and value_json['Input_Coefficient'] != None %}{{value_json['Input_Coefficient']|string() +' %'}}{% else %}{{ this.state }}{% endif %}"  # noqa: E501
     __output_coef_val_tpl = "{% if 'Output_Coefficient' in value_json and value_json['Output_Coefficient'] != None %}{{value_json['Output_Coefficient']|string() +' %'}}{% else %}{{ this.state }}{% endif %}"  # noqa: E501
 
     __info_defs = {
@@ -398,6 +406,8 @@ class Infos:
         Register.NO_INPUTS:       {'name': ['inverter', 'No_Inputs'],              'level': logging.DEBUG, 'unit': ''},  # noqa: E501
         Register.MAX_DESIGNED_POWER: {'name': ['inverter',  'Max_Designed_Power'], 'level': logging.INFO,  'unit': 'W',    'ha': {'dev': 'inverter', 'dev_cla': None, 'stat_cla': None, 'id': 'designed_power_', 'val_tpl': __designed_power_val_tpl, 'name': 'Max Designed Power', 'icon': LIGHTNING, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.RATED_POWER:        {'name': ['inverter',  'Rated_Power'],        'level': logging.DEBUG, 'unit': 'W',    'ha': {'dev': 'inverter', 'dev_cla': None, 'stat_cla': None, 'id': 'rated_power_',    'val_tpl': __rated_power_val_tpl,    'name': 'Rated Power',        'icon': LIGHTNING, 'ent_cat': 'diagnostic'}},  # noqa: E501
+        Register.WORK_MODE:          {'name': ['inverter',  'Work_Mode'],          'level': logging.DEBUG, 'unit': '',     'ha': {'dev': 'inverter', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'work_mode_', 'name': 'Work Mode', 'val_tpl': __work_mode_val_tpl,  'icon': 'mdi:power', 'ent_cat': 'diagnostic'}},  # noqa: E501
+        Register.INPUT_COEFFICIENT:  {'name': ['inverter',  'Input_Coefficient'],  'level': logging.DEBUG, 'unit': '%',    'ha': {'dev': 'inverter', 'dev_cla': None, 'stat_cla': None, 'id': 'input_coef_',    'val_tpl': __input_coef_val_tpl,    'name': 'Input Coefficient', 'icon': LIGHTNING, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.OUTPUT_COEFFICIENT: {'name': ['inverter',  'Output_Coefficient'], 'level': logging.INFO,  'unit': '%',    'ha': {'dev': 'inverter', 'dev_cla': None, 'stat_cla': None, 'id': 'output_coef_',    'val_tpl': __output_coef_val_tpl,    'name': 'Output Coefficient', 'icon': LIGHTNING, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.PV1_MANUFACTURER: {'name': ['inverter', 'PV1_Manufacturer'],      'level': logging.DEBUG, 'unit': ''},  # noqa: E501
         Register.PV1_MODEL:        {'name': ['inverter', 'PV1_Model'],             'level': logging.DEBUG, 'unit': ''},  # noqa: E501
@@ -415,6 +425,7 @@ class Infos:
         Register.DSP_STATUS:       {'name': ['inverter', 'DSP_STATUS'],            'level': logging.DEBUG, 'unit': ''},  # noqa: E501
         # proxy:
         Register.INVERTER_CNT:       {'name': ['proxy', 'Inverter_Cnt'],       'singleton': True,   'ha': {'dev': 'proxy', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'inv_count_',     'fmt': FMT_INT, 'name': 'Active Inverter Connections',    'icon': COUNTER}},  # noqa: E501
+        Register.CLOUD_CONN_CNT:     {'name': ['proxy', 'Cloud_Conn_Cnt'],     'singleton': True,   'ha': {'dev': 'proxy', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'cloud_conn_count_', 'fmt': FMT_INT, 'name': 'Active Cloud Connections',    'icon': COUNTER}},  # noqa: E501
         Register.UNKNOWN_SNR:        {'name': ['proxy', 'Unknown_SNR'],        'singleton': True,   'ha': {'dev': 'proxy', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'unknown_snr_',   'fmt': FMT_INT, 'name': 'Unknown Serial No',    'icon': COUNTER, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.UNKNOWN_MSG:        {'name': ['proxy', 'Unknown_Msg'],        'singleton': True,   'ha': {'dev': 'proxy', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'unknown_msg_',   'fmt': FMT_INT, 'name': 'Unknown Msg Type',     'icon': COUNTER, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.INVALID_DATA_TYPE:  {'name': ['proxy', 'Invalid_Data_Type'],  'singleton': True,   'ha': {'dev': 'proxy', 'comp': 'sensor', 'dev_cla': None, 'stat_cla': None, 'id': 'inv_data_type_', 'fmt': FMT_INT, 'name': 'Invalid Data Type',    'icon': COUNTER, 'ent_cat': 'diagnostic'}},  # noqa: E501
@@ -495,9 +506,12 @@ class Infos:
         Register.IP_ADDRESS:         {'name': ['controller', 'IP_Address'],         'level': logging.DEBUG, 'unit': '',     'ha': {'dev': 'controller', 'dev_cla': None,       'stat_cla': None,          'id': 'ip_address_',           'fmt': '| string',        'name': 'IP Address', 'icon': WIFI, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.POLLING_INTERVAL:   {'name': ['controller', 'Polling_Interval'],   'level': logging.DEBUG, 'unit': 's',    'ha': {'dev': 'controller', 'dev_cla': None,       'stat_cla': None,          'id': 'polling_intval_', 'fmt': FMT_STRING_SEC, 'name': 'Polling Interval', 'icon': UPDATE, 'ent_cat': 'diagnostic'}},  # noqa: E501
         Register.SENSOR_LIST:        {'name': ['controller', 'Sensor_List'],        'level': logging.INFO,  'unit': ''},  # noqa: E501
-        Register.SSID:               {'name': ['controller', 'WiFi_SSID'],          'level': logging.DEBUG,  'unit': ''},  # noqa: E501
+        Register.SSID:               {'name': ['controller', 'WiFi_SSID'],          'level': logging.DEBUG, 'unit': ''},  # noqa: E501
 
-        Register.INV_UNKNOWN_1:      {'name': ['inv_unknown', 'Unknown_1'],          'level': logging.DEBUG, 'unit': ''},  # noqa: E501
+        Register.OUTPUT_SHUTDOWN:    {'name': ['other', 'Output_Shutdown'],         'level': logging.DEBUG, 'unit': ''},  # noqa: E501
+        Register.RATED_LEVEL:        {'name': ['other', 'Rated_Level'],             'level': logging.DEBUG, 'unit': ''},  # noqa: E501
+        Register.GRID_VOLT_CAL_COEF: {'name': ['other', 'Grid_Volt_Cal_Coef'],      'level': logging.DEBUG, 'unit': ''},  # noqa: E501
+        Register.INV_UNKNOWN_1:      {'name': ['inv_unknown', 'Unknown_1'],         'level': logging.DEBUG, 'unit': ''},  # noqa: E501
 
     }
 

app/src/inverter_base.py

@@ -31,12 +31,16 @@ class InverterBase(InverterIfc, Proxy):
 
     def __init__(self, reader: StreamReader, writer: StreamWriter,
                  config_id: str, prot_class,
-                 client_mode: bool = False):
+                 client_mode: bool = False,
+                 remote_prot_class=None):
         Proxy.__init__(self)
         self._registry.append(weakref.ref(self))
         self.addr = writer.get_extra_info('peername')
         self.config_id = config_id
-        self.prot_class = prot_class
+        if remote_prot_class:
+            self.prot_class = remote_prot_class
+        else:
+            self.prot_class = prot_class
         self.__ha_restarts = -1
         self.remote = StreamPtr(None)
         ifc = AsyncStreamServer(reader, writer,
@@ -45,7 +49,7 @@ class InverterBase(InverterIfc, Proxy):
                                 self.remote)
 
         self.local = StreamPtr(
-            self.prot_class(self.addr, ifc, True, client_mode), ifc
+            prot_class(self.addr, ifc, True, client_mode), ifc
         )
 
     def __enter__(self):

app/src/messages.py

@@ -104,12 +104,14 @@ class Message(ProtocolIfc):
         self._registry.append(weakref.ref(self))
 
         self.server_side = server_side
-        if server_side:
-            self.mb = Modbus(send_modbus_cb, mb_timeout)
-        else:
-            self.mb = None
         self.ifc = ifc
         self.node_id = node_id
+        if server_side:
+            self.mb = Modbus(send_modbus_cb, mb_timeout)
+            self.mb_timer = Timer(self.mb_timout_cb, self.node_id)
+        else:
+            self.mb = None
+            self.mb_timer = None
         self.header_valid = False
         self.header_len = 0
         self.data_len = 0
@@ -119,7 +121,6 @@ class Message(ProtocolIfc):
         self.state = State.init
         self.shutdown_started = False
         self.modbus_elms = 0    # for unit tests
-        self.mb_timer = Timer(self.mb_timout_cb, self.node_id)
         self.mb_timeout = self.MB_REGULAR_TIMEOUT
         self.mb_first_timeout = self.MB_START_TIMEOUT
         '''timer value for next Modbus polling request'''
@@ -188,8 +189,8 @@ class Message(ProtocolIfc):
             if self.db.get_db_value(Register.OUTPUT_POWER, 999) < 2:
                 self.db.set_db_def_value(Register.INVERTER_STATUS, 0)
                 self.new_data['env'] = True
+            self.mb_timer.close()
         self.state = State.closed
-        self.mb_timer.close()
         self.ifc.rx_set_cb(None)
         self.ifc.prot_set_timeout_cb(None)
         self.ifc.prot_set_init_new_client_conn_cb(None)

app/src/modbus.py

@@ -42,7 +42,13 @@ class Modbus():
     mb_reg_mapping = {
         0x2000: {'reg': Register.BOOT_STATUS,          'fmt': '!H'},                 # noqa: E501
         0x2001: {'reg': Register.DSP_STATUS,           'fmt': '!H'},                 # noqa: E501
+        0x2003: {'reg': Register.WORK_MODE,            'fmt': '!H'},
+        0x2006: {'reg': Register.OUTPUT_SHUTDOWN,      'fmt': '!H'},
         0x2007: {'reg': Register.MAX_DESIGNED_POWER,   'fmt': '!H', 'ratio':  1},  # noqa: E501
+        0x2008: {'reg': Register.RATED_LEVEL,          'fmt': '!H'},
+        0x2009: {'reg': Register.INPUT_COEFFICIENT,    'fmt': '!H', 'ratio':  100/1024},  # noqa: E501
+        0x200a: {'reg': Register.GRID_VOLT_CAL_COEF,   'fmt': '!H'},
+
         0x202c: {'reg': Register.OUTPUT_COEFFICIENT,   'fmt': '!H', 'ratio':  100/1024},  # noqa: E501
 
         0x3000: {'reg': Register.INVERTER_STATUS,      'fmt': '!H'},                 # noqa: E501

app/src/modbus_tcp.py

@@ -57,15 +57,17 @@ class ModbusTcp():
                 # logging.info(f"SerialNo:{inv['monitor_sn']} host:{client['host']} port:{client['port']}")  # noqa: E501
                 loop.create_task(self.modbus_loop(client['host'],
                                                   client['port'],
-                                                  inv['monitor_sn']))
+                                                  inv['monitor_sn'],
+                                                  client['forward']))
 
-    async def modbus_loop(self, host, port, snr: int) -> None:
+    async def modbus_loop(self, host, port,
+                          snr: int, forward: bool) -> None:
         '''Loop for receiving messages from the TSUN cloud (client-side)'''
         while True:
             try:
                 async with ModbusConn(host, port) as inverter:
                     stream = inverter.local.stream
-                    await stream.send_start_cmd(snr, host)
+                    await stream.send_start_cmd(snr, host, forward)
                     await stream.ifc.loop()
                     logger.info(f'[{stream.node_id}:{stream.conn_no}] '
                                 f'Connection closed - Shutdown: '

app/tests/test_infos.py

@@ -17,13 +17,13 @@ def test_statistic_counter():
     assert val == None or val == 0
 
     i.static_init()                # initialize counter
-    assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 0, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0, "AT_Command_Blocked": 0, "Modbus_Command": 0}})
+    assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 0, "Cloud_Conn_Cnt": 0, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0, "AT_Command_Blocked": 0, "Modbus_Command": 0}})
                                             
     val = i.dev_value(Register.INVERTER_CNT)  # valid and initiliazed addr
     assert val == 0
 
     i.inc_counter('Inverter_Cnt')
-    assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 1, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0, "AT_Command_Blocked": 0, "Modbus_Command": 0}})
+    assert json.dumps(i.stat) == json.dumps({"proxy": {"Inverter_Cnt": 1, "Cloud_Conn_Cnt": 0, "Unknown_SNR": 0, "Unknown_Msg": 0, "Invalid_Data_Type": 0, "Internal_Error": 0,"Unknown_Ctrl": 0, "OTA_Start_Msg": 0, "SW_Exception": 0, "Invalid_Msg_Format": 0, "AT_Command": 0, "AT_Command_Blocked": 0, "Modbus_Command": 0}})
     val = i.dev_value(Register.INVERTER_CNT)
     assert val == 1
 

app/tests/test_infos_g3.py

@@ -421,7 +421,7 @@ def test_must_incr_total(inv_data_seq2, inv_data_seq2_zero):
         if key == 'total' or key == 'inverter' or key == 'env':
             assert update == True
             tests +=1   
-    assert tests==8
+    assert tests==12
     assert json.dumps(i.db['total']) == json.dumps({'Daily_Generation': 1.7, 'Total_Generation': 17.36})
     assert json.dumps(i.db['input']) == json.dumps({"pv1": {"Voltage": 33.6, "Current": 1.91, "Power": 64.5, "Daily_Generation": 1.08, "Total_Generation": 9.74}, "pv2": {"Voltage": 33.5, "Current": 1.36, "Power": 45.7, "Daily_Generation": 0.62, "Total_Generation": 7.62}, "pv3": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}, "pv4": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}})
     assert json.dumps(i.db['env']) == json.dumps({"Inverter_Status": 1, "Inverter_Temp": 23})
@@ -435,7 +435,7 @@ def test_must_incr_total(inv_data_seq2, inv_data_seq2_zero):
     assert json.dumps(i.db['total']) == json.dumps({'Daily_Generation': 1.7, 'Total_Generation': 17.36})
     assert json.dumps(i.db['input']) == json.dumps({"pv1": {"Voltage": 33.6, "Current": 1.91, "Power": 64.5, "Daily_Generation": 1.08, "Total_Generation": 9.74}, "pv2": {"Voltage": 33.5, "Current": 1.36, "Power": 45.7, "Daily_Generation": 0.62, "Total_Generation": 7.62}, "pv3": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}, "pv4": {"Voltage": 0.0, "Current": 0.0, "Power": 0.0}})
     assert json.dumps(i.db['env']) == json.dumps({"Inverter_Status": 1, "Inverter_Temp": 23})
-    assert json.dumps(i.db['inverter']) == json.dumps({"Rated_Power": 600, "Max_Designed_Power": -1, "Output_Coefficient": 100.0, "No_Inputs": 2})
+    assert json.dumps(i.db['inverter']) == json.dumps({"Rated_Power": 600, "BOOT_STATUS": 0, "DSP_STATUS": 21930, "Work_Mode": 0, "Max_Designed_Power": -1, "Input_Coefficient": -0.1, "Output_Coefficient": 100.0, "No_Inputs": 2})
         
     tests = 0
     for key, update in i.parse (inv_data_seq2_zero):
@@ -501,8 +501,8 @@ def test_new_data_types(inv_data_new):
         else:
             assert False
 
-    assert tests==5
+    assert tests==7
-    assert json.dumps(i.db['inverter']) == json.dumps({"Manufacturer": 0})
+    assert json.dumps(i.db['inverter']) == json.dumps({"Manufacturer": 0, "DSP_STATUS": 0})
     assert json.dumps(i.db['input']) == json.dumps({"pv1": {}})
     assert json.dumps(i.db['events']) == json.dumps({"Inverter_Alarm": 0, "Inverter_Fault": 0})
 

app/tests/test_infos_g3p.py

@@ -110,7 +110,7 @@ def test_parse_4210(inverter_data: bytes):
 
     assert json.dumps(i.db) == json.dumps({
          "controller": {"Sensor_List": "02b0", "Power_On_Time": 2051}, 
-         "inverter": {"Serial_Number": "Y17E00000000000E", "Version": "V4.0.10", "Rated_Power": 600, "BOOT_STATUS": 0, "DSP_STATUS": 21930, "Max_Designed_Power": 2000, "Output_Coefficient": 100.0}, 
+         "inverter": {"Serial_Number": "Y17E00000000000E", "Version": "V4.0.10", "Rated_Power": 600, "BOOT_STATUS": 0, "DSP_STATUS": 21930, "Work_Mode": 0, "Max_Designed_Power": 2000, "Input_Coefficient": 100.0, "Output_Coefficient": 100.0}, 
          "env": {"Inverter_Status": 1, "Detect_Status_1": 2, "Detect_Status_2": 0, "Inverter_Temp": 14}, 
          "events": {"Inverter_Alarm": 0, "Inverter_Fault": 0, "Inverter_Bitfield_1": 0, "Inverter_bitfield_2": 0},
          "grid": {"Voltage": 224.8, "Current": 0.73, "Frequency": 50.05, "Output_Power": 165.8}, 
@@ -119,7 +119,8 @@ def test_parse_4210(inverter_data: bytes):
                    "pv3": {"Voltage": 34.6, "Current": 1.89, "Power": 65.5, "Daily_Generation": 0.05, "Total_Generation": 31.89}, 
                    "pv4": {"Voltage": 1.7, "Current": 0.01, "Power": 0.0, "Total_Generation": 15.58}}, 
          "total": {"Daily_Generation": 0.11, "Total_Generation": 101.36},
-         "inv_unknown": {"Unknown_1": 512}
+         "inv_unknown": {"Unknown_1": 512},
+         "other": {"Output_Shutdown": 65535, "Rated_Level": 3, "Grid_Volt_Cal_Coef": 1024}
         })
 
 def test_build_4210(inverter_data: bytes):

app/tests/test_inverter_g3p.py

@@ -144,7 +144,7 @@ async def test_mqtt_publish(config_conn, patch_open_connection):
     with InverterG3P(FakeReader(), FakeWriter(), client_mode=False) as inverter:
         stream = inverter.local.stream
         await inverter.async_publ_mqtt()  # check call with invalid unique_id   
-        stream._SolarmanV5__set_serial_no(snr= 123344)
+        stream._set_serial_no(snr= 123344)
 
         stream.new_data['inverter'] = True
         stream.db.db['inverter'] = {}
@@ -171,7 +171,7 @@ async def test_mqtt_err(config_conn, patch_open_connection, patch_mqtt_err):
 
     with InverterG3P(FakeReader(), FakeWriter(), client_mode=False) as inverter:
         stream = inverter.local.stream
-        stream._SolarmanV5__set_serial_no(snr= 123344)    
+        stream._set_serial_no(snr= 123344)    
         stream.new_data['inverter'] = True
         stream.db.db['inverter'] = {}
         await inverter.async_publ_mqtt()
@@ -188,7 +188,7 @@ async def test_mqtt_except(config_conn, patch_open_connection, patch_mqtt_except
 
     with InverterG3P(FakeReader(), FakeWriter(), client_mode=False) as inverter:
         stream = inverter.local.stream
-        stream._SolarmanV5__set_serial_no(snr= 123344)
+        stream._set_serial_no(snr= 123344)
 
         stream.new_data['inverter'] = True
         stream.db.db['inverter'] = {}

app/tests/test_modbus_tcp.py

@@ -52,6 +52,10 @@ def config_conn(test_hostname, test_port):
                             'proxy_node_id': 'test_1',
                             'proxy_unique_id': ''
                         },
+                        'solarman':{
+                            'host': 'access1.solarmanpv.com',
+                            'port': 10000
+                        },
                         'inverters':{
                             'allow_all': True,
                             "R170000000000001":{
@@ -65,7 +69,8 @@ def config_conn(test_hostname, test_port):
                                 'sensor_list': 0x2b0,
                                 'client_mode':{
                                     'host': '192.168.0.1', 
-                                    'port': 8899
+                                    'port': 8899,
+                                    'forward': True
                                 }  
                             }
                         }

app/tests/test_solarman.py

@@ -6,7 +6,7 @@ import logging
 import random
 from math import isclose
 from app.src.async_stream import AsyncIfcImpl, StreamPtr
-from app.src.gen3plus.solarman_v5 import SolarmanV5
+from app.src.gen3plus.solarman_v5 import SolarmanV5, SolarmanBase
 from app.src.config import Config
 from app.src.infos import Infos, Register
 from app.src.modbus import Modbus
@@ -37,6 +37,9 @@ class FakeIfc(AsyncIfcImpl):
         super().__init__()
         self.remote = StreamPtr(None)
 
+    async def create_remote(self):
+        await asyncio.sleep(0)
+
 class MemoryStream(SolarmanV5):
     def __init__(self, msg, chunks = (0,), server_side: bool = True):
         _ifc = FakeIfc()
@@ -109,7 +112,7 @@ class MemoryStream(SolarmanV5):
         c.ifc.remote.stream = self
         return c
 
-    def _SolarmanV5__flush_recv_msg(self) -> None:
+    def _SolarmanBase__flush_recv_msg(self) -> None:
         self.msg_recvd.append(
             {
                 'control': self.control,
@@ -117,7 +120,7 @@ class MemoryStream(SolarmanV5):
                 'data_len': self.data_len
             }
         )
-        super()._SolarmanV5__flush_recv_msg()
+        super()._SolarmanBase__flush_recv_msg()
         self.msg_count += 1
 
 
@@ -1102,7 +1105,7 @@ def test_sync_start_ind(config_tsun_inv1, sync_start_ind_msg, sync_start_rsp_msg
     assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0
 
     m.seq.server_side = False  # simulate forawding to TSUN cloud
-    m._update_header(m.ifc.fwd_fifo.peek())
+    m._SolarmanBase__update_header(m.ifc.fwd_fifo.peek())
     assert str(m.seq) == '0d:0e'  # value after forwarding indication
     assert m.ifc.fwd_fifo.get()==sync_start_fwd_msg
 
@@ -1768,7 +1771,7 @@ async def test_start_client_mode(config_tsun_inv1, str_test_ip):
     assert m.no_forwarding == False
     assert m.mb_timer.tim == None
     assert asyncio.get_running_loop() == m.mb_timer.loop
-    await m.send_start_cmd(get_sn_int(), str_test_ip, m.mb_first_timeout)
+    await m.send_start_cmd(get_sn_int(), str_test_ip, False, m.mb_first_timeout)
     assert m.sent_pdu==bytearray(b'\xa5\x17\x00\x10E\x01\x00!Ce{\x02\xb0\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x030\x00\x000J\xde\xf1\x15')
     assert m.db.get_db_value(Register.IP_ADDRESS) == str_test_ip
     assert isclose(m.db.get_db_value(Register.POLLING_INTERVAL), 0.5)
@@ -1803,3 +1806,30 @@ def test_timeout(config_tsun_inv1):
     assert SolarmanV5.MAX_DEF_IDLE_TIME == m._timeout()
     m.state = State.closed
     m.close()
+
+def test_fnc_dispatch():
+    def msg():
+        return
+    
+    _ = config_tsun_inv1
+    m = MemoryStream(b'')
+    m.switch[1] = msg
+    m.switch[2] = "msg"
+
+    _obj, _str = m.get_fnc_handler(1)
+    assert _obj == msg
+    assert _str == "'msg'"
+
+    _obj, _str = m.get_fnc_handler(2)
+    assert _obj == m.msg_unknown
+    assert _str == "'msg'"
+
+    _obj, _str = m.get_fnc_handler(3)
+    assert _obj == m.msg_unknown
+    assert _str == "'msg_unknown'"
+
+def test_timestamp():
+    m = MemoryStream(b'')
+    ts = m._timestamp()
+    ts_emu = m._emu_timestamp()
+    assert ts == ts_emu + 24*60*60

app/tests/test_solarman_emu.py

@@ -0,0 +1,230 @@
+import pytest
+import asyncio
+from app.src.async_stream import AsyncIfcImpl, StreamPtr
+from app.src.gen3plus.solarman_v5 import SolarmanV5, SolarmanBase
+from app.src.gen3plus.solarman_emu import SolarmanEmu
+from app.src.infos import Infos, Register
+from app.tests.test_solarman import FakeIfc, MemoryStream, get_sn_int, get_sn, correct_checksum, config_tsun_inv1, msg_modbus_rsp
+from app.tests.test_infos_g3p import str_test_ip, bytes_test_ip
+
+timestamp = 0x3224c8bc
+
+class InvStream(MemoryStream):
+    def __init__(self, msg=b''):
+        super().__init__(msg)
+
+    def _emu_timestamp(self):
+        return timestamp
+
+class CldStream(SolarmanEmu):
+    def __init__(self, inv: InvStream):
+        _ifc = FakeIfc()
+        _ifc.remote.stream = inv
+        super().__init__(('test.local', 1234), _ifc, server_side=False, client_mode=False)
+        self.__msg = b''
+        self.__msg_len = 0
+        self.__offs = 0
+        self.msg_count = 0
+        self.msg_recvd = []
+
+    def _emu_timestamp(self):
+        return timestamp
+    
+    def append_msg(self, msg):
+        self.__msg += msg
+        self.__msg_len += len(msg)    
+
+    def _read(self) -> int:
+        copied_bytes = 0
+        try:    
+            if (self.__offs < self.__msg_len):
+                self.ifc.rx_fifo += self.__msg[self.__offs:]
+                copied_bytes = self.__msg_len - self.__offs
+                self.__offs = self.__msg_len
+        except Exception:
+            pass   # ignore exceptions here
+        return copied_bytes
+
+    def _SolarmanBase__flush_recv_msg(self) -> None:
+        self.msg_recvd.append(
+            {
+                'control': self.control,
+                'seq': str(self.seq),
+                'data_len': self.data_len
+            }
+        )
+        super()._SolarmanBase__flush_recv_msg()
+        self.msg_count += 1
+
+@pytest.fixture
+def device_ind_msg(bytes_test_ip): # 0x4110
+    msg  = b'\xa5\xd4\x00\x10\x41\x00\x01' +get_sn()  +b'\x02\xbc\xc8\x24\x32'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x05\x3c\x78\x01\x00\x01\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + bytes_test_ip
+    msg += b'\x0f\x00\x01\xb0'
+    msg += b'\x02\x0f\x00\xff\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xfe\xfe\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += correct_checksum(msg)
+    msg += b'\x15'
+    return msg
+
+@pytest.fixture
+def inverter_ind_msg():  # 0x4210
+    msg  = b'\xa5\x99\x01\x10\x42\x00\x01' +get_sn()  +b'\x01\xb0\x02\xbc\xc8'
+    msg += b'\x24\x32\x3c\x00\x00\x00\xa0\x47\xe4\x33\x01\x00\x03\x08\x00\x00'
+    msg += b'\x59\x31\x37\x30\x30\x30\x30\x30\x30\x30\x30\x30\x30\x30\x30\x31'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x01\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x40\x10\x08\xc8\x00\x49\x13\x8d\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00'
+    msg += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00'
+    msg += b'\x04\x00\x00\x01\xff\xff\x00\x01\x00\x06\x00\x68\x00\x68\x05\x00'
+    msg += b'\x09\xcd\x07\xb6\x13\x9c\x13\x24\x00\x01\x07\xae\x04\x0f\x00\x41'
+    msg += b'\x00\x0f\x0a\x64\x0a\x64\x00\x06\x00\x06\x09\xf6\x12\x8c\x12\x8c'
+    msg += b'\x00\x10\x00\x10\x14\x52\x14\x52\x00\x10\x00\x10\x01\x51\x00\x05'
+    msg += b'\x00\x00\x00\x01\x13\x9c\x0f\xa0\x00\x4e\x00\x66\x03\xe8\x04\x00'
+    msg += b'\x09\xce\x07\xa8\x13\x9c\x13\x26\x00\x00\x00\x00\x00\x00\x00\x00'
+    msg += b'\x00\x00\x00\x00\x04\x00\x04\x00\x00\x00\x00\x00\xff\xff\x00\x00'
+    msg += b'\x00\x00\x00\x00'
+    msg += correct_checksum(msg)
+    msg += b'\x15'
+    return msg
+
+@pytest.fixture
+def inverter_rsp_msg():  # 0x1210
+    msg  = b'\xa5\x0a\x00\x10\x12\x02\02' +get_sn()  +b'\x01\x01'
+    msg += b'\x00\x00\x00\x00'
+    msg += b'\x3c\x00\x00\x00'
+    msg += correct_checksum(msg)
+    msg += b'\x15'
+    return msg
+
+@pytest.fixture
+def heartbeat_ind():
+    msg  = b'\xa5\x01\x00\x10G\x00\x01\x00\x00\x00\x00\x00Y\x15'
+    return msg
+
+def test_emu_init_close():
+    # received a message with wrong start byte plus an valid message
+    # the complete receive buffer must be cleared to 
+    # find the next valid message
+    inv = InvStream()
+    cld = CldStream(inv)
+    cld.close()
+
+
+@pytest.mark.asyncio
+async def test_emu_start(config_tsun_inv1, msg_modbus_rsp, str_test_ip, device_ind_msg):
+    _ = config_tsun_inv1
+    assert asyncio.get_running_loop()
+    inv = InvStream(msg_modbus_rsp)
+
+    assert asyncio.get_running_loop() == inv.mb_timer.loop
+    await inv.send_start_cmd(get_sn_int(), str_test_ip, True, inv.mb_first_timeout)
+    inv.read()         # read complete msg, and dispatch msg
+    assert not inv.header_valid  # must be invalid, since msg was handled and buffer flushed
+    assert inv.msg_count == 1
+    assert inv.control == 0x1510
+
+    cld = CldStream(inv)
+    cld.ifc.update_header_cb(inv.ifc.fwd_fifo.peek())
+    assert inv.ifc.fwd_fifo.peek() == device_ind_msg
+    cld.close()
+
+def test_snd_hb(config_tsun_inv1, heartbeat_ind):
+    _ = config_tsun_inv1
+    inv = InvStream()
+    cld = CldStream(inv)
+
+    # await inv.send_start_cmd(get_sn_int(), str_test_ip, False, inv.mb_first_timeout)
+    cld.send_heartbeat_cb(0)
+    assert cld.ifc.tx_fifo.peek() == heartbeat_ind
+    cld.close()
+
+@pytest.mark.asyncio
+async def test_snd_inv_data(config_tsun_inv1, inverter_ind_msg, inverter_rsp_msg):
+    _ = config_tsun_inv1
+    inv = InvStream()
+    inv.db.set_db_def_value(Register.INVERTER_STATUS, 1)
+    inv.db.set_db_def_value(Register.DETECT_STATUS_1, 2)
+    inv.db.set_db_def_value(Register.VERSION, 'V4.0.10')
+    inv.db.set_db_def_value(Register.GRID_VOLTAGE, 224.8)
+    inv.db.set_db_def_value(Register.GRID_CURRENT, 0.73)
+    inv.db.set_db_def_value(Register.GRID_FREQUENCY, 50.05)
+    assert asyncio.get_running_loop() == inv.mb_timer.loop
+    await inv.send_start_cmd(get_sn_int(), str_test_ip, False, inv.mb_first_timeout)
+    inv.db.set_db_def_value(Register.DATA_UP_INTERVAL, 17)  # set test value
+
+    cld = CldStream(inv)
+    cld.time_ofs = 0x33e447a0
+    cld.last_sync = cld._emu_timestamp() - 60
+    cld.pkt_cnt = 0x802
+    assert cld.data_up_inv == 17  # check test value
+    cld.data_up_inv = 0.1         # speedup test first data msg
+    cld._init_new_client_conn()
+    cld.data_up_inv = 0.5         # timeout for second data msg
+    await asyncio.sleep(0.2)
+    assert cld.ifc.tx_fifo.get() == inverter_ind_msg
+
+    cld.append_msg(inverter_rsp_msg)
+    cld.read()         # read complete msg, and dispatch msg
+
+    assert not cld.header_valid  # must be invalid, since msg was handled and buffer flushed
+    assert cld.msg_count == 1
+    assert cld.header_len==11
+    assert cld.snr == 2070233889
+    assert cld.unique_id == '2070233889'
+    assert cld.msg_recvd[0]['control']==0x1210
+    assert cld.msg_recvd[0]['seq']=='02:02'
+    assert cld.msg_recvd[0]['data_len']==0x0a
+    assert '02b0' == cld.db.get_db_value(Register.SENSOR_LIST, None)
+    assert cld.db.stat['proxy']['Unknown_Msg'] == 0
+
+    cld.close()
+
+@pytest.mark.asyncio
+async def test_rcv_invalid(config_tsun_inv1, inverter_ind_msg, inverter_rsp_msg):
+    _ = config_tsun_inv1
+    inv = InvStream()
+    assert asyncio.get_running_loop() == inv.mb_timer.loop
+    await inv.send_start_cmd(get_sn_int(), str_test_ip, False, inv.mb_first_timeout)
+    inv.db.set_db_def_value(Register.DATA_UP_INTERVAL, 17)  # set test value
+
+    cld = CldStream(inv)
+    cld._init_new_client_conn()
+
+    cld.append_msg(inverter_ind_msg)
+    cld.read()         # read complete msg, and dispatch msg
+
+    assert not cld.header_valid  # must be invalid, since msg was handled and buffer flushed
+    assert cld.msg_count == 1
+    assert cld.header_len==11
+    assert cld.snr == 2070233889
+    assert cld.unique_id == '2070233889'
+    assert cld.msg_recvd[0]['control']==0x4210
+    assert cld.msg_recvd[0]['seq']=='00:01'
+    assert cld.msg_recvd[0]['data_len']==0x199
+    assert '02b0' == cld.db.get_db_value(Register.SENSOR_LIST, None)
+    assert cld.db.stat['proxy']['Unknown_Msg'] == 1
+
+
+    cld.close()