agentsociety.environment.environment

Simulator: Urban Simulator

Module Contents

Classes

EnvironmentConfig

Configuration for the simulation environment.

Environment

The environment, including map data, simulator clients, and environment variables.

EnvironmentStarter

The entrypoint of the simulator, used to initialize the simulator and map.

Functions

_generate_yaml_config

Data

__all__

POI_START_ID

API

agentsociety.environment.environment.__all__

[‘Environment’, ‘EnvironmentStarter’, ‘EnvironmentConfig’]

agentsociety.environment.environment.POI_START_ID

700000000

class agentsociety.environment.environment.EnvironmentConfig

Bases: pydantic.BaseModel

Configuration for the simulation environment.

model_config

‘ConfigDict(…)’

start_tick: int

‘Field(…)’

Starting tick of one day, in seconds

metric_interval: int

‘Field(…)’

Interval of metrics, in ticks

weather: str

‘Field(…)’

Current weather condition in the environment

temperature: str

‘Field(…)’

Current temperature in the environment

workday: bool

‘Field(…)’

Indicates if it’s a working day

other_information: str

‘Field(…)’

Additional environment information

to_prompts() → dict[str, Any]

Convert the environment config to prompts

class agentsociety.environment.environment.Environment(map_data: agentsociety.environment.mapdata.MapData, server_addr: Optional[str], environment_config: agentsociety.environment.environment.EnvironmentConfig, citizen_ids: set[int] = set(), firm_ids: set[int] = set(), bank_ids: set[int] = set(), nbs_ids: set[int] = set(), government_ids: set[int] = set())

The environment, including map data, simulator clients, and environment variables.

Initialization

Initialize the Environment.

  • Args:

    • map_data: MapData, map data

    • server_addr: str, server address

    • environment_config: EnvironmentConfig, environment config

    • citizen_ids: set[int], citizen ids

    • firm_ids: set[int], firm ids

    • bank_ids: set[int], bank ids

    • nbs_ids: set[int], nbs ids

    • government_ids: set[int], government ids

    • syncer: ray.ObjectRef, syncer for get_tick

poi_cate

None

poi categories

_log_list

[]

log list

_lock

‘Lock(…)’

lock for simulator

_tick

0

number of simulated ticks

_aoi_message: dict[int, dict[int, list[str]]]

None

aoi message

init() → Any
close() → Any

Close the Environment.

_create_poi_id_2_aoi_id()
property map
property city_client
property economy_client
property projector
get_log_list()
clear_log_list()
get_poi_cate()
get_aoi_ids()
register_aoi_message(agent_id: int, target_aoi: Union[int, list[int]], content: str)

Register aoi message

  • Args:

    • target_aoi (Union[int, list[int]]): The ID of the target aoi.

    • content (str): The content of the message to send.

cancel_aoi_message(agent_id: int, target_aoi: Union[int, list[int]])

Cancel aoi message

property environment: dict[str, str]

Get the current state of environment variables.

set_environment(environment: dict[str, str])

Set the entire dictionary of environment variables.

  • Args:

    • environment (Dict[str, str]): Key-value pairs of environment variables.

sense(key: str) → Any

Retrieve the value of an environment variable by its key.

  • Args:

    • key (str): The key of the environment variable.

  • Returns:

    • Any: The value of the corresponding key, or an empty string if not found.

sense_aoi(aoi_id: int) → str

Retrieve the value of an environment variable by its key.

update_environment(key: str, value: Any)

Update the value of a single environment variable.

  • Args:

    • key (str): The key of the environment variable.

    • value (Any): The new value to set.

get_environment() → str
get_poi_categories(center: Optional[Union[tuple[float, float], shapely.geometry.Point]] = None, radius: Optional[float] = None) → list[str]

Retrieve unique categories of Points of Interest (POIs) around a central point.

  • Args:

    • center (Optional[Union[Tuple[float, float], Point]]): The central point as a tuple or Point object. Defaults to (0, 0) if not provided.

    • radius (Optional[float]): The search radius in meters. If not provided, all POIs are considered.

  • Returns:

    • List[str]: A list of unique POI category names.

set_tick(tick: int)
get_tick() → int
get_datetime(format_time: bool = False, format: str = '%H:%M:%S') → Union[tuple[int, int], tuple[int, str]]

Get the current time of the simulator.

By default, returns the number of seconds since midnight. Supports formatted output.

  • Args:

    • format_time (bool): Whether to return the time in a formatted string. Defaults to False.

    • format (str): The format string for formatting the time. Defaults to “%H:%M:%S”.

  • Returns:

    • Union[tuple[int, int], tuple[int, str]]: The current simulation (day, time) either as an integer representing seconds since midnight or as a formatted string.

async get_person(person_id: int) → dict

Retrieve information about a specific person by ID.

  • Args:

    • person_id (int): The ID of the person to retrieve information for.

  • Returns:

    • Dict: Information about the specified person.

async add_person(dict_person: dict) → dict

Add a new person to the simulation.

  • Args:

    • dict_person (dict): The person object to add.

  • Returns:

    • Dict: Response from adding the person.

async set_aoi_schedules(person_id: int, target_positions: Union[list[Union[int, tuple[int, int]]], Union[int, tuple[int, int]]], departure_times: Optional[list[float]] = None, modes: Optional[list[pycityproto.city.trip.v2.trip_pb2.TripMode]] = None)

Set schedules for a person to visit Areas of Interest (AOIs).

  • Args:

    • person_id (int): The ID of the person whose schedule is being set.

    • target_positions (Union[List[Union[int, Tuple[int, int]]], Union[int, Tuple[int, int]]]): A list of AOI or POI IDs or tuples of (AOI ID, POI ID) that the person will visit.

    • departure_times (Optional[List[float]]): Departure times for each trip in the schedule. If not provided, current time will be used for all trips.

    • modes (Optional[List[int]]): Travel modes for each trip. Defaults to TRIP_MODE_DRIVE_ONLY if not specified.

async reset_person_position(person_id: int, aoi_id: Optional[int] = None, poi_id: Optional[int] = None, lane_id: Optional[int] = None, s: Optional[float] = None)

Reset the position of a person within the simulation.

  • Args:

    • person_id (int): The ID of the person whose position is being reset.

    • aoi_id (Optional[int]): The ID of the Area of Interest (AOI) where the person should be placed.

    • poi_id (Optional[int]): The ID of the Point of Interest (POI) within the AOI.

    • lane_id (Optional[int]): The ID of the lane on which the person should be placed.

    • s (Optional[float]): The longitudinal position along the lane.

get_around_poi(center: Union[tuple[float, float], shapely.geometry.Point], radius: float, poi_type: Union[str, list[str]]) → list[dict]

Get Points of Interest (POIs) around a central point based on type.

  • Args:

    • center (Union[Tuple[float, float], Point]): The central point as a tuple or Point object.

    • radius (float): The search radius in meters.

    • poi_type (Union[str, List[str]]): The category or categories of POIs to filter by.

  • Returns:

    • List[Dict]: A list of dictionaries containing information about the POIs found.

class agentsociety.environment.environment.EnvironmentStarter(map_config: agentsociety.environment.mapdata.MapConfig, environment_config: agentsociety.environment.environment.EnvironmentConfig, s3config: agentsociety.s3.S3Config, log_dir: str, home_dir: str)

Bases: agentsociety.environment.environment.Environment

The entrypoint of the simulator, used to initialize the simulator and map.

  • Description:

    • This class is the core of the simulator, responsible for initializing and managing the simulation environment.

    • It reads parameters from a configuration dictionary, initializes map data, and starts or connects to a simulation server as needed.

Initialization

Environment config

  • Args:

    • map_config (MapConfig): Map config

    • simulator_config (SimulatorConfig): Simulator config

    • environment_config (EnvironmentConfig): Environment config

to_init_args()
property syncer
async init()

Initialize the environment including the syncer and the simulator.

async close()

Terminate the simulation process if it’s running.

async step(n: int)
async get_metrics() → list[Tuple[str, float, int]]

Get the metrics of the environment for recording.

  • Returns:

    • List[Tuple[str, float, int]]: A list of tuples, each containing the metric name, value, and step.

agentsociety.environment.environment._generate_yaml_config(map_file) → str