util.py

   1 from __future__ import division
   2
   3 import collections
   4 import hashlib
   5 import itertools
   6 import random
   7
   8 from twisted.internet import defer, reactor
   9 from twisted.python import failure
  10 from twisted.web import resource, server
  11
  12 class DeferredResource(resource.Resource):
  13     def render(self, request):
  14         def finish(x):
  15             if x is not None:
  16                 request.write(x)
  17             request.finish()
  18
  19         def finish_error(fail):
  20             request.setResponseCode(500) # won't do anything if already written to
  21             request.write('---ERROR---')
  22             request.finish()
  23             fail.printTraceback()
  24
  25         defer.maybeDeferred(resource.Resource.render, self, request).addCallbacks(finish, finish_error)
  26         return server.NOT_DONE_YET
  27
  28 class Event(object):
  29     def __init__(self):
  30         self.observers = {}
  31         self.one_time_observers = {}
  32         self.id_generator = itertools.count()
  33
  34     def watch(self, func):
  35         id = self.id_generator.next()
  36         self.observers[id] = func
  37         return id
  38     def unwatch(self, id):
  39         self.observers.pop(id)
  40
  41     def watch_one_time(self, func):
  42         id = self.id_generator.next()
  43         self.one_time_observers[id] = func
  44         return id
  45     def unwatch_one_time(self, id):
  46         self.one_time_observers.pop(id)
  47
  48     def happened(self, event=None):
  49         for func in self.observers.itervalues():
  50             func(event)
  51
  52         one_time_observers = self.one_time_observers
  53         self.one_time_observers = {}
  54         for func in one_time_observers.itervalues():
  55             func(event)
  56
  57     def get_deferred(self):
  58         df = defer.Deferred()
  59         self.watch_one_time(df.callback)
  60         return df
  61
  62 class Variable(object):
  63     def __init__(self, value):
  64         self.value = value
  65         self.changed = Event()
  66
  67     def set(self, value):
  68         if value == self.value:
  69             return
  70
  71         self.value = value
  72         self.changed.happened(value)
  73
  74 def sleep(t):
  75     d = defer.Deferred()
  76     reactor.callLater(t, d.callback, None)
  77     return d
  78
  79 def median(x):
  80     # don't really need a complex algorithm here
  81     y = sorted(x)
  82     left = (len(y) - 1)//2
  83     right = len(y)//2
  84     return (y[left] + y[right])/2
  85
  86 class StringBuffer(object):
  87     'Buffer manager with great worst-case behavior'
  88
  89     def __init__(self, data=''):
  90         self.buf = collections.deque([data])
  91         self.buf_len = len(data)
  92         self.pos = 0
  93
  94     def __len__(self):
  95         return self.buf_len - self.pos
  96
  97     def add(self, data):
  98         self.buf.append(data)
  99         self.buf_len += len(data)
 100
 101     def get(self, wants):
 102         if self.buf_len - self.pos < wants:
 103             raise IndexError('not enough data')
 104         data = []
 105         while wants:
 106             seg = self.buf[0][self.pos:self.pos+wants]
 107             self.pos += len(seg)
 108             while self.buf and self.pos >= len(self.buf[0]):
 109                 x = self.buf.popleft()
 110                 self.buf_len -= len(x)
 111                 self.pos -= len(x)
 112
 113             data.append(seg)
 114             wants -= len(seg)
 115         return ''.join(data)
 116
 117 def _DataChunker(receiver):
 118     wants = receiver.next()
 119     buf = StringBuffer()
 120
 121     while True:
 122         if len(buf) >= wants:
 123             wants = receiver.send(buf.get(wants))
 124         else:
 125             buf.add((yield))
 126 def DataChunker(receiver):
 127     '''
 128     Produces a function that accepts data that is input into a generator
 129     (receiver) in response to the receiver yielding the size of data to wait on
 130     '''
 131     x = _DataChunker(receiver)
 132     x.next()
 133     return x.send
 134
 135 class ReplyMatcher(object):
 136     def __init__(self, func, timeout=5):
 137         self.func = func
 138         self.timeout = timeout
 139         self.map = {}
 140
 141     def __call__(self, id):
 142         try:
 143             self.func(id)
 144             uniq = random.randrange(2**256)
 145             df = defer.Deferred()
 146             def timeout():
 147                 df, timer = self.map[id].pop(uniq)
 148                 df.errback(failure.Failure(defer.TimeoutError()))
 149                 if not self.map[id]:
 150                     del self.map[id]
 151             self.map.setdefault(id, {})[uniq] = (df, reactor.callLater(self.timeout, timeout))
 152             return df
 153         except:
 154             import traceback
 155             traceback.print_exc()
 156
 157     def got_response(self, id, resp):
 158         if id not in self.map:
 159             return
 160         for df, timer in self.map.pop(id).itervalues():
 161             timer.cancel()
 162             df.callback(resp)
 163
 164 class GenericDeferrer(object):
 165     def __init__(self, max_id, func, timeout=5):
 166         self.max_id = max_id
 167         self.func = func
 168         self.timeout = timeout
 169         self.map = {}
 170
 171     def __call__(self, *args, **kwargs):
 172         while True:
 173             id = random.randrange(self.max_id)
 174             if id not in self.map:
 175                 break
 176         df = defer.Deferred()
 177         def timeout():
 178             self.map.pop(id)
 179             df.errback(failure.Failure(defer.TimeoutError()))
 180         timer = reactor.callLater(self.timeout, timeout)
 181         self.func(id, *args, **kwargs)
 182         self.map[id] = df, timer
 183         return df
 184
 185     def got_response(self, id, resp):
 186         if id not in self.map:
 187             return
 188         df, timer = self.map.pop(id)
 189         timer.cancel()
 190         df.callback(resp)
 191
 192 class DeferredCacher(object):
 193     def __init__(self, func, backing=None):
 194         if backing is None:
 195             backing = {}
 196
 197         self.func = func
 198         self.backing = backing
 199         self.waiting = {}
 200
 201     @defer.inlineCallbacks
 202     def __call__(self, key):
 203         if key in self.waiting:
 204             yield self.waiting[key]
 205
 206         if key in self.backing:
 207             defer.returnValue(self.backing[key])
 208         else:
 209             self.waiting[key] = defer.Deferred()
 210             try:
 211                 value = yield self.func(key)
 212             finally:
 213                 self.waiting.pop(key).callback(None)
 214
 215         self.backing[key] = value
 216         defer.returnValue(value)
 217
 218 def pubkey_to_address(pubkey, testnet):
 219     if len(pubkey) != 65:
 220         raise ValueError('invalid pubkey')
 221     version = 111 if testnet else 0
 222     key_hash = chr(version) + hashlib.new('ripemd160', hashlib.sha256(pubkey).digest()).digest()
 223     checksum = hashlib.sha256(hashlib.sha256(key_hash).digest()).digest()[:4]
 224     return base58_encode(key_hash + checksum)
 225
 226 def base58_encode(data):
 227     return '1'*(len(data) - len(data.lstrip(chr(0)))) + natural_to_string(string_to_natural(data), '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz')
 228
 229 def natural_to_string(n, alphabet=None, min_width=1):
 230     if alphabet is None:
 231         s = '%x' % (n,)
 232         if len(s) % 2:
 233             s = '\0' + x
 234         return s.decode('hex').rjust(min_width, '\x00')
 235     res = []
 236     while n:
 237         n, x = divmod(n, len(alphabet))
 238         res.append(alphabet[x])
 239     res.reverse()
 240     return ''.join(res).rjust(min_width, '\x00')
 241
 242 def string_to_natural(s, alphabet=None):
 243     if alphabet is None:
 244         s = s.encode('hex')
 245         return int(s, 16)
 246     if not s or (s != alphabet[0] and s.startswith(alphabet[0])):
 247         raise ValueError()
 248     return sum(alphabet.index(char) * len(alphabet)**i for i, char in enumerate(reversed(s)))
 249
 250
 251 class DictWrapper(object):
 252     def encode_key(self, key):
 253         return key
 254     def decode_key(self, encoded_key):
 255         return encoded_key
 256     def encode_value(self, value):
 257         return value
 258     def decode_value(self, encoded_value):
 259         return encoded_value
 260
 261     def __init__(self, inner):
 262         self.inner = inner
 263
 264     def __len__(self):
 265         return len(self.inner)
 266
 267     def __contains__(self, key):
 268         return self.encode_key(key) in self.inner
 269
 270     def __getitem__(self, key):
 271         return self.decode_value(self.inner[self.encode_key(key)])
 272
 273     def __setitem__(self, key, value):
 274         self.inner[self.encode_key(key)] = self.encode_value(value)
 275
 276     def __delitem__(self, key):
 277         del self.inner[self.encode_key(key)]
 278
 279     def keys(self):
 280         return map(self.decode_key, self.inner.keys())