Main Model

abses.core.model.MainModel

MainModel(
    parameters=DictConfig({}),
    human_class=BaseHuman,
    nature_class=BaseNature,
    run_id=None,
    seed=None,
    rng=None,
    experiment=None,
    **kwargs
)

Bases: Model, BaseStateManager

Base class of a main ABSESpy model.

A MainModel instance represents the core simulation environment that coordinates human and natural subsystems.

Attributes:

Name	Type	Description
name	str	Name of the model (defaults to lowercase class name).
settings	DictConfig	Structured parameters for all model components. Allows nested access like model.nature.params.parameter_name.
human	HumanSystemProtocol	The Human subsystem module.
nature	NatureSystemProtocol	The Nature subsystem module.
time	TimeDriver	Time driver controlling simulation progression.
params	DictConfig	Model parameters (alias: .p).
run_id	int | None	Identifier for current model run (useful in batch runs).
agents	_ModelAgentsContainer	Container for all active agents. Provides methods for agent management.
actors	ActorsListProtocol	List of all agents currently on the earth (in a PatchCell).
outpath	Path	Directory path for model outputs.
version	str	Current version of the model.
datasets	DictConfig	Available datasets (alias: .ds).
plot	DictConfig	Visualization interface for the model.

Parameters:

Name	Type	Description	Default
parameters	DictConfig	Configuration dictionary for model parameters.	DictConfig({})
human_class	Type[HumanSystemProtocol]	Class to use for human subsystem (defaults to BaseHuman).	BaseHuman
nature_class	Type[NatureSystemProtocol]	Class to use for nature subsystem (defaults to BaseNature).	BaseNature
run_id	Optional[int]	Identifier for this model run.	None
outpath		Directory path for model outputs.	required
experiment	Optional[ExperimentProtocol]	Associated experiment instance.	None
**kwargs	Optional[Any]	Additional model parameters.	{}

Raises:

Type	Description
AssertionError	If human_class or nature_class are not valid subclasses.

Source code in abses/core/model.py

def __init__(
    self,
    parameters: DictConfig = DictConfig({}),
    human_class: Type[HumanSystemProtocol] = BaseHuman,
    nature_class: Type[NatureSystemProtocol] = BaseNature,
    run_id: Optional[int] = None,
    seed: Optional[int] = None,
    rng: Optional[RNGLike | SeedLike] = None,
    experiment: Optional[ExperimentProtocol] = None,
    **kwargs: Optional[Any],
) -> None:
    """Initializes a new MainModel instance.

    Args:
        parameters: Configuration dictionary for model parameters.
        human_class: Class to use for human subsystem (defaults to BaseHuman).
        nature_class: Class to use for nature subsystem (defaults to BaseNature).
        run_id: Identifier for this model run.
        outpath: Directory path for model outputs.
        experiment: Associated experiment instance.
        **kwargs: Additional model parameters.

    Raises:
        AssertionError: If human_class or nature_class are not valid subclasses.
    """
    self._names: Set[str] = set()
    Model.__init__(self, seed=seed, rng=rng)
    BaseStateManager.__init__(self)
    self._exp = experiment
    self._run_id: Optional[int] = run_id
    # Filter out None values from kwargs for type safety
    clean_kwargs = {k: v for k, v in kwargs.items() if v is not None}
    normalized_params = normalize_config(parameters)
    apply_validation(normalized_params)

    self._settings = merge_parameters(normalized_params, **clean_kwargs)
    self._time = TimeDriver(model=self)
    self._steps: int = 0  # Initialize steps counter for type checking
    self._setup_subsystems(human_class, nature_class)
    self._agents_handler = _ModelAgentsContainer(
        model=self, max_len=kwargs.get("max_agents", None)
    )

    tracker_cfg = self._settings.get("tracker", {})
    tracker_backend = create_tracker(tracker_cfg, model=self)
    collector_cfg = prepare_collector_config(tracker_cfg)
    self.datacollector: ABSESpyDataCollector = ABSESpyDataCollector(
        reports=collector_cfg,
        tracker=tracker_backend,
        run_id=run_id,
    )

    # Setup logging BEFORE initialize() so user logs in initialize() are captured
    log_cfg = self.settings.get("log", {})
    if log_cfg:
        self._setup_logger(log_cfg)

    # Call initialize on model first
    self.initialize()
    # Then initialize subsystems
    self.do_each("_initialize", order=DEFAULT_INIT_ORDER)
    self.set_state(State.INIT)

name cached property

name

Get the model's name.

Returns:

Type	Description
str	Model name from settings, or class name if not specified.

outpath cached property

outpath

Get the model's output directory path.

Returns:

Type	Description
Path	Output path from settings, or current directory/model_name if not specified.

version cached property

version

Get the model's version string.

Returns:

Type	Description
str	Version from settings, or 'v0' if not specified.

steps property writable

steps

Get the number of steps to run.

Returns:

Type	Description
int	The configured number of simulation steps.

exp property

exp

Returns the associated experiment.

run_id property

run_id

The run id of the current model. It's useful in batch run. When running a single model, the run id is None.

settings property

settings

Structured configuration for all model components.

Allows nested parameter access. Example: If settings = {'nature': {'test': 3}}, Access via: - model.nature.params.test - model.nature.p.test

Returns:

Type	Description
DictConfig	DictConfig containing all model settings.

agents property

agents

Container managing all agents in the model.

Provides methods for: - Accessing agents: agents.select() - Creating agents: agents.new(Actor, num=3) - Registering agent types: agents.register(Actor) - Triggering events: agents.trigger()

Returns:

Type	Description
_ModelAgentsContainer	The model's agent container instance.

actors property

actors

List of all agents currently on the earth.

Returns:

Type	Description
ActorsListProtocol	ActorsList containing all agents in PatchCells.

human property

human

The Human subsystem.

nature property

nature

The Nature subsystem.

time property writable

time

The time driver & controller

params property

params

The global parameters of this model.

datasets property

datasets

Available datasets for the model.

Returns:

Type	Description
DictConfig	DictConfig containing dataset configurations.

do_each

do_each(func, order=DEFAULT_RUN_ORDER, **kwargs)

执行每个子系统

Parameters:

Name	Type	Description	Default
func	str | Callable	函数名或可调用对象	required
order	Tuple[SubSystemName, ...]	子系统顺序	DEFAULT_RUN_ORDER
**kwargs	Any	其他参数	{}

Source code in abses/core/model.py

def do_each(
    self,
    func: str | Callable,
    order: Tuple[SubSystemName, ...] = DEFAULT_RUN_ORDER,
    **kwargs: Any,
) -> Dict[SubSystemName, Any]:
    """执行每个子系统

    Args:
        func: 函数名或可调用对象
        order: 子系统顺序
        **kwargs: 其他参数
    """
    _obj = {"model": self, "nature": self.nature, "human": self.human}
    result = {}
    for name in order:
        if name not in _obj:
            raise ValueError(f"{name} is not a valid component.")
        if isinstance(func, str):
            callable_func = getattr(_obj[name], func)
        else:
            callable_func = func
        if not callable(callable_func):
            raise ValueError(f"{name}.{func} is not callable.")
        callable_func(**kwargs)
        result[name] = _obj[name]
    return result

add_name

add_name(name, check=None)

Add a name to the model's name registry with optional validation.

This method registers names for model components and can enforce uniqueness or existence checks.

Parameters:

Name	Type	Description	Default
name	str	The name to add to the registry.	required
check	Optional[HowCheckName]	Optional validation mode: - 'unique': Raise error if name already exists - 'exists': Raise error if name doesn't exist - None: No validation (default)	None

Raises:

Type	Description
ValueError	If check method is invalid, or if validation fails.

Source code in abses/core/model.py

def add_name(self, name: str, check: Optional[HowCheckName] = None) -> None:
    """Add a name to the model's name registry with optional validation.

    This method registers names for model components and can enforce uniqueness
    or existence checks.

    Parameters:
        name: The name to add to the registry.
        check: Optional validation mode:
            - 'unique': Raise error if name already exists
            - 'exists': Raise error if name doesn't exist
            - None: No validation (default)

    Raises:
        ValueError: If check method is invalid, or if validation fails.
    """
    if check not in ["unique", "exists"] and check is not None:
        raise ValueError(f"Invalid check name method: {check}")
    in_names = name in self._names
    if check == "unique" and in_names:
        raise ValueError(f"Name '{name}' already exists.")
    if check == "exists" and not in_names:
        raise ValueError(f"Name '{name}' does not exist.")
    self._names.add(name)

run_model

run_model(steps=None, order=DEFAULT_RUN_ORDER)

Executes the model simulation.

Runs through the following phases: 1. Setup phase (model.setup()) 2. Step phase (model.step()) - repeated 3. End phase (model.end())

Parameters:

Name	Type	Description	Default
steps	Optional[int]	Number of steps to run. If None, runs until self.running is False.	None

Source code in abses/core/model.py

def run_model(
    self,
    steps: Optional[int] = None,
    order: Tuple[SubSystemName, ...] = DEFAULT_RUN_ORDER,
) -> None:
    """Executes the model simulation.

    Runs through the following phases:
    1. Setup phase (model.setup())
    2. Step phase (model.step()) - repeated
    3. End phase (model.end())

    Args:
        steps: Number of steps to run. If None, runs until self.running is False.
    """
    run_times = 0
    self.do_each("setup", order=order)
    while self.running is True:
        self.do_each("step", order=order)
        run_times += 1
        if steps is not None and run_times >= steps:
            break
    self.do_each("end", order=order)

setup

setup()

Users can custom what to do when the model is setup and going to start running.

Source code in abses/core/model.py

def setup(self) -> None:
    """Users can custom what to do when the model is setup and going to start running."""

step

step()

A step of the model. By default, collect data at each step.

Source code in abses/core/model.py

def step(self) -> None:
    """A step of the model.
    By default, collect data at each step.
    """
    self.datacollector.collect(self)

end

end()

Users can custom what to do when the model is end.

Source code in abses/core/model.py

def end(self) -> None:
    """Users can custom what to do when the model is end."""
    # End tracker run if available
    if self.datacollector.tracker is not None:
        self.datacollector.tracker.end_run()