import socket,numpy,time,_thread,json,requests,zlib,gnupg,binascii has_mqtt=False try: import paho.mqtt.client as mqtt import paho.mqtt.publish as publish except: has_mqtt=False else: has_mqtt=True from uuid import getnode gpg=gnupg.GPG() _STRINGREMOVE=str.maketrans(dict.fromkeys("/:;\\{}(){]%&")) class meas_data: def __init__(self,var_name,ring_length=100,sigma=2,device=socket.gethostname(),sensor="CPU",i2c=0,store_file="/home/pi/data",multiplicator=1,digits=5,check_last=0,mac=0,sql_min_wait=5,rsa_key="",check_last_shrinking=0.99,mean_count=5): self.set_varname(var_name) self.value=[] self.mval=[] self.set_mean_count(mean_count) self.sensebox="" self.senseid="" self.sense_url="https://ingress.opensensemap.org/" self.sense_intervall=300 self.sense_last_time=0 self.sql_min_wait=sql_min_wait self.sqlhost="" self.sqlport=0 self.sql_last_transmit=0 self.bsense=False self.bsql=False self.bchanged=False self.bfile=False self.store_file=store_file self.sigma=sigma self.set_device(device) self.set_sensor(sensor) self.set_digits(digits) self.set_check_last(check_last,check_last_shrinking) self.set_i2c(i2c) self.set_multiplicator(multiplicator) self.mqtt_broker="" self.mqtt_port=1883 self.mqtt_topic="" self.mqtt_bool=False self.hash=0 self.set_mac() self.set_id() self.set_rsa(rsa_key) self.sql_out="" self.json_out={"hash":self.hash,"signature":"","payload":{"device":self.device,"varname":self.var_name,"i2c":self.i2c,"sensor":self.sensor,"mac":self.mac,"measures":{}}} def sign(self): self.sql_out=self.json_out if self.brsa and not self.brsa_sql: self.sql_out={"signed_gpg":gpg.sign(json.dumps(self.json_out),keyid=self.rsa['keyid']).data.decode('utf-8'),"hash":self.hash} def set_rsa(self,rsa_key): self.brsa=False self.brsa_sql=False if len(rsa_key)>0: self.brsa=False try: gpgkey=gpg.list_keys(keys=rsa_key.translate(_STRINGREMOVE)) except: self.brsa=False else: if len(gpgkey)>0: self.rsa=gpgkey[0] self.brsa=True else: self.brsa=False def set_id(self): idstring=str(hex(getnode()))[2:-1]+self.device+str(self.i2c)+self.sensor+self.var_name self.hash=zlib.crc32(idstring.encode("utf-8")) % (1<<32) def get_id(self): print(self.hash) def set_mac(self): self.mac=format(getnode(),"x") def set_i2c(self,i2c): if (isinstance(i2c,(int,float)) and not isinstance(i2c,bool)): self.i2c=i2c else: self.i2c=0 def set_varname(self,var_name): tvn=var_name.translate(_STRINGREMOVE) if len(tvn)==0: tvn="var" self.var_name=tvn def set_check_last(self,check_last,check_last_shrinking): if (isinstance(check_last,(int,float)) and not isinstance(check_last,bool))and(isinstance(check_last_shrinking,(int,float)) and not isinstance(check_last_shrinking,bool)): self.checklast_min=0 self.checklast_max=0 self.checklast_shrinking=1 if check_last > 0: self.bchecklast=True self.checklast=check_last if (check_last_shrinking < 1) and (check_last_shrinking > 0): self.checklast_shrinking=check_last_shrinking else: self.bchecklast=False self.ckecklast=0 def set_digits(self,digits): self.digits=1 if (isinstance(digits,(int,float)) and not isinstance(digits,bool)): try: td=int(digits)-1 except: print("could not set digits") else: if td<0: td=1 if td>20: td=20 self.digits=td def set_ring_length(self,ring_length): self.ring_length=60 if (isinstance(ring_length,(int,float)) and not isinstance(ring_length,bool)): trl=int(ring_length) if trl<0: trl=(-1)*trl if trl==0: trl = 60 self.ring_length=trl def set_mean_count(self,mean_count): self.mean_count=5 if (isinstance(mean_count,(int,float)) and not isinstance(mean_count,bool)): tmc=int(mean_count) if tmc < 0: tmc=(-1)*tmc if tmc == 0: tmc=self.ring_length if mean_count>self.ring_length: mean_count=self.ring_length self.mean_count=mean_count def set_sensebox(self,sensebox,senseid,intervall=300): tsb=sensebox.translate(_STRINGREMOVE) tsb=tsb[0:24] self.sensebox=tsb tsi=senseid.translate(_STRINGREMOVE) tsi=tsi[0:24] self.senseid=tsi self.sense_intervall=int(intervall) self.bsense=True self.sense_url="https://ingress.opensensemap.org/boxes/%s/%s" % (tsb,tsi) print(self.sense_url) def set_sql(self,host,port=8080,min_wait=5): self.bsql=True th=host.translate(_STRINGREMOVE) self.sqlhost=th self.sqlport=8080 if (isinstance(port,(int,float)) and not isinstance(port,bool)): self.sqlport=int(port) if self.sqlport<1024 or self.sqlport > 2**16: self.bsql=False self.sqlurl="http://"+self.sqlhost+":"+str(self.sqlport)+"/data/"+str(self.hash) if isinstance(min_wait,(int,float)) and not isinstance(min_wait,bool): self.sql_min_wait=min_wait else: self.sql_min_wait=0 self.brsa_sql=False if self.brsa: skey=gpg.list_keys(keys="@"+th) if len(skey)>0: self.brsa_sql=True self.sql_rsa=skey[1] else: self.brsa_sql=False def set_mqtt(self,broker="localhost",port=1883,topic=""): mt=topic.translate(_STRINGREMOVE) if len(mt)==0: mt=self.device+"/"+self.sensor+"/"+self.var_name else: mt=mt.translate(":;\\{}(){]%&") mb=broker.translate(_STRINGREMOVE) mport=1883 if (isinstance(port,(int,float)) and not isinstance(port,bool)): mport=int(port) self.mqtt_broker=mb self.mqtt_port=mport self.mqtt_topic=mt self.mqtt_bool=has_mqtt if has_mqtt: if mport < 1024: mport=1883 try: mqtt.connect(host=mb,port=mport) except: print("no connection to broker") self.mqtt=False #else: else: self.mqtt_bool=False def set_file_log(self,store_file="/home/pi/data"): if len(store_file)>1: self.store_file=store_file+"_{:d}.txt" try: f1=open(self.store_file+"_{:d}.txt".format(int(time.time()/3600)),"a") except: print("could not open data storage") self.bfile=False else: self.bfile=True def show_def(self): print(self.var_name) print(self.value) def set_device(self,device): self.device=device.translate(_STRINGREMOVE) def set_sensor(self,sensor): ts=sensor.translate(_STRINGREMOVE) self.sensor=ts def set_multiplicator(self,multiplicator): tmult=float(multiplicator) if tmult < 1: tmult=tmult*(-1) self.mult=tmult def append(self,value): if (isinstance(value,(int,float)) and not isinstance(value,bool)): tdif=-1 if value>0: try: tdig=10**(round(numpy.log10(value))-self.digits) except: print("") else: tdig=1 if tdig>=0: try: tv=round(value/tdig)*tdig except: tv=value btv=True if self.bchecklast and (len(self.value)>0): # check if new value is within range between min - max if (tv <= self.checklast_max) and (tv >= self.checklast_min): btv=False minmax=(self.checklast_max-self.checklast_min)*(1-self.checklast_shrinking)/2 if minmax>0: self.checklast_max=self.checklast_max-minmax self.checklast_min=self.checklast_min+minmax if self.checklast_min>self.checklast_max: self.checklaft_min=self.checklast_max else: # check if new value is exact within last values testvalue=self.value[-1*min(self.checklast,len(self.value)):] if isinstance(testvalue,list): if tv in testvalue: btv=False else: if tv==testvalue: btv=False if btv: self.value.append(tv) if self.bchecklast: # calc new min/max values checklast_amount=min(self.checklast,len(self.value)) btv=True testvalue=[] startcheck=len(self.value)-1 # get last unique values while btv: tv=self.value[startcheck] if not tv in testvalue: testvalue.append(tv) startcheck=startcheck-1 if (startcheck<=0) or (len(testvalue)>self.checklast): btv=False if isinstance(testvalue,list): if len(testvalue)>=3: # simple check for outlier tvrmin=testvalue tvrmax=testvalue tvrmin.remove(min(tvrmin)) tvrmax.remove(max(tvrmin)) tvrminm=numpy.mean(tvrmin) tvrmaxm=numpy.mean(tvrmax) tvrmins=numpy.std(tvrmin)*3 tvrmaxs=numpy.std(tvrmax)*3 if min(testvalue)<(tvrminm-tvrmins): testvalue.remove(min(testvalue)) if max(testvalue)>(tvrmaxm+tvrmaxs): testvalue.remove(max(testvalue)) self.checklast_min=min(testvalue) self.checklast_max=max(testvalue) else: self.checklast_min=testvalue self.checklast_max=testvalue if self.mqtt_bool: _thread.start_new_thread(self.send_mqtt,(0,)) if self.bsql: _thread.start_new_thread(self.send_sql,(0,)) if (len(self.mval)==0) and (len(self.value)>1): self.stat_mean=numpy.mean(self.value) self.stat_std=numpy.std(self.value)/numpy.sqrt(len(self.value)-1) self.mval.append(value) if self.bsense: if len(self.mval)==self.mean_count: mmean=numpy.mean(self.mval) nstd=self.stat_std+numpy.std(self.mval)/numpy.sqrt(len(self.mval)-1) bsave=self.bstoreeach if abs(mmean-self.stat_mean) > (self.sigma*nstd): bsave=True if bsave: if self.stat_val_std==0: self.stat_val_std=abs(self.stat_val_mean/100) nstd=10**(round(numpy.log10(numpy.maximum(0.001,numpy.std(self.mval))))-1) mmean=round(mmean/nstd,0)*nstd self.act_value=int(self.mult*mmean) self.act_std=int(self.mult*numpy.std(self.mval)) self.act_time=int(1000*time.time()) if self.bfile: _thread.start_new_thread(self.send_file,(0,)) if self.bsense: _thread.start_new_thread(self.upload_osm,(0,)) self.mval=[] if len(self.value)>=self.ring_length: self.value=self.value[((-1)*self.ring_length):] def get_ring(self): return(self.value) def get_act_value(self): return(self.act_value) def send_file(self,trigger=0): try: f1=open(self.store_file.format(int(time.time()/3600)),"a") except: print("could not open log file") else: f1.write("time:{0};".format(int(self.act_time))+self.var_name+":{0}".format(self.act_value)+"\n") f1.close() def send_sql(self,trigger=0): sv=(-99) if len(self.value)>0: if len(self.value)>1: if self.value[-1]!=self.value[-2]: sv=self.value[-1] else: sv=self.value[-1] if sv >(-99): act_time=int(1000*time.time()) self.json_out['payload']['measures'][act_time]=int(sv*self.mult) if ((act_time-self.sql_last_transmit)>(self.sql_min_wait*1000)) and (len(self.json_out['payload']['measures'])>0): self.sign() try: self._r=requests.post(self.sqlurl,json=json.dumps(self.sql_out)) except: self._r={"status_code":404} else: if self._r.status_code==200: self.json_out['payload']['measures']={} self.sql_last_transmit=time.time()*1000 def upload_osm(self,trigger=0): if (self.act_time-self.sense_last_time)>(self.sense_intervall*1000): r = requests.post(self.sense_url,json={'value': float(self.act_value)/self.mult}) if (r.status_code != requests.codes.ok) & (r.status_code != 201): print("Error %d: %s" % (r.status_code,r.text)) else: self.sense_last_time=self.act_time def send_mqtt(self,trigger=0): sv=(-99) if len(self.value)>0: if len(self.value)>1: if self.value[-1]!=self.value[-2]: sv=self.value[-1] else: sv=self.value[-1] if sv >(-99): try: publish.single(self.mqtt_topic,sv,hostname=self.mqtt_broker,port=self.mqtt_port) except: print("could not send mqtt")