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

use AsyncIfc class with FIFO

Browse Source
This commit is contained in:
Stefan Allius
2024-09-22 10:40:30 +02:00
parent af81aef07c
commit b7c63b5cf8
12 changed files with 380 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,6 +5,7 @@ from datetime import datetime
from tzlocal import get_localzone
if __name__ == "app.src.gen3.talent":
from app.src.async_ifc import AsyncIfc
from app.src.messages import hex_dump_memory, Message, State
from app.src.modbus import Modbus
from app.src.my_timer import Timer
@@ -12,6 +13,7 @@ if __name__ == "app.src.gen3.talent":
from app.src.gen3.infos_g3 import InfosG3
from app.src.infos import Register
else: # pragma: no cover
from async_ifc import AsyncIfc
from messages import hex_dump_memory, Message, State
from modbus import Modbus
from my_timer import Timer
@@ -44,8 +46,10 @@ class Talent(Message):
MB_REGULAR_TIMEOUT = 60
TXT_UNKNOWN_CTRL = 'Unknown Ctrl'
def __init__(self, server_side: bool, id_str=b''):
def __init__(self, server_side: bool, ifc: "AsyncIfc", id_str=b''):
super().__init__(server_side, self.send_modbus_cb, mb_timeout=15)
ifc.read.reg_trigger(self.read)
self.ifc = ifc
self.await_conn_resp_cnt = 0
self.id_str = id_str
self.contact_name = b''
@@ -103,6 +107,7 @@ class Talent(Message):
self.log_lvl.clear()
self.state = State.closed
self.mb_timer.close()
self.ifc.read.reg_trigger(None)
super().close()
def __set_serial_no(self, serial_no: str):
@@ -138,10 +143,10 @@ class Talent(Message):
self._read()
while True:
if not self.header_valid:
self.__parse_header(self._recv_buffer, len(self._recv_buffer))
self.__parse_header(self.ifc.read.peek(), len(self.ifc.read))
if self.header_valid and \
len(self._recv_buffer) >= (self.header_len + self.data_len):
len(self.ifc.read) >= (self.header_len + self.data_len):
if self.state == State.init:
self.state = State.received # received 1st package
@@ -149,11 +154,10 @@ class Talent(Message):
if callable(log_lvl):
log_lvl = log_lvl()
hex_dump_memory(log_lvl, f'Received from {self.addr}:'
f' BufLen: {len(self._recv_buffer)}'
f' HdrLen: {self.header_len}'
f' DtaLen: {self.data_len}',
self._recv_buffer, len(self._recv_buffer))
self.ifc.read.logging(log_lvl, f'Received from {self.addr}:'
f' BufLen: {len(self.ifc.read)}'
f' HdrLen: {self.header_len}'
f' DtaLen: {self.data_len}')
self.__set_serial_no(self.id_str.decode("utf-8"))
self.__dispatch_msg()
@@ -165,9 +169,9 @@ class Talent(Message):
'''add the actual receive msg to the forwarding queue'''
tsun = Config.get('tsun')
if tsun['enabled']:
buffer = self._recv_buffer
buflen = self.header_len+self.data_len
self._forward_buffer += buffer[:buflen]
buffer = self.ifc.read.peek(buflen)
self._forward_buffer += buffer
hex_dump_memory(logging.DEBUG, 'Store for forwarding:',
buffer, buflen)
@@ -178,21 +182,20 @@ class Talent(Message):
def forward_snd(self) -> None:
'''add the build send msg to the forwarding queue'''
tsun = Config.get('tsun')
rest = self.ifc.write.get(self.send_msg_ofs)
buffer = self.ifc.write.get(len(self.ifc.write))
if tsun['enabled']:
_len = len(self._send_buffer) - self.send_msg_ofs
struct.pack_into('!l', self._send_buffer, self.send_msg_ofs,
_len-4)
buffer = self._send_buffer[self.send_msg_ofs:]
_len = len(buffer)
struct.pack_into('!l', buffer, 0, _len-4)
buflen = _len
self._forward_buffer += buffer[:buflen]
self._forward_buffer += buffer
hex_dump_memory(logging.INFO, 'Store for forwarding:',
buffer, buflen)
fnc = self.switch.get(self.msg_id, self.msg_unknown)
logger.info(self.__flow_str(self.server_side, 'forwrd') +
f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
self._send_buffer = self._send_buffer[:self.send_msg_ofs]
self.ifc.write += rest
def send_modbus_cb(self, modbus_pdu: bytearray, log_lvl: int, state: str):
if self.state != State.up:
@@ -201,15 +204,13 @@ class Talent(Message):
return
self.__build_header(0x70, 0x77)
self._send_buffer += b'\x00\x01\xa3\x28' # magic ?
self._send_buffer += struct.pack('!B', len(modbus_pdu))
self._send_buffer += modbus_pdu
self.ifc.write += b'\x00\x01\xa3\x28' # magic ?
self.ifc.write += struct.pack('!B', len(modbus_pdu))
self.ifc.write += modbus_pdu
self.__finish_send_msg()
hex_dump_memory(log_lvl, 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:]
self.ifc.write.logging(log_lvl, f'Send Modbus {state}:{self.addr}:')
self.ifc.write()
def _send_modbus_cmd(self, func, addr, val, log_lvl) -> None:
if self.state != State.up:
@@ -237,9 +238,9 @@ class Talent(Message):
self.msg_id = 0
self.await_conn_resp_cnt += 1
self.__build_header(0x91)
self._send_buffer += struct.pack(f'!{len(contact_name)+1}p'
f'{len(contact_mail)+1}p',
contact_name, contact_mail)
self.ifc.write += struct.pack(f'!{len(contact_name)+1}p'
f'{len(contact_mail)+1}p',
contact_name, contact_mail)
self.__finish_send_msg()
return True
@@ -323,7 +324,7 @@ class Talent(Message):
self.inc_counter('Invalid_Msg_Format')
# erase broken recv buffer
self._recv_buffer = bytearray()
self.ifc.read.clear()
return
hdr_len = 5+id_len+2
@@ -344,16 +345,17 @@ class Talent(Message):
def __build_header(self, ctrl, msg_id=None) -> None:
if not msg_id:
msg_id = self.msg_id
self.send_msg_ofs = len(self._send_buffer)
self._send_buffer += struct.pack(f'!l{len(self.id_str)+1}pBB',
0, self.id_str, ctrl, msg_id)
self.send_msg_ofs = len(self.ifc.write)
self.ifc.write += struct.pack(f'!l{len(self.id_str)+1}pBB',
0, self.id_str, ctrl, msg_id)
fnc = self.switch.get(msg_id, self.msg_unknown)
logger.info(self.__flow_str(self.server_side, 'tx') +
f' Ctl: {int(ctrl):#02x} Msg: {fnc.__name__!r}')
def __finish_send_msg(self) -> None:
_len = len(self._send_buffer) - self.send_msg_ofs
struct.pack_into('!l', self._send_buffer, self.send_msg_ofs, _len-4)
_len = len(self.ifc.write) - self.send_msg_ofs
struct.pack_into('!l', self.ifc.write.peek(), self.send_msg_ofs,
_len-4)
def __dispatch_msg(self) -> None:
fnc = self.switch.get(self.msg_id, self.msg_unknown)
@@ -367,7 +369,7 @@ class Talent(Message):
f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}')
def __flush_recv_msg(self) -> None:
self._recv_buffer = self._recv_buffer[(self.header_len+self.data_len):]
self.ifc.read.get(self.header_len+self.data_len)
self.header_valid = False
'''
@@ -377,7 +379,7 @@ class Talent(Message):
if self.ctrl.is_ind():
if self.server_side and self.__process_contact_info():
self.__build_header(0x91)
self._send_buffer += b'\x01'
self.ifc.write += b'\x01'
self.__finish_send_msg()
# don't forward this contact info here, we will build one
# when the remote connection is established
@@ -391,18 +393,19 @@ class Talent(Message):
self.forward()
def __process_contact_info(self) -> bool:
result = struct.unpack_from('!B', self._recv_buffer, self.header_len)
buf = self.ifc.read.peek()
result = struct.unpack_from('!B', buf, self.header_len)
name_len = result[0]
if self.data_len == 1: # this is a response withone status byte
return False
if self.data_len >= name_len+2:
result = struct.unpack_from(f'!{name_len+1}pB', self._recv_buffer,
result = struct.unpack_from(f'!{name_len+1}pB', buf,
self.header_len)
self.contact_name = result[0]
mail_len = result[1]
logger.info(f'name: {self.contact_name}')
result = struct.unpack_from(f'!{mail_len+1}p', self._recv_buffer,
result = struct.unpack_from(f'!{mail_len+1}p', buf,
self.header_len+name_len+1)
self.contact_mail = result[0]
logger.info(f'mail: {self.contact_mail}')
@@ -417,12 +420,12 @@ class Talent(Message):
ts = self._timestamp()
logger.debug(f'time: {ts:08x}')
self.__build_header(0x91)
self._send_buffer += struct.pack('!q', ts)
self.ifc.write += struct.pack('!q', ts)
self.__finish_send_msg()
elif self.data_len >= 8:
ts = self._timestamp()
result = struct.unpack_from('!q', self._recv_buffer,
result = struct.unpack_from('!q', self.ifc.read.peek(),
self.header_len)
self.ts_offset = result[0]-ts
if self.remote_stream:
@@ -446,10 +449,10 @@ class Talent(Message):
self.db.set_db_def_value(Register.POLLING_INTERVAL,
self.mb_timeout)
self.__build_header(0x99)
self._send_buffer += b'\x02'
self.ifc.write += b'\x02'
self.__finish_send_msg()
result = struct.unpack_from('!Bq', self._recv_buffer,
result = struct.unpack_from('!Bq', self.ifc.read.peek(),
self.header_len)
resp_code = result[0]
ts = result[1]+self.ts_offset
@@ -457,11 +460,11 @@ class Talent(Message):
f' tsun-time: {ts:08x}'
f' offset: {self.ts_offset}')
self.__build_header(0x91)
self._send_buffer += struct.pack('!Bq', resp_code, ts)
self.ifc.write += struct.pack('!Bq', resp_code, ts)
self.forward_snd()
return
elif self.ctrl.is_resp():
result = struct.unpack_from('!B', self._recv_buffer,
result = struct.unpack_from('!B', self.ifc.read.peek(),
self.header_len)
resp_code = result[0]
logging.debug(f'TimeActRespCode: {resp_code}')
@@ -473,7 +476,8 @@ class Talent(Message):
self.forward()
def parse_msg_header(self):
result = struct.unpack_from('!lB', self._recv_buffer, self.header_len)
result = struct.unpack_from('!lB', self.ifc.read.peek(),
self.header_len)
data_id = result[0] # len of complete message
id_len = result[1] # len of variable id string
@@ -481,7 +485,7 @@ class Talent(Message):
msg_hdr_len = 5+id_len+9
result = struct.unpack_from(f'!{id_len+1}pBq', self._recv_buffer,
result = struct.unpack_from(f'!{id_len+1}pBq', self.ifc.read.peek(),
self.header_len + 4)
timestamp = result[2]
@@ -494,7 +498,7 @@ class Talent(Message):
def msg_collector_data(self):
if self.ctrl.is_ind():
self.__build_header(0x99)
self._send_buffer += b'\x01'
self.ifc.write += b'\x01'
self.__finish_send_msg()
self.__process_data()
@@ -509,7 +513,7 @@ class Talent(Message):
def msg_inverter_data(self):
if self.ctrl.is_ind():
self.__build_header(0x99)
self._send_buffer += b'\x01'
self.ifc.write += b'\x01'
self.__finish_send_msg()
self.__process_data()
self.state = State.up # allow MODBUS cmds
@@ -529,7 +533,7 @@ class Talent(Message):
def __process_data(self):
msg_hdr_len, ts = self.parse_msg_header()
for key, update in self.db.parse(self._recv_buffer, self.header_len
for key, update in self.db.parse(self.ifc.read.peek(), self.header_len
+ msg_hdr_len, self.node_id):
if update:
self._set_mqtt_timestamp(key, self._utcfromts(ts))
@@ -549,7 +553,7 @@ class Talent(Message):
msg_hdr_len = 5
result = struct.unpack_from('!lBB', self._recv_buffer,
result = struct.unpack_from('!lBB', self.ifc.read.peek(),
self.header_len)
modbus_len = result[1]
return msg_hdr_len, modbus_len
@@ -558,7 +562,7 @@ class Talent(Message):
msg_hdr_len = 6
result = struct.unpack_from('!lBBB', self._recv_buffer,
result = struct.unpack_from('!lBBB', self.ifc.read.peek(),
self.header_len)
modbus_len = result[2]
return msg_hdr_len, modbus_len
@@ -579,8 +583,8 @@ class Talent(Message):
self.__msg_modbus(hdr_len)
def __msg_modbus(self, hdr_len):
data = self._recv_buffer[self.header_len:
self.header_len+self.data_len]
data = self.ifc.read.peek()[self.header_len:
self.header_len+self.data_len]
if self.ctrl.is_req():
if self.remote_stream.mb.recv_req(data[hdr_len:],