diff --git a/cbpi/controller/log_file_controller.py b/cbpi/controller/log_file_controller.py new file mode 100644 index 0000000..968929f --- /dev/null +++ b/cbpi/controller/log_file_controller.py @@ -0,0 +1,224 @@ +import datetime +import glob +import logging +import os +from logging.handlers import RotatingFileHandler +from time import strftime, localtime +import pandas as pd +import zipfile +import base64 +import urllib3 +from cbpi.api import * +from cbpi.api.config import ConfigType +from cbpi.api.base import CBPiBase +import asyncio + + +class LogController: + + def __init__(self, cbpi): + ''' + + :param cbpi: craftbeerpi object + ''' + self.cbpi = cbpi + self.logger = logging.getLogger(__name__) + self.configuration = False + self.datalogger = {} + + 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") + if self.logfiles == "Yes": + if name not in self.datalogger: + max_bytes = int(self.cbpi.config.get("SENSOR_LOG_MAX_BYTES", 100000)) + backup_count = int(self.cbpi.config.get("SENSOR_LOG_BACKUP_COUNT", 3)) + + data_logger = logging.getLogger('cbpi.sensor.%s' % name) + data_logger.propagate = False + data_logger.setLevel(logging.DEBUG) + handler = RotatingFileHandler('./logs/sensor_%s.log' % name, maxBytes=max_bytes, backupCount=backup_count) + data_logger.addHandler(handler) + self.datalogger[name] = data_logger + + formatted_time = strftime("%Y-%m-%d %H:%M:%S", localtime()) + self.datalogger[name].info("%s,%s" % (formatted_time, str(value))) + if self.influxdb == "Yes": + 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: + 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) + except Exception as e: + logging.error("InfluxDB write Error: {}".format(e)) + + + + async def get_data(self, names, sample_rate='60s'): + logging.info("Start Log for {}".format(names)) + ''' + :param names: name as string or list of names as string + :param sample_rate: rate for resampling the data + :return: + ''' + # make string to array + if isinstance(names, list) is False: + names = [names] + + # remove duplicates + names = set(names) + + + result = None + + def dateparse(time_in_secs): + ''' + Internal helper for date parsing + :param time_in_secs: + :return: + ''' + return datetime.datetime.strptime(time_in_secs, '%Y-%m-%d %H:%M:%S') + + def datetime_to_str(o): + if isinstance(o, datetime.datetime): + return o.__str__() + + for name in names: + # get all log names + all_filenames = glob.glob('./logs/sensor_%s.log*' % name) + # concat all logs + df = pd.concat([pd.read_csv(f, parse_dates=True, date_parser=dateparse, index_col='DateTime', names=['DateTime', name], header=None) for f in all_filenames]) + logging.info("Read all files for {}".format(names)) + # resample if rate provided + if sample_rate is not None: + df = df[name].resample(sample_rate).max() + logging.info("Sampled now for {}".format(names)) + df = df.dropna() + # take every nth row so that total number of rows does not exceed max_rows * 2 + max_rows = 500 + total_rows = df.shape[0] + if (total_rows > 0) and (total_rows > max_rows): + nth = int(total_rows/max_rows) + if nth > 1: + df = df.iloc[::nth] + + if result is None: + result = df + else: + result = pd.merge(result, df, how='outer', left_index=True, right_index=True) + + data = {"time": df.index.tolist()} + + if len(names) > 1: + for name in names: + data[name] = result[name].interpolate(limit_direction='both', limit=10).tolist() + else: + data[name] = result.interpolate().tolist() + + logging.info("Send Log for {}".format(names)) + + return data + + async def get_data2(self, ids) -> dict: + def dateparse(time_in_secs): + return datetime.datetime.strptime(time_in_secs, '%Y-%m-%d %H:%M:%S') + + result = dict() + for id in ids: + # df = pd.read_csv("./logs/sensor_%s.log" % id, parse_dates=True, date_parser=dateparse, index_col='DateTime', names=['DateTime',"Values"], header=None) + # concat all logs + all_filenames = glob.glob('./logs/sensor_%s.log*' % id) + df = pd.concat([pd.read_csv(f, parse_dates=True, date_parser=dateparse, index_col='DateTime', names=['DateTime', 'Values'], header=None) for f in all_filenames]) + df = df.resample('60s').max() + df = df.dropna() + result[id] = {"time": df.index.astype(str).tolist(), "value":df.Values.tolist()} + return result + + + + def get_logfile_names(self, name:str ) -> list: + ''' + Get all log file names + :param name: log name as string. pattern /logs/sensor_%s.log* + :return: list of log file names + ''' + + return [os.path.basename(x) for x in glob.glob('./logs/sensor_%s.log*' % name)] + + def clear_log(self, name:str ) -> str: + + all_filenames = glob.glob('./logs/sensor_%s.log*' % name) + for f in all_filenames: + os.remove(f) + + if name in self.datalogger: + del self.datalogger[name] + + + def get_all_zip_file_names(self, name: str) -> list: + + ''' + Return a list of all zip file names + :param name: + :return: + ''' + + return [os.path.basename(x) for x in glob.glob('./logs/*-sensor-%s.zip' % name)] + + def clear_zip(self, name:str ) -> None: + """ + clear all zip files for a sensor + :param name: sensor name + :return: None + """ + + all_filenames = glob.glob('./logs/*-sensor-%s.zip' % name) + for f in all_filenames: + os.remove(f) + + def zip_log_data(self, name: str) -> str: + """ + :param name: sensor name + :return: zip_file_name + """ + + formatted_time = strftime("%Y-%m-%d-%H_%M_%S", localtime()) + file_name = './logs/%s-sensor-%s.zip' % (formatted_time, name) + zip = zipfile.ZipFile(file_name, 'w', zipfile.ZIP_DEFLATED) + all_filenames = glob.glob('./logs/sensor_%s.log*' % name) + for f in all_filenames: + zip.write(os.path.join(f)) + zip.close() + return os.path.basename(file_name) + +