haive.games.among_us.state¶
Comprehensive state management for Among Us social deduction gameplay.
This module provides the core state model for managing Among Us game sessions, tracking player positions, tasks, sabotages, and game progression. The state system supports complex spatial navigation, role-based mechanics, and real-time event handling for authentic social deduction experiences.
The state management includes: - Dynamic map initialization with rooms and vents - Player position and status tracking - Task completion monitoring - Sabotage event management - Meeting and voting mechanics - Win condition evaluation - Observation and memory systems
Examples
Initializing a game state:
state = AmongUsState()
state.initialize_map() # Creates Skeld map by default
# Add players
state.player_states["player1"] = PlayerState(
id="player1",
role=PlayerRole.CREWMATE,
location="cafeteria"
)
Checking win conditions:
winner = state.check_win_condition()
if winner == "crewmates":
print("All tasks completed!")
elif winner == "impostors":
print("Impostors have taken control!")
Managing sabotages:
# Create reactor meltdown
sabotage = SabotageEvent(
type="reactor",
location="reactor",
timer=30
)
state.sabotages.append(sabotage)
# Check if critical
active = state.get_active_sabotage()
if active and active.is_critical():
print(f"Emergency! {active.timer}s remaining!")
Note
The state model extends MultiPlayerGameState and integrates with LangGraph for distributed game session management.
Classes¶
Comprehensive state model for Among Us game sessions. |
Module Contents¶
- class haive.games.among_us.state.AmongUsState(/, **data)[source]¶
Bases:
haive.games.framework.multi_player.state.MultiPlayerGameState
Comprehensive state model for Among Us game sessions.
This class manages all aspects of an Among Us game state, including spatial layout, player tracking, task management, and game progression. It provides methods for querying game state, updating player positions, managing events, and evaluating win conditions.
The state model supports: - Multiple map layouts (currently Skeld, with Polus/Mira HQ planned) - Real-time player position and status tracking - Task assignment and completion monitoring - Sabotage event lifecycle management - Emergency meeting and voting systems - Observation and memory tracking for deduction - Flexible win condition evaluation
- Parameters:
data (Any)
- map_locations¶
List of valid room IDs for the current map. Used for movement validation and spawn points.
- map_name¶
Name of the current map (e.g., “skeld”, “polus”). Determines room layout and task locations.
- rooms¶
Mapping of room IDs to Room objects. Defines spatial layout and connections.
- vents¶
Mapping of vent IDs to Vent objects. Enables impostor movement through vent network.
- player_states¶
Mapping of player IDs to PlayerState objects. Tracks all player information including role and position.
- tasks¶
Mapping of task IDs to Task objects. Defines available tasks and their completion status.
- sabotages¶
List of active and resolved sabotage events. Tracks sabotage history and current emergencies.
- eliminated_players¶
List of player IDs who have been eliminated. Used for ghost mechanics and win conditions.
- meeting_active¶
Whether an emergency meeting is in progress. Pauses gameplay and enables discussion/voting.
- meeting_caller¶
Player ID who called the current meeting. None if meeting was triggered by body report.
- reported_body¶
Player ID of body that triggered meeting. None if meeting was called via emergency button.
- votes¶
Mapping of voter ID to target ID for current meeting. Tracks voting progress and determines ejection.
- game_phase¶
Current phase of gameplay. Controls available actions and UI state.
- impostor_count¶
Number of living impostors. Cached for efficient win condition checks.
- crewmate_count¶
Number of living crewmates. Cached for efficient win condition checks.
- discussion_history¶
List of discussion events and messages. Provides context for AI decision-making.
- kill_cooldowns¶
Mapping of impostor ID to remaining cooldown. Prevents rapid elimination chains.
Examples
Creating a new game state:
state = AmongUsState( map_name="skeld", impostor_count=2, crewmate_count=8 ) state.initialize_map()
Checking task progress:
progress = state.get_task_completion_percentage() if progress >= 100: print("Crewmates win by tasks!")
Managing meetings:
state.meeting_active = True state.meeting_caller = "player1" state.votes["player1"] = "player3" # Voting for player3 state.votes["player2"] = "skip" # Voting to skip
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- add_observation(player_id, observation)[source]¶
Add an observation to a player’s memory.
Observations form the basis of deduction in Among Us. This method records what a player has witnessed for later analysis.
- Parameters:
- Return type:
None
Examples
Witnessing movement:
state.add_observation( "player1", "Saw player3 enter electrical at 14:32" )
Task verification:
state.add_observation( "player2", "Watched player4 complete medbay scan" )
- add_observation_to_all_in_room(room_id, observation, exclude_players=None)[source]¶
Add an observation to all players in a specific room.
Used for events that are visible to everyone in a location, such as sabotages, eliminations, or visual tasks.
- Parameters:
- Return type:
None
Examples
Elimination witnessed:
state.add_observation_to_all_in_room( "electrical", "Player5 eliminated player2!", exclude_players=["player5"] # Killer doesn't observe self )
Visual task:
state.add_observation_to_all_in_room( "medbay", "Player3 completed scan (confirmed crewmate)" )
- check_win_condition()[source]¶
Check if either side has achieved victory.
Evaluates all win conditions for both teams:
Crewmate victories: - All tasks completed - All impostors eliminated
Impostor victories: - Impostor count >= crewmate count - Critical sabotage timer expired
- Returns:
Winning team or None.
- Return type:
Optional[Literal[“crewmates”, “impostors”]]
Examples
Task victory:
# All tasks done winner = state.check_win_condition() assert winner == "crewmates"
Impostor majority:
# 2 impostors, 2 crewmates remaining winner = state.check_win_condition() assert winner == "impostors"
Sabotage victory:
# Reactor meltdown timer reaches 0 winner = state.check_win_condition() assert winner == "impostors"
- decrement_cooldowns()[source]¶
Decrement all active kill cooldowns by 1 second.
Should be called each game tick/second to update cooldowns. Automatically removes cooldowns that reach 0.
Examples
Game loop integration:
# In game tick handler state.decrement_cooldowns() # Check if any impostors can now kill for impostor_id in impostor_ids: if state.get_player_cooldown(impostor_id) == 0: # Enable kill button in UI pass
- Return type:
None
- get_active_sabotage()[source]¶
Get the currently active sabotage event.
Only one sabotage can be active at a time. This method returns the first unresolved sabotage found.
- Returns:
Active sabotage or None.
- Return type:
Optional[SabotageEvent]
Examples
Emergency response:
sabotage = state.get_active_sabotage() if sabotage and sabotage.is_critical(): print(f"EMERGENCY: {sabotage.type} - {sabotage.timer}s left!")
Checking if sabotage is possible:
if state.get_active_sabotage() is None: # Impostors can trigger new sabotage pass
- get_alive_players()[source]¶
Get list of alive player IDs.
Returns only players who have not been eliminated, useful for voting counts, task assignments, and win condition checks.
- Returns:
IDs of all living players.
- Return type:
List[str]
Examples
During meeting:
alive = state.get_alive_players() print(f"{len(alive)} players can vote")
Win condition check:
alive = state.get_alive_players() if len(alive) <= 2: # Check for impostor majority pass
- get_connected_rooms(room_id)[source]¶
Get all rooms directly connected to the given room.
Returns only accessible connections (not blocked by sabotage). Used for pathfinding and movement validation.
- Parameters:
room_id (str) – Source room to check connections from.
- Returns:
IDs of all accessible connected rooms.
- Return type:
List[str]
Examples
Movement options:
adjacent = state.get_connected_rooms("cafeteria") print(f"Can move to: {', '.join(adjacent)}")
Sabotage effects:
# During O2 sabotage, some doors may be sealed rooms = state.get_connected_rooms("admin") if len(rooms) < normal_count: print("Some exits are blocked!")
- get_connected_vents(vent_id)[source]¶
Get all vents connected to the given vent.
Used for impostor movement through the vent network. All vent connections are always accessible (no blocking).
- Parameters:
vent_id (str) – Source vent to check connections from.
- Returns:
IDs of all connected vents.
- Return type:
List[str]
Examples
Vent navigation:
connected = state.get_connected_vents("electrical_vent") for vent_id in connected: vent = state.get_vent(vent_id) print(f"Can emerge in: {vent.location}")
- get_player_cooldown(player_id)[source]¶
Get a player’s kill cooldown in seconds.
- Parameters:
player_id (str) – ID of the player (impostor) to check.
- Returns:
Remaining cooldown seconds (0 if can kill).
- Return type:
Examples
Checking kill availability:
cooldown = state.get_player_cooldown("impostor1") if cooldown == 0: print("Kill ability ready") else: print(f"Kill on cooldown: {cooldown}s")
- get_room(room_id)[source]¶
Get a room by its unique identifier.
- Parameters:
room_id (str) – Unique room identifier (e.g., “cafeteria”).
- Returns:
Room object if found, None otherwise.
- Return type:
Optional[Room]
Examples
Checking room properties:
room = state.get_room("electrical") if room and room.has_security_camera: print("This room is monitored")
- get_task_completion_percentage()[source]¶
Calculate overall task completion percentage.
Computes the percentage of all tasks that have been completed across all crewmates. This is a primary win condition - crewmates win immediately when this reaches 100%.
- Returns:
Percentage of completed tasks (0-100).
- Return type:
Examples
Progress tracking:
progress = state.get_task_completion_percentage() print(f"Tasks: {progress:.1f}% complete")
Win condition:
if state.get_task_completion_percentage() >= 100: return "crewmates_win_by_tasks"
Note
Returns 100.0 if no tasks exist (edge case handling).
- get_vent(vent_id)[source]¶
Get a vent by its unique identifier.
- Parameters:
vent_id (str) – Unique vent identifier (e.g., “electrical_vent”).
- Returns:
Vent object if found, None otherwise.
- Return type:
Optional[Vent]
Examples
Checking vent connections:
vent = state.get_vent("cafeteria_vent") if vent: connected = [c.target_vent_id for c in vent.connections] print(f"Can travel to: {connected}")
- get_vents_in_room(room_id)[source]¶
Get all vents located in a specific room.
Used for impostor movement options and vent camping detection.
- Parameters:
room_id (str) – Room to search for vents.
- Returns:
All vents in the specified room.
- Return type:
List[Vent]
Examples
Impostor options:
vents = state.get_vents_in_room("electrical") if vents and player.role == PlayerRole.IMPOSTOR: print(f"Can enter {len(vents)} vent(s) here")
- initialize_map()[source]¶
Initialize the map with rooms and vents based on the map name.
Creates the spatial layout for the selected map, including all rooms, connections, and vent networks. Currently supports the Skeld map with plans for Polus and Mira HQ.
The initialization includes: - Room creation with connections and properties - Vent network setup with travel times - Map location list for spawn points
Examples
Default Skeld initialization:
state = AmongUsState(map_name="skeld") state.initialize_map() assert "cafeteria" in state.rooms assert len(state.vents) == 10 # Skeld has 10 vents
Future map support:
state = AmongUsState(map_name="polus") state.initialize_map() # Will support Polus layout
Note
This method clears any existing rooms and vents before creating the new map layout.
- Return type:
None
- set_player_cooldown(player_id, cooldown)[source]¶
Set a player’s kill cooldown.
Typically called after an impostor eliminates someone.
- Parameters:
- Return type:
None
Examples
After elimination:
# Standard 30 second cooldown state.set_player_cooldown("impostor1", 30)
Custom game settings:
# Fast-paced game with 10s cooldown state.set_player_cooldown("impostor1", 10)
- classmethod validate_map_name(v)[source]¶
Validate map name is supported.
- Parameters:
v (str) – Map name to validate.
- Returns:
Validated map name in lowercase.
- Return type:
- Raises:
ValueError – If map name is not supported.
- property game_statistics: dict[str, Any]¶
Calculate comprehensive game statistics.
- Returns:
Statistics including player counts, progress, etc.
- Return type:
Dict[str, Any]
Examples
Displaying game stats:
stats = state.game_statistics print(f"Tasks: {stats['task_progress']:.1f}%") print(f"Impostors: {stats['alive_impostors']}/{stats['total_impostors']}")
- model_config¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].