242 lines
8.1 KiB
Python
242 lines
8.1 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any, TypedDict
|
|
from uuid import UUID
|
|
|
|
from flask import abort
|
|
|
|
from NetUtils import ClientStatus, Hint, NetworkItem, SlotType
|
|
from WebHostLib import cache
|
|
from WebHostLib.api import api_endpoints
|
|
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
|
|
|
|
|
|
class PlayerLocationsTotal(TypedDict):
|
|
team: int
|
|
player: int
|
|
total_locations: int
|
|
|
|
|
|
@api_endpoints.route("/tracker/<suuid:tracker>")
|
|
@cache.memoize(timeout=60)
|
|
def tracker_data(tracker: UUID) -> dict[str, Any]:
|
|
"""
|
|
Outputs json data to <root_path>/api/tracker/<id of current session tracker>.
|
|
|
|
:param tracker: UUID of current session tracker.
|
|
|
|
:return: 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()
|
|
|
|
player_aliases: list[PlayerAlias] = []
|
|
"""Slot aliases of all players."""
|
|
for team, players in all_players.items():
|
|
for player in players:
|
|
player_aliases.append({"team": team, "player": player, "alias": tracker_data.get_player_alias(team, player)})
|
|
|
|
player_items_received: list[PlayerItemsReceived] = []
|
|
"""Items received by each player."""
|
|
for team, players in all_players.items():
|
|
for player in players:
|
|
player_items_received.append(
|
|
{"team": team, "player": player, "items": tracker_data.get_player_received_items(team, player)})
|
|
|
|
player_checks_done: list[PlayerChecksDone] = []
|
|
"""ID of all locations checked by each player."""
|
|
for team, players in all_players.items():
|
|
for player in players:
|
|
player_checks_done.append(
|
|
{"team": team, "player": player, "locations": sorted(tracker_data.get_player_checked_locations(team, player))})
|
|
|
|
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."""
|
|
|
|
hints: list[PlayerHints] = []
|
|
"""Hints that all players have used or received."""
|
|
for team, players in tracker_data.get_all_slots().items():
|
|
for player in players:
|
|
player_hints = sorted(tracker_data.get_player_hints(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:
|
|
hints[member - 1]["hints"] += player_hints
|
|
|
|
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():
|
|
for player in players:
|
|
activity_timers.append({"team": team, "player": player, "time": None})
|
|
|
|
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[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():
|
|
for player in players:
|
|
connection_timers.append({"team": team, "player": player, "time": None})
|
|
|
|
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
|
|
|
|
player_status: list[PlayerStatus] = []
|
|
"""The current client status for each player."""
|
|
for team, players in all_players.items():
|
|
for player in players:
|
|
player_status.append({"team": team, "player": player, "status": tracker_data.get_player_client_status(team, player)})
|
|
|
|
return {
|
|
"aliases": player_aliases,
|
|
"player_items_received": player_items_received,
|
|
"player_checks_done": player_checks_done,
|
|
"total_checks_done": total_checks_done,
|
|
"hints": hints,
|
|
"activity_timers": activity_timers,
|
|
"connection_timers": connection_timers,
|
|
"player_status": player_status,
|
|
}
|
|
|
|
|
|
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()
|
|
|
|
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():
|
|
for player in players:
|
|
slot_info = tracker_data.get_slot_info(player)
|
|
if slot_info.type != SlotType.group or not slot_info.group_members:
|
|
continue
|
|
groups.append(
|
|
{
|
|
"slot": player,
|
|
"name": slot_info.name,
|
|
"members": list(slot_info.group_members),
|
|
})
|
|
break
|
|
|
|
player_locations_total: list[PlayerLocationsTotal] = []
|
|
for team, players in all_players.items():
|
|
for player in players:
|
|
player_locations_total.append(
|
|
{"team": team, "player": player, "total_locations": len(tracker_data.get_player_locations(player))})
|
|
|
|
return {
|
|
"groups": groups,
|
|
"datapackage": tracker_data._multidata["datapackage"],
|
|
"player_locations_total": player_locations_total,
|
|
}
|
|
|
|
# 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
|