libs.commands.graph.summary

  1"""
  2libs/commands/graph/summary.py
  3"""
  4
  5import logging
  6from typing import TYPE_CHECKING, Literal, Optional, TypedDict, cast
  7
  8import matplotlib.pyplot as plt
  9import pandas as pd
 10import plotly.express as px  # type: ignore
 11
 12import libs.global_value as g
 13from libs.data import loader
 14from libs.datamodels import GameInfo
 15from libs.functions import compose, message
 16from libs.types import StyleOptions
 17from libs.utils import formatter, graphutil, textutil
 18
 19if TYPE_CHECKING:
 20    from pathlib import Path
 21
 22    from integrations.protocols import MessageParserProtocol
 23
 24
 25class GraphParams(TypedDict, total=False):
 26    """グラフ生成パラメータ"""
 27
 28    graph_type: Literal["point", "rank", "point_hbar"]
 29    title_text: str
 30    xlabel_text: Optional[str]
 31    ylabel_text: Optional[str]
 32    total_game_count: int
 33    target_data: pd.DataFrame
 34    pivot: pd.DataFrame
 35    horizontal: bool  # 横棒切替許可フラグ
 36    save_file: str
 37
 38
 39def point_plot(m: "MessageParserProtocol"):
 40    """ポイント推移グラフを生成する
 41
 42    Args:
 43        m (MessageParserProtocol): メッセージデータ
 44    """
 45
 46    # データ収集
 47    game_info = GameInfo()
 48    target_data, df = _data_collection()
 49
 50    if target_data.empty:  # 描写対象が0人の場合は終了
 51        m.post.headline = {"0": message.random_reply(m, "no_hits")}
 52        m.status.result = False
 53        return
 54
 55    # 集計
 56    if g.params.get("search_word"):
 57        pivot_index = "comment"
 58    else:
 59        pivot_index = "playtime"
 60
 61    pivot = pd.pivot_table(
 62        df, index=pivot_index, columns="name", values="point_sum"
 63    ).ffill()
 64    pivot = pivot.reindex(  # 並び替え
 65        target_data["name"].to_list(), axis="columns"
 66    )
 67
 68    # グラフ生成
 69    graphutil.setup()
 70    graph_params = GraphParams(
 71        graph_type="point",
 72        total_game_count=game_info.count,
 73        target_data=target_data,
 74        pivot=pivot,
 75        horizontal=True,
 76    )
 77
 78    match g.adapter.conf.plotting_backend:
 79        case "plotly":
 80            graph_params.update({"save_file": "graph.html"})
 81            save_file = _graph_generation_plotly(graph_params)
 82        case _:
 83            graph_params.update({"save_file": "graph.png"})
 84            save_file = _graph_generation(graph_params)
 85
 86    file_title = graph_params.get("title_text", "").split()[0]
 87    m.post.headline = {f"{file_title}グラフ": message.header(game_info, m)}
 88    m.set_data(
 89        file_title, save_file,
 90        StyleOptions(use_comment=True, header_hidden=True, key_title=False),
 91    )
 92
 93
 94def rank_plot(m: "MessageParserProtocol"):
 95    """順位変動グラフを生成する
 96
 97    Args:
 98        m (MessageParserProtocol): メッセージデータ
 99    """
100
101    # データ収集
102    game_info = GameInfo()
103    target_data, df = _data_collection()
104
105    if target_data.empty:  # 描写対象が0人の場合は終了
106        m.post.headline = {"0": message.random_reply(m, "no_hits")}
107        m.status.result = False
108        return
109
110    if g.params.get("search_word"):
111        pivot_index = "comment"
112    else:
113        pivot_index = "playtime"
114
115    # 集計
116    pivot = pd.pivot_table(
117        df, index=pivot_index, columns="name", values="point_sum"
118    ).ffill()
119    pivot = pivot.reindex(  # 並び替え
120        target_data["name"].to_list(), axis="columns"
121    )
122    pivot = pivot.rank(method="dense", ascending=False, axis=1)
123
124    # グラフ生成
125    graphutil.setup()
126    graph_params = GraphParams(
127        graph_type="rank",
128        total_game_count=game_info.count,
129        target_data=target_data,
130        pivot=pivot,
131        horizontal=False,
132    )
133
134    match g.adapter.conf.plotting_backend:
135        case "plotly":
136            graph_params.update({"save_file": "graph.html"})
137            save_file = _graph_generation_plotly(graph_params)
138        case _:
139            graph_params.update({"save_file": "graph.png"})
140            save_file = _graph_generation(graph_params)
141
142    file_title = graph_params.get("title_text", "").split()[0]
143    m.post.headline = {f"{file_title}グラフ": message.header(game_info, m)}
144    m.set_data(
145        file_title, save_file,
146        StyleOptions(use_comment=True, header_hidden=True, key_title=False),
147    )
148
149
150def _data_collection() -> tuple[pd.DataFrame, pd.DataFrame]:
151    """データ収集
152
153    Returns:
154        tuple[pd.DataFrame, pd.DataFrame]:
155        - pd.DataFrame: 収集したデータのサマリ
156        - pd.DataFrame: 集計範囲のデータ
157    """
158
159    # データ収集
160    g.params.update(fourfold=True)  # 直近Nは4倍する(縦持ちなので4人分)
161
162    target_data = pd.DataFrame()
163    if g.params.get("individual"):  # 個人集計
164        df = loader.read_data("SUMMARY_GAMEDATA")
165        if df.empty:
166            return (target_data, df)
167
168        target_data["name"] = df.groupby("name", as_index=False).last()["name"]
169        target_data["last_point"] = df.groupby("name", as_index=False).last()["point_sum"]
170        target_data["game_count"] = df.groupby("name", as_index=False).max(numeric_only=True)["count"]
171
172        # 足切り
173        target_list = list(
174            target_data.query("game_count >= @g.params['stipulated']")["name"]
175        )
176        _ = target_list  # ignore PEP8 F841
177        target_data = target_data.query("name == @target_list").copy()
178        df = df.query("name == @target_list").copy()
179    else:  # チーム集計
180        df = loader.read_data("SUMMARY_GAMEDATA")
181        if df.empty:
182            return (target_data, df)
183
184        target_data["last_point"] = df.groupby("name").last()["point_sum"]
185        target_data["game_count"] = (
186            df.groupby("name").max(numeric_only=True)["count"]
187        )
188        target_data["name"] = target_data.index
189        target_data = target_data.sort_values("last_point", ascending=False)
190
191    # 順位付け
192    target_data["position"] = target_data["last_point"].rank(ascending=False).astype(int)
193
194    if g.params.get("anonymous"):
195        mapping_dict = formatter.anonymous_mapping(df["name"].unique().tolist())
196        df["name"] = df["name"].replace(mapping_dict)
197        target_data["name"] = target_data["name"].replace(mapping_dict)
198
199    # 凡例用文字列生成
200    target_data["legend"] = target_data.apply(
201        lambda x: f"{x["position"]}位： {x["name"]} ({x["last_point"]:+.1f}pt / {x["game_count"]:.0f}G)".replace("-", "▲"), axis=1
202    )
203
204    return (target_data.sort_values("position"), df)
205
206
207def _graph_generation(graph_params: GraphParams) -> "Path":
208    """グラフ生成共通処理(matplotlib用)
209
210    Args:
211        args (GraphParams): グラフ生成パラメータ
212
213    Returns:
214        Path: 保存先ファイル名
215    """
216
217    save_file = textutil.save_file_path(graph_params["save_file"])
218    target_data = graph_params["target_data"]
219    df = graph_params["pivot"]
220
221    if (all(df.count() == 1) or g.params["collection"] == "all") and graph_params["horizontal"]:
222        graph_params["graph_type"] = "point_hbar"
223        color: list = []
224        for _, v in target_data.iterrows():
225            if v["last_point"] > 0:
226                color.append("deepskyblue")
227            else:
228                color.append("orangered")
229
230        _graph_title(graph_params)
231        tmpdf = pd.DataFrame(
232            {"point": target_data["last_point"].to_list()[::-1]},
233            index=target_data["legend"].to_list()[::-1],
234        )
235
236        tmpdf.plot.barh(
237            figsize=(8, 2 + tmpdf.count().iloc[0] / 5),
238            y="point",
239            xlabel=graph_params["xlabel_text"],
240            color=color[::-1],
241        ).get_figure()
242
243        plt.legend().remove()
244        plt.gca().yaxis.tick_right()
245
246        # X軸修正
247        xlocs, xlabs = plt.xticks()
248        new_xlabs = [xlab.get_text().replace("−", "▲") for xlab in xlabs]
249        plt.xticks(list(xlocs[1:-1]), new_xlabs[1:-1])
250
251        logging.debug("plot data:\n%s", tmpdf)
252    else:
253        _graph_title(graph_params)
254        df.plot(
255            figsize=(8, 6),
256            xlabel=str(graph_params["xlabel_text"]),
257            ylabel=str(graph_params["ylabel_text"]),
258            marker="." if len(df) < 20 else None,
259            linewidth=2 if len(df) < 40 else 1,
260        ).get_figure()
261
262        # 凡例
263        plt.legend(
264            target_data["legend"].to_list(),
265            bbox_to_anchor=(1, 1),
266            loc="upper left",
267            borderaxespad=0.5,
268            ncol=int(len(target_data) / 25 + 1),
269        )
270
271        # X軸修正
272        plt.xticks(**graphutil.xticks_parameter(df.index.to_list()))
273
274        # Y軸修正
275        ylocs, ylabs = plt.yticks()
276        new_ylabs = [ylab.get_text().replace("−", "▲") for ylab in ylabs]
277        plt.yticks(list(ylocs[1:-1]), new_ylabs[1:-1])
278        logging.debug("plot data:\n%s", df)
279
280    # メモリ調整
281    match graph_params["graph_type"]:
282        case "point_hbar":
283            plt.axvline(x=0, linewidth=0.5, ls="dashed", color="grey")
284        case "point":
285            plt.axhline(y=0, linewidth=0.5, ls="dashed", color="grey")
286        case "rank":
287            lab = range(len(target_data) + 1)
288            if len(lab) > 10:
289                plt.yticks(list(map(int, lab))[1::2], list(map(str, lab))[1::2])
290            else:
291                plt.yticks(list(map(int, lab))[1:], list(map(str, lab))[1:])
292            plt.gca().invert_yaxis()
293
294    plt.title(
295        graph_params["title_text"],
296        fontsize=16,
297    )
298
299    plt.savefig(save_file, bbox_inches="tight")
300    return save_file
301
302
303def _graph_generation_plotly(graph_params: GraphParams) -> "Path":
304    """グラフ生成共通処理(plotly用)
305
306    Args:
307        args (GraphParams): グラフ生成パラメータ
308
309    Returns:
310        Path: 保存先ファイル名
311    """
312
313    save_file = textutil.save_file_path(graph_params["save_file"])
314    target_data = cast(pd.DataFrame, graph_params["target_data"])
315    df = graph_params["pivot"]
316
317    if (all(df.count() == 1) or g.params["collection"] == "all") and graph_params["horizontal"]:
318        graph_params["graph_type"] = "point_hbar"
319        df_t = df.T
320        df_t.columns = ["point"]
321        df_t["rank"] = df_t["point"].rank(ascending=False, method="dense").astype("int")
322        df_t["positive"] = df_t["point"] > 0
323        fig = px.bar(
324            df_t,
325            orientation="h",
326            color="positive",
327            color_discrete_map={True: "blue", False: "red"},
328            x=df_t["point"],
329            y=target_data["legend"],
330        )
331    else:
332        df.columns = target_data["legend"].to_list()  # 凡例用ラベル生成
333        fig = px.line(df, markers=True)
334
335    # グラフレイアウト調整
336    _graph_title(graph_params)
337    fig.update_layout(
338        width=1280,
339        height=800,
340        title={
341            "text": graph_params["title_text"],
342            "font": {"size": 30},
343            "xref": "paper",
344            "xanchor": "center",
345            "x": 0.5,
346        },
347        xaxis_title={
348            "text": graph_params["xlabel_text"],
349            "font": {"size": 18},
350        },
351        yaxis_title={
352            "text": graph_params["ylabel_text"],
353            "font": {"size": 18},
354        },
355        legend_title=None,
356    )
357
358    # 軸/目盛調整
359    match graph_params["graph_type"]:
360        case "point_hbar":
361            fig.update_traces(hovertemplate="%{y}<extra></extra>")
362            fig.update_layout(showlegend=False)
363            fig.update_yaxes(
364                autorange="reversed",
365                side="right",
366                title=None,
367            )
368        case "point":
369            # マーカー
370            if all(df.count() > 20):
371                fig.update_traces(mode="lines")
372            # ライン
373            if len(fig.data) > 40:
374                fig.update_traces(mode="lines", line={"width": 1})
375        case "rank":
376            # Y軸目盛
377            lab = list(range(len(target_data) + 1))
378            fig.update_yaxes(
379                autorange="reversed",
380                zeroline=False,
381                tickvals=lab[1:] if len(lab) < 10 else lab[1::2],
382            )
383            # マーカー
384            if all(df.count() == 1):
385                fig.update_traces(marker={"size": 10})
386            elif all(df.count() > 20):
387                fig.update_traces(mode="lines")
388            # ライン
389            if len(fig.data) > 40:
390                fig.update_traces(mode="lines", line={"width": 1})
391
392    fig.write_html(save_file, full_html=False)
393    return save_file
394
395
396def _graph_title(graph_params: GraphParams):
397    """グラフタイトル生成
398
399    Args:
400        args (GraphParams): グラフ生成パラメータ
401    """
402
403    if g.params.get("target_count"):
404        kind = "ymd_o"
405        graph_params.update({"xlabel_text": f"集計日（{graph_params["total_game_count"]} ゲーム）"})
406        match graph_params.get("graph_type"):
407            case "point":
408                graph_params.update({"title_text": f"ポイント推移 (直近 {g.params["target_count"]} ゲーム)"})
409            case "rank":
410                graph_params.update({"title_text": f"順位変動 (直近 {g.params["target_count"]} ゲーム)"})
411            case "point_hbar":
412                graph_params.update({"title_text": f"通算ポイント (直近 {g.params["target_count"]} ゲーム)"})
413    else:
414        match g.params.get("collection"):
415            case "daily":
416                kind = "ymd_o"
417                graph_params.update({"xlabel_text": f"集計日（{graph_params["total_game_count"]} ゲーム）"})
418            case "monthly":
419                kind = "jym_o"
420                graph_params.update({"xlabel_text": f"集計月（{graph_params["total_game_count"]} ゲーム）"})
421            case "yearly":
422                kind = "jy_o"
423                graph_params.update({"xlabel_text": f"集計年（{graph_params["total_game_count"]} ゲーム）"})
424            case "all":
425                kind = "ymdhm"
426                graph_params.update({"xlabel_text": f"ゲーム数：{graph_params["total_game_count"]} ゲーム"})
427            case _:
428                kind = "ymdhm"
429                if g.params.get("search_word"):
430                    graph_params.update({"xlabel_text": f"ゲーム数：{graph_params["total_game_count"]} ゲーム"})
431                else:
432                    graph_params.update({"xlabel_text": f"ゲーム終了日時（{graph_params["total_game_count"]} ゲーム）"})
433
434    match graph_params.get("graph_type", "point"):
435        case "point":
436            graph_params.update({
437                "ylabel_text": "通算ポイント",
438                "title_text": compose.text_item.date_range(kind, "通算ポイント", "ポイント推移"),
439            })
440        case "rank":
441            graph_params.update({
442                "ylabel_text": "順位 (通算ポイント順)",
443                "title_text": compose.text_item.date_range(kind, "順位", "順位変動"),
444            })
445        case "point_hbar":
446            graph_params.update({
447                "ylabel_text": None,
448                "title_text": compose.text_item.date_range(kind, "通算ポイント", "通算ポイント"),
449            })
450            if graph_params["total_game_count"] == 1:
451                graph_params.update({"xlabel_text": "ポイント"})
452            else:
453                graph_params.update({"xlabel_text": f"ポイント（ゲーム数：{graph_params["total_game_count"]} ゲーム）"})