xeals/holy-heck-i-really-like-stats
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor parser into class; add newtyping

Browse Source
This commit is contained in:
xeals 2023-04-04 22:21:33 +10:00
parent 8a348eca84
commit 37a2df3039
Signed by: xeals
GPG Key ID: A498C7AF27EC6B5C
2 changed files with 315 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
flake.nix
View File

@ -13,10 +13,17 @@
buildInputs = buildInputs =
let let
python = pkgs.python3.withPackages (ps: [ python = pkgs.python3.withPackages (ps: [
ps.mypy
ps.requests ps.requests
ps.types-requests
]); ]);
in in
[ python pkgs.sqlite ]; [
python
pkgs.sqlite
python.pkgs.python-lsp-server
python.pkgs.pylsp-mypy
];
}; };
}); });
} }

324
main.py
View File

@ -14,22 +14,22 @@ import sys
import typing as t import typing as t
logging.TRACE = 5 logging.TRACE = 5 # type: ignore
logging.addLevelName(logging.TRACE, "TRACE") logging.addLevelName(logging.TRACE, "TRACE") # type: ignore
logging.Logger.trace = partialmethod(logging.Logger.log, logging.TRACE) logging.Logger.trace = partialmethod(logging.Logger.log, logging.TRACE) # type: ignore
logging.trace = partial(logging.log, logging.TRACE) logging.trace = partial(logging.log, logging.TRACE) # type: ignore
class LogFormatter(logging.Formatter): class LogFormatter(logging.Formatter):
format = "%(name)s [%(levelname)s] %(message)s" _format = "%(name)s [%(levelname)s] %(message)s"
FORMATS = { FORMATS = {
logging.TRACE: f"\x1b[30;20m{format}\x1b[0m", logging.TRACE: f"\x1b[30;20m{_format}\x1b[0m", # type: ignore
logging.DEBUG: f"\x1b[38;20m{format}\x1b[0m", logging.DEBUG: f"\x1b[38;20m{_format}\x1b[0m",
logging.INFO: f"\x1b[34;20m{format}\x1b[0m", logging.INFO: f"\x1b[34;20m{_format}\x1b[0m",
logging.WARNING: f"\x1b[33;20m{format}\x1b[0m", logging.WARNING: f"\x1b[33;20m{_format}\x1b[0m",
logging.ERROR: f"\x1b[31;20m{format}\x1b[0m", logging.ERROR: f"\x1b[31;20m{_format}\x1b[0m",
logging.CRITICAL: f"\x1b[31;1m{format}\x1b[0m", logging.CRITICAL: f"\x1b[31;1m{_format}\x1b[0m",
} }
def format(self, record): def format(self, record):
@ -45,29 +45,6 @@ _ch.setFormatter(LogFormatter())
LOG.addHandler(_ch) LOG.addHandler(_ch)
TEAMS = {}
_logged_teams = []
def team(player: str) -> str:
"""Maps a username to a defined team."""
if player in TEAMS:
return TEAMS[player]
else:
if not player in _logged_teams and player:
LOG.warning(f"missing team mapping for {player}")
_logged_teams.append(player)
return player
class safelist(list):
def get(self, index, default=None):
try:
return self.__getitem__(index)
except IndexError:
return default
def _init_db(conn: sqlite3.Connection): def _init_db(conn: sqlite3.Connection):
def namedtuple_factory(cursor, row): def namedtuple_factory(cursor, row):
fields = [column[0] for column in cursor.description] fields = [column[0] for column in cursor.description]
@ -112,12 +89,45 @@ def _init_db(conn: sqlite3.Connection):
) )
def parse_log(game: str, log: str, into: sqlite3.Connection): # Either the value "p1" or "p2"
conn = into PlayerTag = t.NewType("PlayerTag", str)
# A player's name
Player = t.NewType("Player", str)
# A player prefixed with a PlayerTag
TaggedPlayer = t.NewType("TaggedPlayer", str)
# A Pokemon identified by its nickname, if any
Pokemon = t.NewType("Pokemon", str)
# A Pokemon specie
PokemonSpecie = t.NewType("PokemonSpecie", str)
# A Pokemon prefixed with a PlayerTag
TaggedPokemon = t.NewType("TaggedPokemon", str)
TEAMS: dict[Player, Player] = {}
_logged_teams: list[Player] = []
def team(player: Player) -> Player:
"""Maps a username to a defined team."""
if player in TEAMS:
return TEAMS[player]
else:
if not player in _logged_teams and player:
LOG.warning(f"missing team mapping for {player}")
_logged_teams.append(player)
return player
class LogParser:
turn = 0 turn = 0
players = {} players: dict[PlayerTag, Player] = {}
hp = {} hp: dict[TaggedPokemon, int] = {}
# ("p2a: Edward", "p1a: Meteo") # ("p2a: Edward", "p1a: Meteo")
# memorises the user of the move that causes environment setting or status, # memorises the user of the move that causes environment setting or status,
@ -130,165 +140,185 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
# ("p1a: Meteo", "brn") => "p2a: Edward" # ("p1a: Meteo", "brn") => "p2a: Edward"
last_status_set: dict[tuple[str, str], str] = {} last_status_set: dict[tuple[str, str], str] = {}
def split_pokemon(user: str) -> tuple[str, str]: def __init__(self, game: str, into: sqlite3.Connection):
"""Splits a Pokemon identifier of the form `pXa: Pokemon` into the self.game = game
player's name (as marked by the player log) and "Pokemon". self.conn: sqlite3.Connection = into
Note that all Pokemon are referred to by their nicknames, and will def split_pokemon(self, user: TaggedPokemon) -> tuple[Player, Pokemon]:
require resolving to obtain the Pokemon specie.""" """Splits a TaggedPokemon into the owning player and the Pokemon."""
[player, name] = user.split(": ") [player, pokemon] = user.split(": ")
return players[player.strip("ab")], name return self.players[PlayerTag(player.strip("ab"))], Pokemon(pokemon)
def specie_from_parts(player: str, nickname: str) -> str: @t.overload
def specie(self, pokemon: Pokemon, player: Player) -> PokemonSpecie:
"""Resolves the species of a nicknamed Pokemon.""" """Resolves the species of a nicknamed Pokemon."""
...
@t.overload
def specie(self, pokemon: TaggedPokemon) -> PokemonSpecie:
"""Resolves the species of a Pokemon given its Showdown identifier (used
in split_pokemon)."""
...
def specie(
self, pokemon: Pokemon | TaggedPokemon, player: t.Optional[Player] = None
) -> PokemonSpecie:
if not player:
[player, pokemon] = self.split_pokemon(TaggedPokemon(pokemon))
return ( return (
conn.execute( self.conn.execute(
""" """
SELECT specie SELECT specie
FROM nicknames FROM nicknames
WHERE (game, player, pokemon) = (?, ?, ?) WHERE (game, player, pokemon) = (?, ?, ?)
LIMIT 1 LIMIT 1
""", """,
(game, team(player), nickname), (self.game, team(player), pokemon),
) )
.fetchall()[0] .fetchall()[0]
.specie .specie
) )
def specie(pokemon: str) -> str: def _reset(self):
"""Resolves the species of a Pokemon given its Showdown identifier (used self.turn = 0
in split_pokemon).""" self.players.clear()
return specie_from_parts(*split_pokemon(pokemon))
def _log_appearance(self, name: TaggedPokemon, specie: str):
# Also includes gender and formes.
trimmed_specie = PokemonSpecie(specie.split(", ")[0])
player, nickname = self.split_pokemon(name)
self.conn.execute(
"""
INSERT INTO nicknames(game, player, pokemon, specie)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(self.game, team(player), nickname, trimmed_specie),
)
def parse(self, log: str):
self._reset()
for line in log.split("\n"): for line in log.split("\n"):
chunks = line.split("|")[1:] chunks = line.split("|")[1:]
if not chunks: if not chunks:
continue continue
LOG.trace(line) LOG.trace(line) # type: ignore
match chunks: match chunks:
# t.Literal, PlayerTag, Player
case ["player", id, username, *rest]: case ["player", id, username, *rest]:
players[id] = username self.players[PlayerTag(id)] = Player(username)
# t.Literal, str
case ["turn", turn]: case ["turn", turn]:
turn = int(turn) self.turn = int(turn)
# t.Literal, TaggedPokemon, str, TaggedPokemon
case ["move", user_, move, target_]:
user = TaggedPokemon(user_)
target = TaggedPokemon(target_)
case ["move", user, move, target]:
last_move = (user, target) last_move = (user, target)
player, _ = split_pokemon(user) player, _ = self.split_pokemon(user)
conn.execute( self.conn.execute(
""" """
INSERT INTO moves(game, turn, player, pokemon, move, target) INSERT INTO moves(game, turn, player, pokemon, move, target)
VALUES (?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT DO NOTHING ON CONFLICT DO NOTHING
""", """,
( (
game, self.game,
turn, self.turn,
team(player), team(player),
specie(user), self.specie(user),
move, move,
specie(target), self.specie(target),
), ),
) )
case ["drag", name, specie_, status, *rest]: # t.Literal, TaggedPokemon, str, str
hp[name] = int(status.split("/")[0]) case ["drag", name_, specie, status, *rest]:
name = TaggedPokemon(name_)
self.hp[name] = int(status.split("/")[0])
self._log_appearance(name, specie)
# t.Literal, TaggedPokemon, str
case ["replace", name, specie]:
self._log_appearance(name, specie)
# t.Literal, TaggedPokemon, str, str, t.Optional[str]
case ["switch", name, specie, status, *rest]:
self.hp[name] = int(status.split("/")[0])
# Also includes gender and formes. # Also includes gender and formes.
trimmed_specie = specie_.split(", ")[0] trimmed_specie = specie.split(", ")[0]
player, nickname = self.split_pokemon(name)
player, nickname = split_pokemon(name) self._log_appearance(name, specie)
conn.execute( self.conn.execute(
"""
INSERT INTO nicknames(game, player, pokemon, specie)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, team(player), nickname, trimmed_specie),
)
case ["replace", name, specie_]:
# Also includes gender and formes.
trimmed_specie = specie_.split(", ")[0]
player, nickname = split_pokemon(name)
conn.execute(
"""
INSERT INTO nicknames(game, player, pokemon, specie)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, team(player), nickname, trimmed_specie),
)
case ["switch", name, specie_, status, *rest]:
hp[name] = int(status.split("/")[0])
# Also includes gender and formes.
trimmed_specie = specie_.split(", ")[0]
player, nickname = split_pokemon(name)
conn.execute(
""" """
INSERT INTO switches(game, turn, player, pokemon) INSERT INTO switches(game, turn, player, pokemon)
VALUES (?, ?, ?, ?) VALUES (?, ?, ?, ?)
ON CONFLICT DO NOTHING ON CONFLICT DO NOTHING
""", """,
(game, turn, team(player), trimmed_specie), (self.game, self.turn, team(player), trimmed_specie),
)
conn.execute(
"""
INSERT INTO nicknames(game, player, pokemon, specie)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, team(player), nickname, trimmed_specie),
) )
case ["faint", pokemon]: # t.Literal, TaggedPokemon
conn.execute( case ["faint", pokemon_]:
pokemon = TaggedPokemon(pokemon_)
self.conn.execute(
""" """
INSERT INTO knockouts(game, turn, player, pokemon) INSERT INTO knockouts(game, turn, player, pokemon)
VALUES(?, ?, ?, ?) VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING ON CONFLICT DO NOTHING
""", """,
(game, turn, team(player), specie(pokemon)), (self.game, self.turn, team(player), self.specie(pokemon)),
) )
# t.Literal, Player
case ["win", player]: case ["win", player]:
conn.execute( self.conn.execute(
""" """
UPDATE games UPDATE games
SET winner = ? SET winner = ?
WHERE id = ? WHERE id = ?
""", """,
(team(player), game), (team(player), self.game),
) )
# t.Literal, TaggedPlayer, str
case ["-sidestart", side, env]: case ["-sidestart", side, env]:
if not last_move: if not last_move:
LOG.warning(f"missing previous move for {line}") LOG.warning(f"missing previous move for {line}")
continue continue
LOG.debug(f"{line} <- {last_move}") LOG.debug(f"{line} <- {last_move}")
last_env_set[(side[0:1], env.replace("move: ", ""))] = last_move[0] self.last_env_set[
(side[0:1], env.replace("move: ", ""))
] = last_move[0]
# t.Literal, TaggedPokemon, str
case ["-status", mon, cond]: case ["-status", mon, cond]:
if not last_move or last_move[1] != mon: if not last_move or last_move[1] != mon:
LOG.warning(f"missing previous move for {line}") LOG.warning(f"missing previous move for {line}")
continue continue
LOG.debug(f"{line} <- {last_move}") LOG.debug(f"{line} <- {last_move}")
last_status_set[(mon, cond)] = last_move[0] self.last_status_set[(mon, cond)] = last_move[0]
# t.Literal, TaggedPokemon, str
case ["-damage", pokemon, status]: case ["-damage", pokemon, status]:
# mon takes direct (non-hazard/condition) damage # Pokemon takes direct (non-hazard/condition) damage; status
# status can be a percentage 70/100 with or without condition, # can be a percentage "70/100" with or without condition, or
# or "0 fnt" # "0 fnt"
new_hp = int(re.split("[/ ]", status)[0]) new_hp = int(re.split("[/ ]", status)[0])
LOG.debug(f"{pokemon} dropped to {new_hp} from {hp[pokemon]}") LOG.debug(f"{pokemon} dropped to {new_hp} from {self.hp[pokemon]}")
LOG.debug(f"source: {last_move}") LOG.debug(f"source: {last_move}")
# resolve to damage source # resolve to damage source
@ -299,43 +329,45 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
) )
continue continue
damage_source = last_move[0] damage_source = last_move[0]
source_player, source_nickname = split_pokemon(damage_source) source_player, source_nickname = self.split_pokemon(damage_source)
conn.execute( self.conn.execute(
""" """
INSERT INTO damage(game, player, pokemon, value) INSERT INTO damage(game, player, pokemon, value)
VALUES(?, ?, ?, ?) VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING ON CONFLICT DO NOTHING
""", """,
( (
game, self.game,
team(source_player), team(source_player),
specie(damage_source), self.specie(damage_source),
hp[pokemon] - new_hp, self.hp[pokemon] - new_hp,
), ),
) )
hp[pokemon] = new_hp self.hp[pokemon] = new_hp
case ["-damage", pokemon, status, from_]: # t.Literal, TaggedPokemon, str, str
# mon takes indirect damage case ["-damage", pokemon_, status, from_]:
# status can be a percentage 70/100 with or without condition, pokemon = TaggedPokemon(pokemon_)
# or "0 fnt"
# Pokemon takes indirect damage; status can be a percentage
# "70/100" with or without condition, or "0 fnt"
new_hp = int(re.split("[/ ]", status)[0]) new_hp = int(re.split("[/ ]", status)[0])
LOG.debug(f"{pokemon} dropped to {new_hp} from {from_}") LOG.debug(f"{pokemon} dropped to {new_hp} from {from_}")
LOG.debug(f"tracing reason for {line}") LOG.debug(f"tracing reason for {line}")
reason = from_.replace("[from] ", "") reason = from_.replace("[from] ", "")
source = None source: TaggedPokemon | str | None = None
source_is_pokemon = True source_is_pokemon = True
test_hazard = last_env_set.get((pokemon[0:1], reason)) test_hazard = self.last_env_set.get((pokemon[0:1], reason))
if test_hazard: if test_hazard:
source = test_hazard source = test_hazard
LOG.debug(f"identified hazard source {source}") LOG.debug(f"identified hazard source {source}")
test_status = last_status_set.get((pokemon, reason)) test_status = self.last_status_set.get((pokemon, reason))
if test_status: if test_status:
source = test_status source = test_status
LOG.debug(f"identified move source {source}") LOG.debug(f"identified move source {source}")
@ -350,29 +382,29 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
LOG.error(f"missing reason for {line}") LOG.error(f"missing reason for {line}")
continue continue
player, nickname = split_pokemon(pokemon) player, nickname = self.split_pokemon(pokemon)
if source.startswith("p1") or source.startswith("p2"): if source.startswith("p1") or source.startswith("p2"):
source_player, _ = split_pokemon(source) source_player, _ = self.split_pokemon(TaggedPokemon(source))
else: else:
source_player = None source_player = None # type: ignore
source_is_pokemon = False source_is_pokemon = False
if source_player: if source_player:
conn.execute( self.conn.execute(
""" """
INSERT INTO indirect_damage(game, player, pokemon, value) INSERT INTO indirect_damage(game, player, pokemon, value)
VALUES(?, ?, ?, ?) VALUES(?, ?, ?, ?)
""", """,
( (
game, self.game,
team(source_player), team(source_player),
specie(source), self.specie(TaggedPokemon(source)),
hp[pokemon] - new_hp, self.hp[pokemon] - new_hp,
), ),
) )
if status == "0 fnt": if status == "0 fnt":
conn.execute( self.conn.execute(
""" """
INSERT INTO indirect_knockouts( INSERT INTO indirect_knockouts(
game, turn, player, pokemon, game, turn, player, pokemon,
@ -381,18 +413,20 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
ON CONFLICT DO NOTHING ON CONFLICT DO NOTHING
""", """,
( (
game, self.game,
turn, self.turn,
team(player), team(player),
specie(pokemon), self.specie(pokemon),
reason, reason,
specie(source) if source_is_pokemon else source, self.specie(TaggedPokemon(source))
if source_is_pokemon
else source,
team(source_player), team(source_player),
), ),
) )
case ["-heal", pokemon, status, *rest]: case ["-heal", pokemon, status, *rest]:
hp[pokemon] = int(status.split("/")[0]) self.hp[pokemon] = int(status.split("/")[0])
case _: case _:
# LOG.debug(f"unhandled message {chunks[0]}") # LOG.debug(f"unhandled message {chunks[0]}")
@ -434,7 +468,7 @@ def fetch(replay: str, cache: bool = True) -> Replay:
with replay_file.open(mode="w") as f: with replay_file.open(mode="w") as f:
json.dump(data, f) json.dump(data, f)
return Replay(**data) return Replay(**data) # type: ignore
def main(args): def main(args):
@ -506,7 +540,7 @@ def main(args):
), ),
) )
parse_log(replay.id, replay.log, into=db) LogParser(replay.id, db).parse(replay.log)
db.commit() db.commit()
finally: finally: