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

make class Proxy to a derivation of class AsyncStream

Browse Source
This commit is contained in:
Stefan Allius
2023-10-05 18:47:59 +02:00
parent 4c51a159af
commit 3f14f5cb9e
2 changed files with 73 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
app/src/async_stream.py
View File

@@ -1,22 +1,18 @@
import logging, traceback, aiomqtt, json import logging, traceback
from config import Config from config import Config
from messages import Message, hex_dump_memory from messages import Message, hex_dump_memory
from mqtt import Mqtt
logger = logging.getLogger('conn') logger = logging.getLogger('conn')
logger_mqtt = logging.getLogger('mqtt')
class AsyncStream(Message): class AsyncStream(Message):
def __init__(self, proxy, reader, writer, addr, stream=None, server_side=True): def __init__(self, reader, writer, addr, remote_stream, server_side: bool) -> None:
super().__init__() super().__init__()
self.proxy = proxy
self.reader = reader self.reader = reader
self.writer = writer self.writer = writer
self.remoteStream = stream self.remoteStream = remote_stream
self.addr = addr
self.server_side = server_side self.server_side = server_side
self.mqtt = Mqtt() self.addr = addr
self.unique_id = 0 self.unique_id = 0
self.node_id = '' self.node_id = ''
@@ -24,26 +20,26 @@ class AsyncStream(Message):
Our puplic methods Our puplic methods
''' '''
def set_serial_no(self, serial_no : str): def set_serial_no(self, serial_no : str):
logger_mqtt.info(f'SerialNo: {serial_no}') logger.info(f'SerialNo: {serial_no}')
if self.unique_id != serial_no: if self.unique_id != serial_no:
inverters = Config.get('inverters') inverters = Config.get('inverters')
#logger_mqtt.debug(f'Inverters: {inverters}') #logger.debug(f'Inverters: {inverters}')
if serial_no in inverters: if serial_no in inverters:
logger_mqtt.debug(f'SerialNo {serial_no} allowed!') logger.debug(f'SerialNo {serial_no} allowed!')
inv = inverters[serial_no] inv = inverters[serial_no]
self.node_id = inv['node_id'] self.node_id = inv['node_id']
self.sug_area = inv['suggested_area'] self.sug_area = inv['suggested_area']
else: else:
logger_mqtt.debug(f'SerialNo {serial_no} not known!') logger.debug(f'SerialNo {serial_no} not known!')
self.node_id = '' self.node_id = ''
self.sug_area = '' self.sug_area = ''
if not inverters['allow_all']: if not inverters['allow_all']:
self.unique_id = None self.unique_id = None
logger_mqtt.error('ignore message from unknow inverter!') logger.error('ignore message from unknow inverter!')
return return
self.unique_id = serial_no self.unique_id = serial_no
@@ -53,18 +49,6 @@ class AsyncStream(Message):
self.discovery_prfx = ha['discovery_prefix'] + '/' self.discovery_prfx = ha['discovery_prefix'] + '/'
async def register_home_assistant(self):
if self.server_side:
try:
for data_json, component, id in self.db.ha_confs(self.entitiy_prfx + self.node_id, self.unique_id, self.sug_area):
logger_mqtt.debug(f'Register: {data_json}')
await self.mqtt.publish(f"{self.discovery_prfx}{component}/{self.node_id}{id}/config", data_json)
except Exception:
logging.error(
f"Proxy: Exception:\n"
f"{traceback.format_exc()}")
async def loop(self) -> None: async def loop(self) -> None:
@@ -79,7 +63,7 @@ class AsyncStream(Message):
if self.unique_id: if self.unique_id:
await self.__async_write() await self.__async_write()
await self.__async_forward() await self.__async_forward()
await self.__async_publ_mqtt() await self.async_publ_mqtt()
except (ConnectionResetError, except (ConnectionResetError,
@@ -104,7 +88,6 @@ class AsyncStream(Message):
logger.debug(f'in AsyncStream.close() {self.addr}') logger.debug(f'in AsyncStream.close() {self.addr}')
self.writer.close() self.writer.close()
super().close() # call close handler in the parent class super().close() # call close handler in the parent class
self.proxy = None # clear our refernce to the proxy, to avoid memory leaks
if self.remoteStream: # if we have knowledge about a remote stream, we del the references between the two streams if self.remoteStream: # if we have knowledge about a remote stream, we del the references between the two streams
self.remoteStream.remoteStream = None self.remoteStream.remoteStream = None
@@ -132,8 +115,7 @@ class AsyncStream(Message):
async def __async_forward(self) -> None: async def __async_forward(self) -> None:
if self._forward_buffer: if self._forward_buffer:
if not self.remoteStream: if not self.remoteStream:
tsun = Config.get('tsun') await self.async_create_remote() # only implmeneted for server side => syncServerStream
self.remoteStream = await self.proxy.CreateClientStream (self, tsun['host'], tsun['port'])
if self.remoteStream: if self.remoteStream:
hex_dump_memory(logging.DEBUG, f'Forward to {self.remoteStream.addr}:', self._forward_buffer, len(self._forward_buffer)) hex_dump_memory(logging.DEBUG, f'Forward to {self.remoteStream.addr}:', self._forward_buffer, len(self._forward_buffer))
@@ -141,21 +123,12 @@ class AsyncStream(Message):
await self.remoteStream.writer.drain() await self.remoteStream.writer.drain()
self._forward_buffer = bytearray(0) self._forward_buffer = bytearray(0)
async def __async_publ_mqtt(self) -> None: async def async_create_remote(self) -> None:
if self.server_side: pass
db = self.db.db
# check if new inverter or collector infos are available or when the home assistant has changed the status back to online async def async_publ_mqtt(self) -> None:
if (self.new_data.keys() & {'inverter', 'collector'}) or self.mqtt.home_assistant_restarted: pass
await self.register_home_assistant()
self.mqtt.home_assistant_restarted = False # clear flag
for key in self.new_data:
if self.new_data[key] and key in db:
data_json = json.dumps(db[key])
logger_mqtt.info(f'{key}: {data_json}')
await self.mqtt.publish(f"{self.entitiy_prfx}{self.node_id}{key}", data_json)
self.new_data[key] = False
def __del__ (self): def __del__ (self):
logger.debug ("AsyncStream __del__") logger.debug ("AsyncStream __del__")

72
app/src/proxy.py
View File

@@ -1,42 +1,52 @@
import asyncio, logging, traceback import asyncio, logging, traceback, json
from config import Config
from async_stream import AsyncStream from async_stream import AsyncStream
from mqtt import Mqtt
class Proxy: logger = logging.getLogger('conn')
def __init__ (proxy, reader, writer, addr):
proxy.ServerStream = AsyncStream(proxy, reader, writer, addr)
proxy.ClientStream = None
async def server_loop(proxy, addr):
class Proxy(AsyncStream):
def __init__ (self, reader, writer, addr):
super().__init__(reader, writer, addr, None, True)
self.mqtt = Mqtt()
async def server_loop(self, addr):
'''Loop for receiving messages from the inverter (server-side)''' '''Loop for receiving messages from the inverter (server-side)'''
logging.info(f'Accept connection from {addr}') logging.info(f'Accept connection from {addr}')
await proxy.ServerStream.loop() await self.loop()
logging.info(f'Server loop stopped for {addr}') logging.info(f'Server loop stopped for {addr}')
# if the server connection closes, we also disconnect the connection to te TSUN cloud # if the server connection closes, we also disconnect the connection to te TSUN cloud
if proxy.ClientStream: if self.remoteStream:
logging.debug ("disconnect client connection") logging.debug ("disconnect client connection")
proxy.ClientStream.disc() self.remoteStream.disc()
async def client_loop(proxy, addr): async def client_loop(self, addr):
'''Loop for receiving messages from the TSUN cloud (client-side)''' '''Loop for receiving messages from the TSUN cloud (client-side)'''
await proxy.ClientStream.loop() await self.remoteStream.loop()
logging.info(f'Client loop stopped for {addr}') logging.info(f'Client loop stopped for {addr}')
# if the client connection closes, we don't touch the server connection. Instead we erase the client # 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 # connection stream, thus on the next received packet from the inverter, we can establish a new connection
# to the TSUN cloud # to the TSUN cloud
proxy.ClientStream = None self.remoteStream = None
async def CreateClientStream (proxy, stream, host, port): async def async_create_remote(self) -> None:
'''Establish a client connection to the TSUN cloud''' '''Establish a client connection to the TSUN cloud'''
tsun = Config.get('tsun')
host = tsun['host']
port = tsun['port']
addr = (host, port) addr = (host, port)
try: try:
logging.info(f'Connected to {addr}') logging.info(f'Connected to {addr}')
connect = asyncio.open_connection(host, port) connect = asyncio.open_connection(host, port)
reader, writer = await connect reader, writer = await connect
proxy.ClientStream = AsyncStream(proxy, reader, writer, addr, stream, server_side=False) self.remoteStream = AsyncStream(reader, writer, addr, self, False)
asyncio.create_task(proxy.client_loop(addr)) asyncio.create_task(self.client_loop(addr))
except ConnectionRefusedError as error: except ConnectionRefusedError as error:
logging.info(f'{error}') logging.info(f'{error}')
@@ -44,7 +54,37 @@ class Proxy:
logging.error( logging.error(
f"Proxy: Exception for {addr}:\n" f"Proxy: Exception for {addr}:\n"
f"{traceback.format_exc()}") f"{traceback.format_exc()}")
return proxy.ClientStream
async def async_publ_mqtt(self) -> None:
db = self.db.db
# check if new inverter or collector infos are available or when the home assistant has changed the status back to online
if (self.new_data.keys() & {'inverter', 'collector'}): #or self.mqtt.home_assistant_restarted:
await self.__register_home_assistant()
#self.mqtt.home_assistant_restarted = False # clear flag
for key in self.new_data:
if self.new_data[key] and key in db:
data_json = json.dumps(db[key])
logger.info(f'{key}: {data_json}')
await self.mqtt.publish(f"{self.entitiy_prfx}{self.node_id}{key}", data_json)
self.new_data[key] = False
async def __register_home_assistant(self):
try:
for data_json, component, id in self.db.ha_confs(self.entitiy_prfx + self.node_id, self.unique_id, self.sug_area):
logger.debug(f'MQTT Register: {data_json}')
await self.mqtt.publish(f"{self.discovery_prfx}{component}/{self.node_id}{id}/config", data_json)
except Exception:
logging.error(
f"Proxy: Exception:\n"
f"{traceback.format_exc()}")
def close(self):
logger.debug(f'in AsyncServerStream.close() {self.addr}')
super().close() # call close handler in the parent class
def __del__ (proxy): def __del__ (proxy):
logging.debug ("Proxy __del__") logging.debug ("Proxy __del__")
super().__del__()