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

Merge branch 'refactoring-async-stream' of https://github.com/s-allius/tsun-gen3-proxy into titan-scan

Browse Source
This commit is contained in:
Stefan Allius
2024-09-29 20:27:28 +02:00
parent 4c15495670 41aeac4168
commit b57ded1a73
19 changed files with 798 additions and 895 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,61 +1,16 @@
import logging
from asyncio import StreamReader, StreamWriter
if __name__ == "app.src.gen3.connection_g3":
from app.src.async_stream import AsyncStreamServer
from app.src.async_stream import AsyncStreamClient, StreamPtr
from app.src.gen3.talent import Talent
else: # pragma: no cover
from async_stream import AsyncStreamServer
from async_stream import AsyncStreamClient, StreamPtr
from gen3.talent import Talent
logger = logging.getLogger('conn')
class ConnectionG3(Talent):
async def async_create_remote(self) -> None:
pass # virtual interface # pragma: no cover
async def async_publ_mqtt(self) -> None:
pass # virtual interface # pragma: no cover
def healthy(self) -> bool:
logger.debug('ConnectionG3 healthy()')
return self._ifc.healthy()
def __init__(self, addr, ifc, server_side, id_str=b'') -> None:
super().__init__(addr, server_side, ifc, id_str)
def close(self):
self._ifc.close()
Talent.close(self)
# logger.info(f'AsyncStream refs: {gc.get_referrers(self)}')
class ConnectionG3Server(ConnectionG3):
def __init__(self, reader: StreamReader, writer: StreamWriter,
addr, rstream: 'ConnectionG3Client',
id_str=b'') -> None:
server_side = True
self.remote = StreamPtr(rstream)
self._ifc = AsyncStreamServer(reader, writer,
self.async_publ_mqtt,
self.async_create_remote,
self.remote)
self.conn_no = self._ifc.get_conn_no()
self.addr = addr
Talent.__init__(self, server_side, self._ifc, id_str)
class ConnectionG3Client(ConnectionG3):
def __init__(self, reader: StreamReader, writer: StreamWriter,
addr, rstream: 'ConnectionG3Server',
id_str=b'') -> None:
server_side = False
self.remote = StreamPtr(rstream)
self._ifc = AsyncStreamClient(reader, writer,
self.remote)
self.conn_no = self._ifc.get_conn_no()
self.addr = addr
Talent.__init__(self, server_side, self._ifc, id_str)
super().close()

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

@@ -2,58 +2,40 @@ import logging
from asyncio import StreamReader, StreamWriter
if __name__ == "app.src.gen3.inverter_g3":
from app.src.inverter import Inverter
from app.src.gen3.connection_g3 import ConnectionG3Server
from app.src.gen3.connection_g3 import ConnectionG3Client
from app.src.inverter_base import InverterBase
from app.src.async_stream import StreamPtr
from app.src.async_stream import AsyncStreamServer
from app.src.gen3.connection_g3 import ConnectionG3
else: # pragma: no cover
from inverter import Inverter
from gen3.connection_g3 import ConnectionG3Server
from gen3.connection_g3 import ConnectionG3Client
from inverter_base import InverterBase
from async_stream import StreamPtr
from async_stream import AsyncStreamServer
from gen3.connection_g3 import ConnectionG3
logger_mqtt = logging.getLogger('mqtt')
class InverterG3(Inverter, ConnectionG3Server):
'''class Inverter is a derivation of an Async_Stream
The class has some class method for managing common resources like a
connection to the MQTT broker or proxy error counter which are common
for all inverter connection
Instances of the class are connections to an inverter and can have an
optional link to an remote connection to the TSUN cloud. A remote
connection dies with the inverter connection.
class methods:
class_init(): initialize the common resources of the proxy (MQTT
broker, Proxy DB, etc). Must be called before the
first inverter instance can be created
class_close(): release the common resources of the proxy. Should not
be called before any instances of the class are
destroyed
methods:
server_loop(addr): Async loop method for receiving messages from the
inverter (server-side)
client_loop(addr): Async loop method for receiving messages from the
TSUN cloud (client-side)
async_create_remote(): Establish a client connection to the TSUN cloud
async_publ_mqtt(): Publish data to MQTT broker
close(): Release method which must be called before a instance can be
destroyed
'''
class InverterG3(InverterBase):
def __init__(self, reader: StreamReader, writer: StreamWriter, addr):
Inverter.__init__(self)
ConnectionG3Server.__init__(self, reader, writer, addr, None)
super().__init__()
self.addr = addr
self.remote = StreamPtr(None)
ifc = AsyncStreamServer(reader, writer,
self.async_publ_mqtt,
self.async_create_remote,
self.remote)
self.remote = StreamPtr(None)
self.local = StreamPtr(
ConnectionG3(addr, ifc, True)
)
async def async_create_remote(self) -> None:
await Inverter.async_create_remote(
self, 'tsun', ConnectionG3Client)
await InverterBase.async_create_remote(
self, 'tsun', ConnectionG3)
def close(self) -> None:
logging.debug(f'InverterG3.close() {self.addr}')
ConnectionG3Server.close(self)
self.local.stream.close()
# logging.info(f'Inverter refs: {gc.get_referrers(self)}')

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

@@ -46,13 +46,15 @@ class Talent(Message):
MB_REGULAR_TIMEOUT = 60
TXT_UNKNOWN_CTRL = 'Unknown Ctrl'
def __init__(self, server_side: bool, ifc: "AsyncIfc", id_str=b''):
def __init__(self, addr, server_side: bool, ifc: "AsyncIfc", id_str=b''):
super().__init__(server_side, self.send_modbus_cb, mb_timeout=15)
ifc.rx_set_cb(self.read)
ifc.prot_set_timeout_cb(self._timeout)
ifc.prot_set_init_new_client_conn_cb(self._init_new_client_conn)
ifc.prot_set_update_header_cb(self._update_header)
self.addr = addr
self.ifc = ifc
self.conn_no = ifc.get_conn_no()
self.await_conn_resp_cnt = 0
self.id_str = id_str
self.contact_name = b''
@@ -93,6 +95,10 @@ class Talent(Message):
'''
Our puplic methods
'''
def healthy(self) -> bool:
logger.debug('Talent healthy()')
return self.ifc.healthy()
def close(self) -> None:
logging.debug('Talent.close()')
if self.server_side:
@@ -110,6 +116,7 @@ class Talent(Message):
self.log_lvl.clear()
self.state = State.closed
self.mb_timer.close()
self.ifc.close()
self.ifc.rx_set_cb(None)
self.ifc.prot_set_timeout_cb(None)
self.ifc.prot_set_init_new_client_conn_cb(None)

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

@@ -1,61 +1,18 @@
import logging
from asyncio import StreamReader, StreamWriter
if __name__ == "app.src.gen3plus.connection_g3p":
from app.src.async_stream import AsyncStreamServer
from app.src.async_stream import AsyncStreamClient, StreamPtr
from app.src.gen3plus.solarman_v5 import SolarmanV5
else: # pragma: no cover
from async_stream import AsyncStreamServer
from async_stream import AsyncStreamClient, StreamPtr
from gen3plus.solarman_v5 import SolarmanV5
logger = logging.getLogger('conn')
class ConnectionG3P(SolarmanV5):
async def async_create_remote(self) -> None:
pass # virtual interface # pragma: no cover
async def async_publ_mqtt(self) -> None:
pass # virtual interface # pragma: no cover
def healthy(self) -> bool:
logger.debug('ConnectionG3P healthy()')
return self._ifc.healthy()
def __init__(self, addr, ifc, server_side,
client_mode: bool = False) -> None:
super().__init__(addr, server_side, client_mode, ifc)
def close(self):
self._ifc.close()
SolarmanV5.close(self)
super().close()
# logger.info(f'AsyncStream refs: {gc.get_referrers(self)}')
class ConnectionG3PServer(ConnectionG3P):
def __init__(self, reader: StreamReader, writer: StreamWriter,
addr, rstream: 'ConnectionG3PClient',
client_mode: bool) -> None:
server_side = True
self.remote = StreamPtr(rstream)
self._ifc = AsyncStreamServer(reader, writer,
self.async_publ_mqtt,
self.async_create_remote,
self.remote)
self.conn_no = self._ifc.get_conn_no()
self.addr = addr
SolarmanV5.__init__(self, server_side, client_mode, self._ifc)
class ConnectionG3PClient(ConnectionG3P):
def __init__(self, reader: StreamReader, writer: StreamWriter,
addr, rstream: 'ConnectionG3PServer') -> None:
server_side = False
client_mode = False
self.remote = StreamPtr(rstream)
self._ifc = AsyncStreamClient(reader, writer, self.remote)
self.conn_no = self._ifc.get_conn_no()
self.addr = addr
SolarmanV5.__init__(self, server_side, client_mode, self._ifc)

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

@@ -2,60 +2,40 @@ import logging
from asyncio import StreamReader, StreamWriter
if __name__ == "app.src.gen3plus.inverter_g3p":
from app.src.inverter import Inverter
from app.src.gen3plus.connection_g3p import ConnectionG3PServer
from app.src.gen3plus.connection_g3p import ConnectionG3PClient
from app.src.inverter_base import InverterBase
from app.src.async_stream import StreamPtr
from app.src.async_stream import AsyncStreamServer
from app.src.gen3plus.connection_g3p import ConnectionG3P
else: # pragma: no cover
from inverter import Inverter
from gen3plus.connection_g3p import ConnectionG3PServer
from gen3plus.connection_g3p import ConnectionG3PClient
from inverter_base import InverterBase
from async_stream import StreamPtr
from async_stream import AsyncStreamServer
from gen3plus.connection_g3p import ConnectionG3P
logger_mqtt = logging.getLogger('mqtt')
class InverterG3P(Inverter, ConnectionG3PServer):
'''class Inverter is a derivation of an Async_Stream
The class has some class method for managing common resources like a
connection to the MQTT broker or proxy error counter which are common
for all inverter connection
Instances of the class are connections to an inverter and can have an
optional link to an remote connection to the TSUN cloud. A remote
connection dies with the inverter connection.
class methods:
class_init(): initialize the common resources of the proxy (MQTT
broker, Proxy DB, etc). Must be called before the
first inverter instance can be created
class_close(): release the common resources of the proxy. Should not
be called before any instances of the class are
destroyed
methods:
server_loop(addr): Async loop method for receiving messages from the
inverter (server-side)
client_loop(addr): Async loop method for receiving messages from the
TSUN cloud (client-side)
async_create_remote(): Establish a client connection to the TSUN cloud
async_publ_mqtt(): Publish data to MQTT broker
close(): Release method which must be called before a instance can be
destroyed
'''
class InverterG3P(InverterBase):
def __init__(self, reader: StreamReader, writer: StreamWriter, addr,
client_mode: bool = False):
Inverter.__init__(self)
ConnectionG3PServer.__init__(
self, reader, writer, addr, None, client_mode=client_mode)
super().__init__()
self.addr = addr
self.remote = StreamPtr(None)
ifc = AsyncStreamServer(reader, writer,
self.async_publ_mqtt,
self.async_create_remote,
self.remote)
self.local = StreamPtr(
ConnectionG3P(addr, ifc, True, client_mode)
)
async def async_create_remote(self) -> None:
await Inverter.async_create_remote(
self, 'solarman', ConnectionG3PClient)
await InverterBase.async_create_remote(
self, 'solarman', ConnectionG3P)
def close(self) -> None:
logging.debug(f'InverterG3P.close() {self.addr}')
ConnectionG3PServer.close(self)
self.local.stream.close()
# logger.debug (f'Inverter refs: {gc.get_referrers(self)}')

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

@@ -62,14 +62,17 @@ class SolarmanV5(Message):
HDR_FMT = '<BLLL'
'''format string for packing of the header'''
def __init__(self, server_side: bool, client_mode: bool, ifc: "AsyncIfc"):
def __init__(self, addr, server_side: bool, client_mode: bool,
ifc: "AsyncIfc"):
super().__init__(server_side, self.send_modbus_cb, mb_timeout=8)
ifc.rx_set_cb(self.read)
ifc.prot_set_timeout_cb(self._timeout)
ifc.prot_set_init_new_client_conn_cb(self._init_new_client_conn)
ifc.prot_set_update_header_cb(self._update_header)
self.addr = addr
self.ifc = ifc
self.conn_no = ifc.get_conn_no()
self.header_len = 11 # overwrite construcor in class Message
self.control = 0
self.seq = Sequence(server_side)
@@ -152,6 +155,10 @@ class SolarmanV5(Message):
'''
Our puplic methods
'''
def healthy(self) -> bool:
logger.debug('SolarmanV5 healthy()')
return self.ifc.healthy()
def close(self) -> None:
logging.debug('Solarman.close()')
if self.server_side:
@@ -169,6 +176,7 @@ class SolarmanV5(Message):
self.log_lvl.clear()
self.state = State.closed
self.mb_timer.close()
self.ifc.close()
self.ifc.rx_set_cb(None)
self.ifc.prot_set_timeout_cb(None)
self.ifc.prot_set_init_new_client_conn_cb(None)

107
app/src/inverter.py
View File

@@ -1,8 +1,6 @@
import asyncio
import logging
import traceback
import json
from aiomqtt import MqttCodeError
if __name__ == "app.src.inverter":
from app.src.config import Config
@@ -17,6 +15,28 @@ logger_mqtt = logging.getLogger('mqtt')
class Inverter():
'''class Inverter is a baseclass
The class has some class method for managing common resources like a
connection to the MQTT broker or proxy error counter which are common
for all inverter connection
Instances of the class are connections to an inverter and can have an
optional link to an remote connection to the TSUN cloud. A remote
connection dies with the inverter connection.
class methods:
class_init(): initialize the common resources of the proxy (MQTT
broker, Proxy DB, etc). Must be called before the
first inverter instance can be created
class_close(): release the common resources of the proxy. Should not
be called before any instances of the class are
destroyed
methods:
async_create_remote(): Establish a client connection to the TSUN cloud
async_publ_mqtt(): Publish data to MQTT broker
'''
@classmethod
def class_init(cls) -> None:
logging.debug('Inverter.class_init')
@@ -84,86 +104,3 @@ class Inverter():
logging.info('Close MQTT Task')
loop.run_until_complete(cls.mqtt.close())
cls.mqtt = None
def __init__(self):
self.__ha_restarts = -1
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)
host = tsun['host']
port = tsun['port']
addr = (host, port)
try:
logging.info(f'[{self.node_id}] Connect to {addr}')
connect = asyncio.open_connection(host, port)
reader, writer = await connect
if hasattr(self, 'id_str'):
self.remote.stream = conn_class(
reader, writer, addr, self, self.id_str)
else:
self.remote.stream = conn_class(
reader, writer, addr, self)
logging.info(f'[{self.remote.stream.node_id}:'
f'{self.remote.stream.conn_no}] '
f'Connected to {addr}')
asyncio.create_task(self.remote.ifc.client_loop(addr))
except (ConnectionRefusedError, TimeoutError) as error:
logging.info(f'{error}')
except Exception:
self.inc_counter('SW_Exception')
logging.error(
f"Inverter: Exception for {addr}:\n"
f"{traceback.format_exc()}")
async def async_publ_mqtt(self) -> None:
'''publish data to MQTT broker'''
if not self.unique_id:
return
# check if new inverter or collector infos are available or when the
# home assistant has changed the status back to online
try:
if (('inverter' in self.new_data and self.new_data['inverter'])
or ('collector' in self.new_data and
self.new_data['collector'])
or self.mqtt.ha_restarts != self.__ha_restarts):
await self._register_proxy_stat_home_assistant()
await self.__register_home_assistant()
self.__ha_restarts = self.mqtt.ha_restarts
for key in self.new_data:
await self.__async_publ_mqtt_packet(key)
for key in Infos.new_stat_data:
await Inverter._async_publ_mqtt_proxy_stat(key)
except MqttCodeError as error:
logging.error(f'Mqtt except: {error}')
except Exception:
self.inc_counter('SW_Exception')
logging.error(
f"Inverter: Exception:\n"
f"{traceback.format_exc()}")
async def __async_publ_mqtt_packet(self, key):
db = self.db.db
if key in db and self.new_data[key]:
data_json = json.dumps(db[key])
node_id = self.node_id
logger_mqtt.debug(f'{key}: {data_json}')
await self.mqtt.publish(f'{self.entity_prfx}{node_id}{key}', data_json) # noqa: E501
self.new_data[key] = False
async def __register_home_assistant(self) -> None:
'''register all our topics at home assistant'''
for data_json, component, node_id, id in self.db.ha_confs(
self.entity_prfx, self.node_id, self.unique_id,
self.sug_area):
logger_mqtt.debug(f"MQTT Register: cmp:'{component}'"
f" node_id:'{node_id}' {data_json}")
await self.mqtt.publish(f"{self.discovery_prfx}{component}"
f"/{node_id}{id}/config", data_json)
self.db.reg_clr_at_midnight(f'{self.entity_prfx}{self.node_id}')

108
app/src/inverter_base.py Normal file
View File

@@ -0,0 +1,108 @@
import asyncio
import logging
import traceback
import json
from aiomqtt import MqttCodeError
if __name__ == "app.src.inverter_base":
from app.src.inverter import Inverter
from app.src.async_stream import AsyncStreamClient
from app.src.config import Config
from app.src.infos import Infos
else: # pragma: no cover
from inverter import Inverter
from async_stream import AsyncStreamClient
from config import Config
from infos import Infos
logger_mqtt = logging.getLogger('mqtt')
class InverterBase(Inverter):
def __init__(self):
self.__ha_restarts = -1
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)
host = tsun['host']
port = tsun['port']
addr = (host, port)
stream = self.local.stream
try:
logging.info(f'[{stream.node_id}] Connect to {addr}')
connect = asyncio.open_connection(host, port)
reader, writer = await connect
ifc = AsyncStreamClient(reader, writer,
self.remote)
if hasattr(stream, 'id_str'):
self.remote.stream = conn_class(
addr, ifc, False, stream.id_str)
else:
self.remote.stream = conn_class(
addr, ifc, False)
logging.info(f'[{self.remote.stream.node_id}:'
f'{self.remote.stream.conn_no}] '
f'Connected to {addr}')
asyncio.create_task(self.remote.ifc.client_loop(addr))
except (ConnectionRefusedError, TimeoutError) as error:
logging.info(f'{error}')
except Exception:
Infos.inc_counter('SW_Exception')
logging.error(
f"Inverter: Exception for {addr}:\n"
f"{traceback.format_exc()}")
async def async_publ_mqtt(self) -> None:
'''publish data to MQTT broker'''
stream = self.local.stream
if not stream.unique_id:
return
# check if new inverter or collector infos are available or when the
# home assistant has changed the status back to online
try:
if (('inverter' in stream.new_data and stream.new_data['inverter'])
or ('collector' in stream.new_data and
stream.new_data['collector'])
or self.mqtt.ha_restarts != self.__ha_restarts):
await self._register_proxy_stat_home_assistant()
await self.__register_home_assistant(stream)
self.__ha_restarts = self.mqtt.ha_restarts
for key in stream.new_data:
await self.__async_publ_mqtt_packet(stream, key)
for key in Infos.new_stat_data:
await Inverter._async_publ_mqtt_proxy_stat(key)
except MqttCodeError as error:
logging.error(f'Mqtt except: {error}')
except Exception:
Infos.inc_counter('SW_Exception')
logging.error(
f"Inverter: Exception:\n"
f"{traceback.format_exc()}")
async def __async_publ_mqtt_packet(self, stream, key):
db = stream.db.db
if key in db and stream.new_data[key]:
data_json = json.dumps(db[key])
node_id = stream.node_id
logger_mqtt.debug(f'{key}: {data_json}')
await self.mqtt.publish(f'{self.entity_prfx}{node_id}{key}', data_json) # noqa: E501
stream.new_data[key] = False
async def __register_home_assistant(self, stream) -> None:
'''register all our topics at home assistant'''
for data_json, component, node_id, id in stream.db.ha_confs(
self.entity_prfx, stream.node_id, stream.unique_id,
stream.sug_area):
logger_mqtt.debug(f"MQTT Register: cmp:'{component}'"
f" node_id:'{node_id}' {data_json}")
await self.mqtt.publish(f"{self.discovery_prfx}{component}"
f"/{node_id}{id}/config", data_json)
stream.db.reg_clr_at_midnight(f'{self.entity_prfx}{stream.node_id}')

22
app/src/modbus_tcp.py
View File

@@ -19,24 +19,25 @@ class ModbusConn():
self.host = host
self.port = port
self.addr = (host, port)
self.stream = None
self.inverter = None
async def __aenter__(self) -> 'InverterG3P':
'''Establish a client connection to the TSUN cloud'''
connection = asyncio.open_connection(self.host, self.port)
reader, writer = await connection
self.stream = InverterG3P(reader, writer, self.addr,
client_mode=True)
logging.info(f'[{self.stream.node_id}:{self.stream.conn_no}] '
self.inverter = InverterG3P(reader, writer, self.addr,
client_mode=True)
stream = self.inverter.local.stream
logging.info(f'[{stream.node_id}:{stream.conn_no}] '
f'Connected to {self.addr}')
Infos.inc_counter('Inverter_Cnt')
await self.stream._ifc.publish_outstanding_mqtt()
return self.stream
await self.inverter.local.ifc.publish_outstanding_mqtt()
return self.inverter
async def __aexit__(self, exc_type, exc, tb):
Infos.dec_counter('Inverter_Cnt')
await self.stream._ifc.publish_outstanding_mqtt()
self.stream.close()
await self.inverter.local.ifc.publish_outstanding_mqtt()
self.inverter.close()
class ModbusTcp():
@@ -61,9 +62,10 @@ class ModbusTcp():
'''Loop for receiving messages from the TSUN cloud (client-side)'''
while True:
try:
async with ModbusConn(host, port) as stream:
async with ModbusConn(host, port) as inverter:
stream = inverter.local.stream
await stream.send_start_cmd(snr, host)
await stream._ifc.loop()
await stream.ifc.loop()
logger.info(f'[{stream.node_id}:{stream.conn_no}] '
f'Connection closed - Shutdown: '
f'{stream.shutdown_started}')