xyzshop/plot/refill/ChannelPainter.py

from .DataStream import EChPosmap as pos_map, day_stamp, span_days, time_border, calc_interval
from .ChannelReader import ChannelReader
from .PainterBase import PainterBase
from collections import defaultdict
from matplotlib.figure import Figure
from matplotlib import ticker
from io import BytesIO
import numpy as np
from .algorithm import calc_chratios
import time as time

import logging
logger = logging.getLogger('ChannelPainter')

_all_channels = set()
def _add_channel(channel):
    if channel not in _all_channels:
        _all_channels.add(channel)

def get_channels():
    logger.debug(_all_channels)
    return list(_all_channels)

class ChPathFilter(object):
    def __init__(self,chnames: set = None, card_types: set = None,spec: int = None,show_all: bool=False):
        self._chnames = chnames
        self._card_types = card_types
        self._spec = spec
        self._show_all = show_all
    pass

def ratio_pathes(reader: ChannelReader, 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 name, tup in tuple_pathes.items():
        _add_channel(name)
        mch_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, name, _card_type, _spec)
                if data is not None:
                    column_pos = i * 86400
                    view = mch_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 name, _card_type, _spec, detail_datas
        if all_datas is not None:
            all_datas += mch_datas
        yield name, None, None, mch_datas

    if show_detail == False:
        yield 'all', None, None, all_datas


class ChannelPainter(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()
            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, 5)
        else:
            ratio = 0.00
        lable += f":{succ}/{count}={ratio}%"

        return lable, ratio