Транзакционная работа с БД для избежания блокировок.
[weathermon.git] / 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