agentsociety.environment.economy.econ_client

Module Contents

Classes

EconomyEntityType

EconomyClient

Client side of Economy service.

Functions

_create_aio_channel

Create a gRPC asynchronous channel.

Data

__all__

API

agentsociety.environment.economy.econ_client.__all__

[‘EconomyClient’]

class agentsociety.environment.economy.econ_client.EconomyEntityType

Bases: int, enum.Enum

Unspecified

0

Agent

1

Bank

2

Firm

3

Government

4

NBS

5

agentsociety.environment.economy.econ_client._create_aio_channel(server_address: str, secure: bool = False) → grpc.aio.Channel

Create a gRPC asynchronous channel.

  • Args:

    • server_address (str): The address of the server to connect to.

    • secure (bool, optional): Whether to use a secure connection. Defaults to False.

  • Returns:

    • grpc.aio.Channel: A gRPC asynchronous channel for making RPC calls.

  • Raises:

    • ValueError: If a secure channel is requested but the server address starts with http://.

  • Description:

    • This function creates and returns a gRPC asynchronous channel based on the provided server address and security flag.

    • It ensures that if secure=True, then the server address does not start with http://.

    • If the server address starts with https://, it will automatically switch to a secure connection even if secure=False.

class agentsociety.environment.economy.econ_client.EconomyClient(server_address: str)

Client side of Economy service.

  • Description:

    • This class serves as a client interface to interact with the Economy Simulator via gRPC.

    • It establishes an asynchronous connection and provides methods to communicate with the service.

Initialization

Initialize the EconomyClient.

  • Args:

    • server_address (str): The address of the Economy server to connect to.

  • Attributes:

    • server_address (str): The address of the Economy server.

    • _aio_stub (OrgServiceStub): A gRPC stub used to make remote calls to the Economy service.

  • Description:

    • Initializes the EconomyClient with the specified server address and security preference.

    • Creates an asynchronous gRPC channel using _create_aio_channel.

    • Instantiates a gRPC stub (_aio_stub) for interacting with the Economy service.

get_log_list()
clear_log_list()
__getstate__()

Copy the object’s state from self.dict which contains all our instance attributes. Always use the dict.copy() method to avoid modifying the original state.

__setstate__(state)

Restore instance attributes (i.e., filename and mode) from the unpickled state dictionary.

_get_request_type(id: Union[int, list[int]]) → agentsociety.environment.economy.econ_client.EconomyEntityType
async get_ids()

Get the ids of agents and orgs

set_ids(citizen_ids: set[int], firm_ids: set[int], bank_ids: set[int], nbs_ids: set[int], government_ids: set[int])

Set the ids of agents and orgs

async _get_agent(id: Union[list[int], int]) → Union[dict[str, Any], list[dict[str, Any]]]

Get agent by id

  • Args:

    • id (Union[list[int], int]): The id of the agent.

  • Returns:

    • economyv2.Agent: The agent object.

async _get_bank(id: Union[list[int], int]) → Union[dict[str, Any], list[dict[str, Any]]]

Get bank by id

  • Args:

    • id (Union[list[int], int]): The id of the bank.

  • Returns:

    • economyv2.Bank: The bank object.

async _get_firm(id: Union[list[int], int]) → Union[dict[str, Any], list[dict[str, Any]]]

Get firm by id

  • Args:

    • id (Union[list[int], int]): The id of the firm.

  • Returns:

    • economyv2.Firm: The firm object.

async _get_government(id: Union[list[int], int]) → Union[dict[str, Any], list[dict[str, Any]]]

Get government by id

  • Args:

    • id (Union[list[int], int]): The id of the government.

  • Returns:

    • economyv2.Government: The government object.

async _get_nbs(id: Union[list[int], int]) → Union[dict[str, Any], list[dict[str, Any]]]

Get nbs by id

  • Args:

    • id (Union[list[int], int]): The id of the nbs.

  • Returns:

    • economyv2.Nbs: The nbs object.

async get(id: Union[list[int], int], key: Union[list[str], str]) → Any

Get specific value

  • Args:

    • id (Union[list[int], int]): The id of Org or Agent.

    • key (str): The attribute to fetch.

  • Returns:

    • Any

_merge(original_dict: dict[str, Any], key: Any, value: Any) → bool
async update(id: Union[list[int], int], key: str, value: Union[Any, list[Any]], mode: Union[Literal[replace], Literal[merge]] = 'replace') → Any

Update key-value pair

  • Args:

    • id (Union[list[int], int]): The id of Org or Agent.

    • key (str): The attribute to update.

    • value (Union[Any, list[Any]]): The value to update.

    • mode (Union[Literal[“replace”], Literal[“merge”]], optional): Update mode. Defaults to “replace”.

  • Returns:

    • Any

async add_agents(configs: Union[list[dict], dict])

Add one or more agents to the economy system.

  • Args:

    • configs (Union[list[dict], dict]): A single configuration dictionary or a list of dictionaries, each containing the necessary information to create an agent (e.g., id, currency).

  • Returns:

    • The method does not explicitly return any value but gathers the responses from adding each agent.

  • Description:

    • If a single configuration dictionary is provided, it is converted into a list.

    • For each configuration in the list, a task is created to asynchronously add an agent using the provided configuration.

    • All tasks are executed concurrently, and their results are gathered and returned.

async add_orgs(configs: Union[list[dict], dict])

Add one or more organizations to the economy system.

  • Args:

    • configs (Union[List[Dict], Dict]): A single configuration dictionary or a list of dictionaries, each containing the necessary information to create an organization (e.g., id, type, nominal_gdp, etc.).

  • Returns:

    • List: A list of responses from adding each organization.

  • Raises:

    • KeyError: If a required field is missing from the config dictionary.

  • Description:

    • Ensures configs is always a list, even if only one config is provided.

    • For each configuration in the list, creates a task to asynchronously add an organization using the provided configuration.

    • Executes all tasks concurrently and gathers their results.

async calculate_taxes_due(org_id: int, agent_ids: list[int], incomes: list[float], enable_redistribution: bool)

Calculate the taxes due for agents based on their incomes.

  • Args:

    • org_id (int): The ID of the government organization.

    • agent_ids (List[int]): A list of IDs for the agents whose taxes are being calculated.

    • incomes (List[float]): A list of income values corresponding to each agent.

    • enable_redistribution (bool): Flag indicating whether redistribution is enabled.

  • Returns:

    • Tuple[float, List[float]]: A tuple containing the total taxes due and updated incomes after tax calculation.

async calculate_consumption(org_ids: Union[int, list[int]], agent_id: int, demands: list[int], consumption_accumulation: bool = True)

Calculate consumption for agents based on their demands.

  • Args:

    • org_ids (Union[int, list[int]]): The ID of the firm providing goods or services.

    • agent_id (int): The ID of the agent whose consumption is being calculated.

    • demands (List[int]): A list of demand quantities corresponding to each agent.

    • consumption_accumulation (bool): Weather accumulation.

  • Returns:

    • Tuple[int, List[float]]: A tuple containing the remaining inventory and updated currencies for each agent.

async calculate_real_gdp(nbs_id: int)
async calculate_interest(bank_id: int, agent_ids: list[int])

Calculate interest for agents based on their accounts.

  • Args:

    • bank_id (int): The ID of the bank.

    • agent_ids (List[int]): A list of IDs for the agents whose interests are being calculated.

  • Returns:

    • Tuple[float, List[float]]: A tuple containing the total interest and updated currencies for each agent.

async remove_agents(agent_ids: Union[int, list[int]])

Remove one or more agents from the system.

  • Args:

    • org_ids (Union[int, List[int]]): A single ID or a list of IDs for the agents to be removed.

async remove_banks(bank_ids: Union[int, list[int]])

Remove one or more banks from the system.

  • Args:

    • bank_ids (Union[int, List[int]]): A single ID or a list of IDs for the banks to be removed.

async remove_firms(firm_ids: Union[int, list[int]])

Remove one or more firms from the system.

  • Args:

    • firm_ids (Union[int, List[int]]): A single ID or a list of IDs for the firms to be removed.

async remove_governments(government_ids: Union[int, list[int]])

Remove one or more governments from the system.

  • Args:

    • government_ids (Union[int, List[int]]): A single ID or a list of IDs for the governments to be removed.

async remove_nbs(nbs_ids: Union[int, list[int]])

Remove one or more nbs from the system.

  • Args:

    • nbs_id (Union[int, List[int]]): A single ID or a list of IDs for the NBS to be removed.

async save(file_path: str) → tuple[list[int], list[int], list[int], list[int], list[int]]

Save the current state of all economy entities to a specified file.

  • Args:

    • file_path (str): The path to the file where the economy entities will be saved.

  • Returns:

    • Tuple[list[int], list[int], list[int], list[int], list[int]]: A tuple containing lists of agent IDs and organization IDs that were saved.

async load(file_path: str)

Load the state of economy entities from a specified file.

  • Args:

    • file_path (str): The path to the file from which the economy entities will be loaded.

  • Returns:

    • Tuple[list[int], list[int], list[int], list[int], list[int]]: A tuple containing lists of agent IDs and organization IDs that were loaded.

async get_bank_ids() → list[int]

Get the IDs of bank entities.

  • Returns:

    • List[int]: A list of bank IDs.

async get_nbs_ids() → list[int]

Get the IDs of NBS entities.

  • Returns:

    • List[int]: A list of NBS IDs.

async get_government_ids() → list[int]

Get the IDs of government entities.

  • Returns:

    • List[int]: A list of government IDs.

async get_firm_ids() → list[int]

Get the IDs of firm entities.

  • Returns:

    • List[int]: A list of firm IDs.

async delta_update_bank(bank_id: int, delta_interest_rate: float = 0.0, delta_currency: float = 0.0, add_citizen_ids: list[int] = [], remove_citizen_ids: list[int] = [])

Incrementally update a bank’s properties.

  • Args:

    • bank_id (int): The ID of the bank to update.

    • delta_interest_rate (float, optional): Change in interest rate, can be positive or negative. Defaults to 0.0.

    • delta_currency (float, optional): Change in currency amount, can be positive or negative. Defaults to 0.00.

    • add_citizen_ids (list[int], optional): List of citizen IDs to add to the bank. Defaults to empty list.

    • remove_citizen_ids (list[int], optional): List of citizen IDs to remove from the bank. Defaults to empty list.

  • Returns: This method does not return any value.

  • Description:

    • This method sends a request to the economy service via gRPC to incrementally update a bank’s properties.

    • Can update interest rate, currency amount, and associated citizens simultaneously.

    • All parameters are optional except bank_id - only provide values for properties you want to change.

async delta_update_nbs(nbs_id: int, delta_nominal_gdp: Optional[dict[str, float]] = None, delta_real_gdp: Optional[dict[str, float]] = None, delta_unemployment: Optional[dict[str, float]] = None, delta_wages: Optional[dict[str, float]] = None, delta_prices: Optional[dict[str, float]] = None, delta_working_hours: Optional[dict[str, float]] = None, delta_depression: Optional[dict[str, float]] = None, delta_consumption_currency: Optional[dict[str, float]] = None, delta_income_currency: Optional[dict[str, float]] = None, delta_locus_control: Optional[dict[str, float]] = None, delta_currency: float = 0.0, add_citizen_ids: Optional[list[int]] = None, remove_citizen_ids: Optional[list[int]] = None)

Incrementally update a National Bureau of Statistics (NBS) entity’s properties.

  • Args:

    • nbs_id (int): The ID of the NBS entity to update.

    • delta_nominal_gdp (dict[str,float]): Changes to nominal GDP metrics by category.

    • delta_real_gdp (dict[str,float]): Changes to real GDP metrics by category.

    • delta_unemployment (dict[str,float]): Changes to unemployment metrics by category.

    • delta_wages (dict[str,float]): Changes to wage metrics by category.

    • delta_prices (dict[str,float]): Changes to price metrics by category.

    • delta_working_hours (dict[str,float]): Changes to working hours metrics by category.

    • delta_depression (dict[str,float]): Changes to depression metrics by category.

    • delta_consumption_currency (dict[str,float]): Changes to consumption currency metrics by category.

    • delta_income_currency (dict[str,float]): Changes to income currency metrics by category.

    • delta_locus_control (dict[str,float]): Changes to locus of control metrics by category.

    • delta_currency (float): Change in currency amount, can be positive or negative.

    • add_citizen_ids (list[int]): List of citizen IDs to add to the NBS entity.

    • remove_citizen_ids (list[int]): List of citizen IDs to remove from the NBS entity.

  • Returns: This method does not return any value.

  • Description:

    • This method sends a request to the economy service via gRPC to incrementally update an NBS entity’s properties.

    • Each delta parameter represents changes to be applied to the corresponding statistic tracked by the NBS.

    • Dictionary parameters contain category-specific changes where keys are categories and values are the delta amounts.

    • Can update multiple statistical metrics and associated citizens simultaneously.

async delta_update_government(government_id: int, delta_bracket_cutoffs: Optional[list[float]] = None, delta_bracket_rates: Optional[list[float]] = None, delta_currency: float = 0.0, add_citizen_ids: Optional[list[int]] = None, remove_citizen_ids: Optional[list[int]] = None)

Incrementally update a Government entity’s properties.

  • Args:

    • government_id (int): The ID of the Government entity to update.

    • delta_bracket_cutoffs (list[float]): Changes to tax bracket cutoff values.

    • delta_bracket_rates (list[float]): Changes to tax bracket rate values.

    • delta_currency (float): Change in currency amount, can be positive or negative.

    • add_citizen_ids (list[int]): List of citizen IDs to add to the Government entity.

    • remove_citizen_ids (list[int]): List of citizen IDs to remove from the Government entity.

  • Returns: This method does not return any value.

  • Description:

    • This method sends a request to the economy service via gRPC to incrementally update a Government entity’s properties.

    • Can update tax bracket information, currency amount, and associated citizens simultaneously.

    • The delta parameters represent changes to be applied to the corresponding values currently stored in the Government entity.

async delta_update_firms(firm_id: Union[int, list[int]], delta_price: Optional[Union[float, list[float]]] = None, delta_inventory: Optional[Union[int, list[int]]] = None, delta_demand: Optional[Union[float, list[float]]] = None, delta_sales: Optional[Union[float, list[float]]] = None, delta_currency: Optional[Union[float, list[float]]] = None, add_employee_ids: Optional[Union[list[int], list[list[int]]]] = None, remove_employee_ids: Optional[Union[list[int], list[list[int]]]] = None)

Incrementally update one or more firms’ properties.

Args: firm_id: Single firm ID or list of firm IDs delta_price: Change in price(s) delta_inventory: Change in inventory level(s) delta_demand: Change in demand level(s) delta_sales: Change in sales amount(s) delta_currency: Change in currency amount(s) add_employee_ids: Employee IDs to add (single list or list of lists) remove_employee_ids: Employee IDs to remove (single list or list of lists)

async delta_update_agents(agent_id: Union[int, list[int]], new_firm_id: Optional[Union[int, list[int]]] = None, delta_currency: Optional[Union[float, list[float]]] = None, delta_skill: Optional[Union[float, list[float]]] = None, delta_consumption: Optional[Union[float, list[float]]] = None, delta_income: Optional[Union[float, list[float]]] = None)

Update agent attributes with delta values.

_prepare_parameter(param: Any, length: int) → list[Any]

Helper method to prepare parameters for batch processing.