sensor branch

LICENSE

@@ -1,21 +0,0 @@
-MIT License
-
-Copyright (c) 2016 Mateusz Trojak
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.

avg_values.py

@@ -0,0 +1,49 @@
+import random
+from statistics import mean 
+import time
+
+class AverageValue():
+    def __init__(self): #initialise list values as zero
+        self.list_2= list()
+        self.list_10= list()
+        self.mean_2 =0.0
+        self.mean_10 =0.0
+        self.max_2 =0.0
+        self.max_10 =0.0
+    
+    #insert values into list to calculate 2 minutes average    
+    def insert_value_list2(self, val):
+      
+        if len(self.list_2)>(2*60):
+            self.list_2.pop(0)
+            self.list_2.append(val)
+        else:
+            self.list_2.append(val)
+        return
+    #insert values into list to calculate 10 minutes average   
+    def insert_value_list10(self, val):
+      
+        if len(self.list_10)>(10*60):
+            self.list_10.pop(0)
+            self.list_10.append(val)
+        else:
+            self.list_10.append(val)
+        return
+    
+    #calculate and return 2 minutes average  and max
+    def get_m2(self):
+        self.mean_2 =round(mean(self.list_2),2)
+        return self.mean_2
+    def get_max2(self):
+        self.max_2 =round(max(self.list_2),2)
+        return self.max_2
+        
+    #calculate and return 10 minutes average and max
+    def get_m10(self):
+        self.mean_10 =round(mean(self.list_10),2)
+        return self.mean_10
+    def get_max10(self):
+        self.max_10 =round(max(self.list_10),2)
+        return self.max_10
+ 
+

config.ini

@@ -0,0 +1,92 @@
+[GENERAL]
+debug_mode=False
+path=/var/lib/cloud9/Data/
+
+[DB_SETTINGS]
+db_file_name= WMA001.db
+tablename= WMA001
+days_archive = 3
+
+
+[MQTT_SETTINGS]
+NumberOfHosts = 2
+NumberOfPublishers = 3
+topic_debug=DPW/STR/BUILDING/WIND/DEBUG
+password = netweb123
+qos=1
+
+[Host1]
+port = 1883
+connect = True
+name = 10.0.1.103
+client_name = WMA002
+
+[Host2]
+port = 1884
+connect = True
+name = 192.168.187.121
+client_name = WMA002_DISPLAY
+    
+[Publisher1]
+;#host to which will publish
+host_n=1 
+;#wether will publish or not
+pubflag=True
+#MQTT topic
+topic = DPW/STR/CRAN2/WIND/
+;#name of thread (if created)/name of the publisher
+threadName=Pub_sensor 
+;#read from archive
+arc_flag=True     
+;#execute in daemon thread
+daemonTh=False   
+;#source from date
+Source=sensor      
+;#only true for the sensor source (say if data will be written when not published)
+store_flag = True   
+;#publish from newest(pseudo online data) or oldest (stored data from last)
+order=newest        
+;#number of line publications per execution
+n=1                
+
+
+[Publisher2]
+host_n=1 
+pubflag=False
+topic = DPW/STR/CRAN2/WIND/
+threadName=Pub_archive
+arc_flag=True
+daemonTh=True
+Source=archive
+store_flag = False
+order=oldest
+n=1000
+
+[Publisher3]
+host_n=2
+pubflag=True
+topic = DPW/STR/CRAN2/WIND/
+threadName=sensor_to_display
+arc_flag=False
+daemonTh=False
+Source=sensor to display
+store_flag = False
+order=''
+n=0
+
+
+[RS485_SETTINGS]
+
+baudrate = 9600
+parity   = N
+stopbits = 1
+mode = rtu
+close_port_after_each_call=True
+debug_serial=False
+timeout = 0.1
+precalculate_read_size = True
+handle_local_echo = False
+uart=UART4
+port_name=/dev/ttyO4
+slave_address=1
+

configfile.py

@@ -0,0 +1,40 @@
+
+
+from configparser import ConfigParser
+
+# instantiate
+class Configuration():
+    def __init__(self):
+        self.config = ConfigParser()
+    
+    # parse existing file
+        self.config.read('/var/lib/cloud9/WMA/config.ini')
+    
+    # read General settings
+        self.debug_mode = self.config.getboolean('GENERAL','debug_mode')
+        self.path = self.config.get('GENERAL','path')
+        
+    #read DB settings
+        self.db_file_name = self.config.get('DB_SETTINGS','db_file_name')
+        self.table_name = self.config.get('DB_SETTINGS','tablename')
+        self.days_archive = self.config.getint('DB_SETTINGS','days_archive')
+    # read MQTT settings
+    
+        self.qos = self.config.getint('MQTT_SETTINGS','qos')
+        self.topic_debug = self.config.get('MQTT_SETTINGS','topic_debug')
+        self.numberOfHosts = self.config.getint('MQTT_SETTINGS','NumberOfHosts')
+        self.numberOfPublishers = self.config.getint('MQTT_SETTINGS','NumberOfPublishers')
+
+    # read rs 485 settings
+    
+        self.baudrate = self.config.getint('RS485_SETTINGS','baudrate')         # Baud
+        self.parity   = self.config.get('RS485_SETTINGS','parity')
+        self.stopbits = bytesize = self.config.getint('RS485_SETTINGS','stopbits')
+        self.mode = self.config.get('RS485_SETTINGS','mode')
+        self.close_port_after_each_call= self.config.getboolean('RS485_SETTINGS','close_port_after_each_call')
+        self.debug_serial=self.config.getboolean('RS485_SETTINGS','debug_serial')
+        self.timeout =self.config.getfloat('RS485_SETTINGS','timeout')
+        self.handle_local_echo = self.config.getboolean('RS485_SETTINGS','handle_local_echo')
+        self.uart=self.config.get('RS485_SETTINGS','uart')
+        self.port_name=self.config.get('RS485_SETTINGS','port_name')
+        self.slave_address=self.config.getint('RS485_SETTINGS','slave_address')

db_handle.py

@@ -0,0 +1,74 @@
+import sqlite3
+import configfile
+from os import listdir
+from os.path import isfile, join
+from os import remove
+from json import loads
+import threading
+import time
+
+class Sqlite3_DB():
+    def __init__(self): #constructor to create database connection
+
+        conf = configfile.Configuration()
+        self.dbname=conf.db_file_name
+        self.conn = sqlite3.connect(self.dbname)
+        self.cursor = self.conn.cursor()
+        self.tablename = conf.table_name
+        self.days= conf.days_archive
+        try: #create table if still not exists
+            self.cursor.execute("CREATE TABLE %s(time_date datetime,sensor_data json)"%(conf.table_name))
+            self.conn.commit()
+        except:
+           self.tablecreated=1;
+        self.path=conf.path
+        self.count = 0;
+
+    #function to count the number of rows in SQL database
+    def sql_count(self):
+        self.cursor.execute("SELECT COUNT(sensor_data) FROM %s;"%(self.tablename))
+        numberOfRows = self.cursor.fetchone()[0]
+        return   numberOfRows
+    
+    #function to select the oldest files of the database
+    def select_oldest(self,number):
+        oldest_data= list()
+        for row in self.cursor.execute("SELECT sensor_data FROM %s LIMIT %d;"%(self.tablename,number)): #DESC FOR NEWES, ASC FOR OLDEST        
+                oldest_data.append(row);
+        return(oldest_data)
+        
+    #function to select the newest files of the database   
+    def select_newest(self,number):
+        newest_data= list()
+        for row in self.cursor.execute("SELECT sensor_data FROM %s ORDER BY rowid DESC LIMIT %d;"%(self.tablename,number)): #DESC FOR NEWES, ASC FOR OLDEST        
+            newest_data.append(row);
+        return(newest_data)
+   
+   
+    def remove_oldests_from_db(self,number):#remove number of old lines from DB
+        while(self.conn.in_transaction):#wait for previous transactions to be over
+            continue;
+        try:    
+            self.cursor.execute("DELETE FROM %s LIMIT %d;"%(self.tablename,number)) # remove by order the oldest ones
+            i=0
+            self.conn.commit()
+            while(self.conn.in_transaction):
+                time.sleep(0.0001)
+
+        except:
+            print("Error on Deleting")
+            return;
+        
+    def remove_newest_from_db(self,number):#remove number of new lines from DB
+        while(self.conn.in_transaction):#wait for previous transactions to be over
+            continue;
+        try:    
+            self.cursor.execute("DELETE FROM %s ORDER BY rowid DESC LIMIT %d;"%(self.tablename,number)) # remove by order the newest ones
+            i=0
+            self.conn.commit()
+            while(self.conn.in_transaction):
+                time.sleep(0.001)
+
+        except:
+            print("Error on Deleting")
+            return;

docker-compose.yml

@@ -1,60 +0,0 @@
-grafana:
-    image: grafana/grafana:latest
-    ports: 
-        - 3000:3000
-    links:
-        - influxdb:influxdb
-    environment:
-        GF_SECURITY_ADMIN_USER: admin
-        GF_SECURITY_ADMIN_PASSWORD: admin
-        GF_SECURITY_SECRET_KEY: grafana
-        GF_USERS_ALLOW_SIGN_UP: "true"
-        GF_USERS_ALLOW_ORG_CREATE: "true"
-        GF_AUTH_ANONYMOUS_ENABLED: "true"
-        GF_AUTH_ANONYMOUS_ORG_NAME: grafana
-#        GF_DASHBOARDS_JSON_ENABLED: "true"
-#        GF_DASHBOARDS_JSON_PATH: /opt/grafana
-    volumes_from:
-        - grafana-data
-    restart: always
-    user: root
-
-grafana-data:
-    image: busybox
-    tty: true
-    volumes:
-        - /var/lib/grafana
-        - /var/log/grafana
-        - /var/lib/grafana/plugins
-
-influxdb:
-    build: ./influxdb/
-    ports:
-        - 8083:8083
-        - 8086:8086
-    environment:
-        INFLUX_DATABASE: "telegraf"
-        INLFUX_ADMIN_USER: "grafana"
-        INFLUX_ADMIN_PASS: "grafana"
-    volumes_from:
-        - influxdb-data
-
-influxdb-data:
-    image: busybox
-    tty: true
-    volumes:
-        - /var/lib/influxdb
-
-telegraf:
-    build: ./telegraf/
-    links:
-        - influxdb:influxdb
-    environment:
-        HOST_NAME: "telegraf"
-        INFLUXDB_HOST: "influxdb"
-        INFLUXDB_PORT: "8086"
-        DATABASE: "telegraf"
-    tty: true
-    volumes:
-        - /var/run/docker.sock:/var/run/docker.sock
-    privileged: true

influxdb/Dockerfile

@@ -1,14 +0,0 @@
-FROM influxdb:latest
-MAINTAINER Mateusz Trojak <mateusz.trojak@gmail.com>
-LABEL version="1.1"
-LABEL description="InfluxDB docker image with custom setup"
-
-USER root
-
-ADD influxdb.template.conf /influxdb.template.conf
-
-ADD run.sh /run.sh
-RUN chmod +x /run.sh
-
-CMD ["/run.sh"]
-

influxdb/influxdb.template.conf

@@ -1,122 +0,0 @@
-reporting-disabled = false
-bind-address = ":8088"
-
-[meta]
-  dir = "/var/lib/influxdb/meta"
-  retention-autocreate = true
-  logging-enabled = true
-
-[data]
-  dir = "/var/lib/influxdb/data"
-  engine = "tsm1"
-  wal-dir = "/var/lib/influxdb/wal"
-  wal-logging-enabled = true
-  query-log-enabled = true
-  cache-max-memory-size = 524288000
-  cache-snapshot-memory-size = 26214400
-  cache-snapshot-write-cold-duration = "1h0m0s"
-  compact-full-write-cold-duration = "24h0m0s"
-  max-points-per-block = 0
-  max-series-per-database = 1000000
-  trace-logging-enabled = false
-
-[coordinator]
-  write-timeout = "10s"
-  max-concurrent-queries = 0
-  query-timeout = "0"
-  log-queries-after = "0"
-  max-select-point = 0
-  max-select-series = 0
-  max-select-buckets = 0
-
-[retention]
-  enabled = true
-  check-interval = "30m0s"
-
-[shard-precreation]
-  enabled = true
-  check-interval = "10m0s"
-  advance-period = "30m0s"
-
-[admin]
-  enabled = true
-  bind-address = ":8083"
-  https-enabled = false
-  https-certificate = "/etc/ssl/influxdb.pem"
-
-[monitor]
-  store-enabled = true
-  store-database = "_internal"
-  store-interval = "10s"
-
-[subscriber]
-  enabled = true
-  http-timeout = "30s"
-
-[http]
-  enabled = true
-  bind-address = ":8086"
-  auth-enabled = false
-  log-enabled = true
-  write-tracing = false
-  https-enabled = false
-  https-certificate = "/etc/ssl/influxdb.pem"
-  https-private-key = ""
-  max-row-limit = 10000
-  max-connection-limit = 0
-  shared-secret = ""
-  realm = "InfluxDB"
-
-[[graphite]]
-  enabled = false
-  bind-address = ":2003"
-  database = "graphite"
-  retention-policy = ""
-  protocol = "tcp"
-  batch-size = 5000
-  batch-pending = 10
-  batch-timeout = "1s"
-  consistency-level = "one"
-  separator = "."
-  udp-read-buffer = 0
-
-[[collectd]]
-  enabled = false
-  bind-address = ":25826"
-  database = "collectd"
-  retention-policy = ""
-  batch-size = 5000
-  batch-pending = 10
-  batch-timeout = "10s"
-  read-buffer = 0
-  typesdb = "/usr/share/collectd/types.db"
-
-[[opentsdb]]
-  enabled = false
-  bind-address = ":4242"
-  database = "opentsdb"
-  retention-policy = ""
-  consistency-level = "one"
-  tls-enabled = false
-  certificate = "/etc/ssl/influxdb.pem"
-  batch-size = 1000
-  batch-pending = 5
-  batch-timeout = "1s"
-  log-point-errors = true
-
-[[udp]]
-  enabled = false
-  bind-address = ":8089"
-  database = "udp"
-  retention-policy = ""
-  batch-size = 5000
-  batch-pending = 10
-  read-buffer = 0
-  batch-timeout = "1s"
-  precision = ""
-
-[continuous_queries]
-  log-enabled = true
-  enabled = true
-  run-interval = "1s"
-

influxdb/run.sh

@@ -1,32 +0,0 @@
-#!/bin/bash
-
-set -m
-CONFIG_TEMPLATE="/influxdb.template.conf"
-CONFIG_FILE="/etc/influxdb/influxdb.conf"
-CURR_TIMESTAMP=`date +%s`
-
-INFLUX_HOST="localhost"
-INFLUX_API_PORT="8086"
-[ "${INFLUX_ADMIN_USER}" = "" ] &&
-        INFLUX_ADMIN_USER="grafana"
-[ "${INFLUX_ADMIN_PASS}" = "" ] &&
-        INFLUX_ADMIN_PASS="grafana"
-[ "${INFLUX_DATABASE}" = "" ] &&
-        JENKINS_BIN="telegraf"
-
-mv -v $CONFIG_FILE $CONFIG_FILE.$CURR_TIMESTAMP
-cp -v $CONFIG_TEMPLATE $CONFIG_FILE
-
-exec influxd -config=$CONFIG_FILE 1>>/var/log/influxdb/influxdb.log 2>&1 &
-sleep 5
-
-USER_EXISTS=`influx -host=${INFLUX_HOST} -port=${INFLUX_API_PORT} -execute="SHOW USERS" | awk '{print $1}' | grep "${INFLUX_ADMIN_USER}" | wc -l`
-
-if [ -n ${USER_EXISTS} ]
-then
-  influx -host=${INFLUX_HOST} -port=${INFLUX_API_PORT} -execute="CREATE USER ${INFLUX_ADMIN_USER} WITH PASSWORD '${INFLUX_ADMIN_PASS}' WITH ALL PRIVILEGES"
-  influx -host=${INFLUX_HOST} -port=${INFLUX_API_PORT} -username=${INFLUX_ADMIN_USER} -password="${INFLUX_ADMIN_PASS}" -execute="create database ${INFLUX_DATABASE}"
-  influx -host=${INFLUX_HOST} -port=${INFLUX_API_PORT} -username=${INFLUX_ADMIN_USER} -password="${INFLUX_ADMIN_PASS}" -execute="grant all PRIVILEGES on ${INFLUX_DATABASE} to ${INLFUX_ADMIN_USER}"
-fi 
-
-tail -f /var/log/influxdb/influxdb.log

jsonwind.py

@@ -0,0 +1,40 @@
+import json
+
+
+#returns standard json pack for wind data
+def json_wma(time,speed, direction,temperature,speed_m2,speed_max2,speed_m10,speed_max10):
+    
+    wind = {"time": time, "windspeed": speed, "winddirection" : direction, "temperature": temperature,
+            "speed2AVG": speed_m2,"speed2max": speed_max2,"speed10AVG": speed_m10,"speed10max": speed_max10
+    };
+    
+
+    wind_info = json.dumps(wind)
+    
+    return wind_info
+    
+def json_concat(json_pack,topic):
+    concat=json.loads(json_pack)
+    topic_json = {"topic": topic};
+    concat.update(topic_json)
+    return json.dumps(concat)
+    
+    
+#returns standard json pack for wind data when in debug mode
+
+def json_debug(time,speed, direction,error_n,kompasswinkel):
+    
+    j_debug = {"time": time, "windspeed": speed, "winddirection" : direction, "nErrorsSerial": error_n,
+            "kompasswinkel": kompasswinkel
+    };
+    
+
+
+    debug_info = json.dumps(j_debug)
+    
+    return debug_info
+
+
+
+
+

main.py

@@ -0,0 +1,173 @@
+
+"""
+Created on Tue May  5 15:37:56 2020
+
+@author: nits
+connection:
+        green and white = B
+        yellow and brown = A
+
+"""
+
+
+import windsensor
+from jsonwind import json_wma,json_debug,json_concat
+from json import loads
+import sighand
+import mqtt_comm
+import time
+import avg_values
+import configfile
+import db_handle
+from threading import Timer
+from os import listdir
+from os.path import isfile, join
+from os import remove
+import threading
+from threadf import Publisher
+import threadf
+
+threadf.store_count=0
+
+def main():
+    #extra threads and parse file
+    conf = configfile.Configuration()
+    
+    #initialization of variables and objects
+    
+    
+    signal_handler=sighand.SignalHandler()
+    wind = windsensor.WindSensor()
+    speed_mean= avg_values.AverageValue()
+    
+#    topic = conf.topic
+    days=conf.days_archive
+    topic_debug = conf.topic_debug
+    debug_mode = conf.debug_mode
+    qos_mqtt=conf.qos
+    start = time.time()
+    numberOfHosts=conf.numberOfHosts
+    numberOfPublishers=conf.numberOfPublishers
+    
+    #check the hosts and publisher objects
+    
+    print("Nr of hosts: ", numberOfHosts)
+    hosts = []
+    print("Nr of publishers: ", numberOfPublishers)
+    publishers = []
+
+    for i in range(1, numberOfHosts+1):
+        name = "Host"+str(i)
+        port = conf.config[name].getint('port')
+        ident = conf.config[name].get('client_name')
+        connect = conf.config[name].getboolean('connect')
+        host_name = conf.config[name].get('name')
+        
+        host = mqtt_comm.MQTT_Client(host_name, ident, port,connect)
+        hosts.append(host)
+        
+    for host in hosts:
+        if(host.connect):
+            host.client_connect()
+            host.client.loop_start()
+    #create the publisher objects and link them to the hosts
+    for i in range(1, numberOfPublishers+1):
+        name = "Publisher"+str(i)
+        
+        pubflag=conf.config[name].getboolean('pubflag')
+        arcflag=conf.config[name].getboolean('arc_flag')
+        daemonTh=conf.config[name].getboolean('daemonTh')
+        st_flag=conf.config[name].getboolean('store_flag')
+        topic = conf.config[name].get('topic')
+        host_n=conf.config[name].getint('host_n')
+        thName=conf.config[name].get('threadName')
+        source=conf.config[name].get('source')
+        order=conf.config[name].get('order')
+        n=conf.config[name].getint('n')
+        print("publisher",i,"daemon =",daemonTh,"archive=",arcflag,"publish=",pubflag)
+        Publisher_n = threadf.Publisher(pubflag,arcflag,topic,hosts[host_n-1],qos_mqtt,daemonTh,thName,source,st_flag,order,n)
+        publishers.append(Publisher_n)
+        
+        
+        
+    #connect to SQL
+    while True:
+        try:
+            sqinit=db_handle.Sqlite3_DB();
+            break;
+        except db_handle.sqlite3.Error as er:
+            print('SQLite error: %s' % (' '.join(er.args)))
+            print("Exception class is: ", er.__class__)
+            time.sleep(3)
+    threadf.store_count=sqinit.sql_count();
+    print("Archive has INITIALLY ",threadf.store_count," rows")
+    
+    time_json=""
+    while (True):
+       # if(threadf.quit==False):    
+        if(time.time() - start > 1.0 or ((time_json)is "")): #Wait 1 second to run loop, if sensor comm fails, will try faster so data won't be lost
+            if(threadf.lock.locked()):
+                threadf.lock.release()#release lock to run the main thread
+                #print("Lock released for running the main routines")
+            threadf.lock.acquire() 
+            start = time.time()
+           
+            #load data from sensors and calculate mean
+            
+            time_act = wind.readDateTime()
+            windspeed_act = wind.readWindSpeed()
+            wind_direct= wind.readWindDirection()
+            wind_temp=wind.readTemp()
+            
+            speed_mean.insert_value_list2(windspeed_act)
+            speed_mean.insert_value_list10(windspeed_act)
+            speed_m2=speed_mean.get_m2()
+            speed_max2=speed_mean.get_max2()
+            speed_m10=speed_mean.get_m10()
+            speed_max10=speed_mean.get_max10()
+            
+            rm_number=1800    #number of removed lines when requested, 1800 = half an hour
+            if(threadf.store_count>(days*86400+rm_number)):#number of samples per day(size in lines)
+                startc = time.time()
+                
+                print('Too many files on the archive, removing older than',days,'days size')
+                sqinit.remove_oldests_from_db(rm_number);#clean DB if DB is too large
+                threadf.store_count=threadf.store_count-rm_number;
+                timec=time.time()-startc
+                print("time taken to remove",rm_number," lines from the old data was: ",timec)#computes and displays the time taken 
+            
+
+    
+            json_pack = json_wma(time_act,windspeed_act,wind_direct,wind_temp,speed_m2,speed_max2,speed_m10,speed_max10)#generate json pack with values
+            time_json=loads(json_pack)["time"];
+            if (time_json is not ""):
+                publishers[0].write_line_db_sub(json_pack)#write line to db using the first publisher object
+                
+            for pub in publishers:
+                pub.publish(json_concat(json_pack,pub.topic))
+            #write line to db
+       
+
+
+                    
+            #debug mode
+            if (debug_mode ==1):
+                errorcounter = wind.errorcounter
+                kompasswinkel_act = wind.readCompassAngle()
+                json_pack_debug = json_debug(time_act,windspeed_act,wind_direct,errorcounter,kompasswinkel_act) #print information on console
+                print(json_pack_debug)
+            
+            if(threadf.lock.locked()):#release thread lock
+                threadf.lock.release()
+            #check if sensor was read, in case not, it will skip the sleep in order to get the reading faster
+            
+          #  print("Cycle duration:",time.time()-start,"s")# print time
+            try:
+                if(time_json is not ""):
+                    time.sleep(1-(time.time()-start)) #treats exception if time bigger than 1.0s
+            except:
+                print("Cycle>1s lasted:",time.time()-start,"seconds")
+                
+       
+if __name__ == "__main__":
+    main()

mqtt_comm.py

@@ -0,0 +1,82 @@
+import paho.mqtt.client as mqtt
+import time
+import configfile
+from time import gmtime, strftime 
+import difflib
+
+
+
+        
+        
+class MQTT_Client():
+  
+    def __init__(self,host_name, client_name, port,connect): #construtor initialises variables and callback functions
+        conf = configfile.Configuration()
+        #callback functions definition(here because it doesn't work anywhere else)
+        def on_connect(client, userdata, flags, rc):
+            self.connected =1
+            if(rc==1):
+                print("Connected with result code "+str(rc))
+        def on_message(client, userdata, msg):
+            print(msg.topic+" "+str(msg.payload))
+        def on_publish(client, userdata, mid):
+            self.published =1
+            self.last_message=mid;
+        def on_disconnect(client, userdata, rc):
+            self.connected =0
+            print("MQTT Disconnected") 
+            
+        #initialize variables
+        self.connect=connect
+        self.client_id =client_name 
+        self.host_in = host_name
+        self.port_n = port
+        
+        self.published = 0
+        self.connected = 0
+        self.last_message = 0
+        self.client = mqtt.Client(client_id=client_name)
+        self.client.on_connect = on_connect
+        self.client.on_message = on_message
+        self.client.on_publish = on_publish
+        self.client.on_disconnect=on_disconnect
+      
+
+
+    #create connection to MQTT server
+    def client_connect(self):
+        try:
+            print("Connecting to host: ",self.host_in)
+            self.client.connect(host=self.host_in,port=self.port_n,keepalive=60)
+        except Exception as error:
+            print("Exception:",error);
+    
+
+            
+class Host():
+    def __init__(self, Port, Topic, Name, PublishFlag):
+        self.port = Port;
+        self.topic = Topic;
+        self.name = Name
+        self.publishFlag = PublishFlag
+        
+    def getTopic(self):
+        return self.topic
+    
+    def getPort(self):
+        return self.port
+    
+    def getName(self):
+        return self.name
+    
+    def publish(self):
+        if self.publishFlag:
+            print("I publish to ", self.topic)
+    
+
+ 
+    
+
+    
+
+

sighand.py

@@ -0,0 +1,21 @@
+import signal
+import traceback
+
+
+class SignalHandler:
+
+    def __init__(self):
+
+        # register signal handlers
+        signal.signal(signal.SIGINT, self.exit_gracefully)
+        signal.signal(signal.SIGTERM, self.exit_gracefully)
+
+
+
+    def exit_gracefully(self, signum, frame):
+        
+        print('captured signal %d' % signum)
+        traceback.print_stack(frame)
+        #print signal code
+
+        raise(SystemExit)

telegraf/Dockerfile

@@ -1,14 +0,0 @@
-FROM telegraf:latest
-MAINTAINER Mateusz Trojak <mateusz.trojak@gmail.com>
-LABEL version="1.1"
-LABEL description="Telegraf docker image with custom setup"
-
-USER root
-
-ADD telegraf.template.conf /telegraf.template.conf
-
-ADD run.sh /run.sh
-RUN chmod +x /*.sh
-
-CMD ["/run.sh"]
-

telegraf/run.sh

@@ -1,15 +0,0 @@
-#!/bin/bash
-
-set -m
-CONFIG_TEMPLATE="/telegraf.template.conf"
-CONFIG_FILE="/etc/telegraf/telegraf.conf"
-
-sed -e "s/\${HOST_NAME}/$HOST_NAME/" \
- -e "s!\${INFLUXDB_HOST}!$INFLUXDB_HOST!" \
- -e "s/\${INFLUXDB_PORT}/$INFLUXDB_PORT/" \
- -e "s/\${DATABASE}/$DATABASE/" \
- $CONFIG_TEMPLATE > $CONFIG_FILE
-
-echo "=> Starting Telegraf ..."
-
-exec telegraf --debug -config /etc/telegraf/telegraf.conf 

telegraf/telegraf.template.conf

@@ -1,581 +0,0 @@
-# Telegraf Configuration
-#
-# Telegraf is entirely plugin driven. All metrics are gathered from the
-# declared inputs, and sent to the declared outputs.
-#
-# Plugins must be declared in here to be active.
-# To deactivate a plugin, comment out the name and any variables.
-#
-# Use 'telegraf -config telegraf.conf -test' to see what metrics a config
-# file would generate.
-#
-# Environment variables can be used anywhere in this config file, simply prepend
-# them with $. For strings the variable must be within quotes (ie, "$STR_VAR"),
-# for numbers and booleans they should be plain (ie, $INT_VAR, $BOOL_VAR)
-
-# Global tags can be specified here in key="value" format.
-[global_tags]
-  # dc = "us-east-1" # will tag all metrics with dc=us-east-1
-  # rack = "1a"
-  ## Environment variables can be used as tags, and throughout the config file
-  # user = "$USER"
-
-
-# Configuration for telegraf agent
-[agent]
-  ## Default data collection interval for all inputs
-  interval = "1s"
-  ## Rounds collection interval to 'interval'
-  ## ie, if interval="10s" then always collect on :00, :10, :20, etc.
-  round_interval = true
-
-  ## Telegraf will send metrics to outputs in batches of at
-  ## most metric_batch_size metrics.
-  metric_batch_size = 1000
-  ## For failed writes, telegraf will cache metric_buffer_limit metrics for each
-  ## output, and will flush this buffer on a successful write. Oldest metrics
-  ## are dropped first when this buffer fills.
-  #metric_buffer_limit = 86000
-
-  ## Collection jitter is used to jitter the collection by a random amount.
-  ## Each plugin will sleep for a random time within jitter before collecting.
-  ## This can be used to avoid many plugins querying things like sysfs at the
-  ## same time, which can have a measurable effect on the system.
-  collection_jitter = "0s"
-
-  ## Default flushing interval for all outputs. You shouldn't set this below
-  ## interval. Maximum flush_interval will be flush_interval + flush_jitter
-  flush_interval = "10s"
-  ## Jitter the flush interval by a random amount. This is primarily to avoid
-  ## large write spikes for users running a large number of telegraf instances.
-  ## ie, a jitter of 5s and interval 10s means flushes will happen every 10-15s
-  flush_jitter = "0s"
-
-  ## By default, precision will be set to the same timestamp order as the
-  ## collection interval, with the maximum being 1s.
-  ## Precision will NOT be used for service inputs, such as logparser and statsd.
-  ## Valid values are "ns", "us" (or "µs"), "ms", "s".
-  precision = ""
-  ## Run telegraf in debug mode
-  debug = true
-  ## Run telegraf in quiet mode
-  quiet = false
-  ## Override default hostname, if empty use os.Hostname()
-  hostname = "${HOST_NAME}"
-  ## If set to true, do no set the "host" tag in the telegraf agent.
-  omit_hostname = false
-
-
-###############################################################################
-#                            OUTPUT PLUGINS                                   #
-###############################################################################
-
-# Configuration for influxdb server to send metrics to
-[[outputs.influxdb]]
-  ## The full HTTP or UDP endpoint URL for your InfluxDB instance.
-  ## Multiple urls can be specified as part of the same cluster,
-  ## this means that only ONE of the urls will be written to each interval.
-  # urls = ["udp://localhost:8089"] # UDP endpoint example
-  urls = ["http://${INFLUXDB_HOST}:${INFLUXDB_PORT}"] # required
-  ## The target database for metrics (telegraf will create it if not exists).
-  #database = "sensordata"
-
-  ## Retention policy to write to. Empty string writes to the default rp.
-  retention_policy = ""
-  ## Write consistency (clusters only), can be: "any", "one", "quorum", "all"
-  write_consistency = "any"
-
-  ## Write timeout (for the InfluxDB client), formatted as a string.
-  ## If not provided, will default to 5s. 0s means no timeout (not recommended).
-  timeout = "5s"
-  username = "telegraf"
-  password = "metricsnws"
-  ## Set the user agent for HTTP POSTs (can be useful for log differentiation)
-  user_agent = "telegraf"
-  ## Set UDP payload size, defaults to InfluxDB UDP Client default (512 bytes)
-  udp_payload = 512
-
-  ## Optional SSL Config
-  # ssl_ca = "/etc/telegraf/ca.pem"
-  # ssl_cert = "/etc/telegraf/cert.pem"
-  # ssl_key = "/etc/telegraf/key.pem"
-  ## Use SSL but skip chain & host verification
-  # insecure_skip_verify = false
-
-# # Configuration for Amon Server to send metrics to.
-# [[outputs.amon]]
-#   ## Amon Server Key
-#   server_key = "my-server-key" # required.
-#
-#   ## Amon Instance URL
-#   amon_instance = "https://youramoninstance" # required
-#
-#   ## Connection timeout.
-#   # timeout = "5s"
-
-
-# # Configuration for the AMQP server to send metrics to
-# [[outputs.amqp]]
-#   ## AMQP url
-#   url = "amqp://localhost:5672/influxdb"
-#   ## AMQP exchange
-#   exchange = "telegraf"
-#   ## Auth method. PLAIN and EXTERNAL are supported
-#   # auth_method = "PLAIN"
-#   ## Telegraf tag to use as a routing key
-#   ##  ie, if this tag exists, it's value will be used as the routing key
-#   routing_tag = "host"
-#
-#   ## InfluxDB retention policy
-#   # retention_policy = "default"
-#   ## InfluxDB database
-#   # database = "telegraf"
-#   ## InfluxDB precision
-#   # precision = "s"
-#
-#   ## Optional SSL Config
-#   # ssl_ca = "/etc/telegraf/ca.pem"
-#   # ssl_cert = "/etc/telegraf/cert.pem"
-#   # ssl_key = "/etc/telegraf/key.pem"
-#   ## Use SSL but skip chain & host verification
-#   # insecure_skip_verify = false
-#
-#   ## Data format to output.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
-#   data_format = "influx"
-
-
-# # Configuration for AWS CloudWatch output.
-# [[outputs.cloudwatch]]
-#   ## Amazon REGION
-#   region = 'us-east-1'
-#
-#   ## Amazon Credentials
-#   ## Credentials are loaded in the following order
-#   ## 1) Assumed credentials via STS if role_arn is specified
-#   ## 2) explicit credentials from 'access_key' and 'secret_key'
-#   ## 3) shared profile from 'profile'
-#   ## 4) environment variables
-#   ## 5) shared credentials file
-#   ## 6) EC2 Instance Profile
-#   #access_key = ""
-#   #secret_key = ""
-#   #token = ""
-#   #role_arn = ""
-#   #profile = ""
-#   #shared_credential_file = ""
-#
-#   ## Namespace for the CloudWatch MetricDatums
-#   namespace = 'InfluxData/Telegraf'
-
-
-# # Configuration for DataDog API to send metrics to.
-# [[outputs.datadog]]
-#   ## Datadog API key
-#   apikey = "my-secret-key" # required.
-#
-#   ## Connection timeout.
-#   # timeout = "5s"
-
-
-# # Send telegraf metrics to file(s)
-# [[outputs.file]]
-#   ## Files to write to, "stdout" is a specially handled file.
-#   files = ["stdout", "/tmp/metrics.out"]
-#
-#   ## Data format to output.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
-#   data_format = "influx"
-
-
-# # Configuration for Graphite server to send metrics to
-# [[outputs.graphite]]
-#   ## TCP endpoint for your graphite instance.
-#   ## If multiple endpoints are configured, output will be load balanced.
-#   ## Only one of the endpoints will be written to with each iteration.
-#   servers = ["localhost:2003"]
-#   ## Prefix metrics name
-#   prefix = ""
-#   ## Graphite output template
-#   ## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
-#   template = "host.tags.measurement.field"
-#   ## timeout in seconds for the write connection to graphite
-#   timeout = 2
-
-
-# # Send telegraf metrics to graylog(s)
-# [[outputs.graylog]]
-#   ## Udp endpoint for your graylog instance.
-#   servers = ["127.0.0.1:12201", "192.168.1.1:12201"]
-
-
-# # Configuration for sending metrics to an Instrumental project
-# [[outputs.instrumental]]
-#   ## Project API Token (required)
-#   api_token = "API Token" # required
-#   ## Prefix the metrics with a given name
-#   prefix = ""
-#   ## Stats output template (Graphite formatting)
-#   ## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#graphite
-#   template = "host.tags.measurement.field"
-#   ## Timeout in seconds to connect
-#   timeout = "2s"
-#   ## Display Communcation to Instrumental
-#   debug = false
-
-
-# # Configuration for MQTT server to send metrics to
-# [[outputs.mqtt]]
-#   servers = ["localhost:1883"] # required.
-#
-#   ## MQTT outputs send metrics to this topic format
-#   ##    "<topic_prefix>/<hostname>/<pluginname>/"
-#   ##   ex: prefix/web01.example.com/mem
-#   topic_prefix = "telegraf"
-#
-#   ## username and password to connect MQTT server.
-#   # username = "telegraf"
-#   # password = "metricsmetricsmetricsmetrics"
-#
-#   ## Optional SSL Config
-#   # ssl_ca = "/etc/telegraf/ca.pem"
-#   # ssl_cert = "/etc/telegraf/cert.pem"
-#   # ssl_key = "/etc/telegraf/key.pem"
-#   ## Use SSL but skip chain & host verification
-#   # insecure_skip_verify = false
-#
-#   ## Data format to output.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
-#   data_format = "influx"
-
-
-###############################################################################
-#                            INPUT PLUGINS                                    #
-###############################################################################
-# # Read metrics from MQTT topic(s)
-[[inputs.mqtt_consumer]]
-servers = ["tcp://10.0.1.103:1883"]
-#   ## MQTT QoS, must be 0, 1, or 2
-qos = 1
-#
-#   ## Topics to subscribe to
-topics = [
-	"DPW/STR/BUILDING/WIND",
-	]
-#
-#   # if true, messages that can't be delivered while the subscriber is offline
-#   # will be delivered when it comes back (such as on service restart).
-#   # NOTE: if true, client_id MUST be set
-persistent_session = true
-#   # If empty, a random client ID will be generated.
-client_id = "telegraf_WMA2"
-#
-#   ## username and password to connect MQTT server.
-	username = "telegraf"
-	password = "telegraf"
-#
-#   ## Optional SSL Config
-#   # ssl_ca = "/etc/telegraf/ca.pem"
-#   # ssl_cert = "/etc/telegraf/cert.pem"
-#   # ssl_key = "/etc/telegraf/key.pem"
-#   ## Use SSL but skip chain & host verification
-#   # insecure_skip_verify = false
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-	data_format = "json"
-	json_time_key = "time"
-	json_time_format = "2006-01-02 15:04:05"
-
-
-# Read metrics about cpu usage
-
-[[inputs.cpu]]
-  ## Whether to report per-cpu stats or not
- ## percpu = true
-  ## Whether to report total system cpu stats or not
- ## totalcpu = true
-  ## Comment this line if you want the raw CPU time metrics
-##  fielddrop = ["time_*"]
-
-
-# Read metrics about disk usage by mount point
-[[inputs.disk]]
-  ## By default, telegraf gather stats for all mountpoints.
-  ## Setting mountpoints will restrict the stats to the specified mountpoints.
-  # mount_points = ["/"]
-
-  ## Ignore some mountpoints by filesystem type. For example (dev)tmpfs (usually
-  ## present on /run, /var/run, /dev/shm or /dev).
-  ignore_fs = ["tmpfs", "devtmpfs"]
-
-
-# Read metrics about disk IO by device
-[[inputs.diskio]]
-  ## By default, telegraf will gather stats for all devices including
-  ## disk partitions.
-  ## Setting devices will restrict the stats to the specified devices.
-  # devices = ["sda", "sdb"]
-  ## Uncomment the following line if you need disk serial numbers.
-  # skip_serial_number = false
-
-
-# Get kernel statistics from /proc/stat
-#[[inputs.kernel]]
-  # no configuration
-
-
-# Read metrics about memory usage
-#[[inputs.mem]]
-  # no configuration
-
-
-# Get the number of processes and group them by status
-#[[inputs.processes]]
-  # no configuration
-
-
-# Read metrics about swap memory usage
-#[[inputs.swap]]
-  # no configuration
-
-
-# Read metrics about system load & uptime
-#[[inputs.system]]
-  # no configuration
-
-
-# # Read stats from aerospike server(s)
-# [[inputs.aerospike]]
-#   ## Aerospike servers to connect to (with port)
-#   ## This plugin will query all namespaces the aerospike
-#   ## server has configured and get stats for them.
-#   servers = ["localhost:3000"]
-
-
-# # Read Apache status information (mod_status)
-# [[inputs.apache]]
-#   ## An array of Apache status URI to gather stats.
-#   ## Default is "http://localhost/server-status?auto".
-#   urls = ["http://localhost/server-status?auto"]
-
-
-# # Read InfluxDB-formatted JSON metrics from one or more HTTP endpoints
-# [[inputs.influxdb]]
-#   ## Works with InfluxDB debug endpoints out of the box,
-#   ## but other services can use this format too.
-#   ## See the influxdb plugin's README for more details.
-#
-#   ## Multiple URLs from which to read InfluxDB-formatted JSON
-#   ## Default is "http://localhost:8086/debug/vars".
-#   urls = [
-#     "http://localhost:8086/debug/vars"
-#   ]
-
-
-###############################################################################
-#                            SERVICE INPUT PLUGINS                            #
-###############################################################################
-
-# # Read metrics from Kafka topic(s)
-# [[inputs.kafka_consumer]]
-#   ## topic(s) to consume
-#   topics = ["telegraf"]
-#   ## an array of Zookeeper connection strings
-#   zookeeper_peers = ["localhost:2181"]
-#   ## Zookeeper Chroot
-#   zookeeper_chroot = ""
-#   ## the name of the consumer group
-#   consumer_group = "telegraf_metrics_consumers"
-#   ## Offset (must be either "oldest" or "newest")
-#   offset = "oldest"
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # Stream and parse log file(s).
-# [[inputs.logparser]]
-#   ## Log files to parse.
-#   ## These accept standard unix glob matching rules, but with the addition of
-#   ## ** as a "super asterisk". ie:
-#   ##   /var/log/**.log     -> recursively find all .log files in /var/log
-#   ##   /var/log/*/*.log    -> find all .log files with a parent dir in /var/log
-#   ##   /var/log/apache.log -> only tail the apache log file
-#   files = ["/var/log/apache/access.log"]
-#   ## Read file from beginning.
-#   from_beginning = false
-#
-#   ## Parse logstash-style "grok" patterns:
-#   ##   Telegraf built-in parsing patterns: https://goo.gl/dkay10
-#   [inputs.logparser.grok]
-#     ## This is a list of patterns to check the given log file(s) for.
-#     ## Note that adding patterns here increases processing time. The most
-#     ## efficient configuration is to have one pattern per logparser.
-#     ## Other common built-in patterns are:
-#     ##   %{COMMON_LOG_FORMAT}   (plain apache & nginx access logs)
-#     ##   %{COMBINED_LOG_FORMAT} (access logs + referrer & agent)
-#     patterns = ["%{COMBINED_LOG_FORMAT}"]
-#     ## Name of the outputted measurement name.
-#     measurement = "apache_access_log"
-#     ## Full path(s) to custom pattern files.
-#     custom_pattern_files = []
-#     ## Custom patterns can also be defined here. Put one pattern per line.
-#     custom_patterns = '''
-#     '''
-
-
-
-# # Read metrics from NATS subject(s)
-# [[inputs.nats_consumer]]
-#   ## urls of NATS servers
-#   servers = ["nats://localhost:4222"]
-#   ## Use Transport Layer Security
-#   secure = false
-#   ## subject(s) to consume
-#   subjects = ["telegraf"]
-#   ## name a queue group
-#   queue_group = "telegraf_consumers"
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # Read NSQ topic for metrics.
-# [[inputs.nsq_consumer]]
-#   ## An string representing the NSQD TCP Endpoint
-#   server = "localhost:4150"
-#   topic = "telegraf"
-#   channel = "consumer"
-#   max_in_flight = 100
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # Statsd Server
-# [[inputs.statsd]]
-#   ## Address and port to host UDP listener on
-#   service_address = ":8125"
-#   ## Delete gauges every interval (default=false)
-#   delete_gauges = false
-#   ## Delete counters every interval (default=false)
-#   delete_counters = false
-#   ## Delete sets every interval (default=false)
-#   delete_sets = false
-#   ## Delete timings & histograms every interval (default=true)
-#   delete_timings = true
-#   ## Percentiles to calculate for timing & histogram stats
-#   percentiles = [90]
-#
-#   ## separator to use between elements of a statsd metric
-#   metric_separator = "_"
-#
-#   ## Parses tags in the datadog statsd format
-#   ## http://docs.datadoghq.com/guides/dogstatsd/
-#   parse_data_dog_tags = false
-#
-#   ## Statsd data translation templates, more info can be read here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md#graphite
-#   # templates = [
-#   #     "cpu.* measurement*"
-#   # ]
-#
-#   ## Number of UDP messages allowed to queue up, once filled,
-#   ## the statsd server will start dropping packets
-#   allowed_pending_messages = 10000
-#
-#   ## Number of timing/histogram values to track per-measurement in the
-#   ## calculation of percentiles. Raising this limit increases the accuracy
-#   ## of percentiles but also increases the memory usage and cpu time.
-#   percentile_limit = 1000
-
-
-# # Stream a log file, like the tail -f command
-# [[inputs.tail]]
-#   ## files to tail.
-#   ## These accept standard unix glob matching rules, but with the addition of
-#   ## ** as a "super asterisk". ie:
-#   ##   "/var/log/**.log"  -> recursively find all .log files in /var/log
-#   ##   "/var/log/*/*.log" -> find all .log files with a parent dir in /var/log
-#   ##   "/var/log/apache.log" -> just tail the apache log file
-#   ##
-#   ## See https://github.com/gobwas/glob for more examples
-#   ##
-#   files = ["/var/mymetrics.out"]
-#   ## Read file from beginning.
-#   from_beginning = false
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # Generic TCP listener
-# [[inputs.tcp_listener]]
-#   ## Address and port to host TCP listener on
-#   service_address = ":8094"
-#
-#   ## Number of TCP messages allowed to queue up. Once filled, the
-#   ## TCP listener will start dropping packets.
-#   allowed_pending_messages = 10000
-#
-#   ## Maximum number of concurrent TCP connections to allow
-#   max_tcp_connections = 250
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # Generic UDP listener
-# [[inputs.udp_listener]]
-#   ## Address and port to host UDP listener on
-#   service_address = ":8092"
-#
-#   ## Number of UDP messages allowed to queue up. Once filled, the
-#   ## UDP listener will start dropping packets.
-#   allowed_pending_messages = 10000
-#
-#   ## Data format to consume.
-#   ## Each data format has it's own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-#   data_format = "influx"
-
-
-# # A Webhooks Event collector
-# [[inputs.webhooks]]
-#   ## Address and port to host Webhook listener on
-#   service_address = ":1619"
-#
-#   [inputs.webhooks.github]
-#     path = "/github"
-#
-#   [inputs.webhooks.mandrill]
-#     path = "/mandrill"
-#
-#   [inputs.webhooks.rollbar]
-#     path = "/rollbar"
-

threadf.py

@@ -0,0 +1,152 @@
+from json import loads
+import db_handle
+from jsonwind import json_concat
+import mqtt_comm
+import time
+import threading
+global store_count #global variable, so no need to run sql_count
+global lock
+lock = threading.Lock()
+#thread functions are defined here, as well as global variables
+class Publisher():
+    def __init__(self,publish_flag,pub_arc_flag,topic,MQTT_client,qos,daemonTh,ThName,source,store_flag,order,n):
+        global lock
+        self.topic=topic;
+        self.qos=qos
+        self.source=source
+        self.client_mqtt = MQTT_client
+        self.publish_flag=publish_flag
+        self.pub_arc_flag=pub_arc_flag
+        self.daemon=daemonTh
+        self.threadName= ThName
+        self.source=source
+        self.store_flag = store_flag
+        self.order =order
+        self.n = n
+        if (daemonTh==True):
+            self.t1=threading.Thread
+            
+        
+#function to assure that data has been published
+    def getTopic(self):
+        return self.topic
+        
+    def publish_and_wait(self,count,message):
+        if (loads(message)["time"] is not ""):
+            if (self.client_mqtt.connected):
+                (r_code,msg_id)=self.client_mqtt.client.publish(self.topic,message,qos = self.qos)
+                if(r_code==0 and self.source =='archive'): 
+                    if(msg_id%500==0):
+                        print("Message",msg_id, "was delivered with success from the", self.source)
+                    return True
+                if(r_code==0 and self.source is not'archive'):
+                    print("Message",msg_id, "was delivered with success from the", self.source)
+                    return True
+                if(r_code is not 0 and self.store_flag):
+                    print("Error no:",r_code,"saving message to archive")
+                    self.write_line_db_sub(message)
+                    return False
+            else:
+                if(self.store_flag):
+                    self.write_line_db_sub(message)
+        
+    def publish(self,message=None):
+        global store_count
+        if (self.publish_flag ==True and self.pub_arc_flag == False):
+            self.publish_and_wait(store_count,json_concat(message,self.topic))
+        if (self.publish_flag ==True and self.pub_arc_flag==True and self.daemon==True and store_count>2):
+            self.publish_from_archive()
+        if (self.publish_flag ==True and self.pub_arc_flag==True and self.daemon==False):
+            self.check_publish_stored_files()
+
+    
+    #function to write one line to the DB
+    def write_line_db_sub(self,msg_pack):
+        global store_count
+        global lock
+        lock.acquire()
+        data = loads(msg_pack);
+        sqlite_db=db_handle.Sqlite3_DB();
+        sqlite_db.cursor.execute("INSERT into %s values (?, ?);"%(sqlite_db.tablename),[data["time"] , msg_pack])
+        sqlite_db.conn.commit();
+        store_count=store_count+1
+        if (store_count >1):
+            print("Message ",store_count," stored");
+        sqlite_db.conn.close()
+        lock.release()
+        return;
+        
+    #function to publish from archive lines and delete them after   
+
+    def check_publish_stored_files(self):
+
+        if(self.client_mqtt.connected):
+            starter = time.time()#count time to execute the function
+            global store_count
+            sqlite_db=db_handle.Sqlite3_DB()
+           
+            if(store_count>0):
+                if (self.order=='oldest'):
+                    print("SQL Archive has",store_count,"rows.")
+                if (store_count >self.n):
+                    end= self.n;
+                else:
+                    end= store_count
+                i=0
+                
+                if (self.order=='oldest'): # if the published files are gonna be the oldest (publish from archive)
+                    
+                    for row in sqlite_db.select_oldest(end):
+                        lock.acquire()
+                        if(self.publish_and_wait(store_count,json_concat(row[0],self.topic))):
+                            store_count = store_count-1
+                            i=i+1
+                            time.sleep(0.002)
+                        if (lock.locked()):
+                            lock.release()
+
+                    sqlite_db.remove_oldests_from_db(i)
+                    time_publish=time.time()-starter
+                    
+                   # print("############################################################")
+                   # print("time taken to publish",end,self.order," numbers was",time_publish)
+                   # print("############################################################")
+
+                        
+                if (self.order=='newest'): # if the published files are gonna be the newest (publish actual data from sensor)
+
+                    end = 1 #only allows 1 line to be selected, since there's no sleep
+                    for row in sqlite_db.select_newest(end):
+                        if(self.publish_and_wait(store_count,json_concat(row[0],self.topic))):
+                            store_count = store_count-1
+                            i=i+1
+                    sqlite_db.remove_newest_from_db(i)#delete line   
+            sqlite_db.conn.commit()        
+            sqlite_db.conn.close()
+      
+    
+
+        return
+    #thread that runs when publishing from archive    
+    def publish_from_archive(self):
+            ThNames= [];
+            for thread in threading.enumerate():
+                ThNames.append(thread.name);
+            if (self.threadName in ThNames):
+                print("Thread",self.threadName,"is busy")#check if not empty
+            else:
+                self.t1 = threading.Thread(name =self.threadName,target=self.check_publish_stored_files, args=())
+                self.t1.setDaemon(True);
+                self.t1.start()
+            
+    def write_line_db_sub(self,msg_pack):
+        global store_count
+        data = loads(msg_pack);
+        sqlite_db=db_handle.Sqlite3_DB();
+        sqlite_db.cursor.execute("INSERT into %s values (?, ?);"%(sqlite_db.tablename),[data["time"] , msg_pack])
+        sqlite_db.conn.commit();
+        store_count=store_count+1
+        if (store_count>1):
+            print("Message ",store_count," stored");
+        sqlite_db.conn.close()
+        return;

windsensor.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Created on Tue May  5 14:52:57 2020
+
+@author: nits
+"""
+
+import minimalmodbus
+import time
+
+import configfile
+conf = configfile.Configuration()
+import Adafruit_BBIO.UART as UART
+from datetime import datetime
+
+
+class WindSensor():
+
+    def __init__(self): #initialise network connection and parameters
+        conf = configfile.Configuration()
+        UART.setup(conf.uart)
+            
+        self.inst = minimalmodbus.Instrument(conf.port_name, conf.slave_address)  # port name, slave address (in decimal)
+        self.inst.serial.baudrate = conf.baudrate         # Baud
+        self.inst.serial.parity   = conf.parity
+        self.inst.serial.stopbits = conf.stopbits
+        self.inst.mode = conf.mode
+        self.inst.close_port_after_each_call=conf.close_port_after_each_call
+        self.inst.debug=conf.debug_serial
+        self.inst.serial.timeout = conf.timeout
+        self.inst.handle_local_echo = conf.handle_local_echo
+        self.error=0
+        self.errorcounter = 0
+        self.cyclecounter = 0 
+     
+    
+    
+    #read temperature from sensor
+    def readTemp(self):
+        try:
+            temp = self.inst.read_long(30405,4,True)/10
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+            #time.sleep(0.1)
+            return 0
+        return temp
+    
+    #read speed from sensor
+    def readWindSpeed(self):
+        try:
+            temp = self.inst.read_long(30003,4,False)/10
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+            self.error=1
+            
+            return 0
+        return temp
+    #read wind direction from sensor        
+    def readWindDirection(self):
+        try:
+            temp = self.inst.read_long(30203,4,False)/10
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+           # time.sleep(0.1)
+            error=1
+            
+            return 0
+        return temp
+        
+    #read date and time from sensor  
+    def readDateTime(self):
+        try:
+             date = self.inst.read_long(34601,4,False)
+             uhr = self.inst.read_long(34603,4,False)
+             date_time = datetime.strptime(str(date)+str(uhr), '%Y%m%d%H%M%S');
+             date_info=date_time.strftime('%Y-%m-%d %H:%M:%S');
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+            return ""
+        return date_info
+    
+    #read compass angle from sensor    
+    def readCompassAngle(self):
+        try:
+            temp = self.inst.read_long(35081,4,False)/10
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+          
+            error=1
+            return 0
+            
+        return temp
+    #function to read parameter
+    def readP(self,address):
+        try:
+            param = self.inst.read_long(address,3,False)
+            print("The value of the address ",address,"is: ",param)
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+          
+            error=1
+            return 0
+            
+    #function to set parameter
+    def setP(self,address,value):
+        try:
+            param = self.inst.write_register(address,value,functioncode=16)
+            print("The value of the address ",address,"has been changed to: ",value)
+        except Exception as error:
+            print ("[!] Exception occurred: ", error)
+            self.errorcounter = self.errorcounter + 1
+          
+            error=1
+            return 0
+        return 1

wma.service

@@ -0,0 +1,9 @@
+[Unit]
+Description=Service to start WMA Client
+[Service]
+Type=simple
+RestartSec=10
+Restart=always
+ExecStart=/var/lib/cloud9/WMA/wma.sh
+[Install]
+WantedBy=multi-user.target

wma.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+echo Starting Windmessanlage service....
+python3 /var/lib/cloud9/WMA/main.py