never log the config

Class refact

CHANGELOG.md

@@ -7,7 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-###
+## [0.1.0] - 2023-10-06
+
+- refactoring of the connection classes
+- change user id on startup
+- register MQTT topics to home assistant, even if we have multiple inverters
+
+## [0.0.6] - 2023-10-03
+
+- Bump aiomqtt to version 1.2.1
+- Force MQTT registration when the home assistant has set the status to online again
+- fix control byte output in tx trace
+- dealloc async_stream instances in connection termination
 
 ## [0.0.5] - 2023-10-01
 

app/Dockerfile

@@ -37,12 +37,6 @@ ENV SERVICE_NAME=$SERVICE_NAME
 ENV UID=$UID
 ENV GID=$GID 
 
-
-
-RUN addgroup --gid $GID $SERVICE_NAME && \
-    adduser --ingroup $SERVICE_NAME --shell /bin/false --disabled-password --uid $UID $SERVICE_NAME && \
-    mkdir -p /home/$SERVICE_NAME/log /home/$SERVICE_NAME/config && \
-    chown -R $SERVICE_NAME:$SERVICE_NAME /home/$SERVICE_NAME
     
 # set the working directory in the container
 WORKDIR /home/$SERVICE_NAME

app/entrypoint.sh

@@ -3,12 +3,23 @@ set -e
 
 user="$(id -u)"
 echo "######################################################"
-echo "# start: '$SERVICE_NAME' Version:$VERSION"
-echo "# with UserID:$UID, GroupID:$GID"
-echo "######################################################"
+echo "# prepare: '$SERVICE_NAME' Version:$VERSION"
+echo "# for running with UserID:$UID, GroupID:$GID"
+echo "#"
 
 if [ "$user" = '0' ]; then
-	[ -d "/home/$SERVICE_NAME" ] && chown -R $SERVICE_NAME:$SERVICE_NAME /home/$SERVICE_NAME || true
+    mkdir -p /home/$SERVICE_NAME/log /home/$SERVICE_NAME/config
+    
+    if id $SERVICE_NAME ; then
+        echo "user still exists"
+    else
+        addgroup --gid $GID $SERVICE_NAME 2> /dev/null
+        adduser --ingroup $SERVICE_NAME --shell /bin/false --disabled-password --no-create-home --comment "" --uid $UID $SERVICE_NAME
+    fi    
+	chown -R $SERVICE_NAME:$SERVICE_NAME /home/$SERVICE_NAME || true
+    echo "######################################################"
+    echo "#"
+
     exec gosu $SERVICE_NAME "$@"
 else
     exec "$@"

app/requirements.txt

@@ -1,2 +1,2 @@
-    aiomqtt==1.2.0
-    schema
+    aiomqtt==1.2.1
+    schema==0.7.5

app/src/async_stream.py

@@ -1,22 +1,19 @@
-import logging, traceback, aiomqtt, json
+import logging, traceback
 from config import Config
+#import gc
 from messages import Message, hex_dump_memory
-from mqtt import Mqtt
 
 logger = logging.getLogger('conn')
-logger_mqtt = logging.getLogger('mqtt')
 
 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__()
-        self.proxy  = proxy
         self.reader = reader
         self.writer = writer
-        self.remoteStream = stream
-        self.addr = addr
+        self.remoteStream = remote_stream
         self.server_side = server_side
-        self.mqtt = Mqtt()
+        self.addr = addr
         self.unique_id = 0
         self.node_id = ''
         
@@ -24,47 +21,32 @@ class AsyncStream(Message):
     Our puplic methods
     '''
     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: 
        
             inverters = Config.get('inverters')
-            #logger_mqtt.debug(f'Inverters: {inverters}')
+            #logger.debug(f'Inverters: {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]
                 self.node_id = inv['node_id']
                 self.sug_area    = inv['suggested_area']
             else:    
-                logger_mqtt.debug(f'SerialNo {serial_no} not known!')
+                logger.debug(f'SerialNo {serial_no} not known!')
                 self.node_id = ''
                 self.sug_area    = ''
                 if not inverters['allow_all']:
                     self.unique_id = None
             
-                    logger_mqtt.error('ignore message from unknow inverter!')
+                    logger.error('ignore message from unknow inverter!')
                     return
 
             self.unique_id = serial_no
             
-            ha = Config.get('ha')
-            self.entitiy_prfx = ha['entity_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:
@@ -79,7 +61,7 @@ class AsyncStream(Message):
                 if self.unique_id: 
                     await self.__async_write()     
                     await self.__async_forward()
-                    await self.__async_publ_mqtt()
+                    await self.async_publ_mqtt()
                 
       
             except (ConnectionResetError,
@@ -94,12 +76,18 @@ class AsyncStream(Message):
                     f"{traceback.format_exc()}")
                 self.close()
                 return
+            
+    def disc(self) -> None:
+        logger.debug(f'in AsyncStream.disc() {self.addr}')
+        self.writer.close()
+            
         
     def close(self):
-        logger.info(f'in async_stream.close() {self.addr}')
+        logger.debug(f'in AsyncStream.close() {self.addr}')
         self.writer.close()
-        self.proxy = None
-        self.remoteStream = None
+        super().close()         # call close handler in the parent class
+
+#        logger.info (f'AsyncStream refs: {gc.get_referrers(self)}')
 
     
     '''
@@ -123,8 +111,7 @@ class AsyncStream(Message):
     async def __async_forward(self) -> None:
         if self._forward_buffer:
             if not self.remoteStream:
-                tsun = Config.get('tsun')
-                self.remoteStream = await self.proxy.CreateClientStream (self, tsun['host'], tsun['port'])
+                await self.async_create_remote()  # only implmeneted for server side => syncServerStream
                 
             if self.remoteStream:
                 hex_dump_memory(logging.DEBUG, f'Forward to {self.remoteStream.addr}:', self._forward_buffer, len(self._forward_buffer))
@@ -132,21 +119,14 @@ class AsyncStream(Message):
                 await self.remoteStream.writer.drain()                    
                 self._forward_buffer = bytearray(0)
 
-    async def __async_publ_mqtt(self) -> None:
-        if self.server_side:
-            db = self.db.db
+    async def async_create_remote(self) -> None:
+        pass
 
-            if self.new_data.keys() & {'inverter', 'collector'}:
-                await self.register_home_assistant()
+    async def async_publ_mqtt(self) -> None:
+        pass       
 
-            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):
-        logger.debug ("AsyncStream __del__")       
-            
+        logging.debug (f"AsyncStream.__del__  {self.addr}")     
+
 

app/src/config.py

@@ -60,7 +60,7 @@ class Config():
             config['inverters'] = def_config['inverters'] | usr_config['inverters']
 
             cls.config = cls.conf_schema.validate(config)
-            logging.debug(f'Readed config: "{cls.config}" ')
+            #logging.debug(f'Readed config: "{cls.config}" ')
  
         except Exception as error:
             logger.error(f'Config.read: {error}')

app/src/inverter.py

@@ -0,0 +1,103 @@
+import asyncio, logging, traceback, json
+from config import Config
+from async_stream import AsyncStream
+from mqtt import Mqtt
+#import gc
+
+logger = logging.getLogger('conn')
+
+
+
+class Inverter(AsyncStream):
+
+    def __init__ (self, reader, writer, addr):
+        super().__init__(reader, writer, addr, None, True)
+        self.mqtt = Mqtt()
+        self.ha_restarts = 0
+        ha = Config.get('ha')
+        self.entitiy_prfx = ha['entity_prefix'] + '/'
+        self.discovery_prfx = ha['discovery_prefix'] + '/'
+
+
+    async def server_loop(self, addr):
+        '''Loop for receiving messages from the inverter (server-side)'''
+        logger.info(f'Accept connection from  {addr}')        
+        await self.loop()
+        logging.info(f'Server loop stopped for {addr}')
+        
+        # if the server connection closes, we also have to disconnect the connection to te TSUN cloud
+        if self.remoteStream:
+            logging.debug ("disconnect client connection")
+            self.remoteStream.disc()
+        
+    async def client_loop(self, addr):
+        '''Loop for receiving messages from the TSUN cloud (client-side)'''
+        await self.remoteStream.loop()    
+        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
+        # connection stream, thus on the next received packet from the inverter, we can establish a new connection 
+        # to the TSUN cloud
+        self.remoteStream.remoteStream = None    # erase backlink to inverter instance
+        self.remoteStream = None                 # than erase client connection 
+        
+    async def async_create_remote(self) -> None:
+        '''Establish a client connection to the TSUN cloud'''
+        tsun = Config.get('tsun')
+        host = tsun['host']
+        port = tsun['port']     
+        addr = (host, port)
+            
+        try:
+            logging.info(f'Connected to {addr}')
+            connect = asyncio.open_connection(host, port)
+            reader, writer = await connect    
+            self.remoteStream = AsyncStream(reader, writer, addr, self, False)
+            asyncio.create_task(self.client_loop(addr))
+            
+        except ConnectionRefusedError as error:
+            logging.info(f'{error}')
+        except Exception:
+            logging.error(
+                f"Inverter: Exception for {addr}:\n"
+                f"{traceback.format_exc()}")
+        
+    
+
+    async def async_publ_mqtt(self) -> None:
+        '''puplish data to MQTT broker'''
+        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 (('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_home_assistant()
+            self.ha_restarts = self.mqtt.ha_restarts
+
+        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) -> None:
+        '''register all our topics at home assistant'''
+        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"Inverter: Exception:\n"
+                f"{traceback.format_exc()}")
+            
+    def close(self) -> None:
+        logging.debug(f'Inverter.close() {self.addr}')
+        super().close()         # call close handler in the parent class
+#        logger.debug (f'Inverter refs: {gc.get_referrers(self)}')
+
+
+    def __del__ (self):
+        logging.debug ("Inverter.__del__")
+        super().__del__()  

app/src/logging.ini

@@ -25,13 +25,13 @@ qualname=conn
 
 [logger_data]
 level=DEBUG
-handlers=console_handler,file_handler_name1,file_handler_name2
+handlers=file_handler_name1,file_handler_name2
 propagate=0
 qualname=data
 
 [logger_mqtt]
-level=DEBUG
-handlers=console_handler,file_handler_name1,file_handler_name2
+level=INFO
+handlers=console_handler,file_handler_name1
 propagate=0
 qualname=mqtt
 
@@ -43,12 +43,12 @@ qualname=tracer
 
 [handler_console_handler]
 class=StreamHandler
-level=INFO
+level=DEBUG
 formatter=console_formatter
 
 [handler_file_handler_name1]
 class=handlers.TimedRotatingFileHandler
-level=NOTSET
+level=INFO
 formatter=file_formatter
 args=('log/proxy.log', when:='midnight')
 

app/src/messages.py

@@ -100,6 +100,13 @@ class Message(metaclass=IterRegistry):
     '''
     Our puplic methods
     '''
+    def close(self) -> None:
+        # we have refernces to methods of this class in self.switch
+        # so we have to erase self.switch, otherwise this instance can't be 
+        # deallocated by the garbage collector ==> we get a memory leak
+        del self.switch 
+        
+
     def read(self) -> None:
         self._read()
         
@@ -186,7 +193,7 @@ class Message(metaclass=IterRegistry):
         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, self.msg_id)
         fnc = self.switch.get(self.msg_id, self.msg_unknown)
-        logger.info(self.__flow_str(self.server_side, 'tx') + f' Ctl: {int(self.ctrl):#02x} Msg: {fnc.__name__!r}' )
+        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
@@ -287,11 +294,9 @@ class Message(metaclass=IterRegistry):
 
 
     def msg_unknown(self):
+        logger.error (f"Unknow Msg: ID:{self.msg_id}")       
         self.forward(self._recv_buffer, self.header_len+self.data_len)
 
-
-    def __del__ (self):
-        logger.debug ("Messages __del__")       
         
 
 

app/src/mqtt.py

@@ -16,13 +16,22 @@ class Singleton(type):
 
 class Mqtt(metaclass=Singleton):
     client = None
-
+    
     def __init__(self):
         logger_mqtt.debug(f'MQTT: __init__') 
         loop = asyncio.get_event_loop()
         self.task = loop.create_task(self.__loop())
-                
+        self.ha_restarts = 0        
+
+    
+    @property
+    def ha_restarts(self):
+        return self._ha_restarts
         
+    @ha_restarts.setter
+    def ha_restarts(self, value):
+        self._ha_restarts = value
+            
     def __del__(self):
         logger_mqtt.debug(f'MQTT: __del__')
     
@@ -55,7 +64,11 @@ class Mqtt(metaclass=Singleton):
                     async with self.client.messages() as messages:
                         await self.client.subscribe(f"{ha['auto_conf_prefix']}/status")
                         async for message in messages:
-                            logger_mqtt.info(f'Home-Assistant Status: {message.payload.decode("UTF-8")}')
+                            status = message.payload.decode("UTF-8")
+                            logger_mqtt.info(f'Home-Assistant Status: {status}')
+                            if status == 'online':
+                                self.ha_restarts += 1
+
             except aiomqtt.MqttError:
                 logger_mqtt.info(f"Connection lost; Reconnecting in {interval} seconds ...")
                 await asyncio.sleep(interval)

app/src/proxy.py

@@ -1,43 +0,0 @@
-import asyncio, logging, traceback
-from async_stream import AsyncStream
-
-class Proxy:
-    def __init__ (proxy, reader, writer, addr):
-        proxy.ServerStream = AsyncStream(proxy, reader, writer, addr)
-        proxy.ClientStream = None
-
-    async def server_loop(proxy, addr):
-        logging.info(f'Accept connection from {addr}')        
-        await proxy.ServerStream.loop()
-        logging.info(f'Close server connection {addr}')
-        
-        if proxy.ClientStream:
-            logging.debug ("close client connection")
-            proxy.ClientStream.close()
-        
-    async def client_loop(proxy, addr):
-        await proxy.ClientStream.loop()    
-        logging.info(f'Close client connection {addr}')
-        proxy.ServerStream.remoteStream = None
-        proxy.ClientStream = None
-        
-    async def CreateClientStream (proxy, stream, host, port):
-        addr = (host, port)
-            
-        try:
-            logging.info(f'Connected to {addr}')
-            connect = asyncio.open_connection(host, port)
-            reader, writer = await connect    
-            proxy.ClientStream = AsyncStream(proxy, reader, writer, addr, stream, server_side=False)
-            asyncio.create_task(proxy.client_loop(addr))
-            
-        except ConnectionRefusedError as error:
-            logging.info(f'{error}')
-        except Exception:
-            logging.error(
-                f"Proxy: Exception for {addr}:\n"
-                f"{traceback.format_exc()}")
-        return proxy.ClientStream
-        
-    def __del__ (proxy):
-        logging.debug ("Proxy __del__")

app/src/server.py

@@ -1,8 +1,7 @@
 import logging, asyncio, signal, functools, os
-#from logging.handlers import TimedRotatingFileHandler
 from logging import config
 from async_stream import AsyncStream
-from proxy import Proxy
+from inverter import Inverter
 from config import Config
 from mqtt import Mqtt
 
@@ -11,7 +10,7 @@ async def handle_client(reader, writer):
     '''Handles a new incoming connection and starts an async loop'''
 
     addr = writer.get_extra_info('peername')
-    await Proxy(reader, writer, addr).server_loop(addr) 
+    await Inverter(reader, writer, addr).server_loop(addr) 
 
 
 def handle_SIGTERM(loop):
@@ -46,6 +45,7 @@ if __name__ == "__main__":
 
     logging.config.fileConfig('logging.ini')
     logging.info(f'Server "{serv_name} - {version}" will be started')
+    logging.getLogger().setLevel(logging.DEBUG if __debug__ else logging.INFO)
     
     # read config file
     Config.read()