libs.functions.tools.gen_test_data

  1"""
  2libs/functions/tools/gen_test_data.py
  3"""
  4
  5import itertools
  6import logging
  7import random
  8from contextlib import closing
  9from datetime import datetime
 10from typing import Any, cast
 11
 12from tqdm import tqdm
 13
 14import libs.global_value as g
 15from libs.domain.score import GameResult
 16from libs.functions import lookup
 17from libs.functions.tools import score_simulator
 18from libs.utils import dbutil
 19from libs.utils.timekit import ExtendedDatetime as ExtDt
 20
 21
 22def main(season_times: int = 1) -> None:
 23    """
 24    テスト用ゲーム結果生成処理
 25
 26    Args:
 27        season_times (int, optional): 総当たり回数. Defaults to 1.
 28
 29    """
 30    g.cfg.initialization()
 31    lookup.read_memberslist()
 32
 33    # 対戦組み合わせ作成
 34    teams: list[str] = g.cfg.team.lists
 35    position: list[str] = ["先鋒", "次鋒", "中堅", "副将", "大将"]
 36    teams_data: dict[str, list[str]] = {x["team"]: x["members"] for x in g.cfg.team.info}
 37    matchup: list[tuple[str, str, str, str]] = list(itertools.combinations(teams, 4))
 38    teams_count: dict[str, int] = {x: 0 for x in teams}
 39    total_count: int = 0
 40
 41    now = datetime.now().timestamp() - ((len(matchup) + 7) * 86400 * season_times)
 42    dt = now
 43
 44    with closing(dbutil.connection(g.cfg.setting.database_file)) as cur:
 45        cur.execute("delete from result;")
 46        for season in range(1, season_times + 1):
 47            random.shuffle(matchup)
 48            for count, game in tqdm(enumerate(matchup), total=len(matchup), desc=f"season({season}/{season_times})"):
 49                # 対戦メンバーの決定
 50                vs_member = {}
 51                for team_name in game:
 52                    teams_count[team_name] += 1
 53                    team_member = teams_data[team_name]
 54                    random.shuffle(team_member)
 55                    vs_member[team_name] = team_member
 56
 57                # 試合結果
 58                total_count += 1
 59                team_order = list(game)
 60                random.shuffle(team_order)
 61                for idx in range(5):
 62                    # 席順シャッフル
 63                    member = [
 64                        vs_member[team_order[0]][idx],
 65                        vs_member[team_order[1]][idx],
 66                        vs_member[team_order[2]][idx],
 67                        vs_member[team_order[3]][idx],
 68                    ]
 69                    random.shuffle(member)
 70
 71                    # スコアデータ生成
 72                    dt = now + total_count * 86400 + idx * 3600 + random.random()
 73                    vs_score = score_simulator.simulate_game()
 74                    result = GameResult(
 75                        ts=str(dt),
 76                        p1_name=member[0],
 77                        p1_str=str(int(vs_score[0] / 100)),
 78                        p2_name=member[1],
 79                        p2_str=str(int(vs_score[1] / 100)),
 80                        p3_name=member[2],
 81                        p3_str=str(int(vs_score[2] / 100)),
 82                        p4_name=member[3],
 83                        p4_str=str(int(vs_score[3] / 100)),
 84                        comment=f"第{season:02d}期{count + 1:04d}試合_{position[idx]}戦",
 85                        rule_version=g.cfg.rule.get_version(4)[0],
 86                        source="score_simulator",
 87                    )
 88
 89                    # データ投入
 90                    param = {
 91                        "playtime": ExtDt(dt).format(ExtDt.FMT.SQL),
 92                    }
 93                    param.update(cast(dict[str, Any], result.to_dict()))
 94                    cur.execute(dbutil.query("RESULT_INSERT"), param)
 95
 96                    output = f"{position[idx]}: "
 97                    output += result.to_text("detail")
 98                    logging.debug(output)
 99
100        cur.commit()
101
102    ret = dbutil.execute("select team, round(sum(point), 1) as point from individual_results group by team order by point desc;")
103    if ret:
104        logging.info([f"{x.get('team')}: {x.get('point')}pt" for x in ret])
105    else:
106        logging.info("No test data was generated.")