From 7a86b2d90acde8865f24a5b866542a9dafa32ef9 Mon Sep 17 00:00:00 2001 From: "Joel Z. Leibo" Date: Tue, 30 Jul 2024 03:09:29 -0700 Subject: [PATCH] Add two useful game master components for creating custom environments. PiperOrigin-RevId: 657515734 Change-Id: I9065dfdbcdbc4051cd7d9c3a482fb08a2f6c1160 --- concordia/components/game_master/__init__.py | 2 + .../game_master/triggered_function.py | 143 ++++++++++++++++++ .../game_master/triggered_inventory_effect.py | 130 ++++++++++++++++ 3 files changed, 275 insertions(+) create mode 100644 concordia/components/game_master/triggered_function.py create mode 100644 concordia/components/game_master/triggered_inventory_effect.py diff --git a/concordia/components/game_master/__init__.py b/concordia/components/game_master/__init__.py index 8fc8567a..13df985a 100644 --- a/concordia/components/game_master/__init__.py +++ b/concordia/components/game_master/__init__.py @@ -25,3 +25,5 @@ from concordia.components.game_master import schedule from concordia.components.game_master import schelling_diagram_payoffs from concordia.components.game_master import time_display +from concordia.components.game_master import triggered_function +from concordia.components.game_master import triggered_inventory_effect diff --git a/concordia/components/game_master/triggered_function.py b/concordia/components/game_master/triggered_function.py new file mode 100644 index 00000000..b43e5c70 --- /dev/null +++ b/concordia/components/game_master/triggered_function.py @@ -0,0 +1,143 @@ +# Copyright 2023 DeepMind Technologies Limited. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A component to modify inventories based on events.""" + +from collections.abc import Callable, Sequence +import dataclasses +import datetime + +from concordia.agents import basic_agent +from concordia.agents import entity_agent +from concordia.associative_memory import associative_memory +from concordia.components.game_master import current_scene +from concordia.typing import component + +MemoryT = associative_memory.AssociativeMemory +PlayersT = Sequence[basic_agent.BasicAgent | entity_agent.EntityAgent] + + +@dataclasses.dataclass +class PreEventFnArgsT: + """A specification of the arguments to a pre-event function. + + Attributes: + player_name: The name of the player. + player_choice: The choice of the player on the current timestep. + current_scene_type: The type of the current scene. + players: Sequence of player objects. + memory: The game master's associative memory. + """ + + player_name: str + player_choice: str + current_scene_type: str + players: PlayersT + memory: MemoryT + + +@dataclasses.dataclass +class PostEventFnArgsT: + """A specification of the arguments to a post-event function. + + Attributes: + event_statement: The event that resulted from the player's choice. + current_scene_type: The type of the current scene. + players: Sequence of player objects. + memory: The game master's associative memory. + """ + + event_statement: str + current_scene_type: str + players: PlayersT + memory: MemoryT + + +class TriggeredFunction(component.Component): + """A component to modify inventories based on events.""" + + def __init__( + self, + memory: MemoryT, + players: PlayersT, + clock_now: Callable[[], datetime.datetime], + pre_event_fn: Callable[[PreEventFnArgsT], None] | None = None, + post_event_fn: Callable[[PostEventFnArgsT], None] | None = None, + name: str = ' \n', + verbose: bool = False, + ): + """Initialize a component to track how events change inventories. + + Args: + memory: an associative memory + players: sequence of players who have an inventory and will observe it. + clock_now: Function to call to get current time. + pre_event_fn: function to call with the action attempt before + computing the event. + post_event_fn: function to call with the event statement. + name: the name of this component e.g. Possessions, Account, Property, etc + verbose: whether to print the full update chain of thought or not + """ + self._verbose = verbose + + self._memory = memory + self._name = name + self._clock_now = clock_now + + self._pre_event_fn = pre_event_fn + self._post_event_fn = post_event_fn + self._players = players + + self._current_scene = current_scene.CurrentScene( + name='current scene type', + memory=self._memory, + clock_now=self._clock_now, + verbose=self._verbose, + ) + + def name(self) -> str: + """Returns the name of this component.""" + return self._name + + def state(self) -> str: + return '' + + def update(self) -> None: + self._current_scene.update() + + def update_before_event(self, player_action_attempt: str) -> None: + if self._pre_event_fn is None: + return + player_name, choice = player_action_attempt.split(': ') + if player_name not in [player.name for player in self._players]: + return + current_scene_type = self._current_scene.state() + self._pre_event_fn( + PreEventFnArgsT(player_name=player_name, + player_choice=choice, + current_scene_type=current_scene_type, + players=self._players, + memory=self._memory) + ) + + def update_after_event(self, event_statement: str) -> None: + if self._post_event_fn is None: + return + current_scene_type = self._current_scene.state() + self._post_event_fn( + PostEventFnArgsT(event_statement=event_statement, + current_scene_type=current_scene_type, + players=self._players, + memory=self._memory) + ) diff --git a/concordia/components/game_master/triggered_inventory_effect.py b/concordia/components/game_master/triggered_inventory_effect.py new file mode 100644 index 00000000..4ef0f057 --- /dev/null +++ b/concordia/components/game_master/triggered_inventory_effect.py @@ -0,0 +1,130 @@ +# Copyright 2023 DeepMind Technologies Limited. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A component to modify inventories based on events.""" + +from collections.abc import Callable, Sequence +import dataclasses +import datetime + +from concordia.agents import basic_agent +from concordia.agents import entity_agent +from concordia.associative_memory import associative_memory +from concordia.components.game_master import current_scene +from concordia.components.game_master import inventory as inventory_gm_component +from concordia.typing import component + +MemoryT = associative_memory.AssociativeMemory +PlayerT = basic_agent.BasicAgent | entity_agent.EntityAgent +PlayersT = Sequence[PlayerT] +InventoryT = inventory_gm_component.Inventory + + +@dataclasses.dataclass +class PreEventFnArgsT: + """A specification of the arguments to a pre-event function. + + Attributes: + player_name: The name of the player. + player_choice: The choice of the player on the current timestep. + current_scene_type: The type of the current scene. + inventory_component: The inventory component where amounts of items are + stored. + memory: The game master's associative memory. + player: Player object for the acting player. + """ + + player_name: str + player_choice: str + current_scene_type: str + inventory_component: InventoryT + memory: MemoryT + player: PlayerT + + +def _get_player_by_name(player_name: str, players: PlayersT) -> PlayerT | None: + """Get a player object by name. Assumes no duplicate names.""" + for player in players: + if player.name == player_name: + return player + return None + + +class TriggeredInventoryEffect(component.Component): + """A component to modify inventories based on events.""" + + def __init__( + self, + function: Callable[[PreEventFnArgsT], None], + inventory: inventory_gm_component.Inventory, + memory: associative_memory.AssociativeMemory, + players: PlayersT, + clock_now: Callable[[], datetime.datetime], + name: str = ' \n', + verbose: bool = False, + ): + """Initialize a component to track how events change inventories. + + Args: + function: user-provided function that can modify the inventory based on + an action attempt. + inventory: the inventory component to use to get the inventory of players. + memory: an associative memory + players: sequence of players who can trigger an inventory event. + clock_now: Function to call to get current time. + name: the name of this component e.g. Possessions, Account, Property, etc + verbose: whether to print the full update chain of thought or not + """ + self._verbose = verbose + + self._memory = memory + self._name = name + self._clock_now = clock_now + + self._function = function + self._inventory = inventory + self._players = players + + self._current_scene = current_scene.CurrentScene( + name='current scene type', + memory=self._memory, + clock_now=self._clock_now, + verbose=self._verbose, + ) + + def name(self) -> str: + """Returns the name of this component.""" + return self._name + + def state(self) -> str: + return '' + + def update(self) -> None: + self._current_scene.update() + + def update_before_event(self, player_action_attempt: str) -> None: + player_name, choice = player_action_attempt.split(': ') + if player_name not in [player.name for player in self._players]: + return + current_scene_type = self._current_scene.state() + player = _get_player_by_name(player_name, self._players) + self._function( + PreEventFnArgsT( + player_name=player_name, + player_choice=choice, + current_scene_type=current_scene_type, + inventory_component=self._inventory, + memory=self._memory, + player=player) + )