Temperature-Monitor/temperature.py at master · toddq/Temperature-Monitor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
#!/usr/bin/python

import os
import time
import glob
import sys
from threading import Thread
from notification import Notification
from collections import deque

os.system('modprobe w1-gpio')
os.system('modprobe w1-therm')

base_dir = '/sys/bus/w1/devices/'
try:
    device_folder = glob.glob(base_dir + '28*')[0]
except:
    print "No temperature input device found.  Exiting."
    sys.exit(1)
device_file = device_folder + '/w1_slave'

def read_raw_temp():
    f = open(device_file, 'r')
    lines = f.readlines()
    f.close()
    return lines

def get_temp():
    lines = read_raw_temp()
    if lines[0].strip()[-3:] != 'YES':
        #print "raw value was " + lines[0]
        return None
    pos = lines[1].find('t=')
    if pos != -1:
        temp_string = lines[1][pos+2:]
        temp_c = float(temp_string) / 1000.0
        temp_f = temp_c * 9.0 / 5.0 + 32.0
        return temp_c, temp_f
    else:
        print "no temp in line: " + lines[1]
        return None


class Temperature:
    # current value is degrees F, the default
    current_value = 0
    current_value_c = 0;
    historical_values = deque(maxlen=50)
    counter = 0

    # target value is degrees F, the default
    target_value = None
    increasing_target = True
    target_reached = False

    notifications = []

    should_stop = False

    def __init__(self):
        print "Temperature init"
        self.read_value()

    def read_value(self):
        val = get_temp();
        if val != None:
            self.current_value_c = val[0]
            self.current_value = int(val[1])
            self.add_history(time.time(), self.current_value)
            self.check_target()

    def add_history(self, timestamp, value):
    	if self.counter == 5:
            #print "adding value to history " + timestamp + "  " + str(value)
            self.historical_values.append( [timestamp, value] )
            self.counter = 0
        else:
        	self.counter += 1

    def print_history(self):
    	for elem in self.historical_values:
    		#print "%s  %i" % elem[0], elem[1]
    		#print str(elem)
    		print elem[0] + "  " + str(elem[1])

    def set_target(self, target_temp):
        print "setting target to " + str(target_temp)
        self.target_value = target_temp
        self.increasing_target = True if target_temp > self.current_value else False
        self.target_reached = False

    def clear_target(self):
        self.target_value = None
        self.target_reached = False

    def check_target(self):
        if self.target_value and not self.target_reached:
            if (self.increasing_target and self.current_value >= self.target_value) or (not self.increasing_target and self.current_value <= self.target_value):
                self.target_reached = True
                self.fire_target_reached()

    def fire_target_reached(self):
        msg = "Target temp of %i F was reached!" % (self.target_value)
        print msg
        for notification in self.notifications[:]:
            print "notifying " + notification.email
            notification.send(msg)
            self.notifications.remove(notification)

    def add_notification(self, notification):
        # don't add duplicate
        if not [x for x in self.notifications if x.email == notification.email]:
        #if not notification in self.notifications:
            print "adding notification for " + notification.email
            self.notifications.append(notification)
        else:
            print "not adding duplicate notification"

    def loop(self):
        while self.should_stop is False:
          self.read_value()
          time.sleep(5)
        print "Loop exited"

    def start(self):
        self.should_stop = False
        print "Starting thread"
        t = Thread(target=self.loop)
        t.start()

    def stop(self):
        print "Stop requested"
        self.should_stop = True

    def _test_notification(self):
        print "%i F" % (self.current_value)
        time.sleep(1)
        print "target %i F" % (self.current_value+2)
        self.set_target(self.current_value+2)
        self.add_notification(Notification("todd@quessenberry.com"))
        self.current_value = self.current_value + 3
        self.check_target()

    def _test_history(self):
		for i in range(60):
			#print "reading..."
			self.read_value()
			time.sleep(1)
		self.print_history()

if '__main__' == __name__:
    temp = Temperature()
    temp.read_value()
    #temp.start()
    #time.sleep(2)
    temp.set_target(75)
    temp.read_value()
    temp.set_target(-1)
    #for i in range(10):
    #    print "%i F " % (temp.current_value)
    #    time.sleep(2)
    #temp.stop()
    #temp._test_notification()
    #temp.add_notification(Notification("todd@quessenberry.com"))
    #temp.add_notification(Notification("todd@quessenberry.com"))
    #temp._test_history()

    #while True:
    #    print(get_temp())
    #    time.sleep(1)