experiments

agentevac.analysis.experiments

Parameter sweep driver for AgentEvac calibration experiments.

This module builds a Cartesian-product grid of agent parameters and runs one simulation subprocess per grid cell, collecting the resulting metrics and replay files.

Parameter axes: - info_sigma : Gaussian noise standard deviation on margin observations (metres). - info_delay_s : Information delay in seconds (stale observation replay). - theta_trust : Social-signal trust weight ∈ [0, 1]. - scenario : Information regime ("no_notice", "alert_guided", "advice_guided").

Each case is run by spawning agentevac.simulation.main as a subprocess with the appropriate environment variables set (INFO_SIGMA, INFO_DELAY_S, DEFAULT_THETA_TRUST). The SUMO GUI is suppressed (--sumo-binary sumo) for headless batch execution.

Outputs (written to output_dir): - routes_<case_id>.jsonl : Recorded LLM route decisions (for replay). - metrics_<case_id>.json : Run-level KPI summary. - stdout_<case_id>.log : Full stdout + stderr of the subprocess. - experiment_results.json : Aggregated list of all case result dicts. - experiment_results.csv : Flat CSV of key result fields.

build_experiment_grid(sigma_values=None, delay_values=None, trust_values=None, scenario_modes=None, base_overrides=None)

Build a Cartesian-product experiment grid from parameter value lists.

Defaults apply when a parameter list is not provided

• sigma_values : [40.0]
• delay_values : [0.0]
• trust_values : [0.5]
• scenario_modes : ["advice_guided"]

Parameters:

Name	Type	Description	Default
sigma_values	Optional[List[float]]	List of INFO_SIGMA values to sweep.	None
delay_values	Optional[List[float]]	List of INFO_DELAY_S values to sweep.	None
trust_values	Optional[List[float]]	List of DEFAULT_THETA_TRUST values to sweep.	None
scenario_modes	Optional[List[str]]	List of scenario mode strings to sweep.	None
base_overrides	Optional[Dict[str, Any]]	Additional key-value pairs merged into every case dict (e.g., {"messaging_enabled": True}).	None

Returns:

Type	Description
List[Dict[str, Any]]	List of case config dicts, one per grid cell.

Source code in agentevac/analysis/experiments.py

def build_experiment_grid(
    sigma_values: Optional[List[float]] = None,
    delay_values: Optional[List[float]] = None,
    trust_values: Optional[List[float]] = None,
    scenario_modes: Optional[List[str]] = None,
    base_overrides: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    """Build a Cartesian-product experiment grid from parameter value lists.

    Defaults apply when a parameter list is not provided:
        - sigma_values    : [40.0]
        - delay_values    : [0.0]
        - trust_values    : [0.5]
        - scenario_modes  : ["advice_guided"]

    Args:
        sigma_values: List of ``INFO_SIGMA`` values to sweep.
        delay_values: List of ``INFO_DELAY_S`` values to sweep.
        trust_values: List of ``DEFAULT_THETA_TRUST`` values to sweep.
        scenario_modes: List of scenario mode strings to sweep.
        base_overrides: Additional key-value pairs merged into every case dict
            (e.g., ``{"messaging_enabled": True}``).

    Returns:
        List of case config dicts, one per grid cell.
    """
    sigma_seq = sigma_values if sigma_values is not None else [40.0]
    delay_seq = delay_values if delay_values is not None else [0.0]
    trust_seq = trust_values if trust_values is not None else [0.5]
    scenario_seq = scenario_modes if scenario_modes is not None else ["advice_guided"]

    grid: List[Dict[str, Any]] = []
    for info_sigma, info_delay_s, theta_trust, scenario in itertools.product(
        sigma_seq,
        delay_seq,
        trust_seq,
        scenario_seq,
    ):
        case = {
            "info_sigma": float(info_sigma),
            "info_delay_s": float(info_delay_s),
            "theta_trust": float(theta_trust),
            "scenario": str(scenario),
        }
        if base_overrides:
            case.update(dict(base_overrides))
        grid.append(case)
    return grid

export_experiment_results(results, *, output_dir, stem='experiment_results')

Write experiment results to JSON (full) and CSV (flat key subset) files.

The JSON file contains all result fields. The CSV file contains a flattened subset suitable for quick inspection in spreadsheet tools.

Parameters:

Name	Type	Description	Default
results	List[Dict[str, Any]]	List of result dicts from run_parameter_sweep.	required
output_dir	str	Directory to write output files.	required
stem	str	Base filename stem (without extension).	'experiment_results'

Returns:

Type	Description
Dict[str, str]	Dict with "json" and "csv" keys mapping to the written file paths.

Source code in agentevac/analysis/experiments.py

def export_experiment_results(
    results: List[Dict[str, Any]],
    *,
    output_dir: str,
    stem: str = "experiment_results",
) -> Dict[str, str]:
    """Write experiment results to JSON (full) and CSV (flat key subset) files.

    The JSON file contains all result fields.  The CSV file contains a flattened
    subset suitable for quick inspection in spreadsheet tools.

    Args:
        results: List of result dicts from ``run_parameter_sweep``.
        output_dir: Directory to write output files.
        stem: Base filename stem (without extension).

    Returns:
        Dict with ``"json"`` and ``"csv"`` keys mapping to the written file paths.
    """
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    json_path = out_dir / f"{stem}.json"
    csv_path = out_dir / f"{stem}.csv"

    with open(json_path, "w", encoding="utf-8") as fh:
        json.dump(results, fh, ensure_ascii=False, indent=2, sort_keys=True)
        fh.write("\n")

    with open(csv_path, "w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(
            fh,
            fieldnames=[
                "case_id",
                "status",
                "returncode",
                "timeout",
                "elapsed_s",
                "scenario",
                "info_sigma",
                "info_delay_s",
                "theta_trust",
                "replay_path",
                "metrics_path",
                "stdout_log",
            ],
        )
        writer.writeheader()
        for row in results:
            case = row.get("case") or {}
            writer.writerow({
                "case_id": row.get("case_id"),
                "status": row.get("status"),
                "returncode": row.get("returncode"),
                "timeout": row.get("timeout"),
                "elapsed_s": row.get("elapsed_s"),
                "scenario": case.get("scenario"),
                "info_sigma": case.get("info_sigma"),
                "info_delay_s": case.get("info_delay_s"),
                "theta_trust": case.get("theta_trust"),
                "replay_path": row.get("replay_path"),
                "metrics_path": row.get("metrics_path"),
                "stdout_log": row.get("stdout_log"),
            })

    return {
        "json": str(json_path),
        "csv": str(csv_path),
    }

run_experiment_case(case_cfg, *, script_path='agentevac/simulation/main.py', python_executable=None, output_dir='outputs/experiments', sumo_binary='sumo', run_mode='record', timeout_s=None, sumo_seed=None, map_name=None)

Execute one parameter-grid case by spawning a simulation subprocess.

Constructs the CLI command and environment from case_cfg, runs the process, captures stdout/stderr, and extracts the metrics and replay file paths from the subprocess stdout using fixed prefix patterns.

Parameters:

Name	Type	Description	Default
case_cfg	Dict[str, Any]	Case configuration dict (from build_experiment_grid), containing at minimum info_sigma, info_delay_s, theta_trust, and scenario keys.	required
script_path	str	Path to the main simulation script (relative to project root).	'agentevac/simulation/main.py'
python_executable	Optional[str]	Python interpreter to use; defaults to sys.executable.	None
output_dir	str	Directory for output files.	'outputs/experiments'
sumo_binary	str	SUMO binary name; use "sumo" for headless batch runs.	'sumo'
run_mode	str	"record" or "replay".	'record'
timeout_s	Optional[float]	Optional subprocess timeout in seconds.	None

Returns:

Type	Description
Dict[str, Any]	A result dict with fields including case_id, status, returncode,
Dict[str, Any]	elapsed_s, replay_path, metrics_path, events_path,
Dict[str, Any]	stdout_log, and stdout_tail.

Source code in agentevac/analysis/experiments.py

def run_experiment_case(
    case_cfg: Dict[str, Any],
    *,
    script_path: str = "agentevac/simulation/main.py",
    python_executable: Optional[str] = None,
    output_dir: str = "outputs/experiments",
    sumo_binary: str = "sumo",
    run_mode: str = "record",
    timeout_s: Optional[float] = None,
    sumo_seed: Optional[int] = None,
    map_name: Optional[str] = None,
) -> Dict[str, Any]:
    """Execute one parameter-grid case by spawning a simulation subprocess.

    Constructs the CLI command and environment from ``case_cfg``, runs the process,
    captures stdout/stderr, and extracts the metrics and replay file paths from the
    subprocess stdout using fixed prefix patterns.

    Args:
        case_cfg: Case configuration dict (from ``build_experiment_grid``), containing
            at minimum ``info_sigma``, ``info_delay_s``, ``theta_trust``, and
            ``scenario`` keys.
        script_path: Path to the main simulation script (relative to project root).
        python_executable: Python interpreter to use; defaults to ``sys.executable``.
        output_dir: Directory for output files.
        sumo_binary: SUMO binary name; use ``"sumo"`` for headless batch runs.
        run_mode: ``"record"`` or ``"replay"``.
        timeout_s: Optional subprocess timeout in seconds.

    Returns:
        A result dict with fields including ``case_id``, ``status``, ``returncode``,
        ``elapsed_s``, ``replay_path``, ``metrics_path``, ``events_path``,
        ``stdout_log``, and ``stdout_tail``.
    """
    python_bin = python_executable or sys.executable
    script_file = Path(script_path).resolve()
    # Run subprocess from the project root (three levels above this file:
    # agentevac/analysis/experiments.py → agentevac/analysis → agentevac → root).
    project_root = Path(__file__).parents[2]
    out_dir = Path(output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    case_index = int(case_cfg.get("_case_index", 0))
    case_id = str(case_cfg.get("case_id") or _case_id(case_cfg, case_index))
    replay_base = out_dir / f"routes_{case_id}.jsonl"
    metrics_base = out_dir / f"metrics_{case_id}.json"
    params_base = out_dir / f"run_params_{case_id}.json"
    stdout_log = out_dir / f"stdout_{case_id}.log"

    cmd = [
        python_bin,
        str(script_file),
        "--run-mode", run_mode,
        "--scenario", str(case_cfg["scenario"]),
        "--sumo-binary", str(sumo_binary),
        "--events", "off",
        "--events-stdout", "off",
        "--web-dashboard", "off",
        "--overlays", "off",
        "--metrics", "on",
        "--replay-log-path", str(replay_base),
        "--metrics-log-path", str(metrics_base),
        "--params-log-path", str(params_base),
    ]

    messaging_enabled = bool(case_cfg.get("messaging_enabled", True))
    cmd.extend(["--messaging", "on" if messaging_enabled else "off"])
    _map = map_name or case_cfg.get("map_name")
    if _map:
        cmd.extend(["--map", str(_map)])
    print(f"[SIM_CLI] case_id={case_id} {_format_cmd(cmd[2:])}")

    env = os.environ.copy()
    env.update({
        "INFO_SIGMA": str(float(case_cfg["info_sigma"])),
        "INFO_DELAY_S": str(float(case_cfg["info_delay_s"])),
        "DEFAULT_THETA_TRUST": str(float(case_cfg["theta_trust"])),
        "SUMO_BINARY": str(sumo_binary),
    })
    if sumo_seed is not None:
        env["SUMO_SEED"] = str(int(sumo_seed))
    if "DEFAULT_LAMBDA_E" in case_cfg:
        env["DEFAULT_LAMBDA_E"] = str(float(case_cfg["DEFAULT_LAMBDA_E"]))
    if "DEFAULT_LAMBDA_T" in case_cfg:
        env["DEFAULT_LAMBDA_T"] = str(float(case_cfg["DEFAULT_LAMBDA_T"]))

    started_at = time.time()
    try:
        proc = subprocess.run(
            cmd,
            cwd=str(project_root),
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            timeout=timeout_s,
            check=False,
        )
        elapsed_s = time.time() - started_at
        stdout_text = proc.stdout or ""
        status = "ok" if proc.returncode == 0 else "failed"
        timeout_hit = False
        returncode = int(proc.returncode)
    except subprocess.TimeoutExpired as exc:
        elapsed_s = time.time() - started_at
        stdout_text = (exc.stdout or "") if isinstance(exc.stdout, str) else ""
        status = "timeout"
        timeout_hit = True
        returncode = -1

    stdout_log.write_text(stdout_text, encoding="utf-8")

    result = {
        "case_id": case_id,
        "case": dict(case_cfg),
        "command": cmd,
        "status": status,
        "returncode": returncode,
        "timeout": timeout_hit,
        "elapsed_s": round(elapsed_s, 3),
        "stdout_log": str(stdout_log),
        "replay_path": _extract_path(stdout_text, "[REPLAY] mode=record path="),
        "metrics_path": _extract_path(stdout_text, "[METRICS] summary_path="),
        "events_path": _extract_events_path(stdout_text),
        "stdout_tail": stdout_text.splitlines()[-20:],
    }
    return result

run_parameter_sweep(grid, *, script_path='agentevac/simulation/main.py', python_executable=None, output_dir='outputs/experiments', sumo_binary='sumo', run_mode='record', timeout_s=None, sumo_seed=None, map_name=None)

Run all cases in the experiment grid sequentially.

Cases run one at a time (no parallelism) to avoid SUMO port conflicts and to keep resource usage predictable.

Parameters:

Name	Type	Description	Default
grid	List[Dict[str, Any]]	List of case config dicts from build_experiment_grid.	required
script_path	str	Path to the main simulation script.	'agentevac/simulation/main.py'
python_executable	Optional[str]	Python interpreter to use.	None
output_dir	str	Directory for all case output files.	'outputs/experiments'
sumo_binary	str	SUMO binary name.	'sumo'
run_mode	str	"record" or "replay".	'record'
timeout_s	Optional[float]	Per-case subprocess timeout in seconds.	None
map_name	Optional[str]	Map config directory name (e.g., "lytton").	None

Returns:

Type	Description
List[Dict[str, Any]]	List of result dicts (one per grid case) from run_experiment_case.

Source code in agentevac/analysis/experiments.py

def run_parameter_sweep(
    grid: List[Dict[str, Any]],
    *,
    script_path: str = "agentevac/simulation/main.py",
    python_executable: Optional[str] = None,
    output_dir: str = "outputs/experiments",
    sumo_binary: str = "sumo",
    run_mode: str = "record",
    timeout_s: Optional[float] = None,
    sumo_seed: Optional[int] = None,
    map_name: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Run all cases in the experiment grid sequentially.

    Cases run one at a time (no parallelism) to avoid SUMO port conflicts and to
    keep resource usage predictable.

    Args:
        grid: List of case config dicts from ``build_experiment_grid``.
        script_path: Path to the main simulation script.
        python_executable: Python interpreter to use.
        output_dir: Directory for all case output files.
        sumo_binary: SUMO binary name.
        run_mode: ``"record"`` or ``"replay"``.
        timeout_s: Per-case subprocess timeout in seconds.
        map_name: Map config directory name (e.g., ``"lytton"``).

    Returns:
        List of result dicts (one per grid case) from ``run_experiment_case``.
    """
    results: List[Dict[str, Any]] = []
    for idx, raw_case in enumerate(grid, start=1):
        case_cfg = dict(raw_case)
        case_cfg["_case_index"] = idx
        case_cfg["case_id"] = _case_id(case_cfg, idx)
        results.append(
            run_experiment_case(
                case_cfg,
                script_path=script_path,
                python_executable=python_executable,
                output_dir=output_dir,
                sumo_binary=sumo_binary,
                run_mode=run_mode,
                timeout_s=timeout_s,
                sumo_seed=sumo_seed,
                map_name=map_name,
            )
        )
    return results