diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ddf9b0..0ec1d47 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- add timeout monitoring for received packets +- parse Modbus values and store them in the database +- add cron task to request the output power every minute +- GEN3PLUS: add MQTT topics to send AT commands to the inverter +- add MQTT topics to send Modbus commands to the inverter - convert data collect interval to minutes - add postfix for rc and dev versions to the version number - change logging level to DEBUG for some logs diff --git a/README.md b/README.md index 1a26179..342a643 100644 --- a/README.md +++ b/README.md @@ -39,12 +39,15 @@ If you use a Pi-hole, you can also store the host entry in the Pi-hole. ## Features -- supports TSUN GEN3 PLUS inverters: TSOL-MS2000, MS1800 and MS1600 -- supports TSUN GEN3 inverters: TSOL-MS800, MS700, MS600, MS400, MS350 and MS300 +- Supports TSUN GEN3 PLUS inverters: TSOL-MS2000, MS1800 and MS1600 +- Supports TSUN GEN3 inverters: TSOL-MS800, MS700, MS600, MS400, MS350 and MS300 - `MQTT` support - `Home-Assistant` auto-discovery support +- `MODBUS` support via MQTT topics +- `AT Command` support via MQTT topics (GEN3PLUS only) +- Faster DataUp interval sends measurement data to the MQTT broker every minute - Self-sufficient island operation without internet -- runs in a non-root Docker Container +- Runs in a non-root Docker Container ## Home Assistant Screenshots diff --git a/app/proxy.svg b/app/proxy.svg index 588835e..cef1e69 100644 --- a/app/proxy.svg +++ b/app/proxy.svg @@ -4,340 +4,372 @@ - - + + G - + A0 - - - -You can stick notes -on diagrams too! + + + +You can stick notes +on diagrams too! A1 - -Singleton + +Singleton A2 - -Mqtt - -<static>ha_restarts -<static>__client -<static>__cb_MqttIsUp - -<async>publish() -<async>close() + +Mqtt + +<static>ha_restarts +<static>__client +<static>__cb_MqttIsUp + +<async>publish() +<async>close() A1->A2 - - - - - -A10 - -Inverter - -cls.db_stat -cls.entity_prfx -cls.discovery_prfx -cls.proxy_node_id -cls.proxy_unique_id -cls.mqtt:Mqtt - - - - -A2->A10 - - - - -A3 - -IterRegistry - - -__iter__ - - - -A4 - -Message - -server_side:bool -header_valid:bool -header_len:unsigned -data_len:unsigned -unique_id -node_id -sug_area -_recv_buffer:bytearray -_send_buffer:bytearray -_forward_buffer:bytearray -db:Infos -new_data:list - -_read():void<abstract> -close():void -inc_counter():void -dec_counter():void - - - -A3->A4 - - - - - -A5 - -Talent - -await_conn_resp_cnt -id_str -contact_name -contact_mail -switch - -msg_contact_info() -msg_ota_update() -msg_get_time() -msg_collector_data() -msg_inverter_data() -msg_unknown() -close() - - - -A4->A5 - - - - - -A6 - -SolarmanV5 - -control -serial -snr -switch - -msg_unknown() -close() - - - -A4->A6 - - - - - -A7 - -ConnectionG3 - -remoteStream:ConnectionG3 - -close() - - - -A5->A7 - - - - - -A8 - -ConnectionG3P - -remoteStream:ConnectionG3P - -close() - - - -A6->A8 - - - - - -A7->A7 - - -0..1 -has + + A11 - -InverterG3 - -__ha_restarts - -async_create_remote() -close() + +Inverter + +cls.db_stat +cls.entity_prfx +cls.discovery_prfx +cls.proxy_node_id +cls.proxy_unique_id +cls.mqtt:Mqtt + - - -A7->A11 - - + + +A2->A11 + - - -A8->A8 - - -0..1 -has + + +A3 + +Modbus + + +build_msg() +recv_req() +recv_resp() +check_crc() - - -A12 - -InverterG3P - -__ha_restarts - -async_create_remote() -close() + + +A4 + +IterRegistry + + +__iter__ - - -A8->A12 - - + + +A5 + +Message + +server_side:bool +header_valid:bool +header_len:unsigned +data_len:unsigned +unique_id +node_id +sug_area +_recv_buffer:bytearray +_send_buffer:bytearray +_forward_buffer:bytearray +db:Infos +new_data:list + +_read():void<abstract> +close():void +inc_counter():void +dec_counter():void + + + +A4->A5 + + + + + +A6 + +Talent + +await_conn_resp_cnt +id_str +contact_name +contact_mail +db:InfosG3 +mb:Modbus +switch + +msg_contact_info() +msg_ota_update() +msg_get_time() +msg_collector_data() +msg_inverter_data() +msg_unknown() +close() + + + +A5->A6 + + + + + +A7 + +SolarmanV5 + +control +serial +snr +db:InfosG3P +mb:Modbus +switch + +msg_unknown() +close() + + + +A5->A7 + + + + + +A6->A3 + + +1 +has + + + +A8 + +ConnectionG3 + +remoteStream:ConnectionG3 + +close() + + + +A6->A8 + + + + + +A7->A3 + + +1 +has A9 - -AsyncStream - -reader -writer -addr -r_addr -l_addr - -<async>server_loop() -<async>client_loop() -<async>loop -disc() -close() -__async_read() -__async_write() -__async_forward() + +ConnectionG3P + +remoteStream:ConnectionG3P + +close() - + -A9->A7 - - +A7->A9 + + - - -A9->A8 - - + + +A8->A8 + + +0..1 +has - - -A10->A11 - - + + +A12 + +InverterG3 + +__ha_restarts + +async_create_remote() +close() - - -A10->A12 - - + + +A8->A12 + + + + + +A9->A9 + + +0..1 +has A13 - -Infos - -stat -new_stat_data -info_dev - -static_init() -dev_value() -inc_counter() -dec_counter() -ha_proxy_conf -ha_conf -update_db -set_db_def_value -get_db_value -ignore_this_device + +InverterG3P + +__ha_restarts + +async_create_remote() +close() + + + +A9->A13 + + + + + +A10 + +AsyncStream + +reader +writer +addr +r_addr +l_addr + +<async>server_loop() +<async>client_loop() +<async>loop +disc() +close() +__async_read() +async_write() +__async_forward() + + + +A10->A8 + + + + + +A10->A9 + + + + + +A11->A12 + + + + + +A11->A13 + + A14 - -InfosG3 - - -ha_confs() -parse() - - - -A13->A14 - - + +Infos + +stat +new_stat_data +info_dev + +static_init() +dev_value() +inc_counter() +dec_counter() +ha_proxy_conf +ha_conf +update_db +set_db_def_value +get_db_value +ignore_this_device A15 - -InfosG3P - - -ha_confs() -parse() + +InfosG3 + + +ha_confs() +parse() - - -A13->A15 - - + + +A14->A15 + + - + + +A16 + +InfosG3P + + +ha_confs() +parse() + + -A14->A5 - - +A14->A16 + + - + A15->A6 - - + + + + + +A16->A7 + + diff --git a/app/proxy.yuml b/app/proxy.yuml index 7f5be21..7514a93 100644 --- a/app/proxy.yuml +++ b/app/proxy.yuml @@ -4,13 +4,15 @@ [note: You can stick notes on diagrams too!{bg:cornsilk}] [Singleton]^[Mqtt|ha_restarts;__client;__cb_MqttIsUp|publish();close()] - +[Modbus||build_msg();recv_req();recv_resp();check_crc()] [IterRegistry||__iter__]^[Message|server_side:bool;header_valid:bool;header_len:unsigned;data_len:unsigned;unique_id;node_id;sug_area;_recv_buffer:bytearray;_send_buffer:bytearray;_forward_buffer:bytearray;db:Infos;new_data:list|_read():void;close():void;inc_counter():void;dec_counter():void] -[Message]^[Talent|await_conn_resp_cnt;id_str;contact_name;contact_mail;switch|msg_contact_info();msg_ota_update();msg_get_time();msg_collector_data();msg_inverter_data();msg_unknown();;close()] -[Message]^[SolarmanV5|control;serial;snr;switch|msg_unknown();;close()] +[Message]^[Talent|await_conn_resp_cnt;id_str;contact_name;contact_mail;db:InfosG3;mb:Modbus;switch|msg_contact_info();msg_ota_update();msg_get_time();msg_collector_data();msg_inverter_data();msg_unknown();;close()] +[Message]^[SolarmanV5|control;serial;snr;db:InfosG3P;mb:Modbus;switch|msg_unknown();;close()] [Talent]^[ConnectionG3|remoteStream:ConnectionG3|close()] +[Talent]has-1>[Modbus] [SolarmanV5]^[ConnectionG3P|remoteStream:ConnectionG3P|close()] -[AsyncStream|reader;writer;addr;r_addr;l_addr|server_loop();client_loop();loop;disc();close();;__async_read();__async_write();__async_forward()]^[ConnectionG3] +[SolarmanV5]has-1>[Modbus] +[AsyncStream|reader;writer;addr;r_addr;l_addr|server_loop();client_loop();loop;disc();close();;__async_read();async_write();__async_forward()]^[ConnectionG3] [AsyncStream]^[ConnectionG3P] [Inverter|cls.db_stat;cls.entity_prfx;cls.discovery_prfx;cls.proxy_node_id;cls.proxy_unique_id;cls.mqtt:Mqtt|]^[InverterG3|__ha_restarts|async_create_remote();;close()] [Inverter]^[InverterG3P|__ha_restarts|async_create_remote();;close()] diff --git a/app/src/async_stream.py b/app/src/async_stream.py index 6c1136c..28873e8 100644 --- a/app/src/async_stream.py +++ b/app/src/async_stream.py @@ -61,7 +61,7 @@ class AsyncStream(): await self.__async_read() if self.unique_id: - await self.__async_write() + await self.async_write() await self.__async_forward() await self.async_publ_mqtt() @@ -100,9 +100,9 @@ class AsyncStream(): else: raise RuntimeError("Peer closed.") - async def __async_write(self) -> None: + async def async_write(self, headline='Transmit to ') -> None: if self._send_buffer: - hex_dump_memory(logging.INFO, f'Transmit to {self.addr}:', + hex_dump_memory(logging.INFO, f'{headline}{self.addr}:', self._send_buffer, len(self._send_buffer)) self.writer.write(self._send_buffer) await self.writer.drain() @@ -114,7 +114,7 @@ class AsyncStream(): await self.async_create_remote() if self.remoteStream: if self.remoteStream._init_new_client_conn(): - await self.remoteStream.__async_write() + await self.remoteStream.async_write() if self.remoteStream: self.remoteStream._update_header(self._forward_buffer) diff --git a/app/src/gen3/infos_g3.py b/app/src/gen3/infos_g3.py index 7e45634..d3fb987 100644 --- a/app/src/gen3/infos_g3.py +++ b/app/src/gen3/infos_g3.py @@ -161,7 +161,7 @@ class InfosG3(Infos): update = False name = str(f'info-id.0x{addr:x}') - self.tracer.log(level, f'GEN3: {name} : {result}{unit}' - f' update: {update}') + if update: + self.tracer.log(level, f'GEN3: {name} : {result}{unit}') i += 1 diff --git a/app/src/gen3/talent.py b/app/src/gen3/talent.py index 46302ac..45eb081 100644 --- a/app/src/gen3/talent.py +++ b/app/src/gen3/talent.py @@ -5,10 +5,12 @@ from datetime import datetime if __name__ == "app.src.gen3.talent": from app.src.messages import hex_dump_memory, Message + from app.src.modbus import Modbus from app.src.config import Config from app.src.gen3.infos_g3 import InfosG3 else: # pragma: no cover from messages import hex_dump_memory, Message + from modbus import Modbus from config import Config from gen3.infos_g3 import InfosG3 @@ -41,13 +43,20 @@ class Talent(Message): self.contact_name = b'' self.contact_mail = b'' self.db = InfosG3() + self.mb = Modbus() + self.forward_modbus_resp = False + self.closed = False self.switch = { 0x00: self.msg_contact_info, 0x13: self.msg_ota_update, 0x22: self.msg_get_time, 0x71: self.msg_collector_data, + # 0x76: + 0x77: self.msg_modbus, + # 0x78: 0x04: self.msg_inverter_data, } + self.modbus_elms = 0 # for unit tests ''' Our puplic methods @@ -58,6 +67,7 @@ class Talent(Message): # 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.closed = True def __set_serial_no(self, serial_no: str): @@ -115,6 +125,19 @@ class Talent(Message): f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}') return + async def send_modbus_cmd(self, func, addr, val) -> None: + self.forward_modbus_resp = False + self.__build_header(0x70, 0x77) + self._send_buffer += b'\x00\x01\xa3\x28' # fixme + modbus_msg = self.mb.build_msg(Modbus.INV_ADDR, func, addr, val) + self._send_buffer += struct.pack('!B', len(modbus_msg)) + self._send_buffer += modbus_msg + self.__finish_send_msg() + try: + await self.async_write('Send Modbus Command:') + except Exception: + self._send_buffer = bytearray(0) + def _init_new_client_conn(self) -> bool: contact_name = self.contact_name contact_mail = self.contact_mail @@ -190,11 +213,13 @@ class Talent(Message): self.header_valid = True return - def __build_header(self, ctrl) -> None: + def __build_header(self, ctrl, msg_id=None) -> None: + if not msg_id: + msg_id = self.msg_id self.send_msg_ofs = len(self._send_buffer) self._send_buffer += struct.pack(f'!l{len(self.id_str)+1}pBB', - 0, self.id_str, ctrl, self.msg_id) - fnc = self.switch.get(self.msg_id, self.msg_unknown) + 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}') @@ -348,6 +373,40 @@ class Talent(Message): self.inc_counter('Unknown_Ctrl') self.forward(self._recv_buffer, self.header_len+self.data_len) + def parse_modbus_header(self): + + msg_hdr_len = 5 + + result = struct.unpack_from('!lBB', self._recv_buffer, + self.header_len) + modbus_len = result[1] + # logger.debug(f'Ref: {result[0]}') + # logger.debug(f'Modbus MsgLen: {modbus_len} Func:{result[2]}') + return msg_hdr_len, modbus_len + + def msg_modbus(self): + hdr_len, modbus_len = self.parse_modbus_header() + + if self.ctrl.is_req(): + self.forward_modbus_resp = True + self.inc_counter('Modbus_Command') + elif self.ctrl.is_ind(): + # logger.debug(f'Modbus Ind MsgLen: {modbus_len}') + self.modbus_elms = 0 + for key, update in self.mb.recv_resp(self.db, self._recv_buffer[ + self.header_len + hdr_len:self.header_len+self.data_len], + self.node_id): + if update: + self.new_data[key] = True + self.modbus_elms += 1 + + if not self.forward_modbus_resp: + return + else: + logger.warning('Unknown Ctrl') + self.inc_counter('Unknown_Ctrl') + self.forward(self._recv_buffer, self.header_len+self.data_len) + def msg_unknown(self): logger.warning(f"Unknow Msg: ID:{self.msg_id}") self.inc_counter('Unknown_Msg') diff --git a/app/src/gen3plus/infos_g3p.py b/app/src/gen3plus/infos_g3p.py index 0191d04..ed8d9bd 100644 --- a/app/src/gen3plus/infos_g3p.py +++ b/app/src/gen3plus/infos_g3p.py @@ -19,7 +19,7 @@ class RegisterMap: 0x4102001a: {'reg': Register.HEARTBEAT_INTERVAL, 'fmt': ' we get a memory leak self.switch.clear() + self.closed = True def __set_serial_no(self, snr: int): serial_no = str(snr) @@ -293,16 +301,49 @@ class SolarmanV5(Message): self._heartbeat()) self.__finish_send_msg() - def send_at_cmd(self, AT_cmd: str) -> None: + async def send_modbus_cmd(self, func, addr, val) -> None: + self.forward_modbus_resp = False self.__build_header(0x4510) - self._send_buffer += struct.pack(f' None: + self.__build_header(0x4510) + self._send_buffer += struct.pack(f'> 8 @@ -313,21 +354,7 @@ class SolarmanV5(Message): self.new_data[key] = True if inv_update: - db = self.db - MaxPow = db.get_db_value(Register.MAX_DESIGNED_POWER, 0) - Rated = db.get_db_value(Register.RATED_POWER, 0) - Model = None - if MaxPow == 2000: - if Rated == 800 or Rated == 600: - Model = f'TSOL-MS{MaxPow}({Rated})' - else: - Model = f'TSOL-MS{MaxPow}' - elif MaxPow == 1800 or MaxPow == 1600: - Model = f'TSOL-MS{MaxPow}' - if Model: - logger.info(f'Model: {Model}') - self.db.set_db_def_value(Register.EQUIPMENT_MODEL, Model) - + self.__build_model_name() ''' Message handler methods ''' @@ -390,11 +417,43 @@ class SolarmanV5(Message): data = self._recv_buffer[self.header_len:] result = struct.unpack_from(' 4: + # logger.info(f'first byte modbus:{data[14]}') + inv_update = False + for key, update in self.mb.recv_resp(self.db, data[14:-2], + self.node_id): + if update: + if key == 'inverter': + inv_update = True + self.new_data[key] = True + + if inv_update: + self.__build_model_name() + + if not self.forward_modbus_resp: + return + self.__forward_msg() + def msg_hbeat_ind(self): data = self._recv_buffer[self.header_len:] result = struct.unpack_from('>12)}.{(result>>8)&0xf}.{(result>>4)&0xf}{result&0xf}'"}, # noqa: E501 + 0x3009: {'reg': Register.GRID_VOLTAGE, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x300a: {'reg': Register.GRID_CURRENT, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x300b: {'reg': Register.GRID_FREQUENCY, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x300c: {'reg': Register.INVERTER_TEMP, 'fmt': '!H', 'eval': 'result-40'}, # noqa: E501 + # 0x300d + 0x300e: {'reg': Register.RATED_POWER, 'fmt': '!H', 'ratio': 1}, # noqa: E501 + 0x300f: {'reg': Register.OUTPUT_POWER, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3010: {'reg': Register.PV1_VOLTAGE, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3011: {'reg': Register.PV1_CURRENT, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x3012: {'reg': Register.PV1_POWER, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3013: {'reg': Register.PV2_VOLTAGE, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3014: {'reg': Register.PV2_CURRENT, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x3015: {'reg': Register.PV2_POWER, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3016: {'reg': Register.PV3_VOLTAGE, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3017: {'reg': Register.PV3_CURRENT, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x3018: {'reg': Register.PV3_POWER, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x3019: {'reg': Register.PV4_VOLTAGE, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x301a: {'reg': Register.PV4_CURRENT, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + 0x301b: {'reg': Register.PV4_POWER, 'fmt': '!H', 'ratio': 0.1}, # noqa: E501 + 0x301c: {'reg': Register.DAILY_GENERATION, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + # 0x301d: {'reg': Register.TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 + 0x301f: {'reg': Register.PV1_DAILY_GENERATION, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + # 0x3020: {'reg': Register.PV1_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 + 0x3022: {'reg': Register.PV2_DAILY_GENERATION, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + # 0x3023: {'reg': Register.PV2_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 + 0x3025: {'reg': Register.PV3_DAILY_GENERATION, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + # 0x3026: {'reg': Register.PV3_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 + 0x3028: {'reg': Register.PV4_DAILY_GENERATION, 'fmt': '!H', 'ratio': 0.01}, # noqa: E501 + # 0x3029: {'reg': Register.PV4_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501 + } + + def __init__(self): + if not len(self.__crc_tab): + self.__build_crc_tab(CRC_POLY) + self.last_fcode = 0 + self.last_len = 0 + self.last_reg = 0 + self.err = 0 + + def build_msg(self, addr, func, reg, val): + msg = struct.pack('>BBHH', addr, func, reg, val) + msg += struct.pack(' bool: + # logging.info(f'recv_req: first byte modbus:{buf[0]} len:{len(buf)}') + if not self.check_crc(buf): + self.err = 1 + logging.error('Modbus: CRC error') + return False + if buf[0] != self.INV_ADDR: + self.err = 2 + logging.info(f'Modbus: Wrong addr{buf[0]}') + return False + res = struct.unpack_from('>BHH', buf, 1) + self.last_fcode = res[0] + self.last_reg = res[1] + self.last_len = res[2] + self.err = 0 + return True + + def recv_resp(self, info_db, buf: bytearray, node_id: str) -> \ + Generator[tuple[str, bool], None, None]: + # logging.info(f'recv_resp: first byte modbus:{buf[0]} len:{len(buf)}') + if not self.check_crc(buf): + logging.error('Modbus: CRC error') + self.err = 1 + return + if buf[0] != self.INV_ADDR: + logging.info(f'Modbus: Wrong addr {buf[0]}') + self.err = 2 + return + if buf[1] != self.last_fcode: + logging.info(f'Modbus: Wrong fcode {buf[1]} != {self.last_fcode}') + self.err = 3 + return + elmlen = buf[2] >> 1 + if elmlen != self.last_len: + logging.info(f'Modbus: len error {elmlen} != {self.last_len}') + self.err = 4 + return + self.err = 0 + + for i in range(0, elmlen): + val = struct.unpack_from('>H', buf, 3+2*i) + addr = self.last_reg+i + # logging.info(f'Modbus: 0x{addr:04x}: {val[0]}') + if addr in self.map: + row = self.map[addr] + info_id = row['reg'] + result = val[0] + # fmt = row['fmt'] + # res = struct.unpack_from(fmt, buf, addr) + # result = res[0] + + if 'eval' in row: + result = eval(row['eval']) + if 'ratio' in row: + result = round(result * row['ratio'], 2) + + keys, level, unit, must_incr = info_db._key_obj(info_id) + + if keys: + name, update = info_db.update_db(keys, must_incr, result) + yield keys[0], update + else: + name = str(f'info-id.0x{addr:x}') + update = False + if update: + info_db.tracer.log(level, + f'MODBUS[{node_id}]: {name} : {result}' + f'{unit}') + + def check_crc(self, msg) -> bool: + return 0 == self.__calc_crc(msg) + + def __calc_crc(self, buffer: bytes) -> int: + crc = CRC_INIT + + for cur in buffer: + crc = (crc >> 8) ^ self.__crc_tab[(crc ^ cur) & 0xFF] + return crc + + def __build_crc_tab(self, poly) -> None: + for index in range(256): + data = index << 1 + crc = 0 + for _ in range(8, 0, -1): + data >>= 1 + if (data ^ crc) & 1: + crc = (crc >> 1) ^ poly + else: + crc >>= 1 + self.__crc_tab.append(crc) diff --git a/app/src/mqtt.py b/app/src/mqtt.py index 5b2de02..3469201 100644 --- a/app/src/mqtt.py +++ b/app/src/mqtt.py @@ -1,22 +1,15 @@ import asyncio import logging import aiomqtt +import traceback +from modbus import Modbus +from messages import Message from config import Config +from singleton import Singleton logger_mqtt = logging.getLogger('mqtt') -class Singleton(type): - _instances = {} - - def __call__(cls, *args, **kwargs): - logger_mqtt.debug('singleton: __call__') - if cls not in cls._instances: - cls._instances[cls] = super(Singleton, - cls).__call__(*args, **kwargs) - return cls._instances[cls] - - class Mqtt(metaclass=Singleton): __client = None __cb_MqttIsUp = None @@ -65,6 +58,12 @@ class Mqtt(metaclass=Singleton): password=mqtt['passwd']) interval = 5 # Seconds + ha_status_topic = f"{ha['auto_conf_prefix']}/status" + mb_rated_topic = "tsun/+/rated_load" # fixme + mb_reads_topic = "tsun/+/modbus_read_regs" # fixme + mb_inputs_topic = "tsun/+/modbus_read_inputs" # fixme + mb_at_cmd_topic = "tsun/+/at_cmd" # fixme + while True: try: async with self.__client: @@ -74,16 +73,36 @@ class Mqtt(metaclass=Singleton): await self.__cb_MqttIsUp() # async with self.__client.messages() as messages: - await self.__client.subscribe( - f"{ha['auto_conf_prefix']}" - "/status") + await self.__client.subscribe(ha_status_topic) + await self.__client.subscribe(mb_rated_topic) + await self.__client.subscribe(mb_reads_topic) + await self.__client.subscribe(mb_inputs_topic) + await self.__client.subscribe(mb_at_cmd_topic) + async for message in self.__client.messages: - status = message.payload.decode("UTF-8") - logger_mqtt.info('Home-Assistant Status:' - f' {status}') - if status == 'online': - self.ha_restarts += 1 - await self.__cb_MqttIsUp() + if message.topic.matches(ha_status_topic): + status = message.payload.decode("UTF-8") + logger_mqtt.info('Home-Assistant Status:' + f' {status}') + if status == 'online': + self.ha_restarts += 1 + await self.__cb_MqttIsUp() + + if message.topic.matches(mb_rated_topic): + await self.modbus_cmd(message, + Modbus.WRITE_SINGLE_REG, + 1, 0x2008) + + if message.topic.matches(mb_reads_topic): + await self.modbus_cmd(message, + Modbus.READ_REGS, 2) + + if message.topic.matches(mb_inputs_topic): + await self.modbus_cmd(message, + Modbus.READ_INPUTS, 2) + + if message.topic.matches(mb_at_cmd_topic): + await self.at_cmd(message) except aiomqtt.MqttError: if Config.is_default('mqtt'): @@ -101,3 +120,52 @@ class Mqtt(metaclass=Singleton): logger_mqtt.debug("MQTT task cancelled") self.__client = None return + except Exception: + # self.inc_counter('SW_Exception') # fixme + logger_mqtt.error( + f"Exception:\n" + f"{traceback.format_exc()}") + + def each_inverter(self, message, func_name: str): + topic = str(message.topic) + node_id = topic.split('/')[1] + '/' + for m in Message: + if m.server_side and not m.closed and (m.node_id == node_id): + logger_mqtt.debug(f'Found: {node_id}') + fnc = getattr(m, func_name, None) + if callable(fnc): + yield fnc + else: + logger_mqtt.warning(f'Cmd not supported by: {node_id}') + + else: + logger_mqtt.warning(f'Node_id: {node_id} not found') + + async def modbus_cmd(self, message, func, params=0, addr=0, val=0): + topic = str(message.topic) + node_id = topic.split('/')[1] + '/' + # refactor into a loop over a table + payload = message.payload.decode("UTF-8") + logger_mqtt.info(f'InvCnf: {node_id}:{payload}') + for m in Message: + if m.server_side and not m.closed and (m.node_id == node_id): + logger_mqtt.info(f'Found: {node_id}') + fnc = getattr(m, "send_modbus_cmd", None) + res = payload.split(',') + if params != len(res): + logger_mqtt.error(f'Parameter expected: {params}, ' + f'got: {len(res)}') + return + + if callable(fnc): + if params == 1: + val = int(payload) + elif params == 2: + addr = int(res[0], base=16) + val = int(res[1]) # lenght + await fnc(func, addr, val) + + async def at_cmd(self, message): + payload = message.payload.decode("UTF-8") + for fnc in self.each_inverter(message, "send_at_cmd"): + await fnc(payload) diff --git a/app/src/scheduler.py b/app/src/scheduler.py index b5d238d..a1e763b 100644 --- a/app/src/scheduler.py +++ b/app/src/scheduler.py @@ -3,6 +3,8 @@ import json from mqtt import Mqtt from aiocron import crontab from infos import ClrAtMidnight +from modbus import Modbus +from messages import Message logger_mqtt = logging.getLogger('mqtt') @@ -17,7 +19,9 @@ class Schedule: cls.mqtt = Mqtt(None) crontab('0 0 * * *', func=cls.atmidnight, start=True) - # crontab('*/5 * * * *', func=cls.atmidnight, start=True) + + # every minute + crontab('* * * * *', func=cls.regular_modbus_cmds, start=True) @classmethod async def atmidnight(cls) -> None: @@ -28,3 +32,12 @@ class Schedule: logger_mqtt.debug(f'{key}: {data}') data_json = json.dumps(data) await cls.mqtt.publish(f"{key}", data_json) + + @classmethod + async def regular_modbus_cmds(cls): + # logging.info("Regular Modbus requests") + for m in Message: + if m.server_side: + fnc = getattr(m, "send_modbus_cmd", None) + if callable(fnc): + await fnc(Modbus.READ_REGS, 0x3008, 20) diff --git a/app/src/singleton.py b/app/src/singleton.py new file mode 100644 index 0000000..48778b9 --- /dev/null +++ b/app/src/singleton.py @@ -0,0 +1,9 @@ +class Singleton(type): + _instances = {} + + def __call__(cls, *args, **kwargs): + # logger_mqtt.debug('singleton: __call__') + if cls not in cls._instances: + cls._instances[cls] = super(Singleton, + cls).__call__(*args, **kwargs) + return cls._instances[cls] diff --git a/app/tests/test_infos.py b/app/tests/test_infos.py index d3b542e..c3e6ddf 100644 --- a/app/tests/test_infos.py +++ b/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}}) + 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, "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}}) + 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, "Modbus_Command": 0}}) val = i.dev_value(Register.INVERTER_CNT) assert val == 1 diff --git a/app/tests/test_infos_g3p.py b/app/tests/test_infos_g3p.py index 3878dff..4af9cb2 100644 --- a/app/tests/test_infos_g3p.py +++ b/app/tests/test_infos_g3p.py @@ -70,7 +70,7 @@ def test_parse_4110(DeviceData: bytes): pass assert json.dumps(i.db) == json.dumps({ - 'controller': {"Data_Up_Interval": 300, "Collect_Interval": 1, "Heartbeat_Interval": 120, "Signal_Strength": 100, "IP_Adress": "192.168.80.49"}, + 'controller': {"Data_Up_Interval": 300, "Collect_Interval": 1, "Heartbeat_Interval": 120, "Signal_Strength": 100, "IP_Address": "192.168.80.49"}, 'collector': {"Chip_Model": "LSW5BLE_17_02B0_1.05", "Collector_Fw_Version": "V1.1.00.0B"}, }) diff --git a/app/tests/test_modbus.py b/app/tests/test_modbus.py new file mode 100644 index 0000000..b1764e9 --- /dev/null +++ b/app/tests/test_modbus.py @@ -0,0 +1,41 @@ +# test_with_pytest.py +# import pytest, logging +from app.src.modbus import Modbus +from app.src.infos import Infos + +class TestHelper(Modbus): + def __init__(self): + super().__init__() + self.db = Infos() + +def test_modbus_crc(): + mb = Modbus() + assert 0x0b02 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04') + assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b') + assert mb.check_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b') + + assert 0xc803 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00') + assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8') + assert mb.check_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8') + +def test_build_modbus_pdu(): + mb = Modbus() + pdu = mb.build_msg(1,6,0x2000,0x12) + assert pdu == b'\x01\x06\x20\x00\x00\x12\x02\x07' + assert mb.check_crc(pdu) + +def test_build_recv(): + mb = TestHelper() + pdu = mb.build_msg(1,3,0x300e,0x2) + assert pdu == b'\x01\x03\x30\x0e\x00\x02\xaa\xc8' + assert mb.check_crc(pdu) + call = 0 + for key, update in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'): + if key == 'grid': + assert update == True + elif key == 'inverter': + assert update == True + else: + assert False + call += 1 + assert 2 == call diff --git a/app/tests/test_solarman.py b/app/tests/test_solarman.py index 48f5509..e699ae7 100644 --- a/app/tests/test_solarman.py +++ b/app/tests/test_solarman.py @@ -6,6 +6,9 @@ from app.src.gen3plus.solarman_v5 import SolarmanV5 from app.src.config import Config from app.src.infos import Infos, Register + +pytest_plugins = ('pytest_asyncio',) + # initialize the proxy statistics Infos.static_init() @@ -54,6 +57,9 @@ class MemoryStream(SolarmanV5): pass return copied_bytes + async def async_write(self, headline=''): + pass + def _SolarmanV5__flush_recv_msg(self) -> None: super()._SolarmanV5__flush_recv_msg() self.msg_count += 1 @@ -725,7 +731,7 @@ def test_device_rsp(ConfigTsunInv1, DeviceRspMsg): assert m.data_len == 0x0a assert m._recv_buffer==b'' assert m._send_buffer==b'' - assert m._forward_buffer==b'' # DeviceRspMsg + assert m._forward_buffer==DeviceRspMsg assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 m.close() @@ -743,7 +749,7 @@ def test_inverter_rsp(ConfigTsunInv1, InverterRspMsg): assert m.data_len == 0x0a assert m._recv_buffer==b'' assert m._send_buffer==b'' - assert m._forward_buffer==b'' # InverterRspMsg + assert m._forward_buffer==InverterRspMsg assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 m.close() @@ -779,7 +785,7 @@ def test_heartbeat_rsp(ConfigTsunInv1, HeartbeatRspMsg): assert m.data_len == 0x0a assert m._recv_buffer==b'' assert m._send_buffer==b'' - assert m._forward_buffer==b'' # HeartbeatRspMsg + assert m._forward_buffer==HeartbeatRspMsg assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 m.close() @@ -820,7 +826,7 @@ def test_sync_start_rsp(ConfigTsunInv1, SyncStartRspMsg): assert m.data_len == 0x0a assert m._recv_buffer==b'' assert m._send_buffer==b'' - assert m._forward_buffer==b'' # HeartbeatRspMsg + assert m._forward_buffer==SyncStartRspMsg assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 m.close() @@ -856,7 +862,7 @@ def test_sync_end_rsp(ConfigTsunInv1, SyncEndRspMsg): assert m.data_len == 0x0a assert m._recv_buffer==b'' assert m._send_buffer==b'' - assert m._forward_buffer==b'' # HeartbeatRspMsg + assert m._forward_buffer==SyncEndRspMsg assert m.db.stat['proxy']['Invalid_Msg_Format'] == 0 m.close() @@ -942,7 +948,8 @@ def test_build_logger_modell(ConfigTsunAllowAll, DeviceIndMsg): assert 'V1.1.00.0B' == m.db.get_db_value(Register.COLLECTOR_FW_VERSION, 0).rstrip('\00') m.close() -def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, AtCommandIndMsg): +@pytest.mark.asyncio +async def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, AtCommandIndMsg): ConfigTsunAllowAll m = MemoryStream(DeviceIndMsg, (0,), True) m.read() @@ -954,7 +961,7 @@ def test_AT_cmd(ConfigTsunAllowAll, DeviceIndMsg, DeviceRspMsg, AtCommandIndMsg) m._send_buffer = bytearray(0) # clear send buffer for next test m._forward_buffer = bytearray(0) # clear send buffer for next test - m.send_at_cmd('AT+TIME=214028,1,60,120') + await m.send_at_cmd('AT+TIME=214028,1,60,120') assert m._recv_buffer==b'' assert m._send_buffer==AtCommandIndMsg assert m._forward_buffer==b'' diff --git a/app/tests/test_talent.py b/app/tests/test_talent.py index 89fd420..cc9ab85 100644 --- a/app/tests/test_talent.py +++ b/app/tests/test_talent.py @@ -3,6 +3,10 @@ import pytest, logging from app.src.gen3.talent import Talent, Control from app.src.config import Config from app.src.infos import Infos +from app.src.modbus import Modbus + + +pytest_plugins = ('pytest_asyncio',) # initialize the proxy statistics Infos.static_init() @@ -19,6 +23,7 @@ class MemoryStream(Talent): self.__chunk_idx = 0 self.msg_count = 0 self.addr = 'Test: SrvSide' + self.send_msg_ofs = 0 def append_msg(self, msg): self.__msg += msg @@ -50,6 +55,10 @@ class MemoryStream(Talent): self.msg_count += 1 return + async def async_write(self, headline=''): + pass + + @pytest.fixture def MsgContactInfo(): # Contact Info message @@ -170,6 +179,35 @@ def MsgOtaAck(): # Over the air update rewuest from tsun cloud def MsgOtaInvalid(): # Get Time Request message return b'\x00\x00\x00\x14\x10R170000000000001\x99\x13\x01' +@pytest.fixture +def MsgModbusCmd(): + msg = b'\x00\x00\x00\x20\x10R170000000000001' + msg += b'\x70\x77\x00\x01\xa3\x28\x08\x01\x06\x20\x08' + msg += b'\x00\x00\x03\xc8' + return msg + +@pytest.fixture +def MsgModbusRsp(): + msg = b'\x00\x00\x00\x20\x10R170000000000001' + msg += b'\x91\x77\x17\x18\x19\x1a\x08\x01\x06\x20\x08' + msg += b'\x00\x00\x03\xc8' + return msg + +@pytest.fixture +def MsgModbusInv(): + msg = b'\x00\x00\x00\x20\x10R170000000000001' + msg += b'\x99\x77\x17\x18\x19\x1a\x08\x01\x06\x20\x08' + msg += b'\x00\x00\x03\xc8' + return msg + +@pytest.fixture +def MsgModbusResp20(): + msg = b'\x00\x00\x00\x45\x10R170000000000001' + msg += b'\x91\x77\x17\x18\x19\x1a\x2d\x01\x03\x28\x51' + msg += b'\x09\x08\xd3\x00\x29\x13\x87\x00\x3e\x00\x00\x01\x2c\x03\xb4\x00' + msg += b'\x08\x00\x00\x00\x00\x01\x59\x01\x21\x03\xe6\x00\x00\x00\x00\x00' + msg += b'\x00\x00\x00\x00\x00\x00\x00\xdb\x6b' + return msg def test_read_message(MsgContactInfo): m = MemoryStream(MsgContactInfo, (0,)) @@ -740,3 +778,123 @@ def test_proxy_counter(): assert Infos.new_stat_data == {'proxy': True} assert 0 == m.db.stat['proxy']['Unknown_Msg'] m.close() + +def test_msg_modbus_req(ConfigTsunInv1, MsgModbusCmd): + ConfigTsunInv1 + m = MemoryStream(MsgModbusCmd, (0,), False) + m.db.stat['proxy']['Unknown_Ctrl'] = 0 + m.db.stat['proxy']['Modbus_Command'] = 0 + m.read() # read complete msg, and dispatch msg + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + assert m.id_str == b"R170000000000001" + assert m.unique_id == 'R170000000000001' + assert int(m.ctrl)==112 + assert m.msg_id==119 + assert m.header_len==23 + assert m.data_len==13 + assert m._forward_buffer==MsgModbusCmd + assert m._send_buffer==b'' + assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 + assert m.db.stat['proxy']['Modbus_Command'] == 1 + m.close() + +def test_msg_modbus_rsp1(ConfigTsunInv1, MsgModbusRsp): + ConfigTsunInv1 + m = MemoryStream(MsgModbusRsp, (0,), False) + m.db.stat['proxy']['Unknown_Ctrl'] = 0 + m.db.stat['proxy']['Modbus_Command'] = 0 + m.forward_modbus_resp = False + m.read() # read complete msg, and dispatch msg + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + assert m.id_str == b"R170000000000001" + assert m.unique_id == 'R170000000000001' + assert int(m.ctrl)==145 + assert m.msg_id==119 + assert m.header_len==23 + assert m.data_len==13 + assert m._forward_buffer==b'' + assert m._send_buffer==b'' + assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 + assert m.db.stat['proxy']['Modbus_Command'] == 0 + m.close() + +def test_msg_modbus_rsp2(ConfigTsunInv1, MsgModbusRsp): + ConfigTsunInv1 + m = MemoryStream(MsgModbusRsp, (0,), False) + m.db.stat['proxy']['Unknown_Ctrl'] = 0 + m.db.stat['proxy']['Modbus_Command'] = 0 + m.forward_modbus_resp = True + m.read() # read complete msg, and dispatch msg + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + assert m.id_str == b"R170000000000001" + assert m.unique_id == 'R170000000000001' + assert int(m.ctrl)==145 + assert m.msg_id==119 + assert m.header_len==23 + assert m.data_len==13 + assert m._forward_buffer==MsgModbusRsp + assert m._send_buffer==b'' + assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 + assert m.db.stat['proxy']['Modbus_Command'] == 0 + m.close() + +def test_msg_modbus_invalid(ConfigTsunInv1, MsgModbusInv): + ConfigTsunInv1 + m = MemoryStream(MsgModbusInv, (0,), False) + m.db.stat['proxy']['Unknown_Ctrl'] = 0 + m.db.stat['proxy']['Modbus_Command'] = 0 + m.read() # read complete msg, and dispatch msg + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + assert m.id_str == b"R170000000000001" + assert m.unique_id == 'R170000000000001' + assert int(m.ctrl)==153 + assert m.msg_id==119 + assert m.header_len==23 + assert m.data_len==13 + assert m._forward_buffer==MsgModbusInv + assert m._send_buffer==b'' + assert m.db.stat['proxy']['Unknown_Ctrl'] == 1 + assert m.db.stat['proxy']['Modbus_Command'] == 0 + m.close() + +def test_msg_modbus_fragment(ConfigTsunInv1, MsgModbusResp20): + ConfigTsunInv1 + # receive more bytes than expected (7 bytes from the next msg) + m = MemoryStream(MsgModbusResp20+b'\x00\x00\x00\x45\x10\x52\x31', (0,)) + m.db.stat['proxy']['Unknown_Ctrl'] = 0 + m.db.stat['proxy']['Modbus_Command'] = 0 + m.forward_modbus_resp = True + m.mb.last_fcode = 3 + m.mb.last_len = 20 + m.mb.last_reg = 0x3008 + m.read() # read complete msg, and dispatch msg + assert not m.header_valid # must be invalid, since msg was handled and buffer flushed + assert m.msg_count == 1 + assert m.id_str == b"R170000000000001" + assert m.unique_id == 'R170000000000001' + assert int(m.ctrl)==0x91 + assert m.msg_id==119 + assert m.header_len==23 + assert m.data_len==50 + assert m._forward_buffer==MsgModbusResp20 + assert m._send_buffer==b'' + assert m.mb.err == 0 + assert m.modbus_elms == 20-1 # register 0x300d is unknown, so one value can't be mapped + assert m.db.stat['proxy']['Unknown_Ctrl'] == 0 + assert m.db.stat['proxy']['Modbus_Command'] == 0 + m.close() + +@pytest.mark.asyncio +async def test_msg_build_modbus_req(ConfigTsunInv1, MsgModbusCmd): + ConfigTsunInv1 + m = MemoryStream(b'', (0,), False) + m.id_str = b"R170000000000001" + await m.send_modbus_cmd(Modbus.WRITE_SINGLE_REG, 0x2008, 0) + assert 0==m.send_msg_ofs + assert m._forward_buffer==b'' + assert m._send_buffer==MsgModbusCmd + m.close() diff --git a/system_tests/test_tcp_socket.py b/system_tests/test_tcp_socket.py index 606ea68..f01a0a0 100644 --- a/system_tests/test_tcp_socket.py +++ b/system_tests/test_tcp_socket.py @@ -224,7 +224,7 @@ def test_send_inv_data(ClientConnection, MsgTimeStampReq, MsgTimeStampResp, MsgI data = s.recv(1024) except TimeoutError: pass - # time.sleep(32.5) + time.sleep(32.5) # assert data == MsgTimeStampResp try: s.sendall(MsgInvData)