Mercurial > templog

comparison py/uploader.py @ 293:d15dda1b1f76

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
merge
author Matt Johnston <matt@ucc.asn.au>
date Sat, 06 Jul 2019 18:29:45 +0800
parents 28eb733cb803 26eee8591f61
children 6bacd8ca9f8f
comparison
equal deleted inserted replaced
292:28eb733cb803 293:d15dda1b1f76
1 import json 1 import json
2 import hmac 2 import hmac
3 import hashlib
3 import zlib 4 import zlib
4 import binascii 5 import binascii
5 import logging 6 import logging
7 import asyncio
6 8
7 import gevent 9 import aiohttp
8 import requests
9 10
10 import config 11 import config
11 from utils import L,D,EX,W,E 12 from utils import L,D,EX,W,E
12 import utils 13 import utils
13 14
14 class Uploader(gevent.Greenlet): 15 class Uploader(object):
15 def __init__(self, server): 16 def __init__(self, server):
16 gevent.Greenlet.__init__(self)
17 self.server = server 17 self.server = server
18 18
19 requests_log = logging.getLogger("requests") 19 @asyncio.coroutine
20 requests_log.setLevel(logging.WARNING) 20 def run(self):
21 21 # wait for the first read
22 def _run(self): 22 yield from asyncio.sleep(5)
23 gevent.sleep(5)
24 while True: 23 while True:
25 self.do() 24 yield from self.do()
26 self.server.sleep(config.UPLOAD_SLEEP) 25 yield from asyncio.sleep(config.UPLOAD_SLEEP)
27 26
28 def get_tosend(self, readings): 27 def get_tosend(self, readings):
29 tosend = {} 28 tosend = {}
30 29
31 tosend['fridge_on'] = self.server.fridge.is_on() 30 tosend['fridge_on'] = self.server.fridge.is_on()
35 34
36 tosend['wort_name'] = self.server.wort_name 35 tosend['wort_name'] = self.server.wort_name
37 tosend['fridge_name'] = self.server.wort_name 36 tosend['fridge_name'] = self.server.wort_name
38 37
39 tosend['current_params'] = dict(self.server.params) 38 tosend['current_params'] = dict(self.server.params)
39 tosend['current_params_epoch'] = self.server.params.get_epoch()
40 40
41 tosend['start_time'] = self.server.start_time 41 tosend['start_time'] = self.server.start_time
42 tosend['uptime'] = utils.uptime() 42 tosend['uptime'] = utils.uptime()
43 43
44 return tosend 44 return tosend
45 45
46 @asyncio.coroutine
46 def send(self, tosend): 47 def send(self, tosend):
47 js = json.dumps(tosend) 48 js = json.dumps(tosend)
48 js_enc = binascii.b2a_base64(zlib.compress(js)) 49 if self.server.test_mode():
49 mac = hmac.new(config.HMAC_KEY, js_enc).hexdigest() 50 D("Would upload %s to %s" % (js, config.UPDATE_URL))
50 send_data = {'data': js_enc, 'hmac': mac} 51 return
51 r = requests.post(config.UPDATE_URL, data=send_data, timeout=60) 52 js_enc = binascii.b2a_base64(zlib.compress(js.encode())).strip()
52 result = r.text 53 mac = hmac.new(config.HMAC_KEY.encode(), js_enc, hashlib.sha256).hexdigest()
53 if result != 'OK': 54 send_data = {'data': js_enc.decode(), 'hmac': mac}
55 r = yield from asyncio.wait_for(aiohttp.request('post', config.UPDATE_URL, data=send_data), 60)
56 result = yield from asyncio.wait_for(r.text(), 60)
57 if r.status == 200 and result != 'OK':
54 raise Exception("Server returned %s" % result) 58 raise Exception("Server returned %s" % result)
55 59
60 @asyncio.coroutine
56 def do(self): 61 def do(self):
62 try:
57 readings = self.server.take_readings() 63 readings = self.server.take_readings()
58 try:
59 tosend = self.get_tosend(readings) 64 tosend = self.get_tosend(readings)
65 D("tosend >>>%s<<<" % str(tosend))
60 nreadings = len(readings) 66 nreadings = len(readings)
61 self.send(tosend) 67 yield from self.send(tosend)
62 readings = None 68 readings = None
63 D("Sent updated %d readings" % nreadings) 69 D("Sent updated %d readings" % nreadings)
64 except requests.exceptions.RequestException, e: 70 except Exception as e:
65 E("Error in uploader: %s" % str(e))
66 except Exception, e:
67 EX("Error in uploader: %s" % str(e)) 71 EX("Error in uploader: %s" % str(e))
68 finally: 72 finally:
69 if readings is not None: 73 if readings is not None:
70 self.server.pushfront(readings) 74 self.server.pushfront(readings)