Utilities.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. import string
  2. from binascii import hexlify
  3. from os import urandom
  4. import logging
  5. import logging.handlers
  6. import json
  7. import numpy
  8. def random_string(size):
  9. return hexlify(urandom(size)).decode('utf8')
  10. # return ''.join(random.choice(chars) for x in range(size))
  11. def get_exponential_delay(avg_delay, cache=[]):
  12. if cache == []:
  13. cache.extend(list(numpy.random.exponential(avg_delay, 10000)))
  14. return cache.pop()
  15. class StructuredMessage(object):
  16. def __init__(self, metadata):
  17. self.metadata = metadata
  18. def __str__(self):
  19. return ';'.join(str(x) for x in self.metadata) # json.dumps(self.metadata)
  20. def float_equlity(tested, correct=1.0):
  21. return correct * 0.99 < tested < correct * 1.01
  22. def stream_to_file(filename, stream):
  23. with open(filename, 'a') as outfile:
  24. outfile.write(stream.getvalue())
  25. def setup_logger(logger_name, filehandler_name, capacity=50000000):
  26. logger = logging.getLogger(logger_name)
  27. logger.setLevel(logging.DEBUG)
  28. filehandler = logging.FileHandler(filehandler_name)
  29. memoryhandler = logging.handlers.MemoryHandler(
  30. capacity=capacity,
  31. flushLevel=logging.ERROR,
  32. target=filehandler
  33. )
  34. logger.addHandler(memoryhandler)
  35. return logger