sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

increase test coverage

Browse Source
This commit is contained in:
Stefan Allius
2024-05-20 16:53:26 +02:00
parent 6ef6f4cd34
commit eff3e7558b
4 changed files with 190 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/src/gen3/talent.py
View File

@@ -122,7 +122,7 @@ class Talent(Message):
f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
return
def send_modbus_cb(self, modbus_pdu: bytearray, retrans: bool):
def send_modbus_cb(self, modbus_pdu: bytearray, state: str):
if self.state != self.STATE_UP:
return
@@ -131,12 +131,8 @@ class Talent(Message):
self._send_buffer += struct.pack('!B', len(modbus_pdu))
self._send_buffer += modbus_pdu
self.__finish_send_msg()
if retrans:
cmd = 'Retrans'
else:
cmd = 'Command'
hex_dump_memory(logging.INFO, f'Send Modbus {cmd}:{self.addr}:',
hex_dump_memory(logging.INFO, f'Send Modbus {state}:{self.addr}:',
self._send_buffer, len(self._send_buffer))
self.writer.write(self._send_buffer)
self._send_buffer = bytearray(0) # self._send_buffer[sent:]

8
app/src/gen3plus/solarman_v5.py
View File

@@ -301,7 +301,7 @@ class SolarmanV5(Message):
self._heartbeat())
self.__finish_send_msg()
def send_modbus_cb(self, pdu: bytearray, retrans: bool):
def send_modbus_cb(self, pdu: bytearray, state: str):
if self.state != self.STATE_UP:
return
self.__build_header(0x4510)
@@ -309,11 +309,7 @@ class SolarmanV5(Message):
0x2b0, 0, 0, 0)
self._send_buffer += pdu
self.__finish_send_msg()
if retrans:
cmd = 'Retrans'
else:
cmd = 'Command'
hex_dump_memory(logging.INFO, f'Send Modbus {cmd}:{self.addr}:',
hex_dump_memory(logging.INFO, f'Send Modbus {state}:{self.addr}:',
self._send_buffer, len(self._send_buffer))
self.writer.write(self._send_buffer)
self._send_buffer = bytearray(0) # self._send_buffer[sent:]

45
app/src/modbus.py
View File

@@ -1,3 +1,16 @@
'''MODBUS module for TSUN inverter support
TSUN uses the MODBUS in the RTU transmission mode over serial line.
see: https://modbus.org/docs/Modbus_Application_Protocol_V1_1b3.pdf
see: https://modbus.org/docs/Modbus_over_serial_line_V1_02.pdf
A Modbus PDU consists of: 'Function-Code' + 'Data'
A Modbus RTU message consists of: 'Addr' + 'Modbus-PDU' + 'CRC-16'
The inverter is a MODBUS server and the proxy the MODBUS client.
The 16-bit CRC is known as CRC-16-ANSI(reverse)
see: https://en.wikipedia.org/wiki/Computation_of_cyclic_redundancy_checks
'''
import struct
import logging
import asyncio
@@ -8,30 +21,21 @@ if __name__ == "app.src.modbus":
else: # pragma: no cover
from infos import Register
#######
# TSUN uses the Modbus in the RTU transmission mode.
# see: https://modbus.org/docs/Modbus_over_serial_line_V1_02.pdf
#
# A Modbus PDU consists of: 'Function-Code' + 'Data'
# A Modbus RTU message consists of: 'Addr' + 'Modbus-PDU' + 'CRC-16'
#
# The 16-bit CRC is known as CRC-16-ANSI(reverse)
# see: https://en.wikipedia.org/wiki/Computation_of_cyclic_redundancy_checks
#######
CRC_POLY = 0xA001 # (LSBF/reverse)
CRC_INIT = 0xFFFF
class Modbus():
'''Simple MODBUS implementation with TX queue and retransmit timer'''
INV_ADDR = 1
'''MODBUS slave address of the TSUN inverter'''
'''MODBUS server address of the TSUN inverter'''
READ_REGS = 3
'''MODBUS function code: Read Holding Register'''
READ_INPUTS = 4
'''MODBUS function code: Read Input Register'''
WRITE_SINGLE_REG = 6
'''Modbus function code: Write Single Register'''
'''Modbus function code: Write Single Register'''
__crc_tab = []
map = {
@@ -69,12 +73,12 @@ class Modbus():
0x3029: {'reg': Register.PV4_TOTAL_GENERATION, 'fmt': '!L', 'ratio': 0.01}, # noqa: E501
}
def __init__(self, snd_handler: Callable[[bool], None], timeout: int = 1):
def __init__(self, snd_handler: Callable[[str], None], timeout: int = 1):
if not len(self.__crc_tab):
self.__build_crc_tab(CRC_POLY)
self.que = asyncio.Queue(100)
self.snd_handler = snd_handler
'''Send handler to transmit a MODBUS request'''
'''Send handler to transmit a MODBUS RTU request'''
self.rsp_handler = None
'''Response handler to forward the response'''
self.timeout = timeout
@@ -99,14 +103,13 @@ class Modbus():
self.tim = None
def __del__(self):
if type(self.counter) is not None:
logging.info(f'Modbus __del__:\n {self.counter}')
logging.info(f'Modbus __del__:\n {self.counter}')
def build_msg(self, addr: int, func: int, reg: int, val: int) -> None:
"""Build MODBUS RTU message frame and add it to the tx queue
"""Build MODBUS RTU request frame and add it to the tx queue
Keyword arguments:
addr: RTU slave address
addr: RTU server address (inverter)
func: MODBUS function code
reg: 16-bit register number
val: 16 bit value
@@ -120,7 +123,7 @@ class Modbus():
def recv_req(self, buf: bytearray,
rsp_handler: Callable[[None], None] = None) -> bool:
"""Add the received Modbus request to the tx queue
"""Add the received Modbus RTU request to the tx queue
Keyword arguments:
buf: Modbus RTU pdu incl ADDR byte and trailing CRC
@@ -241,7 +244,7 @@ class Modbus():
logging.debug(f'Modbus retrans {self}')
self.retry_cnt += 1
self.__start_timer()
self.snd_handler(self.last_req, retrans=True)
self.snd_handler(self.last_req, state='Retrans')
else:
logging.info(f'Modbus timeout {self}')
self.counter['timeouts'] += 1
@@ -264,7 +267,7 @@ class Modbus():
self.last_len = res[1]
self.retry_cnt = 0
self.__start_timer()
self.snd_handler(self.last_req, retrans=False)
self.snd_handler(self.last_req, state='Command')
except asyncio.QueueEmpty:
pass

237
app/tests/test_modbus.py
View File

@@ -2,7 +2,7 @@
import pytest
import asyncio
from app.src.modbus import Modbus
from app.src.infos import Infos
from app.src.infos import Infos, Register
pytest_plugins = ('pytest_asyncio',)
pytestmark = pytest.mark.asyncio(scope="module")
@@ -13,11 +13,15 @@ class TestHelper(Modbus):
self.db = Infos()
self.pdu = None
self.send_calls = 0
def send_cb(self, pdu: bytearray, retrans: bool):
self.recv_responses = 0
def send_cb(self, pdu: bytearray, state: str):
self.pdu = pdu
self.send_calls += 1
def resp_handler(self):
self.recv_responses += 1
def test_modbus_crc():
'''Check CRC-16 calculation'''
mb = Modbus(None)
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')
@@ -30,101 +34,139 @@ def test_modbus_crc():
assert 0x5c75 == mb._Modbus__calc_crc(b'\x01\x03\x08\x01\x2c\x00\x2c\x02\x2c\x2c\x46')
def test_build_modbus_pdu():
'''Check building and sending a MODBUS RTU'''
mb = TestHelper()
mb.build_msg(1,6,0x2000,0x12)
mb._Modbus__send_next_from_que()
assert mb.pdu == b'\x01\x06\x20\x00\x00\x12\x02\x07'
assert mb._Modbus__check_crc(mb.pdu)
def test_recv_req_crc():
mb = TestHelper()
mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x08')
mb._Modbus__send_next_from_que()
assert mb.last_fcode == 0
assert mb.last_reg == 0
assert mb.last_len == 0
assert mb.err == 1
def test_recv_req_addr():
mb = TestHelper()
mb.recv_req(b'\x02\x06\x20\x00\x00\x12\x02\x34')
mb._Modbus__send_next_from_que()
assert mb.last_addr == 2
assert mb.last_fcode == 6
assert mb.last_addr == 1
assert mb.last_fcode == 6
assert mb.last_reg == 0x2000
assert mb.last_len == 18
assert mb.err == 0
def test_recv_req():
'''Receive a valid request, which must transmitted'''
mb = TestHelper()
mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x07')
mb._Modbus__send_next_from_que()
assert mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x07')
assert mb.last_fcode == 6
assert mb.last_reg == 0x2000
assert mb.last_len == 0x12
assert mb.err == 0
def test_recv_recv_crc():
def test_recv_req_crc_err():
'''Receive a request with invalid CRC, which must be dropped'''
mb = TestHelper()
assert not mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x08')
assert mb.pdu == None
assert mb.last_fcode == 0
assert mb.last_reg == 0
assert mb.last_len == 0
assert mb.err == 1
def test_recv_resp_crc_err():
'''Receive a response with invalid CRC, which must be dropped'''
mb = TestHelper()
# simulate a transmitted request
mb.req_pend = True
mb.last_addr = 1
mb.last_fcode = 3
mb.last_reg == 0x300e
mb.last_len == 2
# check matching response, but with CRC error
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf3', 'test'):
call += 1
assert mb.err == 1
assert 0 == call
assert mb.req_pend == True
# cleanup queue
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_recv_recv_addr():
def test_recv_resp_invalid_addr():
'''Receive a response with wrong server addr, which must be dropped'''
mb = TestHelper()
mb.req_pend = True
# simulate a transmitted request
mb.last_addr = 1
mb.last_fcode = 3
mb.last_reg == 0x300e
mb.last_len == 2
# check not matching response, with wrong server addr
call = 0
for key, update in mb.recv_resp(mb.db, b'\x02\x03\x04\x01\x2c\x00\x46\x88\xf4', 'test'):
call += 1
assert mb.err == 2
assert 0 == call
assert mb.req_pend == True
assert mb.que.qsize() == 0
# cleanup queue
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_recv_recv_fcode():
'''Receive a response with wrong function code, which must be dropped'''
mb = TestHelper()
mb.build_msg(1,4,0x300e,2)
assert mb.que.qsize() == 0
assert mb.req_pend
# check not matching response, with wrong function code
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
call += 1
assert mb.err == 3
assert 0 == call
assert mb.req_pend == True
assert mb.que.qsize() == 0
# cleanup queue
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_recv_recv_len():
def test_recv_resp_len():
'''Receive a response with wrong data length, which must be dropped'''
mb = TestHelper()
mb.build_msg(1,3,0x300e,3)
assert mb.que.qsize() == 0
assert mb.req_pend
assert mb.last_len == 3
# check not matching response, with wrong data length
call = 0
for key, update, _ in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
call += 1
assert mb.err == 4
assert 0 == call
assert mb.req_pend == True
assert mb.que.qsize() == 0
# cleanup queue
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_build_recv():
def test_recv_unexpect_resp():
'''Receive a response when we havb't sent a request'''
mb = TestHelper()
assert not mb.req_pend
# check unexpected response, which must be dropped
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
call += 1
assert mb.err == 5
assert 0 == call
assert mb.req_pend == False
assert mb.que.qsize() == 0
def test_parse_resp():
'''Receive matching response and parse the values'''
mb = TestHelper()
mb.build_msg(1,3,0x3007,6)
assert mb.que.qsize() == 0
@@ -145,49 +187,7 @@ def test_build_recv():
call += 1
assert 0 == mb.err
assert 5 == call
mb.req_pend = True
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
if key == 'grid':
assert update == False
elif key == 'inverter':
assert update == False
elif key == 'env':
assert update == False
else:
assert False
assert exp_result[call] == val
call += 1
assert 0 == mb.err
assert 5 == call
assert mb.que.qsize() == 0
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_build_long():
mb = TestHelper()
mb.build_msg(1,3,0x3022,4)
assert mb.que.qsize() == 0
assert mb.req_pend
assert mb.last_addr == 1
assert mb.last_fcode == 3
assert mb.err == 0
call = 0
exp_result = [3.0, 28841.4, 113.34]
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x08\x01\x2c\x00\x2c\x02\x2c\x2c\x46\x75\x5c', 'test'):
if key == 'input':
assert update == True
assert exp_result[call] == val
else:
assert False
call += 1
assert 0 == mb.err
assert 3 == call
assert mb.que.qsize() == 0
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_queue():
@@ -199,25 +199,28 @@ def test_queue():
assert mb.send_calls == 1
assert mb.pdu == b'\x01\x030"\x00\x04\xeb\x03'
mb.pdu = None
mb._Modbus__send_next_from_que()
assert mb.send_calls == 1
assert mb.pdu == None
assert mb.que.qsize() == 0
# cleanup queue
mb._Modbus__stop_timer()
assert not mb.req_pend
def test_queue2():
'''Check queue handling for build_msg() calls'''
mb = TestHelper()
mb.build_msg(1,3,0x3007,6)
mb.build_msg(1,6,0x2008,4)
assert mb.que.qsize() == 1
assert mb.req_pend
mb.build_msg(1,3,0x3007,6)
assert mb.que.qsize() == 2
assert mb.req_pend
assert mb.send_calls == 1
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
mb._Modbus__send_next_from_que()
assert mb.send_calls == 1
call = 0
exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30]
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
@@ -234,21 +237,87 @@ def test_queue2():
assert 0 == mb.err
assert 5 == call
assert mb.que.qsize() == 1
assert mb.send_calls == 2
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'):
pass
assert mb.que.qsize() == 0
assert mb.send_calls == 3
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
call += 1
assert 0 == mb.err
assert 5 == call
assert mb.que.qsize() == 0
assert not mb.req_pend
def test_queue3():
'''Check queue handling for recv_req() calls'''
mb = TestHelper()
assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t', mb.resp_handler)
assert mb.recv_req(b'\x01\x06\x20\x08\x00\x04\x02\x0b', mb.resp_handler)
assert mb.que.qsize() == 1
assert mb.req_pend
assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t')
assert mb.que.qsize() == 2
assert mb.req_pend
assert mb.send_calls == 1
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
assert mb.recv_responses == 0
call = 0
exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30]
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
if key == 'grid':
assert update == True
elif key == 'inverter':
assert update == True
elif key == 'env':
assert update == True
else:
assert False
assert exp_result[call] == val
call += 1
assert 0 == mb.err
assert 5 == call
assert mb.recv_responses == 1
assert mb.que.qsize() == 1
assert mb.send_calls == 2
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'):
pass
assert 0 == mb.err
assert mb.recv_responses == 2
assert mb.que.qsize() == 0
assert mb.send_calls == 3
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
call += 1
assert 0 == mb.err
assert mb.recv_responses == 2
assert 5 == call
assert mb.que.qsize() == 0
assert not mb.req_pend
@pytest.mark.asyncio
async def test_timeout():
'''Test MODBUS response timeout and RTU retransmitting'''
assert asyncio.get_running_loop()
mb = TestHelper()
mb.max_retries = 2
mb.timeout = 0.1 # 100ms timeout for fast testing, expect a time resolution of at least 10ms
assert asyncio.get_running_loop() == mb.loop
mb.build_msg(1,3,0x3007,6)
mb.build_msg(1,6,0x2008,4)
@@ -260,7 +329,7 @@ async def test_timeout():
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
mb.pdu = None
await asyncio.sleep(1.1) # wait for first timeout and retransmittion
await asyncio.sleep(0.11) # wait for first timeout and retransmittion
assert mb.que.qsize() == 1
assert mb.req_pend
assert mb.retry_cnt == 1
@@ -268,7 +337,7 @@ async def test_timeout():
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
mb.pdu = None
await asyncio.sleep(1.1) # wait for second timeout and retransmittion
await asyncio.sleep(0.11) # wait for second timeout and retransmittion
assert mb.que.qsize() == 1
assert mb.req_pend
assert mb.retry_cnt == 2
@@ -276,7 +345,7 @@ async def test_timeout():
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
mb.pdu = None
await asyncio.sleep(1.1) # wait for third timeout and next pdu
await asyncio.sleep(0.11) # wait for third timeout and next pdu
assert mb.que.qsize() == 0
assert mb.req_pend
assert mb.retry_cnt == 0
@@ -284,10 +353,28 @@ async def test_timeout():
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
mb.max_retries = 0 # next pdu without retranmsission
await asyncio.sleep(1.1) # wait for fourth timout
await asyncio.sleep(0.11) # wait for fourth timout
assert mb.que.qsize() == 0
assert not mb.req_pend
assert mb.retry_cnt == 0
assert mb.send_calls == 4
# assert mb.counter == {}
def test_recv_unknown_data():
'''Receive a response with an unknwon register'''
mb = TestHelper()
assert 0x9000 not in mb.map
mb.map[0x9000] = {'reg': Register.TEST_REG1, 'fmt': '!H', 'ratio': 1}
mb.build_msg(1,3,0x9000,2)
# check matching response, but with CRC error
call = 0
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
call += 1
assert mb.err == 0
assert 0 == call
assert not mb.req_pend
del mb.map[0x9000]