morgan/Grinch-AP
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

WebHost: revamp /api/*tracker/ (#5388)

Browse Source
This commit is contained in:
Fabian Dill
2025-09-22 00:25:12 +02:00
committed by GitHub
parent 68187ba25f
commit fb9011da63
4 changed files with 255 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
WebHostLib/api/tracker.py
View File

@@ -11,6 +11,47 @@ from WebHostLib.models import Room
from WebHostLib.tracker import TrackerData
class PlayerAlias(TypedDict):
team: int
player: int
alias: str | None
class PlayerItemsReceived(TypedDict):
team: int
player: int
items: list[NetworkItem]
class PlayerChecksDone(TypedDict):
team: int
player: int
locations: list[int]
class TeamTotalChecks(TypedDict):
team: int
checks_done: int
class PlayerHints(TypedDict):
team: int
player: int
hints: list[Hint]
class PlayerTimer(TypedDict):
team: int
player: int
time: datetime | None
class PlayerStatus(TypedDict):
team: int
player: int
status: ClientStatus
@api_endpoints.route("/tracker/<suuid:tracker>")
@cache.memoize(timeout=60)
def tracker_data(tracker: UUID) -> dict[str, Any]:
@@ -29,122 +70,77 @@ def tracker_data(tracker: UUID) -> dict[str, Any]:
all_players: dict[int, list[int]] = tracker_data.get_all_players()
class PlayerAlias(TypedDict):
player: int
name: str | None
player_aliases: list[dict[str, int | list[PlayerAlias]]] = []
player_aliases: list[PlayerAlias] = []
"""Slot aliases of all players."""
for team, players in all_players.items():
team_player_aliases: list[PlayerAlias] = []
team_aliases = {"team": team, "players": team_player_aliases}
player_aliases.append(team_aliases)
for player in players:
team_player_aliases.append({"player": player, "alias": tracker_data.get_player_alias(team, player)})
player_aliases.append({"team": team, "player": player, "alias": tracker_data.get_player_alias(team, player)})
class PlayerItemsReceived(TypedDict):
player: int
items: list[NetworkItem]
player_items_received: list[dict[str, int | list[PlayerItemsReceived]]] = []
player_items_received: list[PlayerItemsReceived] = []
"""Items received by each player."""
for team, players in all_players.items():
player_received_items: list[PlayerItemsReceived] = []
team_items_received = {"team": team, "players": player_received_items}
player_items_received.append(team_items_received)
for player in players:
player_received_items.append(
{"player": player, "items": tracker_data.get_player_received_items(team, player)})
player_items_received.append(
{"team": team, "player": player, "items": tracker_data.get_player_received_items(team, player)})
class PlayerChecksDone(TypedDict):
player: int
locations: list[int]
player_checks_done: list[dict[str, int | list[PlayerChecksDone]]] = []
player_checks_done: list[PlayerChecksDone] = []
"""ID of all locations checked by each player."""
for team, players in all_players.items():
per_player_checks: list[PlayerChecksDone] = []
team_checks_done = {"team": team, "players": per_player_checks}
player_checks_done.append(team_checks_done)
for player in players:
per_player_checks.append(
{"player": player, "locations": sorted(tracker_data.get_player_checked_locations(team, player))})
player_checks_done.append(
{"team": team, "player": player, "locations": sorted(tracker_data.get_player_checked_locations(team, player))})
total_checks_done: list[dict[str, int]] = [
total_checks_done: list[TeamTotalChecks] = [
{"team": team, "checks_done": checks_done}
for team, checks_done in tracker_data.get_team_locations_checked_count().items()
]
"""Total number of locations checked for the entire multiworld per team."""
class PlayerHints(TypedDict):
player: int
hints: list[Hint]
hints: list[dict[str, int | list[PlayerHints]]] = []
hints: list[PlayerHints] = []
"""Hints that all players have used or received."""
for team, players in tracker_data.get_all_slots().items():
per_player_hints: list[PlayerHints] = []
team_hints = {"team": team, "players": per_player_hints}
hints.append(team_hints)
for player in players:
player_hints = sorted(tracker_data.get_player_hints(team, player))
per_player_hints.append({"player": player, "hints": player_hints})
slot_info = tracker_data.get_slot_info(team, player)
hints.append({"team": team, "player": player, "hints": player_hints})
slot_info = tracker_data.get_slot_info(player)
# this assumes groups are always after players
if slot_info.type != SlotType.group:
continue
for member in slot_info.group_members:
team_hints[member]["hints"] += player_hints
hints[member - 1]["hints"] += player_hints
class PlayerTimer(TypedDict):
player: int
time: datetime | None
activity_timers: list[dict[str, int | list[PlayerTimer]]] = []
activity_timers: list[PlayerTimer] = []
"""Time of last activity per player. Returned as RFC 1123 format and null if no connection has been made."""
for team, players in all_players.items():
player_timers: list[PlayerTimer] = []
team_timers = {"team": team, "players": player_timers}
activity_timers.append(team_timers)
for player in players:
player_timers.append({"player": player, "time": None})
activity_timers.append({"team": team, "player": player, "time": None})
client_activity_timers: tuple[tuple[int, int], float] = tracker_data._multisave.get("client_activity_timers", ())
for (team, player), timestamp in client_activity_timers:
# use index since we can rely on order
# FIX: key is "players" (not "player_timers")
activity_timers[team]["players"][player - 1]["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
for (team, player), timestamp in tracker_data._multisave.get("client_activity_timers", []):
for entry in activity_timers:
if entry["team"] == team and entry["player"] == player:
entry["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
break
connection_timers: list[dict[str, int | list[PlayerTimer]]] = []
connection_timers: list[PlayerTimer] = []
"""Time of last connection per player. Returned as RFC 1123 format and null if no connection has been made."""
for team, players in all_players.items():
player_timers: list[PlayerTimer] = []
team_connection_timers = {"team": team, "players": player_timers}
connection_timers.append(team_connection_timers)
for player in players:
player_timers.append({"player": player, "time": None})
connection_timers.append({"team": team, "player": player, "time": None})
client_connection_timers: tuple[tuple[int, int], float] = tracker_data._multisave.get(
"client_connection_timers", ())
for (team, player), timestamp in client_connection_timers:
connection_timers[team]["players"][player - 1]["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
for (team, player), timestamp in tracker_data._multisave.get("client_connection_timers", []):
# find the matching entry
for entry in connection_timers:
if entry["team"] == team and entry["player"] == player:
entry["time"] = datetime.fromtimestamp(timestamp, timezone.utc)
break
class PlayerStatus(TypedDict):
player: int
status: ClientStatus
player_status: list[dict[str, int | list[PlayerStatus]]] = []
player_status: list[PlayerStatus] = []
"""The current client status for each player."""
for team, players in all_players.items():
player_statuses: list[PlayerStatus] = []
team_status = {"team": team, "players": player_statuses}
player_status.append(team_status)
for player in players:
player_statuses.append({"player": player, "status": tracker_data.get_player_client_status(team, player)})
player_status.append({"team": team, "player": player, "status": tracker_data.get_player_client_status(team, player)})
return {
**get_static_tracker_data(room),
"aliases": player_aliases,
"player_items_received": player_items_received,
"player_checks_done": player_checks_done,
@@ -153,80 +149,80 @@ def tracker_data(tracker: UUID) -> dict[str, Any]:
"activity_timers": activity_timers,
"connection_timers": connection_timers,
"player_status": player_status,
"datapackage": tracker_data._multidata["datapackage"],
}
@cache.memoize()
def get_static_tracker_data(room: Room) -> dict[str, Any]:
"""
Builds and caches the static data for this active session tracker, so that it doesn't need to be recalculated.
"""
class PlayerGroups(TypedDict):
slot: int
name: str
members: list[int]
class PlayerSlotData(TypedDict):
player: int
slot_data: dict[str, Any]
@api_endpoints.route("/static_tracker/<suuid:tracker>")
@cache.memoize(timeout=300)
def static_tracker_data(tracker: UUID) -> dict[str, Any]:
"""
Outputs json data to <root_path>/api/static_tracker/<id of current session tracker>.
:param tracker: UUID of current session tracker.
:return: Static tracking data for all players in the room. Typing and docstrings describe the format of each value.
"""
room: Room | None = Room.get(tracker=tracker)
if not room:
abort(404)
tracker_data = TrackerData(room)
all_players: dict[int, list[int]] = tracker_data.get_all_players()
class PlayerGroups(TypedDict):
slot: int
name: str
members: list[int]
groups: list[dict[str, int | list[PlayerGroups]]] = []
groups: list[PlayerGroups] = []
"""The Slot ID of groups and the IDs of the group's members."""
for team, players in tracker_data.get_all_slots().items():
groups_in_team: list[PlayerGroups] = []
team_groups = {"team": team, "groups": groups_in_team}
groups.append(team_groups)
for player in players:
slot_info = tracker_data.get_slot_info(team, player)
slot_info = tracker_data.get_slot_info(player)
if slot_info.type != SlotType.group or not slot_info.group_members:
continue
groups_in_team.append(
groups.append(
{
"slot": player,
"name": slot_info.name,
"members": list(slot_info.group_members),
})
class PlayerName(TypedDict):
player: int
name: str
player_names: list[dict[str, str | list[PlayerName]]] = []
"""Slot names of all players."""
for team, players in all_players.items():
per_team_player_names: list[PlayerName] = []
team_names = {"team": team, "players": per_team_player_names}
player_names.append(team_names)
for player in players:
per_team_player_names.append({"player": player, "name": tracker_data.get_player_name(team, player)})
class PlayerGame(TypedDict):
player: int
game: str
games: list[dict[str, int | list[PlayerGame]]] = []
"""The game each player is playing."""
for team, players in all_players.items():
player_games: list[PlayerGame] = []
team_games = {"team": team, "players": player_games}
games.append(team_games)
for player in players:
player_games.append({"player": player, "game": tracker_data.get_player_game(team, player)})
class PlayerSlotData(TypedDict):
player: int
slot_data: dict[str, Any]
slot_data: list[dict[str, int | list[PlayerSlotData]]] = []
"""Slot data for each player."""
for team, players in all_players.items():
player_slot_data: list[PlayerSlotData] = []
team_slot_data = {"team": team, "players": player_slot_data}
slot_data.append(team_slot_data)
for player in players:
player_slot_data.append({"player": player, "slot_data": tracker_data.get_slot_data(team, player)})
break
return {
"groups": groups,
"slot_data": slot_data,
"datapackage": tracker_data._multidata["datapackage"],
}
# It should be exceedingly rare that slot data is needed, so it's separated out.
@api_endpoints.route("/slot_data_tracker/<suuid:tracker>")
@cache.memoize(timeout=300)
def tracker_slot_data(tracker: UUID) -> list[PlayerSlotData]:
"""
Outputs json data to <root_path>/api/slot_data_tracker/<id of current session tracker>.
:param tracker: UUID of current session tracker.
:return: Slot data for all players in the room. Typing completely arbitrary per game.
"""
room: Room | None = Room.get(tracker=tracker)
if not room:
abort(404)
tracker_data = TrackerData(room)
all_players: dict[int, list[int]] = tracker_data.get_all_players()
slot_data: list[PlayerSlotData] = []
"""Slot data for each player."""
for team, players in all_players.items():
for player in players:
slot_data.append({"player": player, "slot_data": tracker_data.get_slot_data(player)})
break
return slot_data

1
WebHostLib/tracker.py
View File

@@ -17,7 +17,6 @@ from .models import GameDataPackage, Room
# Multisave is currently updated, at most, every minute.
TRACKER_CACHE_TIMEOUT_IN_SECONDS = 60
_multidata_cache = {}
_multiworld_trackers: Dict[str, Callable] = {}
_player_trackers: Dict[str, Callable] = {}