libs.commands.graph.summary

  1"""
  2libs/commands/graph/summary.py
  3"""
  4
  5import logging
  6import os
  7
  8import matplotlib.font_manager as fm
  9import matplotlib.pyplot as plt
 10import pandas as pd
 11
 12import libs.global_value as g
 13from cls.timekit import ExtendedDatetime as ExtDt
 14from cls.types import GameInfoDict
 15from libs.data import aggregate, loader
 16from libs.functions import configuration, message
 17from libs.utils import formatter
 18
 19
 20def point_plot() -> tuple[int, str]:
 21    """ポイント推移グラフを生成する
 22
 23    Returns:
 24        tuple[int,str]:
 25        - int: グラフにプロットしたゲーム数
 26        - str: 検索結果が0件のときのメッセージ or グラフ画像保存パス
 27    """
 28
 29    plt.close()
 30    # 初期化
 31    title_text = None
 32    xlabel_text = None
 33
 34    # データ収集
 35    game_info: GameInfoDict = aggregate.game_info()
 36    target_data, df = _data_collection()
 37
 38    if target_data.empty:  # 描写対象が0人の場合は終了
 39        return (len(target_data), message.reply(message="no_hits"))
 40
 41    # グラフタイトル/X軸ラベル
 42    pivot_index = "playtime"
 43    if g.params.get("target_count"):
 44        title_text = f"ポイント推移 (直近 {g.params["target_count"]} ゲーム)"
 45        xlabel_text = f"集計日（総ゲーム数：{game_info["game_count"]}）"
 46    else:
 47        match g.params["collection"]:
 48            case "daily":
 49                xlabel_text = f"集計日（総ゲーム数：{game_info["game_count"]}）"
 50                title_text = message.item_date_range("ymd_o", "通算ポイント", "ポイント推移")
 51            case "monthly":
 52                xlabel_text = f"集計月（総ゲーム数：{game_info["game_count"]}）"
 53                title_text = message.item_date_range("jym_o", "通算ポイント", "ポイント推移")
 54            case "yearly":
 55                xlabel_text = f"集計年（総ゲーム数：{game_info["game_count"]}）"
 56                title_text = message.item_date_range("jy_o", "通算ポイント", "ポイント推移")
 57            case "all":
 58                xlabel_text = f"総ゲーム数：{game_info["game_count"]}"
 59                title_text = message.item_date_range("ymdhm", "通算ポイント", "ポイント推移")
 60            case _:
 61                if g.params.get("search_word"):
 62                    pivot_index = "comment"
 63                    xlabel_text = f"（総ゲーム数：{game_info["game_count"]} ）"
 64                    if game_info["first_comment"] == game_info["last_comment"]:
 65                        title_text = f"通算ポイント ({game_info["first_comment"]})"
 66                    else:
 67                        title_text = f"ポイント推移 ({game_info["first_comment"]} - {game_info["last_comment"]})"
 68                else:
 69                    xlabel_text = f"ゲーム終了日時（{game_info["game_count"]} ゲーム）"
 70                    title_text = message.item_date_range("ymdhm", "通算ポイント", "ポイント推移")
 71                    if ExtDt(g.params["starttime"]).format("ymd") == ExtDt(g.params["onday"]).format("ymd") and game_info["game_count"] == 1:
 72                        title_text = f"獲得ポイント ({ExtDt(g.params["starttime"]).format("ymd")})"
 73
 74    # 集計
 75    if g.params.get("individual"):  # 個人集計
 76        legend = "name"
 77    else:  # チーム集計
 78        legend = "team"
 79
 80    pivot = pd.pivot_table(
 81        df, index=pivot_index, columns=legend, values="point_sum"
 82    ).ffill()
 83    pivot = pivot.reindex(  # 並び替え
 84        target_data[legend].to_list(), axis="columns"
 85    )
 86
 87    # グラフ生成
 88    args = {
 89        "kind": "point",
 90        "title_text": title_text,
 91        "total_game_count": game_info["game_count"],
 92        "target_data": target_data,
 93        "legend": legend,
 94        "xlabel_text": xlabel_text,
 95        "ylabel_text": "通算ポイント",
 96        "horizontal": True,
 97    }
 98    save_file = _graph_generation(pivot, **args)
 99    plt.savefig(save_file, bbox_inches="tight")
100
101    return (game_info["game_count"], save_file)
102
103
104def rank_plot() -> tuple[int, str]:
105    """順位変動グラフを生成する
106
107    Returns:
108        tuple[int,str]:
109        - int: グラフにプロットしたゲーム数
110        - str: 検索結果が0件のときのメッセージ or グラフ画像保存パス
111    """
112
113    plt.close()
114    # 初期化
115    title_text = None
116    xlabel_text = None
117
118    # データ収集
119    game_info: GameInfoDict = aggregate.game_info()
120    target_data, df = _data_collection()
121
122    if target_data.empty:  # 描写対象が0人の場合は終了
123        return (len(target_data), message.reply(message="no_hits"))
124
125    # グラフタイトル/X軸ラベル
126    pivot_index = "playtime"
127    if g.params.get("target_count"):
128        title_text = f"順位変動 (直近 {g.params["target_count"]} ゲーム)"
129        xlabel_text = f"集計日（総ゲーム数：{game_info["game_count"]}）"
130    else:
131        match g.params["collection"]:
132            case "daily":
133                xlabel_text = f"集計日（総ゲーム数：{game_info["game_count"]}）"
134                title_text = message.item_date_range("ymd_o", "順位", "順位変動")
135            case "monthly":
136                xlabel_text = f"集計月（総ゲーム数：{game_info["game_count"]}）"
137                title_text = message.item_date_range("jym", "順位", "順位変動")
138            case "yearly":
139                xlabel_text = f"集計年（総ゲーム数：{game_info["game_count"]}）"
140                title_text = message.item_date_range("jy", "順位", "順位変動")
141            case "all":
142                xlabel_text = f"総ゲーム数：{game_info["game_count"]}"
143                title_text = message.item_date_range("ymdhm", "順位", "順位変動")
144            case _:
145                if g.params.get("search_word"):
146                    pivot_index = "comment"
147                    xlabel_text = f"（総ゲーム数：{game_info["game_count"]} ）"
148                    if game_info["first_comment"] == game_info["last_comment"]:
149                        title_text = f"順位 ({game_info["first_comment"]})"
150                    else:
151                        title_text = f"順位変動 ({game_info["first_comment"]} - {game_info["last_comment"]})"
152                else:
153                    xlabel_text = f"ゲーム終了日時（{game_info["game_count"]} ゲーム）"
154                    title_text = message.item_date_range("ymdhm", "順位", "順位変動")
155                    if ExtDt(g.params["starttime"]).format("ymd") == ExtDt(g.params["onday"]).format("ymd") and game_info["game_count"] == 1:
156                        title_text = f"順位 ({ExtDt(g.params["starttime"]).format("ymd")})"
157
158    # 集計
159    if g.params.get("individual"):  # 個人集計
160        legend = "name"
161    else:  # チーム集計
162        legend = "team"
163
164    pivot = pd.pivot_table(
165        df, index=pivot_index, columns=legend, values="point_sum"
166    ).ffill()
167    pivot = pivot.reindex(  # 並び替え
168        target_data[legend].to_list(), axis="columns"
169    )
170    pivot = pivot.rank(method="dense", ascending=False, axis=1)
171
172    # グラフ生成
173    args = {
174        "kind": "rank",
175        "title_text": title_text,
176        "total_game_count": game_info["game_count"],
177        "target_data": target_data,
178        "legend": legend,
179        "xlabel_text": xlabel_text,
180        "ylabel_text": "順位 (通算ポイント順)",
181        "horizontal": False,
182    }
183    save_file = _graph_generation(pivot, **args)
184    plt.savefig(save_file, bbox_inches="tight")
185
186    return (game_info["game_count"], save_file)
187
188
189def _data_collection() -> tuple[pd.DataFrame, pd.DataFrame]:
190    """データ収集
191
192    Returns:
193        tuple[pd.DataFrame, pd.DataFrame]:
194        - pd.DataFrame: 収集したデータのサマリ
195        - pd.DataFrame: 集計範囲のデータ
196    """
197
198    # データ収集
199    g.params.update(fourfold=True)  # 直近Nは4倍する(縦持ちなので4人分)
200
201    target_data = pd.DataFrame()
202    if g.params.get("individual"):  # 個人集計
203        df = loader.read_data("summary/gamedata.sql")
204        if df.empty:
205            return (target_data, df)
206
207        target_data["name"] = df.groupby("name", as_index=False).last()["name"]
208        target_data["last_point"] = df.groupby("name", as_index=False).last()["point_sum"]
209        target_data["game_count"] = df.groupby("name", as_index=False).max(numeric_only=True)["count"]
210
211        # 足切り
212        target_list = list(
213            target_data.query("game_count >= @g.params['stipulated']")["name"]
214        )
215        _ = target_list  # ignore PEP8 F841
216        target_data = target_data.query("name == @target_list").copy()
217        df = df.query("name == @target_list").copy()
218    else:  # チーム集計
219        df = loader.read_data("summary/gamedata.sql")
220        if df.empty:
221            return (target_data, df)
222
223        target_data["last_point"] = df.groupby("team").last()["point_sum"]
224        target_data["game_count"] = (
225            df.groupby("team").max(numeric_only=True)["count"]
226        )
227        target_data["team"] = target_data.index
228        target_data = target_data.sort_values("last_point", ascending=False)
229
230    # 順位付け
231    target_data["position"] = (
232        target_data["last_point"].rank(ascending=False).astype(int)
233    )
234
235    if g.params.get("anonymous"):
236        col = "team"
237        if g.params.get("individual"):
238            col = "name"
239        mapping_dict = formatter.anonymous_mapping(df[col].unique().tolist())
240        df[col] = df[col].replace(mapping_dict)
241        target_data[col] = target_data[col].replace(mapping_dict)
242
243    return (target_data.sort_values("position"), df)
244
245
246def _graph_generation(df: pd.DataFrame, **kwargs) -> str:
247    """グラフ生成共通処理
248
249    Args:
250        df (pd.DataFrame): _description_
251        kwargs (dict): グラフ生成パラメータ
252
253    Returns:
254        str: 生成したグラフの保存パス
255    """
256
257    save_file = os.path.join(
258        g.cfg.setting.work_dir,
259        f"{g.params["filename"]}.png" if g.params.get("filename") else "graph.png",
260    )
261
262    configuration.graph_setup(plt, fm)
263
264    if (all(df.count() == 1) or g.params["collection"] == "all") and kwargs["horizontal"]:
265        kwargs["kind"] = "barh"
266        lab: list = []
267        color: list = []
268        for _, v in kwargs["target_data"].iterrows():
269            lab.append("{:2d}位：{} ({}pt / {}G)".format(  # pylint: disable=consider-using-f-string
270                v["position"],
271                v[kwargs["legend"]],
272                "{:+.1f}".format(v["last_point"]).replace("-", "▲"),  # pylint: disable=consider-using-f-string
273                v["game_count"],
274            ))
275            if v["last_point"] > 0:
276                color.append("deepskyblue")
277            else:
278                color.append("orangered")
279
280        tmpdf = pd.DataFrame(
281            {"point": kwargs["target_data"]["last_point"].to_list()[::-1]},
282            index=lab[::-1],
283        )
284
285        tmpdf.plot.barh(
286            figsize=(8, 2 + tmpdf.count().iloc[0] / 5),
287            y="point",
288            xlabel=f"総ゲーム数：{kwargs["total_game_count"]}",
289            color=color[::-1],
290        )
291
292        plt.legend().remove()
293        plt.gca().yaxis.tick_right()
294
295        # X軸修正
296        xlocs, xlabs = plt.xticks()
297        new_xlabs = [xlab.get_text().replace("−", "▲") for xlab in xlabs]
298        plt.xticks(list(xlocs[1:-1]), new_xlabs[1:-1])
299
300        logging.info("plot data:\n%s", tmpdf)
301    else:
302        df.plot(
303            figsize=(8, 6),
304            xlabel=kwargs["xlabel_text"],
305            ylabel=kwargs["ylabel_text"],
306            marker="." if len(df) < 50 else None,
307        )
308
309        # 凡例
310        legend_text = []
311        for _, v in kwargs["target_data"].iterrows():
312            legend_text.append("{:2d}位：{} ({}pt / {}G)".format(  # pylint: disable=consider-using-f-string
313                v["position"], v[kwargs["legend"]],
314                "{:+.1f}".format(v["last_point"]).replace("-", "▲"),  # pylint: disable=consider-using-f-string
315                v["game_count"],
316            ))
317
318        plt.legend(
319            legend_text,
320            bbox_to_anchor=(1, 1),
321            loc="upper left",
322            borderaxespad=0.5,
323            ncol=int(len(kwargs["target_data"]) / 25 + 1),
324        )
325
326        # X軸修正
327        plt.xticks(
328            list(range(len(df)))[::int(len(df) / 25) + 1],
329            list(df.index)[::int(len(df) / 25) + 1],
330            rotation=45,
331            ha="right",
332        )
333
334        # Y軸修正
335        ylocs, ylabs = plt.yticks()
336        new_ylabs = [ylab.get_text().replace("−", "▲") for ylab in ylabs]
337        plt.yticks(list(ylocs[1:-1]), new_ylabs[1:-1])
338
339        logging.info("plot data:\n%s", df)
340
341    #
342    match kwargs["kind"]:
343        case "barh":
344            plt.axvline(x=0, linewidth=0.5, ls="dashed", color="grey")
345        case "point":
346            plt.axhline(y=0, linewidth=0.5, ls="dashed", color="grey")
347        case "rank":
348            lab = list(range(len(kwargs["target_data"]) + 1))
349            if len(lab) > 10:
350                plt.yticks(lab[1::2], lab[1::2])
351            else:
352                plt.yticks(lab[1:], lab[1:])
353            plt.gca().invert_yaxis()
354
355    plt.title(
356        kwargs["title_text"],
357        fontsize=16,
358    )
359
360    return save_file