Simplify poll script (remove priority queue) so that it does not crash anymore

poll.py

@@ -6,64 +6,31 @@ import time
 import grequests
 import json
 from argparse import ArgumentParser
-from Queue import PriorityQueue
 
-# TODO: try round-robin queue instead of priority queue
-# TODO: log update times to evaluate queue type
-
-exchanges = {
-    'https://btc-e.com/api/3/ticker/btc_usd':
-        ('btce', ('btc_usd', 'last'), 'USD'),
-    'https://api.gdax.com/products/BTC-USD/ticker':
-        ('coinbase', ('price',), 'USD'),
-    'https://www.bitstamp.net/api/v2/ticker/btcusd/':
-        ('bitstamp', ('last',), 'USD'),
-    'https://api.bitfinex.com/v1/pubticker/btcusd':
-        ('bitfinex', ('last_price',), 'USD'),
-    'https://data.btcchina.com/data/ticker?market=btccny':
-        ('btcchina', ('ticker', 'last'), 'CNY'),
-    'http://api.huobi.com/staticmarket/ticker_btc_json.js':
-        ('huobi', ('ticker', 'last'), 'CNY'),
-    'https://www.okcoin.com/api/v1/ticker.do?symbol=btc_usd':
-        ('okcoin', ('ticker', 'last'), 'CNY'),
-    'https://api.kraken.com/0/public/Ticker?pair=XXBTZEUR':
-        ('kraken', ('result', 'XXBTZEUR', 'c', 0), 'EUR'),
-}
-
-marketcaps = {
-    'https://api.coinmarketcap.com/v1/ticker/bitcoin/':
-        ('coinmarketcap', (0, 'market_cap_usd'), 'USD'),
-}
-
-status = {}
-
-queue = PriorityQueue(len(exchanges) + len(marketcaps))
-
-
-def generate_requests():
-    assert queue.empty()
-
-    for url in exchanges.iterkeys():
-        queue.put((0, url))
-
-    for url in marketcaps.iterkeys():
-        queue.put((0, url))
-
-    while True:
-        scheduled, url = queue.get()
-        wait = scheduled - time.time()
-
-        if wait > 0:
-            interval = status[url][1][1]
-            time.sleep(min(wait, interval))
-
-        yield grequests.get(url, timeout=5)
-
-
-def requeue_after_error(req, e):
+exchanges = [
+    ('https://btc-e.com/api/3/ticker/btc_usd',
+        'btce', ('btc_usd', 'last'), 'USD'),
+    ('https://api.gdax.com/products/BTC-USD/ticker',
+        'coinbase', ('price',), 'USD'),
+    ('https://www.bitstamp.net/api/v2/ticker/btcusd/',
+        'bitstamp', ('last',), 'USD'),
+    ('https://api.bitfinex.com/v1/pubticker/btcusd',
+        'bitfinex', ('last_price',), 'USD'),
+    ('https://data.btcchina.com/data/ticker?market=btccny',
+        'btcchina', ('ticker', 'last'), 'CNY'),
+    ('http://api.huobi.com/staticmarket/ticker_btc_json.js',
+        'huobi', ('ticker', 'last'), 'CNY'),
+    ('https://www.okcoin.com/api/v1/ticker.do?symbol=btc_usd',
+        'okcoin', ('ticker', 'last'), 'CNY'),
+    ('https://api.kraken.com/0/public/Ticker?pair=XXBTZEUR',
+        'kraken', ('result', 'XXBTZEUR', 'c', 0), 'EUR'),
+    ('https://api.coinmarketcap.com/v1/ticker/bitcoin/',
+        'coinmarketcap', (0, 'market_cap_usd'), 'USD'),
+]
+
+
+def handle_error(req, e):
     print >>sys.stderr, 'Error:', e.message
-    status[req.url][1][0] += 0.5
-    queue.put((status[req.url][1][0], req.url))
 
 
 if __name__ == '__main__':
@@ -75,21 +42,14 @@ if __name__ == '__main__':
                         '(default dev/shm/tothemoon)')
     parser.add_argument('-i', '--interval', metavar='SECONDS=2', type=float,
                         default=2, help='exchange poll interval')
-    parser.add_argument('-c', '--cap-interval', metavar='SECONDS=60', type=float,
-                        default=60, help='market cap poll interval')
     parser.add_argument('-l', '--list', action='store_true', default=False,
                         help='list api info')
     args = parser.parse_args()
 
     root = args.dir
 
-    for url, ex in exchanges.iteritems():
-        status[url] = ex, [0, args.interval]
-    for url, cap in marketcaps.iteritems():
-        status[url] = cap, [0, args.cap_interval]
-
     if args.list:
-        for url, ((exid, indices, currency), stat) in status.iteritems():
+        for url, exid, indices, currency in exchanges:
             print exid
             print '  url:     ', url
             print '  currency:', currency
@@ -100,7 +60,7 @@ if __name__ == '__main__':
 
     # clean up created files on exit
     def remove_files():
-        for (exid, indices, currency), stat in status.itervalues():
+        for url, exid, indices, currency in exchanges:
             path = root + '/' + exid
             if os.path.exists(path):
                 os.remove(path)
@@ -122,24 +82,42 @@ if __name__ == '__main__':
     # keep updating URLs from priority queue, adding them back to the queue
     # scheduled after their current interval
     try:
-        for res in grequests.imap(generate_requests(),
-                exception_handler=requeue_after_error):
-            (exid, indices, currency), stat = status[res.request.url]
-            interval = stat[1]
-            stat[0] = scheduled = time.time() + interval
-            queue.put((scheduled, res.request.url))
-
-            try:
-                last = res.json()
-            except ValueError as e:
-                print >>sys.stderr, 'invalid response from %s:' % exid, e.message
-                continue
-
-            for i in indices:
-                last = last[i]
-            last = float(last)
-
-            with open(root + '/' + exid, 'w') as f:
-                json.dump({'last': last, 'currency': currency}, f)
+        last_update = 0
+
+        while True:
+            diff = time.time() - last_update
+            if diff < args.interval:
+                time.sleep(args.interval - diff)
+
+            requests = (grequests.get(ex[0], timeout=args.interval * 1.5)
+                        for ex in exchanges)
+            responses = grequests.map(requests, exception_handler=handle_error)
+
+            for (url, exid, indices, currency), res in zip(exchanges, responses):
+                status = {'last': None, 'currency': currency, 'updated': 0}
+
+                if res is None:
+                    print >>sys.stderr, exid, 'unknown error'
+                elif res.status_code != 200:
+                    print >>sys.stderr, 'server error %d at %s' % (res.status_code, exid)
+                else:
+                    try:
+                        last = res.json()
+
+                        for i in indices:
+                            last = last[i]
+                        last = float(last)
+
+                        status['updated'] = time.time()
+                        status['last'] = last
+                    except ValueError as e:
+                        print >>sys.stderr, 'invalid response from %s:' % exid, e.message
+                    except KeyError as e:
+                        print >>sys.stderr, 'unexpected response content from %s:' % exid, res.text
+
+                with open(root + '/' + exid, 'w') as f:
+                    json.dump(status, f)
+
+            last_update = time.time()
     except KeyboardInterrupt:
         pass