Experimental#

This namespace contains experimental features. These are under development, and their API is not necessarily stable.

Continuous Space#

A Continuous Space class.

class ContinuousSpace(dimensions: ArrayLike, torus: bool = False, random: Random | None = None, n_agents: int = 100)[source]#

Continuous space where each agent can have an arbitrary position.

Create a new continuous space.

Parameters:

  • dimensions – a numpy array like object where each row specifies the minimum and maximum value of that dimension.

  • torus – boolean for whether the space wraps around or not

  • random – a seeded stdlib random.Random instance

  • n_agents – the expected number of agents in the space

Internally, a numpy array is used to store the positions of all agents. This is resized if needed, but you can control the initial size explicitly by passing n_agents.

property agents: AgentSet#

Return an AgentSet with the agents in the space.

calculate_difference_vector(point: ndarray, agents=None) → ndarray[source]#

Calculate the difference vector between the point and all agenents.

Parameters:

  • point – the point to calculate the difference vector for

  • agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

calculate_distances(point: ArrayLike, agents: Iterable[Agent] | None = None, **kwargs) → tuple[ndarray, list][source]#

Calculate the distance between the point and all agents.

Parameters:

  • point – the point to calculate the difference vector for

  • agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

  • kwargs – any additional keyword arguments are passed to scipy’s cdist, which is used only if torus is False. This allows for non-Euclidian distance measures.

get_agents_in_radius(point: ArrayLike, radius: float | int = 1) → tuple[list, ndarray][source]#

Return the agents and their distances within a radius for the point.

get_k_nearest_agents(point: ArrayLike, k: int = 1) → tuple[list, ndarray][source]#

Return the k nearest agents and their distances to the point.

Notes

This method returns exactly k agents, ignoring ties. In case of ties, the earlier an agent is inserted the higher it will rank.

If fewer than k agents are present in the space, all agents are returned and a UserWarning is emitted to indicate that the requested k could not be satisfied. If the space is empty or k <= 0, an empty result is returned without a warning.

in_bounds(point: ArrayLike) → bool[source]#

Check if point is inside the bounds of the space.

torus_correct(point: ArrayLike) → ndarray[source]#

Apply a torus correction to the point.

Continuous space agents.

class HasPositionProtocol(*args, **kwargs)[source]#

Protocol for continuous space position holders.

class ContinuousSpaceAgent(space: ContinuousSpace, model)[source]#

A continuous space agent.

space#

the continuous space in which the agent is located

Type:

ContinuousSpace

position#

the position of the agent

Type:

np.ndarray

Initialize a continuous space agent.

Parameters:

  • space – the continuous space in which the agent is located

  • model – the model to which the agent belongs

property position: ndarray#

Position of the agent.

remove() → None[source]#

Remove and delete the agent from the model and continuous space.

get_neighbors_in_radius(radius: float | int = 1) → tuple[list, ndarray][source]#

Get neighbors within radius.

Parameters:

radius – radius within which to look for neighbors

get_nearest_neighbors(k: int = 1) → tuple[list, ndarray][source]#

Get neighbors within radius.

Parameters:

k – the number of nearest neighbors to return

Continuous Space#

A Continuous Space class.

class ContinuousSpace(dimensions: ArrayLike, torus: bool = False, random: Random | None = None, n_agents: int = 100)[source]#

Continuous space where each agent can have an arbitrary position.

Create a new continuous space.

Parameters:

  • dimensions – a numpy array like object where each row specifies the minimum and maximum value of that dimension.

  • torus – boolean for whether the space wraps around or not

  • random – a seeded stdlib random.Random instance

  • n_agents – the expected number of agents in the space

Internally, a numpy array is used to store the positions of all agents. This is resized if needed, but you can control the initial size explicitly by passing n_agents.

property agents: AgentSet#

Return an AgentSet with the agents in the space.

calculate_difference_vector(point: ndarray, agents=None) → ndarray[source]#

Calculate the difference vector between the point and all agenents.

Parameters:

  • point – the point to calculate the difference vector for

  • agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

calculate_distances(point: ArrayLike, agents: Iterable[Agent] | None = None, **kwargs) → tuple[ndarray, list][source]#

Calculate the distance between the point and all agents.

Parameters:

  • point – the point to calculate the difference vector for

  • agents – the agents to calculate the difference vector of point with. By default, all agents are considered.

  • kwargs – any additional keyword arguments are passed to scipy’s cdist, which is used only if torus is False. This allows for non-Euclidian distance measures.

get_agents_in_radius(point: ArrayLike, radius: float | int = 1) → tuple[list, ndarray][source]#

Return the agents and their distances within a radius for the point.

get_k_nearest_agents(point: ArrayLike, k: int = 1) → tuple[list, ndarray][source]#

Return the k nearest agents and their distances to the point.

Notes

This method returns exactly k agents, ignoring ties. In case of ties, the earlier an agent is inserted the higher it will rank.

If fewer than k agents are present in the space, all agents are returned and a UserWarning is emitted to indicate that the requested k could not be satisfied. If the space is empty or k <= 0, an empty result is returned without a warning.

in_bounds(point: ArrayLike) → bool[source]#

Check if point is inside the bounds of the space.

torus_correct(point: ArrayLike) → ndarray[source]#

Apply a torus correction to the point.

Continuous space agents.

class HasPositionProtocol(*args, **kwargs)[source]#

Protocol for continuous space position holders.

class ContinuousSpaceAgent(space: ContinuousSpace, model)[source]#

A continuous space agent.

space#

the continuous space in which the agent is located

Type:

ContinuousSpace

position#

the position of the agent

Type:

np.ndarray

Initialize a continuous space agent.

Parameters:

  • space – the continuous space in which the agent is located

  • model – the model to which the agent belongs

property position: ndarray#

Position of the agent.

remove() → None[source]#

Remove and delete the agent from the model and continuous space.

get_neighbors_in_radius(radius: float | int = 1) → tuple[list, ndarray][source]#

Get neighbors within radius.

Parameters:

radius – radius within which to look for neighbors

get_nearest_neighbors(k: int = 1) → tuple[list, ndarray][source]#

Get neighbors within radius.

Parameters:

k – the number of nearest neighbors to return

Scenarios#

Base Scenario class.

rescale_samples(samples: ndarray, ranges: ndarray, *, inplace: bool = False) → ndarray[source]#

Rescale samples from the unit interval [0, 1] to parameter ranges.

Parameters:

  • samples (ndarray (n, d)) – Samples drawn from the unit interval.

  • ranges (ndarray (d, 2)) – Parameter ranges given as [[min, max], …].

  • inplace (bool, optional) – If True, the input samples array is modified in place. If False (default), a new array containing the rescaled samples is returned.

  • Returns

  • -------

  • (n (ndarray) – Rescaled samples.

  • d) – Rescaled samples.

  • Notes

  • -----

  • inplace=True (The rescaling is performed using NumPy broadcasting. If)

:param : :param the original samples array is overwritten.:

class Scenario(*, rng: Generator | BitGenerator | int | integer | Sequence[int] | SeedSequence | None = None, scenario_id: int | None = None, replication_id: int = -1, **kwargs)[source]#

A Scenario class for defining model parameters and experiments.

Supports both simple instantiation and type-hinted subclassing:

# Simple usage scenario = Scenario(rng=42, density=0.8, minority_pc=0.5)

# Type-hinted subclass (recommended for complex models) class MyScenario(Scenario):

citizen_density: float = 0.7 cop_vision: int = 7 movement: bool = True

scenario = MyScenario(rng=42, cop_vision=10) # Override defaults

scenario_id#

A unique identifier for this scenario, auto-generated starting from 0

experiment_id#

Identifies the design point (e.g., row in a QMC sample matrix)

replication_id#

Identifies the stochastic replication within a design point

rng#

Random number generator seed value

Notes

All parameters are accessible via attribute access (scenario.param). Class-level attributes in subclasses serve as default values. Scenario instances are frozen after initialisation; parameters cannot be modified. To create replications with derived seeds, use spawn_replications().

Initialize a Scenario.

Parameters:

  • rng – Seed for the random number generator. Accepts any value accepted by numpy.random.default_rng(). scenario.rng is always a Generator after initialisation. The initial seed sequence is stored in scenario.seed_sequence and used by spawn_replications() to derive child seeds.

  • scenario_id – Index of the design point in the experiment matrix.

  • replication_id – Index of the stochastic replication for this design point. defaults to PARENT_REPLICATION_ID (-1). It is advised to use spawn_replications() to create replications for a given scenario, rather than handling this yourself.

  • **kwargs – All other scenario parameters (override class-level defaults).

to_dict() → dict[str, Any][source]#

Return dict representation of the scenario.

Seed_sequence_entropy and seed_sequence_spawn_key together can be used to reconstruct the SeedSequence.

spawn_replications(n: int) → list[Scenario][source]#

Spawn n replications of this scenario with deterministically derived seeds.

Each replication has identical user provided parameters but a unique random number generator and replication_id. The rng is spawned from the original rng of the base scenario instance. Indexing for replication_id starts from 0. This assumes that the instance on which this is called has replication_id == PARENT_REPLICATION_ID. If this assumption does not hold, a user warning is issued.

Parameters:

n – Number of replications to create.

Returns:

A list of n Scenario instances with replication_id 0..n-1.

classmethod from_dataframe(experiments: DataFrame, *, rng: int | integer | Sequence[int] | SeedSequence | None = None, replications: int | None = None) → list[Scenario][source]#

Turn a dataframe into a list of scenarios.

Parameters:

  • experiments – Dataframe containing the parameters for the scenarios.

  • rng – the number of random seeds to use or a list of seeds.

  • replications – the number of replications to create for each scenario

Returns:

a list of scenario instances

If rng is an integer, numpy will be used to generate that many seed values.

classmethod from_ndarray(experiments: ndarray, parameter_names: list[str], *, rng: int | integer | Sequence[int] | SeedSequence | None = None, replications: int | None = None) → list[Scenario][source]#

Turn a numpy array into a list of scenarios.

Parameters:

  • experiments – Dataframe containing the parameters for the scenarios.

  • parameter_names – the names of the parameters

  • rng – the number of random seeds to use or a list of seeds.

  • replications – the number of replications to create for each scenario

Returns:

a list of scenario instances

If rng is an integer, numpy will be used to generate that many seed values.

Classes for running parameter sweeps over scenarios.

class RunConfiguration(model_class: type[Model], until: float | int, model_args: None | list[Any] = None, model_kwargs: None | dict[str, Any] = None, outcomes: None | str | list[str] = None, data_recorder_attr_name='data_recorder')[source]#

Defines how a single Scenario is executed and what is extracted from it.

Can be used as is for simple use cases or subclassed by overriding one or more of the following methods

  • instantiate_model — construct a Model from a Scenario (default: model_class(*model_args, scenario=scenario, **model_kwargs)).

  • run_model — advance the model. Default delegates to model.run_until based on the until attribute. Override for alternative run control

  • extract_output — return a dict with outcome names as key and dataframes as values

Stopping is the model’s responsibility. RunConfiguration only chooses which run primitive to call.

Initialize a RunConfiguration object.

Parameters:

  • model_class – the model class to instantiate

  • until – until which time to run the model

  • model_args – any additional model arguments

  • model_kwargs – any additional model keyword arguments

  • outcomes – the outcomes to extract. If None, extract all outcomes.

  • data_recorder_attr_name – the name of the data recorder attribute to use on the model

instantiate_model(scenario: Scenario) → Model[source]#

Instantiate the model.

run_model(model: Model) → None[source]#

Run the model.

extract_output(model: Model) → dict[str, pd.DataFrame][source]#

Extract output from model.

run_scenarios(scenarios: Iterable[Scenario], config: RunConfiguration, *, executor: Executor | None = None, store: Store | None = None, progress: bool = True) → Store[source]#

Run the scenarios and return a Store object.

Parameters:

  • scenarios – an iterable of scenarios to run Scenarios to execute. For replications, construct these via MyScenario.from_dataframe(df, replications=n) — replication is handled at scenario construction, not here.

  • config – a RunConfiguration instance Per-scenario execution unit. Must be picklable when using a distributed executor (e.g., ProcessPoolExecutor).

  • executor – an executor to run the scenarios Execution backend. If None, scenarios run sequentially in the calling thread (useful for debugging and small experiments). Otherwise, pass a user-constructed executor; its lifetime is the caller’s responsibility (use a with block).

  • store – the Storage backend to use

  • progress – whether to display the progress Display a progress bar via tqdm if installed.

Returns: a Store instance