1 Commits
Author SHA1 Message Date
Nielson Tschá 30e78ce13f sensor branch 2021-03-03 16:39:43 +01:00
16 changed files with 857 additions and 3 deletions
Vendored
BIN
View File
Binary file not shown.
Submodule Display deleted from b3fb08d074
Submodule Sensor deleted from b3fb08d074
+49
View File
@@ -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
+92
View File
@@ -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
+40
View File
@@ -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')
+74
View File
@@ -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;
+40
View File
@@ -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
+173
View File
@@ -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()
+82
View File
@@ -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)
+21
View File
@@ -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)
+152
View File
@@ -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;
Submodule tig-stack deleted from b3fb08d074
+122
View File
@@ -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
+9
View File
@@ -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
Executable
+3
View File
@@ -0,0 +1,3 @@
#!/bin/bash
echo Starting Windmessanlage service....
python3 /var/lib/cloud9/WMA/main.py