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

refactor close handling

Browse Source
This commit is contained in:
Stefan Allius
2024-10-01 19:50:42 +02:00
parent a1441fb4fd
commit 39aba31bbd
15 changed files with 173 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/src/async_ifc.py
View File

@@ -117,7 +117,3 @@ class AsyncIfc(ABC):
@abstractmethod
def prot_set_update_header_cb(self, callback):
pass # pragma: no cover
@abstractmethod
def prot_set_close_header_cb(self, callback):
pass # pragma: no cover

68
app/src/async_stream.py
View File

@@ -35,14 +35,12 @@ class AsyncIfcImpl(AsyncIfc):
self.timeout_cb = None
self.init_new_client_conn_cb = None
self.update_header_cb = None
self.close_cb = None
def close(self):
self.timeout_cb = None
self.fwd_fifo.reg_trigger(None)
self.tx_fifo.reg_trigger(None)
self.rx_fifo.reg_trigger(None)
self.close_cb = None
def set_node_id(self, value: str):
self.node_id = value
@@ -126,9 +124,6 @@ class AsyncIfcImpl(AsyncIfc):
def prot_set_update_header_cb(self, callback):
self.update_header_cb = callback
def prot_set_close_header_cb(self, callback):
self.close_cb = callback
class StreamPtr():
'''Descr StreamPtr'''
@@ -212,7 +207,6 @@ class AsyncStream(AsyncIfcImpl):
f'connection timeout ({dead_conn_to}s) '
f'for {self.l_addr}')
await self.disc()
self.close()
return self
except OSError as error:
@@ -220,14 +214,12 @@ class AsyncStream(AsyncIfcImpl):
f'{error} for l{self.l_addr} | '
f'r{self.r_addr}')
await self.disc()
self.close()
return self
except RuntimeError as error:
logger.info(f'[{self.node_id}:{self.conn_no}] '
f'{error} for {self.l_addr}')
await self.disc()
self.close()
return self
except Exception:
@@ -251,10 +243,7 @@ class AsyncStream(AsyncIfcImpl):
hint: must be called before releasing the connection instance
"""
close_cb = self.close_cb
super().close()
if close_cb:
close_cb()
self._reader.feed_eof() # abort awaited read
if self._writer.is_closing():
return
@@ -304,21 +293,22 @@ class AsyncStream(AsyncIfcImpl):
except OSError as error:
if self.remote.stream:
rmt = self.remote.stream
self.remote.stream = None
logger.error(f'[{rmt.node_id}:{rmt.conn_no}] Fwd: {error} for '
f'l{rmt._ifc.l_addr} | r{rmt._ifc.r_addr}')
await rmt._ifc.disc()
rmt._ifc.close()
rmt = self.remote
logger.error(f'[{rmt.stream.node_id}:{rmt.stream.conn_no}] '
f'Fwd: {error} for '
f'l{rmt.ifc.l_addr} | r{rmt.ifc.r_addr}')
await rmt.ifc.disc()
if rmt.ifc.close_cb:
rmt.ifc.close_cb()
except RuntimeError as error:
if self.remote.stream:
rmt = self.remote.stream
self.remote.stream = None
logger.info(f'[{rmt.node_id}:{rmt.conn_no}] '
f'Fwd: {error} for {rmt._ifc.l_addr}')
await rmt._ifc.disc()
rmt._ifc.close()
rmt = self.remote
logger.info(f'[{rmt.stream.node_id}:{rmt.stream.conn_no}] '
f'Fwd: {error} for {rmt.ifc.l_addr}')
await rmt.ifc.disc()
if rmt.ifc.close_cb:
rmt.ifc.close_cb()
except Exception:
Infos.inc_counter('SW_Exception')
@@ -381,46 +371,22 @@ class AsyncStreamServer(AsyncStream):
except Exception:
pass
def close(self) -> None:
"""close handler for a no waiting disconnect
hint: must be called before releasing the connection instance
"""
logging.info(
f'AsyncStreamServer.close() l{self.l_addr} | r{self.r_addr}')
self.async_create_remote = None
self.async_publ_mqtt = None
super().close()
class AsyncStreamClient(AsyncStream):
def __init__(self, reader: StreamReader, writer: StreamWriter,
rstream: "StreamPtr") -> None:
rstream: "StreamPtr", close_cb) -> None:
AsyncStream.__init__(self, reader, writer, rstream)
self.close_cb = close_cb
async def client_loop(self, _: str) -> None:
'''Loop for receiving messages from the TSUN cloud (client-side)'''
logging.info(f'AsynStream.client_loop{self} rem-> {self.remote}')
await self.loop()
logger.info(f'[{self.node_id}:{self.conn_no}] '
'Client loop stopped for'
f' l{self.l_addr}')
server_ifc = self.remote.ifc
# if the client connection closes, we don't touch the server
# connection. Instead we erase the client connection stream,
# thus on the next received packet from the inverter, we can
# establish a new connection to the TSUN cloud
if server_ifc and server_ifc.remote.ifc == self:
# logging.debug(f'Client l{client_stream.l_addr} refs:'
# f' {gc.get_referrers(client_stream)}')
# than erase client connection
server_ifc.remote.stream = None # erases stream and ifc link
# erase backlink to inverter
self.remote.stream = None
if self.close_cb:
self.close_cb()
async def _async_forward(self) -> None:
"""forward handler transmits data over the remote connection"""

3
app/src/gen3/connection_g3.py
View File

@@ -11,6 +11,3 @@ logger = logging.getLogger('conn')
class ConnectionG3(Talent):
def __init__(self, addr, ifc, server_side, id_str=b'') -> None:
super().__init__(addr, server_side, ifc, id_str)
def close(self):
super().close()

5
app/src/gen3/inverter_g3.py
View File

@@ -33,8 +33,3 @@ class InverterG3(InverterBase):
async def async_create_remote(self) -> None:
await InverterBase.async_create_remote(
self, 'tsun', ConnectionG3)
def close(self) -> None:
logging.debug(f'InverterG3.close() {self.addr}')
self.local.stream.close()
# logging.info(f'Inverter refs: {gc.get_referrers(self)}')

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

@@ -438,8 +438,8 @@ class Talent(Message):
result = struct.unpack_from('!q', self.ifc.rx_peek(),
self.header_len)
self.ts_offset = result[0]-ts
if self.remote.stream:
self.remote.stream.ts_offset = self.ts_offset
if self.ifc.remote.stream:
self.ifc.remote.stream.ts_offset = self.ts_offset
logger.debug(f'tsun-time: {int(result[0]):08x}'
f' proxy-time: {ts:08x}'
f' offset: {self.ts_offset}')
@@ -597,9 +597,8 @@ class Talent(Message):
self.header_len+self.data_len]
if self.ctrl.is_req():
if self.remote.stream.mb.recv_req(data[hdr_len:],
self.remote.stream.
msg_forward):
rstream = self.ifc.remote.stream
if rstream.mb.recv_req(data[hdr_len:], rstream.msg_forward):
self.inc_counter('Modbus_Command')
else:
self.inc_counter('Invalid_Msg_Format')

4
app/src/gen3plus/connection_g3p.py
View File

@@ -12,7 +12,3 @@ class ConnectionG3P(SolarmanV5):
def __init__(self, addr, ifc, server_side,
client_mode: bool = False) -> None:
super().__init__(addr, server_side, client_mode, ifc)
def close(self):
super().close()
# logger.info(f'AsyncStream refs: {gc.get_referrers(self)}')

5
app/src/gen3plus/inverter_g3p.py
View File

@@ -34,8 +34,3 @@ class InverterG3P(InverterBase):
async def async_create_remote(self) -> None:
await InverterBase.async_create_remote(
self, 'solarman', ConnectionG3P)
def close(self) -> None:
logging.debug(f'InverterG3P.close() {self.addr}')
self.local.stream.close()
# logger.debug (f'Inverter refs: {gc.get_referrers(self)}')

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

@@ -603,9 +603,9 @@ class SolarmanV5(Message):
self.forward_at_cmd_resp = True
elif ftype == self.MB_RTU_CMD:
if self.remote.stream.mb.recv_req(data[15:],
self.remote.stream.
__forward_msg):
rstream = self.ifc.remote.stream
if rstream.mb.recv_req(data[15:],
rstream.__forward_msg):
self.inc_counter('Modbus_Command')
else:
logger.error('Invalid Modbus Msg')

30
app/src/inverter_base.py
View File

@@ -22,6 +22,32 @@ class InverterBase(Inverter):
def __init__(self):
self.__ha_restarts = -1
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb) -> None:
logging.info(f'Inverter.__exit__() {self.addr}')
self.__del_remote()
if self.local.stream:
self.local.stream.close()
self.local.stream = None
if self.local.ifc:
self.local.ifc.close()
self.local.ifc = None
def __del__(self) -> None:
logging.info(f'Inverter.__del__() {self.addr}')
def __del_remote(self):
if self.remote.stream:
self.remote.stream.close()
self.remote.stream = None
if self.remote.ifc:
self.remote.ifc.close()
self.remote.ifc = None
async def async_create_remote(self, inv_prot: str, conn_class) -> None:
'''Establish a client connection to the TSUN cloud'''
tsun = Config.get(inv_prot)
@@ -36,8 +62,8 @@ class InverterBase(Inverter):
logging.info(f'[{stream.node_id}] Connect to {addr}')
connect = asyncio.open_connection(host, port)
reader, writer = await connect
ifc = AsyncStreamClient(reader, writer,
self.local)
ifc = AsyncStreamClient(
reader, writer, self.local, self.__del_remote)
if hasattr(stream, 'id_str'):
self.remote.stream = conn_class(

3
app/src/modbus_tcp.py
View File

@@ -27,6 +27,7 @@ class ModbusConn():
reader, writer = await connection
self.inverter = InverterG3P(reader, writer, self.addr,
client_mode=True)
self.inverter.__enter__()
stream = self.inverter.local.stream
logging.info(f'[{stream.node_id}:{stream.conn_no}] '
f'Connected to {self.addr}')
@@ -37,7 +38,7 @@ class ModbusConn():
async def __aexit__(self, exc_type, exc, tb):
Infos.dec_counter('Inverter_Cnt')
await self.inverter.local.ifc.publish_outstanding_mqtt()
self.inverter.close()
self.inverter.__exit__(exc_type, exc, tb)
class ModbusTcp():

3
app/src/server.py
View File

@@ -74,7 +74,8 @@ async def handle_client(reader: StreamReader, writer: StreamWriter, inv_class):
'''Handles a new incoming connection and starts an async loop'''
addr = writer.get_extra_info('peername')
await inv_class(reader, writer, addr).local.ifc.server_loop()
with inv_class(reader, writer, addr) as inv:
await inv.local.ifc.server_loop()
async def handle_shutdown(web_task):