libs.commands.graph.summary

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