123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346 |
- import os
- import time as stime
- import redis
- import h5py
- from os import path
- import re
- from datetime import timedelta
- import numpy as np
- from matplotlib.figure import Figure
- from matplotlib import ticker
- from io import BytesIO
- import logging
- class MchDataCenter(object):
- pos_map = {
- 'commit': 0, 'notify': 1
- }
- def __init__(self):
- self._mquit = False
- self._mRHost = ''
- self._mRPort = 6379
- self._file_name = '/var/www/html/data/stdata/user.hdf5'
- def set_redis(self, rhost, rport):
- self._mRHost = rhost
- self._mRPort = rport
- def stop(self):
- self._mquit = True
- pass
- def prepare_data(self):
- while True:
- try:
- pool = redis.ConnectionPool(host=self._mRHost, port=self._mRPort, db=0)
- r = redis.Redis(connection_pool=pool)
- if path.exists(self._file_name):
- hfive = h5py.File(self._file_name, 'a')
- else:
- hfive = h5py.File(self._file_name, 'w')
- self.read_redis(hfive, r, 'nc_user_monitor_commit', 'commit')
- self.read_redis(hfive, r, 'nc_user_monitor_success', 'notify')
- hfive.close()
- self.del_redis(r, 'nc_user_monitor_commit')
- self.del_redis(r, 'nc_user_monitor_success')
- except Exception as ex:
- print(ex)
- finally:
- for i in range(60):
- if self._mquit == True:
- break
- else:
- stime.sleep(1)
- def del_redis(self, redis, name):
- latest_time = int(stime.time()) - 300
- for item in redis.hscan_iter(name):
- key = str(item[0], encoding="utf-8")
- items = re.split(r'-', key)
- fdel = True
- if len(items) == 5:
- (mchid, quality, card_type, amount, time) = items
- time = int(time)
- if latest_time <= time:
- fdel = False
- if fdel:
- redis.hdel(name, key)
- pass
- def read_redis(self, hfive, redis, name, prefix):
- i = 0
- for item in redis.hscan_iter(name):
- key = str(item[0], encoding="utf-8")
- val = str(item[1], encoding="utf-8")
- print(f'{prefix}:{i}')
- i += 1
- self.parase(hfive, key, val, prefix)
- def parase(self, hfive, text, val, prefix):
- items = re.split(r'-', text)
- if len(items) != 5:
- return False
- (mchid, quality, card_type, amount, time) = items
- pos = self.pos_map[f'{prefix}']
- time = int(time)
- today = self.day_stamp(time)
- path = f'/{today}/{mchid}/{quality}/{card_type}/{amount}'
- if path not in hfive:
- hfive[path] = np.zeros((2, 86400))
- diff = time - today
- if diff < 0:
- print(diff)
- hfive[path][pos, diff] = val
- print(path, pos, diff, val, hfive[path][pos, diff])
- pass
- def day_stamp(self, stamp):
- stamp = int(stamp)
- x = stime.gmtime(stamp + 8 * 3600)
- diff = timedelta(hours=x.tm_hour, minutes=x.tm_min, seconds=x.tm_sec)
- today = stamp - diff.total_seconds()
- return int(today)
- def _days(self, root):
- result = []
- try:
- for name, sub in root.items():
- if isinstance(sub, h5py.Group):
- result.append(name)
- except Exception as ex:
- print(ex)
- finally:
- return result
- def days(self):
- try:
- hfive = h5py.File(self._file_name, 'r')
- root = hfive.require_group('/')
- days = self._days(root)
- hfive.close()
- return days
- except Exception as ex:
- print(ex)
- return []
- def paths(self, time_stamp):
- try:
- day_stamp = self.day_stamp(time_stamp)
- hfive = h5py.File(self._file_name, 'r')
- group = hfive.require_group(f'/{day_stamp}')
- paths = self.dir(group)
- hfive.close()
- return paths
- except Exception as ex:
- print(ex)
- return []
- def dir(self, group):
- result = []
- for name, sub in group.items():
- if isinstance(sub, h5py.Group):
- result.extend(self.dir(sub))
- else:
- result.append(sub.name)
- return result
- def all_none(self, **kwargs):
- for key, val in kwargs.items():
- if val is not None:
- return False
- return True
- def draw_plot(self, start_time, interval=300, **kwargs):
- logger = logging.getLogger('app')
- hfive = h5py.File(self._file_name, 'r')
- try:
- filer_text, paths = self.datasets(hfive, start_time, **kwargs)
- day_stamp = self.day_stamp(start_time)
- start_pos = start_time - day_stamp
- cur_day = self.day_stamp(stime.time())
- if day_stamp == cur_day:
- end_pos = int(stime.time()) - day_stamp
- else:
- end_pos = -1
- fig = Figure(figsize=(16, 8))
- ax = fig.subplots()
- predata = np.zeros((2, 86400))
- x = np.arange(0, 86400, interval)
- sub_count = 0
- for path, data in self.read_data(hfive, paths):
- data = np.array(data)
- predata = predata + data
- ret = self._draw_plot(ax, x, day_stamp, start_pos, end_pos, data, interval, path)
- if ret:
- sub_count += 1
- if sub_count > 1:
- self._draw_plot(ax, x, day_stamp, start_pos, end_pos, predata, interval, filer_text)
- ax.legend()
- ax.grid()
- ax.set_title('success ratio')
- ax.set(xlabel='time', ylabel='ratio')
- fig.autofmt_xdate()
- fig.subplots_adjust(left=0.05, right=0.999, top=0.95, bottom=0.1)
- buf = BytesIO()
- fig.savefig(buf, format="png")
- return buf
- except Exception as ex:
- print(ex)
- finally:
- hfive.close()
- def read_data(self, hfive, paths):
- for path in paths:
- yield path, hfive[path]
- def datasets(self, hfive, start_time, **kwargs):
- logger = logging.getLogger('app')
- day_stamp = self.day_stamp(start_time)
- sday = f'{day_stamp}'
- root = hfive.require_group('/')
- days = self._days(root)
- if sday not in days:
- return False
- group = hfive.require_group(sday)
- dsets = self.dir(group)
- mchid = quality = card_type = amount = None
- for key, val in kwargs.items():
- if val is None:
- continue
- if key == 'mchid':
- mchid = val
- elif key == 'quality':
- quality = f'{val}'
- elif key == 'card_type':
- card_type = f'{val}'
- elif key == 'amount':
- amount = f'{val}'
- else:
- continue
- return self._filter(dsets, mchid=mchid, quality=quality, card_type=card_type, amount=amount)
- def _filter(self, dsets, mchid=None, quality=None, card_type=None, amount=None):
- filer_text = ''
- if mchid is not None:
- filer_text = mchid
- if quality is not None:
- filer_text = filer_text + f"-qua:{quality}"
- if card_type is not None:
- filer_text = filer_text + f"-type:{card_type}"
- if amount is not None:
- filer_text = filer_text + f"-amount:{amount}"
- paths = []
- for text in dsets:
- items = re.split(r'/', text)
- if len(items) != 6:
- return False
- (_, _sday, _mchid, _quality, _card_type, _amount) = items
- if (mchid is not None) and (_mchid != mchid):
- continue
- if (quality is not None) and (_quality != quality):
- continue
- if (card_type is not None) and (_card_type != card_type):
- continue
- if (amount is not None) and (_amount != amount):
- continue
- paths.append(text)
- return filer_text, paths
- def _draw_plot(self, ax, x, day_stamp, start_pos, end_pos, data, interval=300, path=''):
- # 'commit-succ': 0, 'commit-fail': 1, 'notify-succ': 2, 'notify-fail': 3, 'user_succ': 4
- logging.getLogger('app').debug("path=%s", path)
- all = data[0]
- all = all.reshape((-1, interval))
- all = np.sum(all, axis=1)
- ySucc = data[1]
- ySucc = ySucc.reshape((-1, interval))
- ySucc = np.sum(ySucc, axis=1)
- if end_pos == -1:
- pos = np.where(x >= start_pos)
- x = x[pos]
- ySucc = ySucc[pos]
- all = all[pos]
- else:
- pos = np.where(start_pos <= x)
- x = x[pos]
- ySucc = ySucc[pos]
- all = all[pos]
- pos = np.where(x < end_pos)
- x = x[pos]
- ySucc = ySucc[pos]
- all = all[pos]
- succ_count = int(np.sum(ySucc))
- all_count = int(np.sum(all))
- if all_count < 1:
- return False
- pos = np.where(ySucc > all)
- ySucc[pos] = all[pos]
- # if len(opened[0]) == 0:
- # logging.getLogger('app').debug("path=%s,opened=False", path)
- # return False
- ySucc = ySucc / (all + 0.00000001)
- xs = np.array([stime.strftime('%H:%M', stime.localtime(d + day_stamp)) for d in x])
- ax.yaxis.set_major_formatter(ticker.PercentFormatter(xmax=1, decimals=0))
- ax.plot(xs, ySucc, ls='--', marker='o', label=self._label(path, succ_count, all_count))
- return True
- def _label(self, path, count, all):
- ratio = 0.00
- if all > 0:
- ratio = round(count * 100 / all, 2)
- items = re.split(r'/', path)
- if len(items) == 6:
- (_, _sday, _chname, _quality, _card_type, _amount) = items
- card_type = ''
- if _card_type == '1':
- card_type = 'SY'
- elif _card_type == '2':
- card_type = 'SH'
- elif _card_type == '4':
- card_type = 'YD'
- elif _card_type == '5':
- card_type = 'LT'
- elif _card_type == '6':
- card_type = 'DX'
- return f"{_chname}-{_quality}-{card_type}-{_amount}:{count}/{all} = {ratio}%"
- else:
- if path == '' or path is None:
- path = 'average'
- return f"{path}:{count}/{all} = {ratio}%"
- mchDataCenter = MchDataCenter()
|