server/filter_meteo.py

   1 #!/usr/bin/python
   2
   3 import MySQLdb
   4 import ConfigParser
   5 import sys
   6
   7 from pprint import pprint
   8 import datetime
   9
  10 import numpy as np
  11
  12 import scipy.signal
  13
  14 global database
  15
  16 def GetTables():
  17   if database:
  18     c = database.cursor()
  19     c.execute("SELECT s.id sid,p.id pid FROM sensors s,st_parameters p where s.st_id=p.st_id and p.id>=0")
  20     return c.fetchall()
  21   else:
  22     print "No connection to DB"
  23     exit()
  24
  25 def Today():
  26   dt = datetime.datetime.now()
  27   d_truncated = datetime.date(dt.year, dt.month, dt.day)
  28   return d_truncated
  29
  30 def Tomorrow():
  31   dt = Today()
  32   return dt + datetime.timedelta(days=1)
  33
  34 def Yesterday():
  35   dt = Today()
  36   return dt - datetime.timedelta(days=1)
  37
  38 def Prehistoric():
  39   dt = datetime.date(2000,01,01)
  40   return dt
  41
  42 def GetData(sid,pid,fromDate=Yesterday(),toDate=Today()):
  43   if database:
  44     c = database.cursor()
  45     c.execute("SELECT id,timestamp,value FROM sensor_values WHERE sensor_id=%s and parameter_id=%s and timestamp>=%s AND timestamp<%s",[sid,pid,fromDate.strftime('%Y-%m-%d %H:%M:%S'),toDate.strftime('%Y-%m-%d %H:%M:%S')])
  46     return c.fetchall()
  47   else:
  48     print "No connection to DB"
  49     exit()
  50
  51 def FixRecord(id,value):
  52   if database:
  53     c = database.cursor()
  54     command="UPDATE sensor_values  SET value={} WHERE id='{}'".format(value,id)
  55     c.execute(command)
  56   else:
  57     print "No connection to DB"
  58     exit()
  59
  60 def ProcessTable(sid,pid):
  61
  62   if process_all:
  63     data=GetData(sid,pid,Prehistoric(),Today())
  64   elif not current:
  65     data=GetData(sid,pid)
  66   else:
  67     data=GetData(sid,pid,Today(),Tomorrow())
  68
  69   if not data:
  70     return
  71
  72   sID=[]
  73   sTime=[]
  74   sValue=[]
  75   for rec in data:
  76     sID.append(rec[0])
  77     sTime.append(rec[1])
  78     sValue.append(rec[2])
  79   sValue=np.array(sValue)
  80
  81   sValueFilt=scipy.signal.medfilt(sValue,5)
  82
  83   sValueDiff=abs(sValue-sValueFilt)
  84
  85   avg=np.mean(sValueDiff)
  86
  87   for i in range(0,len(sTime)-1):
  88     if sValueDiff[i]>avg*filterThreshold:
  89       print "fixing %s : %5.2f %5.2f %5.2f" % (sTime[i],sValue[i],sValueFilt[i],sValueDiff[i])
  90       FixRecord(sID[i],sValueFilt[i])
  91
  92   database.commit()
  93
  94 if len(sys.argv)==2 and sys.argv[1]=='current':
  95   current=True
  96 else:
  97   current=False
  98
  99 if len(sys.argv)==2 and sys.argv[1]=='all':
 100   process_all=True
 101 else:
 102   process_all=False
 103
 104 try:
 105
 106   cfg = ConfigParser.RawConfigParser(allow_no_value=True)
 107   cfg.readfp(open('/etc/weathermon.conf'))
 108   dbhost = cfg.get("mysql","host")
 109   dbuser = cfg.get("mysql","user")
 110   dbpasswd = cfg.get("mysql","passwd")
 111   dbdb = cfg.get("mysql","db")
 112
 113   filterWindow = int(cfg.get("filter","window"))
 114   filterThreshold = float(cfg.get("filter","threshold"))
 115
 116 except:
 117
 118   print "Error reading configuration file"
 119   exit()
 120
 121 try:
 122
 123   database = MySQLdb.connect(host=dbhost,user=dbuser,passwd=dbpasswd,db=dbdb,use_unicode=True)
 124   database.set_character_set('utf8')
 125   c = database.cursor()
 126   c.execute('SET NAMES utf8;')
 127
 128   print "Connected..."
 129
 130 except:
 131
 132   print "Error connecting database"
 133   exit()
 134
 135 tables = GetTables()
 136
 137 for sid,pid in tables:
 138
 139   print "Processing sensor %d, parameter %d " % (sid,pid)
 140
 141   ProcessTable(sid,pid)
 142
 143 print "Processed "
 144