"""Agent related classes.
Core Objects: Agent.
"""
# Postpone annotation evaluation to avoid NameError from forward references (PEP 563). Remove once Python 3.14+ is required.
from __future__ import annotations
import itertools
from collections.abc import Iterator
from random import Random
from typing import TYPE_CHECKING, ClassVar
import numpy as np
import pandas as pd
if TYPE_CHECKING:
from modelly.core.model import Model
from modelly.core.agentset import AgentSet, _resolve_per_agent_values
from modelly.core.column_access import collect_column_specs, install_column_properties
from modelly.core.columns import Column
[docs]
class Agent[M: Model]:
"""Base class for a model agent in Modelly.
Attributes:
model (Model): A reference to the model instance.
unique_id (int): A unique identifier for this agent.
Notes:
Agents must be hashable to be used in an AgentSet.
In Python 3, defining `__eq__` without `__hash__` makes an object unhashable,
which will break AgentSet usage.
unique_id is unique relative to a model instance and starts from 1
"""
_datasets: ClassVar = set()
_column_specs: ClassVar[dict[str, Column]] = {}
_column_row: int = -1 # instance attribute once the agent has a column row
# True when no class after Agent in the MRO defines __init__, so the
# cooperative super().__init__() would only reach object.__init__.
_plain_init: ClassVar[bool] = True
# True when the class has columns or datasets to update on birth and death.
_extra_hooks: ClassVar[bool] = False
_repr_excluded_fields: ClassVar[set[str]] = {
"model",
"current_action",
"unique_id",
"_column_table",
"_column_row",
}
def __init_subclass__(cls, **kwargs):
"""Called when Agent is subclassed, giving each subclass its own dataset set."""
super().__init_subclass__(**kwargs)
# Each subclass gets its own dataset set
# we use strings on this to avoid memory leaks
# and ensure the retrieved dataset belongs to the same
# model instance as the agent
cls._datasets = set()
# Columns declared on this class or inherited, in MRO order; each one is
# replaced on the class by a fast property (see modelly.core.column_access).
specs = collect_column_specs(cls)
cls._column_specs = specs
install_column_properties(cls, specs)
after_agent = cls.__mro__[cls.__mro__.index(Agent) + 1 : -1]
cls._plain_init = not any("__init__" in vars(base) for base in after_agent)
cls._extra_hooks = bool(specs)
def __init__(self, model: M, *args, **kwargs) -> None:
"""Create a new agent.
Args:
model (Model): The model instance in which the agent exists.
args: Passed on to super.
kwargs: Passed on to super.
Notes:
to make proper use of python's super, in each class remove the arguments and
keyword arguments you need and pass on the rest to super
"""
if args or kwargs:
super().__init__(*args, **kwargs)
elif not self._plain_init:
super().__init__() # a cooperative base class needs the call
self.model: M = model
if model._quiet:
# Model.register_agent written out for the common case (default
# registry, nobody listening): a birth is the hottest path in
# lifecycle-heavy models. Kept in sync with it (tested).
model._all_agents[self] = None
self.unique_id = model.agent_id_counter
model.agent_id_counter += 1
try:
model._agents_by_type[type(self)][self] = None
except KeyError:
model._register_new_type(self)
else:
self.unique_id = None
model.register_agent(self)
if self._extra_hooks:
if self._column_specs:
model.columns.register(self)
for dataset in self._datasets:
model.data_registry[dataset].add_agent(self)
[docs]
@classmethod
def column_specs(cls) -> dict[str, Column]:
"""The ``Column`` declarations of this class (name to column), in table order.
On the class, a column attribute is the fast property that reads the
agent's row; this returns the declarations themselves (dtype, default).
"""
return dict(cls._column_specs)
[docs]
def remove(self) -> None:
"""Remove and delete the agent from the model."""
# try/except rather than contextlib.suppress: the context manager costs
# about 300 ns per call and this runs on every agent death.
model = self.model
try:
if model._quiet:
# Model.deregister_agent written out; see __init__.
del model._agents_by_type[type(self)][self]
del model._all_agents[self]
else:
model.deregister_agent(self)
except KeyError:
pass # already removed
if self._extra_hooks:
if self._column_specs and self._column_row >= 0:
model.columns.unregister(self)
del self._column_table # its Column attributes now name the agent
# ensures models are also removed from datasets
for dataset in self._datasets:
model.data_registry[dataset].remove_agent(self)
[docs]
def step(self) -> None:
"""A single step of the agent."""
def advance(self) -> None: # noqa: D102
pass
[docs]
@classmethod
def create_agents[T: Agent](
cls: type[T], model: Model, n: int, *args, **kwargs
) -> AgentSet[T]:
"""Create N agents.
Args:
model: the model to which the agents belong
args: arguments to pass onto agent instances
each arg is either a single object or a sequence of length n
n: the number of agents to create
kwargs: keyword arguments to pass onto agent instances
each keyword arg is either a single object or a sequence of length n
Returns:
AgentSet containing the agents created.
Warning:
A list, tuple, ndarray, or pandas Series argument is treated as one
value per agent and must have length n; a length mismatch raises
ValueError. This is especially easy to hit with coordinate tuples:
create_agents(model, 2, pos=(10, 20)) does NOT give both agents
pos=(10, 20); it gives agent 0 pos=10 and agent 1 pos=20, since the
tuple's length (2) matches n (2).
To assign the same sequence value to every agent, broadcast it
explicitly as a length-n list of that value, e.g.:
create_agents(model, 2, pos=[(10, 20)] * 2)
Raises:
ValueError: If a sequence argument's length does not match n.
"""
agents = []
if not args and not kwargs:
for _ in range(n):
agents.append(cls(model))
return AgentSet(agents, random=model.random)
pos_iter, kw_keys, kw_iter = _per_agent_iters(args, kwargs, n)
# We rely on range(n) to drive the loop length
if kwargs:
for _, p_args, k_vals in zip(range(n), pos_iter, kw_iter):
agents.append(cls(model, *p_args, **dict(zip(kw_keys, k_vals))))
else:
for _, p_args in zip(range(n), pos_iter):
agents.append(cls(model, *p_args))
return AgentSet(agents, random=model.random)
[docs]
@classmethod
def from_dataframe[T: Agent](
cls: type[T], model: Model, df: pd.DataFrame, **kwargs
) -> AgentSet[T]:
"""Create agents from a pandas DataFrame.
Each row of the DataFrame represents one agent. The DataFrame columns are
mapped to the agent's constructor as keyword arguments. Additional keyword
arguments (`**kwargs`) can be used to set constant attributes for all agents.
Args:
model: The model instance.
df: The pandas DataFrame. Each row represents an agent.
**kwargs: Constant values to pass to every agent's constructor.
Only non-sequence data is allowed in kwargs to avoid ambiguity
with DataFrame columns.
Returns:
AgentSet containing the agents created.
Note:
If you need to pass variable data or sequences, add them as columns
to the DataFrame before calling this method.
"""
for key, value in kwargs.items():
if isinstance(value, (list, np.ndarray, tuple, pd.Series)):
raise TypeError(
f"from_dataframe does not support sequence data in kwargs ('{key}'). "
"Please add this data to the DataFrame before calling from_dataframe."
)
agents = [
cls(model, **{**record, **kwargs})
for record in df.to_dict(orient="records")
]
return AgentSet(agents, random=model.random)
def __str__(self) -> str:
"""Return a human-readable string representation of the agent."""
return f"{self.__class__.__name__}, agent_id = {self.unique_id}"
def __repr__(self) -> str:
"""Return an unambiguous string representation including agent state."""
# Get excluded fields (allows subclasses to override)
excluded = self._repr_excluded_fields
# Get user-defined attributes (exclude private and Modelly fields)
user_attrs = {
k: v
for k, v in self.__dict__.items()
if not k.startswith("_") and k not in excluded
}
if user_attrs:
attr_str = ", ".join(f"{k}={v!r}" for k, v in user_attrs.items())
return f"<{self.__class__.__name__} id={self.unique_id} {attr_str}>"
else:
return f"<{self.__class__.__name__} id={self.unique_id}>"
@property
def random(self) -> Random:
"""Return a seeded stdlib rng."""
return self.model.random
@property
def rng(self) -> np.random.Generator:
"""Return a seeded np.random rng."""
return self.model.rng
@property
def scenario(self):
"""Return the scenario associated with the model."""
return self.model.scenario
def _per_agent_iters(
args: tuple, kwargs: dict, n: int
) -> tuple[Iterator, list, Iterator]:
"""Return the positional iterator, keyword names and keyword iterator of ``create_agents``.
A sequence must have length n (assigned per agent); a length mismatch
raises. Anything else is broadcast.
"""
arg_iters = [_resolve_per_agent_values(arg, n) for arg in args]
kw_keys = list(kwargs.keys())
kw_val_iters = [_resolve_per_agent_values(v, n) for v in kwargs.values()]
# If arg_iters is empty, zip(*[]) returns nothing, so we use repeat(())
pos_iter = zip(*arg_iters) if arg_iters else itertools.repeat(())
kw_iter = zip(*kw_val_iters) if kw_val_iters else itertools.repeat(())
return pos_iter, kw_keys, kw_iter
