Merge pull request #40 from avollkopf/development

Merge changes from development branch

cbpi/__init__.py

@@ -1 +1 @@
-__version__ = "4.0.1.7"
+__version__ = "4.0.1.14"

cbpi/api/base.py

@@ -51,10 +51,10 @@ class CBPiBase(metaclass=ABCMeta):
     def get_actor_state(self,id):
         try:
             actor = self.cbpi.actor.find_by_id(id)
-            return actor.get("instance").get_state()
+            return actor.instance.state
         except:
             logging.error("Failed to read actor state in step - actor {}".format(id))
-            return None
+            return False 
 
     async def actor_on(self,id,power=100):
         

cbpi/controller/actor_controller.py

@@ -9,7 +9,6 @@ class ActorController(BasicController):
         self.update_key = "actorupdate"
 
     async def on(self, id, power=None):
-#        logging.info("Controller_power: {}".format(power))
         try:
             item = self.find_by_id(id)
             if power is None:
@@ -20,7 +19,8 @@ class ActorController(BasicController):
                     power = 100
             if item.instance.state is False:
                 await item.instance.on(power)
-                await self.push_udpate()
+                #await self.push_udpate()
+                self.cbpi.ws.send(dict(topic=self.update_key, data=list(map(lambda item: item.to_dict(), self.data))))
                 self.cbpi.push_update("cbpi/actorupdate/{}".format(id), item.to_dict(), True)
             else:
                 await self.set_power(id, power)
@@ -33,7 +33,8 @@ class ActorController(BasicController):
             item = self.find_by_id(id)
             if item.instance.state is True:
                 await item.instance.off()
-                await self.push_udpate()
+                #await self.push_udpate()
+                self.cbpi.ws.send(dict(topic=self.update_key, data=list(map(lambda item: item.to_dict(), self.data))))
                 self.cbpi.push_update("cbpi/actorupdate/{}".format(id), item.to_dict())
         except Exception as e:
             logging.error("Failed to switch on Actor {} {}".format(id, e), True)
@@ -43,6 +44,7 @@ class ActorController(BasicController):
             item = self.find_by_id(id)
             instance = item.get("instance")
             await instance.toggle()
+            self.cbpi.ws.send(dict(topic=self.update_key, data=list(map(lambda item: item.to_dict(), self.data))))
             self.cbpi.push_update("cbpi/actorupdate/{}".format(id), item.to_dict())
         except Exception as e:
             logging.error("Failed to toggle Actor {} {}".format(id, e))
@@ -58,7 +60,8 @@ class ActorController(BasicController):
         try:
             item = self.find_by_id(id)
             item.power = round(power)
-            await self.push_udpate()
+            #await self.push_udpate()
+            self.cbpi.ws.send(dict(topic=self.update_key, data=list(map(lambda item: item.to_dict(), self.data))))
             self.cbpi.push_update("cbpi/actorupdate/{}".format(id), item.to_dict())
         except Exception as e:
             logging.error("Failed to update Actor {} {}".format(id, e))

cbpi/controller/fermentation_controller.py

@@ -199,7 +199,8 @@ class FermentationController:
 
     def get_state(self):
         if self.data == []:
-            logging.info(self.data)
+            #logging.info(self.data)
+            pass
 
         return {"data": list(map(lambda x: x.to_dict(), self.data)), "types":self.get_types(), "steptypes":self.get_steptypes()}
 

cbpi/controller/log_file_controller.py

@@ -26,15 +26,6 @@ class LogController:
         self.configuration = False
         self.datalogger = {}
 
-#        self.cbpi.config.get does not seem to work here...
-#        self.influxdbaddr="192.168.163.105"
-#        self.influxdbport="8086"
-#        self.influxdbname="cbpi4"
-#        self.influxdburl='http://' + self.influxdbaddr + ':' + str(self.influxdbport) + '/write?db=' + self.influxdbname
-#        self.influxdbuser=""
-#        self.influxdbpwd=""
-#        self.base64string = base64.b64encode(('%s:%s' % (self.influxdbuser,self.influxdbpwd)).encode())
-
     def log_data(self, name: str, value: str) -> None:
         self.logfiles = self.cbpi.config.get("CSVLOGFILES", "Yes")
         self.influxdb = self.cbpi.config.get("INFLUXDB", "No")
@@ -52,22 +43,39 @@ class LogController:
 
             formatted_time = strftime("%Y-%m-%d %H:%M:%S", localtime())
             self.datalogger[name].info("%s,%s" % (formatted_time, value))
-
         if self.influxdb == "Yes":
-            self.influxdb = self.cbpi.config.get("INFLUXDB", "No")
+            self.influxdbcloud = self.cbpi.config.get("INFLUXDBCLOUD", "No")
             self.influxdbaddr = self.cbpi.config.get("INFLUXDBADDR", None)
             self.influxdbport = self.cbpi.config.get("INFLUXDBPORT", None)
             self.influxdbname = self.cbpi.config.get("INFLUXDBNAME", None)
             self.influxdbuser = self.cbpi.config.get("INFLUXDBUSER", None)
             self.influxdbpwd = self.cbpi.config.get("INFLUXDBPWD", None)
+            
+            id = name
+            try:
+                chars = {'ö':'oe','ä':'ae','ü':'ue','Ö':'Oe','Ä':'Ae','Ü':'Ue'}
+                sensor=self.cbpi.sensor.find_by_id(name)
+                if sensor is not None:
+                    itemname=sensor.name.replace(" ", "_")
+                    for char in chars:
+                        itemname = itemname.replace(char,chars[char])
+                    out="measurement,source=" + itemname + ",itemID=" + str(id) + " value="+str(value)
+            except Exception as e:
+                logging.error("InfluxDB ID Error: {}".format(e))
+
+            if self.influxdbcloud == "Yes":
+                self.influxdburl="https://" + self.influxdbaddr + "/api/v2/write?org=" + self.influxdbuser + "&bucket=" + self.influxdbname + "&precision=s"
+                try:
+                    header = {'User-Agent': name, 'Authorization': "Token {}".format(self.influxdbpwd)}
+                    http = urllib3.PoolManager()
+                    req = http.request('POST',self.influxdburl, body=out, headers = header)
+                except Exception as e:
+                    logging.error("InfluxDB cloud write Error: {}".format(e))
+
+            else:
                 self.base64string = base64.b64encode(('%s:%s' % (self.influxdbuser,self.influxdbpwd)).encode())
                 self.influxdburl='http://' + self.influxdbaddr + ':' + str(self.influxdbport) + '/write?db=' + self.influxdbname
-
-
                 try:
-                sensor=self.cbpi.sensor.find_by_id(name)
-                sensorname=sensor.name.replace(" ", "_")
-                out="measurement,source=" + sensorname + "___" + name + " value="+str(value)
                     header = {'User-Agent': name, 'Content-Type': 'application/x-www-form-urlencoded','Authorization': 'Basic %s' % self.base64string.decode('utf-8')}
                     http = urllib3.PoolManager()
                     req = http.request('POST',self.influxdburl, body=out, headers = header)

cbpi/controller/satellite_controller.py

@@ -1,5 +1,4 @@
 
-
 import asyncio
 import json
 from re import M
@@ -8,11 +7,14 @@ from contextlib import AsyncExitStack, asynccontextmanager
 from cbpi import __version__
 import logging
 
-
 class SatelliteController:
 
     def __init__(self, cbpi):
         self.cbpi = cbpi
+        self.kettlecontroller = cbpi.kettle
+        self.fermentercontroller = cbpi.fermenter
+        self.sensorcontroller = cbpi.sensor
+        self.actorcontroller = cbpi.actor
         self.logger = logging.getLogger(__name__)
         self.host = cbpi.static_config.get("mqtt_host", "localhost")
         self.port = cbpi.static_config.get("mqtt_port", 1883)
@@ -23,6 +25,11 @@ class SatelliteController:
             ("cbpi/actor/+/on", self._actor_on),
             ("cbpi/actor/+/off", self._actor_off),
             ("cbpi/actor/+/power", self._actor_power),
+            ("cbpi/updateactor", self._actorupdate),
+            ("cbpi/updatekettle", self._kettleupdate),
+            ("cbpi/updatesensor", self._sensorupdate),
+            ("cbpi/updatefermenter", self._fermenterupdate),
+        
         ]
         self.tasks = set()
 
@@ -63,12 +70,48 @@ class SatelliteController:
                     if power < 0:
                         power = 0
                     await self.cbpi.actor.set_power(topic_key[2],power)
-                    await self.cbpi.actor.actor_update(topic_key[2],power)
+                    #await self.cbpi.actor.actor_update(topic_key[2],power)
                 except:
                     self.logger.warning("Failed to set actor power via mqtt. No valid power in message")
             except:
                 self.logger.warning("Failed to set actor power via mqtt")
 
+    async def _kettleupdate(self, messages):
+        async for message in messages:
+            try:
+                self.kettle=self.kettlecontroller.get_state()
+                for item in self.kettle['data']:
+                    self.cbpi.push_update("cbpi/{}/{}".format("kettleupdate",item['id']), item)
+            except Exception as e:
+                self.logger.warning("Failed to send kettleupdate via mqtt: {}".format(e))
+
+    async def _fermenterupdate(self, messages):
+        async for message in messages:
+            try:
+                self.fermenter=self.fermentercontroller.get_state()
+                for item in self.fermenter['data']:
+                    self.cbpi.push_update("cbpi/{}/{}".format("fermenterupdate",item['id']), item)
+            except Exception as e:
+                self.logger.warning("Failed to send fermenterupdate via mqtt: {}".format(e))
+
+    async def _actorupdate(self, messages):
+        async for message in messages:
+            try:
+                self.actor=self.actorcontroller.get_state()
+                for item in self.actor['data']:
+                    self.cbpi.push_update("cbpi/{}/{}".format("actorupdate",item['id']), item)
+            except Exception as e:
+                self.logger.warning("Failed to send actorupdate via mqtt: {}".format(e))
+
+    async def _sensorupdate(self, messages):
+        async for message in messages:
+            try:
+                self.sensor=self.sensorcontroller.get_state()
+                for item in self.sensor['data']:
+                    self.cbpi.push_update("cbpi/{}/{}".format("sensorupdate",item['id']), item)
+            except Exception as e:
+                self.logger.warning("Failed to send sensorupdate via mqtt: {}".format(e))
+
     def subcribe(self, topic, method):
         task = asyncio.create_task(self._subcribe(topic, method))
         return task

cbpi/craftbeerpi.py

@@ -109,6 +109,7 @@ class CraftBeerPi:
         self.log = LogController(self)
         self.system = SystemController(self)
         self.kettle = KettleController(self)
+        self.fermenter : FermentationController = FermentationController(self)
         self.step : StepController = StepController(self)
         self.recipe : RecipeController = RecipeController(self)
         self.upload : UploadController = UploadController(self)
@@ -117,7 +118,6 @@ class CraftBeerPi:
         if str(self.static_config.get("mqtt", False)).lower() == "true":
             self.satellite: SatelliteController = SatelliteController(self)
         self.dashboard = DashboardController(self)
-        self.fermenter : FermentationController = FermentationController(self)
 
         self.http_step = StepHttpEndpoints(self)
         self.http_recipe = RecipeHttpEndpoints(self)
@@ -285,6 +285,12 @@ class CraftBeerPi:
         
         self._swagger_setup()
 
+        level = logging.INFO
+        logger = logging.getLogger()
+        logger.setLevel(level)
+        for handler in logger.handlers:
+            handler.setLevel(level)
+
         return self.app
 
     def start(self):

cbpi/extension/ConfigUpdate/__init__.py

@@ -44,6 +44,8 @@ class ConfigUpdate(CBPiExtension):
         influxdbname = self.cbpi.config.get("INFLUXDBNAME", None)
         influxdbuser = self.cbpi.config.get("INFLUXDBUSER", None)
         influxdbpwd = self.cbpi.config.get("INFLUXDBPWD", None)
+        influxdbcloud = self.cbpi.config.get("INFLUXDBCLOUD", None)
+        mqttupdate = self.cbpi.config.get("MQTTUpdate", None)
 
 
 
@@ -202,7 +204,7 @@ class ConfigUpdate(CBPiExtension):
             except:
                 logger.warning('Unable to update config')
 
-       ## Check if CSV logfiles is on config 
+       ## Check if influxdb is on config 
         if influxdb is None:
             logger.info("INIT Influxdb")
             try:
@@ -216,7 +218,7 @@ class ConfigUpdate(CBPiExtension):
         if influxdbaddr is None:
             logger.info("INIT Influxdbaddr")
             try:
-                await self.cbpi.config.add("INFLUXDBADDR", "localhost", ConfigType.STRING, "IP Address of your influxdb server")
+                await self.cbpi.config.add("INFLUXDBADDR", "localhost", ConfigType.STRING, "IP Address of your influxdb server (If INFLUXDBCLOUD set to Yes use URL Address of your influxdb cloud server)")
             except:
                 logger.warning('Unable to update config')
 
@@ -232,26 +234,49 @@ class ConfigUpdate(CBPiExtension):
         if influxdbname is None:
             logger.info("INIT Influxdbname")
             try:
-                await self.cbpi.config.add("INFLUXDBNAME", "cbpi4", ConfigType.STRING, "Name of your influxdb database name")
+                await self.cbpi.config.add("INFLUXDBNAME", "cbpi4", ConfigType.STRING, "Name of your influxdb database name (If INFLUXDBCLOUD set to Yes use bucket of your influxdb cloud database)")
             except:
                 logger.warning('Unable to update config')
 
-        ## Check if influxdber is in config
+        ## Check if influxduser is in config
         if influxdbuser is None:
             logger.info("INIT Influxdbuser")
             try:
-                await self.cbpi.config.add("INFLUXDBUSER", " ", ConfigType.STRING, "User name for your influxdb database (only if required)")
+                await self.cbpi.config.add("INFLUXDBUSER", " ", ConfigType.STRING, "User name for your influxdb database (only if required)(If INFLUXDBCLOUD set to Yes use organisation of your influxdb cloud database)")
             except:
                 logger.warning('Unable to update config')
 
-        ## Check if influxdber is in config
+        ## Check if influxdpwd is in config
         if influxdbpwd is None:
             logger.info("INIT Influxdbpwd")
             try:
-                await self.cbpi.config.add("INFLUXDBPWD", " ", ConfigType.STRING, "Password for your influxdb database (only if required)")
+                await self.cbpi.config.add("INFLUXDBPWD", " ", ConfigType.STRING, "Password for your influxdb database (only if required)(If INFLUXDBCLOUD set to Yes use token of your influxdb cloud database)")
             except:
                 logger.warning('Unable to update config')
 
+       ## Check if influxdb cloud is on config 
+        if influxdbcloud is None:
+            logger.info("INIT influxdbcloud")
+            try:
+                await self.cbpi.config.add("INFLUXDBCLOUD", "No", ConfigType.SELECT, "Write sensor data to influxdb cloud (INFLUXDB must set to Yes)", 
+                                                                                                [{"label": "Yes", "value": "Yes"},
+                                                                                                {"label": "No", "value": "No"}])
+            except:
+                logger.warning('Unable to update config')
+
+        if mqttupdate is None:
+            logger.info("INIT MQTT update frequency for Kettles and Fermenters")
+            try:
+                await self.cbpi.config.add("MQTTUpdate", 0, ConfigType.SELECT, "Forced MQTT Update frequency in s for Kettle and Fermenter (no changes in payload required). Restart required after change",
+                                                                                                [{"label": "30", "value": 30},
+                                                                                                {"label": "60", "value": 60},
+                                                                                                {"label": "120", "value": 120},
+                                                                                                {"label": "300", "value": 300},
+                                                                                                {"label": "Never", "value": 0}])
+            except:
+                logger.warning('Unable to update database')
+
+
 def setup(cbpi):
     cbpi.plugin.register("ConfigUpdate", ConfigUpdate)
     pass

cbpi/extension/gpioactor/__init__.py

@@ -61,7 +61,7 @@ class GPIOActor(CBPiActor):
             self.power = power
         else: 
             self.power = 100
-        await self.set_power(self.power)
+#        await self.set_power(self.power)
 
         logger.info("ACTOR %s ON - GPIO %s " %  (self.id, self.gpio))
         GPIO.output(self.gpio, self.get_GPIO_state(1))  
@@ -123,21 +123,21 @@ class GPIOPWMActor(CBPiActor):
         pass
 
     async def on(self, power = None):
-        logging.info("PWM Actor Power: {}".format(power))
+        logging.debug("PWM Actor Power: {}".format(power))
         if power is not None:
             self.power = power
         else:
             self.power = 100
 
-        logging.info("PWM Final Power: {}".format(self.power))    
+        logging.debug("PWM Final Power: {}".format(self.power))    
         
-        logger.info("PWM ACTOR %s ON - GPIO %s - Frequency %s - Power %s" %  (self.id, self.gpio,self.frequency,self.power))
+        logger.debug("PWM ACTOR %s ON - GPIO %s - Frequency %s - Power %s" %  (self.id, self.gpio,self.frequency,self.power))
         try:
             if self.p is None:
                 self.p = GPIO.PWM(int(self.gpio), float(self.frequency))
             self.p.start(self.power)
             self.state = True
-            await self.cbpi.actor.actor_update(self.id,self.power)
+#            await self.cbpi.actor.actor_update(self.id,self.power)
         except:
             pass
 

cbpi/extension/mqtt_util/__init__.py

@@ -0,0 +1,74 @@
+import logging
+import asyncio
+from cbpi.api import *
+from cbpi.api.config import ConfigType
+from cbpi.api.base import CBPiBase
+from cbpi.controller.fermentation_controller import FermentationController
+from cbpi.controller.kettle_controller import KettleController
+
+logger = logging.getLogger(__name__)
+
+class MQTTUtil(CBPiExtension):
+
+    def __init__(self,cbpi):
+        self.cbpi = cbpi
+        self.kettlecontroller = cbpi.kettle
+        self.fermentationcontroller = cbpi.fermenter
+        # sensor and actor update is done anyhow during startup
+#        self.sensorcontroller = cbpi.sensor
+#        self.actorcontroller = cbpi.actor
+
+        self.mqttupdate = int(self.cbpi.config.get("MQTTUpdate", 0))
+        if self.mqttupdate != 0:
+            self._task = asyncio.create_task(self.run())
+            logger.info("INIT MQTTUtil")
+        else:
+            self._task = asyncio.create_task(self.push_once())
+
+    async def push_once(self):
+        # wait some time to ensure that kettlecontroller is started
+        await asyncio.sleep(5)
+        self.push_update()
+
+
+    async def run(self):
+
+        while True:
+            self.push_update()
+            await asyncio.sleep(self.mqttupdate)
+
+    def push_update(self):
+#        try:
+#            self.actor=self.actorcontroller.get_state()
+#            for item in self.actor['data']:
+#                self.cbpi.push_update("cbpi/{}/{}".format("actorupdate",item['id']), item)
+#        except Exception as e:
+#            logging.error(e)
+#            pass
+#        try:
+#            self.sensor=self.sensorcontroller.get_state()
+#            for item in self.sensor['data']:
+#                self.cbpi.push_update("cbpi/{}/{}".format("sensorupdate",item['id']), item)
+#        except Exception as e:
+#            logging.error(e)
+#            pass
+        try:
+            self.kettle=self.kettlecontroller.get_state()
+            for item in self.kettle['data']:
+                self.cbpi.push_update("cbpi/{}/{}".format("kettleupdate",item['id']), item)
+        except Exception as e:
+            logging.error(e)
+            pass
+        try:
+            self.fermenter=self.fermentationcontroller.get_state()
+            for item in self.fermenter['data']:
+                self.cbpi.push_update("cbpi/{}/{}".format("fermenterupdate",item['id']), item)
+        except Exception as e:
+            logging.error(e)
+            pass
+
+
+def setup(cbpi):
+    if str(cbpi.static_config.get("mqtt", False)).lower() == "true":
+        cbpi.plugin.register("MQTTUtil", MQTTUtil)
+    pass

cbpi/extension/mqtt_util/config.yaml

@@ -0,0 +1,3 @@
+name: MQTTUtil
+version: 4
+active: true

setup.py

@@ -2,6 +2,18 @@ from setuptools import setup, find_packages
 from cbpi import __version__
 import platform
 
+# read the contents of your README file
+from os import popen
+
+localsystem = platform.system()
+raspberrypi=False
+if localsystem == "Linux":
+    command="cat /proc/cpuinfo | grep Raspberry"
+    model=popen(command).read()
+    if len(model) != 0:
+        raspberrypi=True
+
+
 setup(name='cbpi',
       version=__version__,
       description='CraftBeerPi',
@@ -30,16 +42,17 @@ setup(name='cbpi',
           "requests==2.25.1",
           "voluptuous==0.12.1",
           "pyfiglet==0.8.post1",
-          'pandas==1.1.5',
           'click==7.1.2',
           'shortuuid==1.0.1',
           'tabulate==0.8.7',
           'asyncio-mqtt',
           'psutil==5.8.0',
-          'numpy==1.20.3',
           'cbpi4ui',
           'importlib_metadata'] + (
-              ['RPi.GPIO==0.7.1a4'] if platform.uname()[1] == "raspberrypi" else [] ),
+              ['RPi.GPIO==0.7.1a4'] if raspberrypi else [] ) +
+              (['numpy==1.22.0'] if (int(platform.python_version_tuple()[1]) >= 9) and (int(platform.python_version_tuple()[0]) == 3) else ['numpy==1.20.3'] ) +
+                (['pandas==1.4.0'] if (int(platform.python_version_tuple()[1]) >= 9) and (int(platform.python_version_tuple()[0]) == 3) else ['pandas==1.1.5'] ),
+
 
         dependency_links=[
         'https://testpypi.python.org/pypi',