from .DataStream import EMchPosmap as pos_map, span_days, time_border, calc_interval from .PainterBase import PainterBase from .MerchantReader import MerchantReader from matplotlib.figure import Figure from matplotlib import ticker from io import BytesIO import numpy as np from .algorithm import calc_mchratios, calc_morder_send import time as time import logging logger = logging.getLogger('painter') _all_mchids = set() def add_mchid(mchid): if mchid not in _all_mchids: _all_mchids.add(mchid) def get_mchids(): return list(_all_mchids) def mch_paths(reader: MerchantReader, tuple_pathes: dict, days: list): count = len(days) for mchid, tup in tuple_pathes.items(): mch_datas = reader.init_data(count) for _card_type, _spec in tup: for i, day in enumerate(days): data = reader.read(day, mchid, _card_type, _spec) if data is not None: column_pos = i * 86400 view = mch_datas[:, column_pos:column_pos + 86400] view += data yield mchid, None, None, mch_datas def allpathes(reader: MerchantReader, tuple_pathes: dict, days: list, spec=None): count = len(days) show_detail = True if len(list(tuple_pathes.keys())) == 1 else False if show_detail == False: all_datas = reader.init_data(count) else: all_datas = None for mchid, tup in tuple_pathes.items(): add_mchid(mchid) ch_datas = reader.init_data(count) for _card_type, _spec in tup: if spec is not None and _spec != spec: continue if show_detail: detail_datas = reader.init_data(count) else: detail_datas = None for i, day in enumerate(days): data = reader.read(day, mchid, _card_type, _spec) if data is not None: column_pos = i * 86400 view = ch_datas[:, column_pos:column_pos + 86400] view += data if show_detail: view = detail_datas[:, column_pos:column_pos + 86400] view += data if show_detail: yield mchid, _card_type, _spec, detail_datas if all_datas is not None: all_datas += ch_datas yield mchid, None, None, ch_datas if show_detail == False: yield 'all', None, None, all_datas class MerchantPainter(PainterBase): def _fig_funs(self): def create(): fig = Figure(figsize=(19, 8)) ax = fig.subplots() ax.set_title('success ratio') ax.set(xlabel='time', ylabel='ratio') return ax, fig def flush(ax, fig, xticks=None, xlables=None, yticks=None, ylables=None): ax.yaxis.set_major_formatter(ticker.PercentFormatter(xmax=1, decimals=4)) if xticks is not None: ax.set_xticks(ticks=xticks) if xlables is not None: ax.set_xticklabels(xlables) if yticks is not None: ax.set_yticks(ticks=yticks) if ylables is not None: ax.set_yticklabels(ylables) fig.autofmt_xdate() ax.grid() # fig.subplots_adjust(left=0.1, right=0.8, top=0.95, bottom=0.1) ax.legend(bbox_to_anchor=(1, 1), loc='upper left') fig.tight_layout() buf = BytesIO() fig.savefig(buf, format="png") return buf return create, flush def _label(self, chname, succ, count, card_type=None, spec=None): _card_type = None 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' elif card_type == 7: _card_type = 'TH' lable = f"{chname}" if _card_type is not None: lable += f"-{_card_type}" if spec is not None: lable += f"-{spec}" if count > 0: ratio = round(succ * 100 / count, 2) else: ratio = 0.00 lable += f":{succ}/{count}={ratio}%" return lable,ratio