ChannelWriter.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  1. # from . import DataHandler #此时是导入文件
  2. from .DataStream import DataWriteStream, ChPosmap, day_stamp
  3. import numpy as np
  4. __all__ = ['ChannelWriter']
  5. class ChannelWriter(DataWriteStream, ChPosmap):
  6. def write(self, method, params):
  7. if method == 'refill_commit':
  8. self._onCommit(params)
  9. elif method == 'notify_succ':
  10. self._onSucc(params)
  11. elif method == 'notify_fail':
  12. self._onFail(params)
  13. else:
  14. pass
  15. def _onCommit(self, params):
  16. def parse(input):
  17. return input['channel_name'], input['time'], input['spec'], input['card_type'], input['channel_amount']
  18. chname, time, spec, card_type, channel_amount = parse(params)
  19. path, pos = self.path_pos(chname, time, spec, card_type)
  20. self.file[path][self._pos_map['commit_count'], pos] += 1
  21. self.file[path][self._pos_map['commit_amounts'], pos] += channel_amount
  22. pass
  23. def _onSucc(self, params):
  24. def parse(input):
  25. return input['channel_name'], input['time'], input['spec'], input['card_type'], input['channel_amount'], input['period']
  26. chname, time, spec, card_type, channel_amount, period = parse(params)
  27. path, pos = self.path_pos(chname, time, spec, card_type)
  28. self.file[path][self._pos_map['succ_count'], pos] += 1
  29. self.file[path][self._pos_map['succ_amounts'], pos] += channel_amount
  30. self.file[path][self._pos_map['succ_periods'], pos] += period
  31. pass
  32. def _onFail(self, params):
  33. def parse(input):
  34. return input['channel_name'], input['time'], input['spec'], input['card_type'], input['channel_amount'], input['period']
  35. chname, time, spec, card_type, channel_amount, period = parse(params)
  36. path, pos = self.path_pos(chname, time, spec, card_type)
  37. self.file[path][self._pos_map['fail_count'], pos] += 1
  38. self.file[path][self._pos_map['fail_amounts'], pos] += channel_amount
  39. self.file[path][self._pos_map['fail_periods'], pos] += period
  40. pass
  41. def path_pos(self, chname, time, spec, card_type):
  42. today = day_stamp(time)
  43. path = f'/{self._version}/{today}/{chname}/{card_type}/{spec}'
  44. hfive = self.file
  45. if path not in hfive:
  46. dim = len(self._pos_map)
  47. hfive[path] = np.zeros((dim, 86400))
  48. return path, time - today