Mercurial > templog

comparison py/uploader.py @ 527:dd8895652485

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