Merge pull request #116 from PiBrewing/development

merge from Development

cbpi/__init__.py

@@ -1,3 +1,3 @@
-__version__ = "4.1.10"
+__version__ = "4.1.11"
 __codename__ = "Groundhog Day"
 

cbpi/controller/satellite_controller.py

@@ -2,7 +2,7 @@
 import asyncio
 import json
 from re import M
-from asyncio_mqtt import Client, MqttError, Will
+from aiomqtt import Client, MqttError, Will
 from contextlib import AsyncExitStack, asynccontextmanager
 from cbpi import __version__
 import logging
@@ -73,7 +73,7 @@ async def listen(self):
             except asyncio.CancelledError:
                 # Cancel
                 self.logger.warning("MQTT Listening Cancelled")
-                #break
+                break
             except MqttError as e:
                 self.logger.error("MQTT Exception: {}".format(e))
             except Exception as e:

cbpi/extension/SensorLogTarget_InfluxDB/__init__.py

@@ -1,7 +1,7 @@
 
 # -*- coding: utf-8 -*-
 import os
-from urllib3 import Timeout, PoolManager
+from urllib3 import Timeout, PoolManager, Retry
 import logging
 from unittest.mock import MagicMock, patch
 import asyncio
@@ -21,6 +21,9 @@ def __init__(self, cbpi): # called from cbpi on start
         if self.influxdb == "No":
             return # never run()
         self._task = asyncio.create_task(self.run()) # one time run() only
+        self.counter = 0
+        self.max_retries = 2
+        self.send=True
 
 
     async def run(self): # called by __init__ once on start if influx is enabled
@@ -40,7 +43,7 @@ async def log_data_to_InfluxDB(self, cbpi, id:str, value:str, timestamp, name):
         self.influxdbuser = self.cbpi.config.get("INFLUXDBUSER", None)
         self.influxdbpwd = self.cbpi.config.get("INFLUXDBPWD", None)
         self.influxdbmeasurement = self.cbpi.config.get("INFLUXDBMEASUREMENT", "measurement")
-        timeout = Timeout(connect=5.0, read=None)
+        timeout = Timeout(connect=2.0, read=None)
         try:
             sensor=self.cbpi.sensor.find_by_id(id)
             if sensor is not None:
@@ -49,28 +52,45 @@ async def log_data_to_InfluxDB(self, cbpi, id:str, value:str, timestamp, name):
         except Exception as e:
             logging.error("InfluxDB ID Error: {}".format(e))
 
-        if self.influxdbcloud == "Yes":
-            self.influxdburl=self.influxdbaddr + "/api/v2/write?org=" + self.influxdbuser + "&bucket=" + self.influxdbname + "&precision=s"
-            try:
-                header = {'User-Agent': id, 'Authorization': "Token {}".format(self.influxdbpwd)}
-                http = PoolManager(timeout=timeout)
-                req = http.request('POST',self.influxdburl, body=out.encode(), headers = header)
-                if req.status != 204:
-                    raise Exception(f'InfluxDB Status code {req.status}')
-            except Exception as e:
-                logging.error("InfluxDB cloud write Error: {}".format(e))
+        if self.influxdbcloud == "Yes" and self.send == True:
+            if self.counter <= self.max_retries:
+                self.influxdburl=self.influxdbaddr + "/api/v2/write?org=" + self.influxdbuser + "&bucket=" + self.influxdbname + "&precision=s"
+                try:
+                    header = {'User-Agent': id, 'Authorization': "Token {}".format(self.influxdbpwd)}
+                    http = PoolManager(timeout=timeout)
+                    req = http.request('POST',self.influxdburl, body=out.encode(), headers = header, retries=Retry(2))
+                    if req.status != 204:
+                        raise Exception(f'InfluxDB Status code {req.status}')
+                except Exception as e:
+                    self.counter += 1
+                    logging.error("InfluxDB cloud write Error #{}: {}".format(self.counter, e))
+            else:
+                logging.warning("Waiting 3 Minutes before connecting to INFLUXDB again")
+                self.send=False
+                await asyncio.sleep(180)
+                self.counter = 0
+                self.send = True
+
+        elif self.influxdbcloud == "No" and self.send == True:
+            if self.counter <= self.max_retries:
+                self.base64string = base64.b64encode(('%s:%s' % (self.influxdbuser,self.influxdbpwd)).encode())
+                self.influxdburl= self.influxdbaddr + '/write?db=' + self.influxdbname
+                try:
+                    header = {'User-Agent': id, 'Content-Type': 'application/x-www-form-urlencoded','Authorization': 'Basic %s' % self.base64string.decode('utf-8')}
+                    http = PoolManager(timeout=timeout)
+                    req = http.request('POST',self.influxdburl, body=out.encode(), headers = header, retries=Retry(2))
+                    if req.status != 204:
+                        raise Exception(f'InfluxDB Status code {req.status}')
+                except Exception as e:
+                    self.counter += 1
+                    logging.error("InfluxDB write Error #{}: {}".format(self.counter, e))
+            else:
+                logging.warning("Waiting 3 Minutes before connecting to INFLUXDB again")
+                self.send=False
+                await asyncio.sleep(180)
+                self.counter = 0
+                self.send = True
 
-        else:
-            self.base64string = base64.b64encode(('%s:%s' % (self.influxdbuser,self.influxdbpwd)).encode())
-            self.influxdburl= self.influxdbaddr + '/write?db=' + self.influxdbname
-            try:
-                header = {'User-Agent': id, 'Content-Type': 'application/x-www-form-urlencoded','Authorization': 'Basic %s' % self.base64string.decode('utf-8')}
-                http = PoolManager(timeout=timeout)
-                req = http.request('POST',self.influxdburl, body=out.encode(), headers = header)
-                if req.status != 204:
-                    raise Exception(f'InfluxDB Status code {req.status}')
-            except Exception as e:
-                logging.error("InfluxDB write Error: {}".format(e))
 
 def setup(cbpi):
     cbpi.plugin.register("SensorLogTargetInfluxDB", SensorLogTargetInfluxDB)

requirements.txt

@@ -1,13 +1,14 @@
 typing-extensions>=4
-aiohttp==3.8.3
+aiohttp==3.8.5
 aiohttp-auth==0.1.1 
 aiohttp-route-decorator==0.1.4 
 aiohttp-security==0.4.0 
 aiohttp-session==2.12.0 
 aiohttp-swagger==1.0.16 
 aiojobs==1.1.0 
 aiosqlite==0.17.0 
-cryptography==40.0.1 
+cryptography==41.0.2
+pyopenssl==23.2.0
 requests==2.31.0 
 voluptuous==0.13.1 
 pyfiglet==0.8.post1 
@@ -18,7 +19,7 @@ numpy==1.24.1
 cbpi4gui
 click==8.1.3
 importlib_metadata==4.11.1
-asyncio-mqtt==0.16.1
+aiomqtt==1.0.0
 psutil==5.9.4
 zipp>=0.5
 colorama==0.4.6

setup.py

@@ -39,22 +39,23 @@
 	    long_description_content_type='text/markdown',
       install_requires=[
           "typing-extensions>=4",
-          "aiohttp==3.8.4",
+          "aiohttp==3.8.5",
           "aiohttp-auth==0.1.1",
           "aiohttp-route-decorator==0.1.4",
           "aiohttp-security==0.4.0",
           "aiohttp-session==2.12.0",
           "aiohttp-swagger==1.0.16",
           "aiojobs==1.1.0 ",
           "aiosqlite==0.17.0",
-          "cryptography==40.0.1",
+          "cryptography==41.0.2",
+          "pyopenssl==23.2.0",
           "requests==2.31.0",
           "voluptuous==0.13.1",
           "pyfiglet==0.8.post1",
           'click==8.1.3',
           'shortuuid==1.0.11',
           'tabulate==0.9.0',
-          'asyncio-mqtt==0.16.1',
+          'aiomqtt==1.0.0',
           'inquirer==3.1.1',
           'colorama==0.4.6',
           'psutil==5.9.4',