from .DataStream import ENetPosmap as pos_map, day_stamp, span_days, time_border, calc_interval from .NetchkReader import NetchkReader from .PainterBase import PainterBase from matplotlib.figure import Figure from matplotlib import ticker from io import BytesIO import numpy as np import time as time import logging logger = logging.getLogger('ChannelPainter') _all_net_channels = set() def _add_net_channel(channel): if channel not in _all_net_channels: _all_net_channels.add(channel) def get_net_channels(): logger.debug(_all_net_channels) return list(_all_net_channels) def net_pathes(reader: NetchkReader, names: set, days: list): count = len(days) if len(names) > 1: _all_data = reader.init_data(count) else: _all_data = None for name in names: _add_net_channel(name) _ch_data = reader.init_data(count) for i, day in enumerate(days): data = reader.read(day, name) if data is not None: column_pos = i * 86400 view = _ch_data[:, column_pos:column_pos + 86400] view += data if _all_data is not None: _all_data += _ch_data yield name, _ch_data if _all_data is not None: yield 'all', _all_data class NetcheckPainter(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, ticks, lables): ax.yaxis.set_major_formatter(ticker.PercentFormatter(xmax=1, decimals=4)) ax.set_xticks(ticks=ticks) ax.set_xticklabels(lables) 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): lable = f"{chname}" if count > 0: ratio = round(succ * 100 / count, 5) else: ratio = 0.00 lable += f":{succ}/{count}={ratio}%" return lable, ratio