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

mass refactor

Browse Source 
- resolve nicknames as we go
- update column names
- better variables
This commit is contained in:
xeals
2023-04-04 21:03:41 +10:00
parent 262a051356
commit 8a348eca84
1 changed files with 145 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
main.py
View File

@@ -79,23 +79,24 @@ def _init_db(conn: sqlite3.Connection):
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS moves(
game, turn, player, name, user, target,
UNIQUE(game, turn, player, user)
game, turn, player, pokemon, move, target,
UNIQUE(game, turn, player, pokemon)
);
CREATE TABLE IF NOT EXISTS switches(
game, turn, player, name,
UNIQUE(game, turn, player, name)
game, turn, player, pokemon,
UNIQUE(game, turn, player, pokemon)
);
CREATE TABLE IF NOT EXISTS nicknames(
game, player, name, specie,
game, player, pokemon, specie,
UNIQUE(game, player, specie)
);
CREATE TABLE IF NOT EXISTS knockouts(
game, turn, player, name,
game, turn, player, pokemon,
UNIQUE(game, turn, player)
);
CREATE TABLE IF NOT EXISTS indirect_knockouts(
game, turn, player, name, source, source_user, source_player,
game, turn, player, pokemon,
reason, source, source_player,
UNIQUE(game, turn, player)
);
CREATE TABLE IF NOT EXISTS games(
@@ -104,9 +105,9 @@ def _init_db(conn: sqlite3.Connection):
);
-- No good way to ensure idempotence for damage; just re-build it.
DROP TABLE IF EXISTS damage;
CREATE TABLE damage(game, player, name, value);
CREATE TABLE damage(game, player, pokemon, value);
DROP TABLE IF EXISTS indirect_damage;
CREATE TABLE indirect_damage(game, player, name, value);
CREATE TABLE indirect_damage(game, player, pokemon, value);
"""
)
@@ -129,10 +130,36 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
# ("p1a: Meteo", "brn") => "p2a: Edward"
last_status_set: dict[tuple[str, str], str] = {}
def resolve_mon(user: str) -> tuple[str, str]:
def split_pokemon(user: str) -> tuple[str, str]:
"""Splits a Pokemon identifier of the form `pXa: Pokemon` into the
player's name (as marked by the player log) and "Pokemon".
Note that all Pokemon are referred to by their nicknames, and will
require resolving to obtain the Pokemon specie."""
[player, name] = user.split(": ")
return players[player.strip("ab")], name
def specie_from_parts(player: str, nickname: str) -> str:
"""Resolves the species of a nicknamed Pokemon."""
return (
conn.execute(
"""
SELECT specie
FROM nicknames
WHERE (game, player, pokemon) = (?, ?, ?)
LIMIT 1
""",
(game, team(player), nickname),
)
.fetchall()[0]
.specie
)
def specie(pokemon: str) -> str:
"""Resolves the species of a Pokemon given its Showdown identifier (used
in split_pokemon)."""
return specie_from_parts(*split_pokemon(pokemon))
for line in log.split("\n"):
chunks = line.split("|")[1:]
if not chunks:
@@ -149,50 +176,85 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
case ["move", user, move, target]:
last_move = (user, target)
player, user = resolve_mon(user)
_, target = resolve_mon(target)
player, _ = split_pokemon(user)
conn.execute(
"""
INSERT INTO moves(game, turn, player, name, user, target)
INSERT INTO moves(game, turn, player, pokemon, move, target)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, turn, team(player), move, user, target),
(
game,
turn,
team(player),
specie(user),
move,
specie(target),
),
)
case ["drag", name, specie, status, *rest]:
case ["drag", name, specie_, status, *rest]:
hp[name] = int(status.split("/")[0])
case ["switch", name, specie, status, *rest]:
# 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 ["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])
player, name = resolve_mon(name)
# Also includes gender and formes.
trimmed_specie = specie_.split(", ")[0]
player, nickname = split_pokemon(name)
conn.execute(
"""
INSERT INTO switches(game, turn, player, name)
INSERT INTO switches(game, turn, player, pokemon)
VALUES (?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, turn, team(player), name),
(game, turn, team(player), trimmed_specie),
)
conn.execute(
"""
INSERT INTO nicknames(game, player, name, specie)
INSERT INTO nicknames(game, player, pokemon, specie)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, team(player), name, specie.split(", ")[0]),
(game, team(player), nickname, trimmed_specie),
)
case ["faint", mon]:
player, mon = resolve_mon(mon)
case ["faint", pokemon]:
conn.execute(
"""
INSERT INTO knockouts(game, turn, player, name)
INSERT INTO knockouts(game, turn, player, pokemon)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, turn, team(player), mon),
(game, turn, team(player), specie(pokemon)),
)
case ["win", player]:
@@ -209,6 +271,7 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
if not last_move:
LOG.warning(f"missing previous move for {line}")
continue
LOG.debug(f"{line} <- {last_move}")
last_env_set[(side[0:1], env.replace("move: ", ""))] = last_move[0]
@@ -216,88 +279,104 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
if not last_move or last_move[1] != mon:
LOG.warning(f"missing previous move for {line}")
continue
LOG.debug(f"{line} <- {last_move}")
last_status_set[(mon, cond)] = last_move[0]
case ["-damage", mon, status]:
case ["-damage", pokemon, status]:
# mon takes direct (non-hazard/condition) damage
# status can be a percentage 70/100 with or without condition,
# or "0 fnt"
new_hp = int(re.split("[/ ]", status)[0])
LOG.debug(f"{mon} dropped to {new_hp} from {hp[mon]}")
LOG.debug(f"{pokemon} dropped to {new_hp} from {hp[pokemon]}")
LOG.debug(f"source: {last_move}")
# resolve to damage source
if last_move[1] != mon:
LOG.warn(
f"{mon} took direct damage but last move was not targeted at them"
if last_move[1] != pokemon:
LOG.warning(
f"{pokemon} took direct damage but last move was not"
" targeted at them"
)
continue
user = last_move[0]
source_player, source_mon = resolve_mon(user)
damage_source = last_move[0]
source_player, source_nickname = split_pokemon(damage_source)
conn.execute(
"""
INSERT INTO damage(game, player, name, value)
INSERT INTO damage(game, player, pokemon, value)
VALUES(?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
(game, team(source_player), source_mon, hp[mon] - new_hp),
(
game,
team(source_player),
specie(damage_source),
hp[pokemon] - new_hp,
),
)
hp[mon] = new_hp
hp[pokemon] = new_hp
case ["-damage", mon, status, from_]:
case ["-damage", pokemon, status, from_]:
# mon takes indirect damage
# status can be a percentage 70/100 with or without condition,
# or "0 fnt"
# mon has fainted from an indirect damage source
#
new_hp = int(re.split("[/ ]", status)[0])
LOG.debug(f"{mon} dropped to {new_hp} from {from_}")
LOG.debug(f"{pokemon} dropped to {new_hp} from {from_}")
LOG.debug(f"tracing source for {line}")
source = from_.replace("[from] ", "")
source_user = None
LOG.debug(f"tracing reason for {line}")
reason = from_.replace("[from] ", "")
test_hazard = last_env_set.get((mon[0:1], source))
source = None
source_is_pokemon = True
test_hazard = last_env_set.get((pokemon[0:1], reason))
if test_hazard:
source_user = test_hazard
LOG.debug(f"identified hazard source {source_user}")
source = test_hazard
LOG.debug(f"identified hazard source {source}")
test_status = last_status_set.get((mon, source))
test_status = last_status_set.get((pokemon, reason))
if test_status:
source_user = test_status
LOG.debug(f"identified move source {source_user}")
source = test_status
LOG.debug(f"identified move source {source}")
if source == "Recoil" or source.startswith("item: "):
LOG.debug(f"identified special source {source}")
source = source.replace("item: ", "")
source_user = "self"
if reason == "Recoil" or reason.startswith("item: "):
LOG.debug(f"identified special source {reason}")
reason = reason.replace("item: ", "")
source = "self"
source_is_pokemon = False
if not source_user:
LOG.error(f"missing source for {line}")
if not source:
LOG.error(f"missing reason for {line}")
continue
player, pkmn = resolve_mon(mon)
if source_user.startswith("p1") or source_user.startswith("p2"):
source_player, source_mon = resolve_mon(source_user)
player, nickname = split_pokemon(pokemon)
if source.startswith("p1") or source.startswith("p2"):
source_player, _ = split_pokemon(source)
else:
source_player = None
source_is_pokemon = False
if source_player:
conn.execute(
"""
INSERT INTO indirect_damage(game, player, name, value)
INSERT INTO indirect_damage(game, player, pokemon, value)
VALUES(?, ?, ?, ?)
""",
(game, team(source_player), source_mon, hp[mon] - new_hp),
(
game,
team(source_player),
specie(source),
hp[pokemon] - new_hp,
),
)
if status == "0 fnt":
conn.execute(
"""
INSERT INTO indirect_knockouts(game, turn, player, name, source, source_user, source_player)
INSERT INTO indirect_knockouts(
game, turn, player, pokemon,
reason, source, source_player)
VALUES(?, ?, ?, ?, ?, ?, ?)
ON CONFLICT DO NOTHING
""",
@@ -305,15 +384,15 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
game,
turn,
team(player),
pkmn,
source,
source_mon,
specie(pokemon),
reason,
specie(source) if source_is_pokemon else source,
team(source_player),
),
)
case ["-heal", mon, status, *rest]:
hp[mon] = int(status.split("/")[0])
case ["-heal", pokemon, status, *rest]:
hp[pokemon] = int(status.split("/")[0])
case _:
# LOG.debug(f"unhandled message {chunks[0]}")