Skip to content

Reference

avenir_goals_scenario.draw_simulations(definition_path, base_year, n_simulations=100, seed=None)

Draw scenario simulations in memory from a scenario definitions file.

Parameters:

Name Type Description Default
definition_path Path | str

Path to the input CSV scenario definitions file.

 required
n_simulations int

Number of simulations drawn per scenario.

 100
seed int | None

Optional integer seed for reproducible draws.

 None
base_year int

If provided, used as the minimum value for target_year draws. Draws below base_year are clamped up to it.

 required

Returns:

Type Description
ScenarioSimulations

A avenir_goals_scenario.models.ScenarioSimulations containing
ScenarioSimulations

all scenarios and their draws.

Raises:

Type Description
FileNotFoundError

If definition_path does not exist.
ValueError

If definition_path is not a valid .csv file or its contents fail schema validation.
Source code in src/avenir_goals_scenario/scenarios.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def draw_simulations(
    definition_path: Path | str,
    base_year: int,
    n_simulations: int = 100,
    seed: int | None = None,
) -> ScenarioSimulations:
    """Draw scenario simulations in memory from a scenario definitions file.

    Args:
        definition_path: Path to the input CSV scenario definitions file.
        n_simulations: Number of simulations drawn per scenario.
        seed: Optional integer seed for reproducible draws.
        base_year: If provided, used as the minimum value for ``target_year``
            draws. Draws below ``base_year`` are clamped up to it.

    Returns:
        A ``avenir_goals_scenario.models.ScenarioSimulations`` containing
        all scenarios and their draws.

    Raises:
        FileNotFoundError: If ``definition_path`` does not exist.
        ValueError: If ``definition_path`` is not a valid ``.csv`` file or its
            contents fail schema validation.
    """
    definition_path = Path(definition_path).expanduser().resolve()
    rng = np.random.default_rng(seed)
    definition = load_scenario_definition(definition_path)
    return gen_simulations(definition, n_simulations=n_simulations, rng=rng, base_year=base_year)

avenir_goals_scenario.write_simulations(simulations, path)

Write scenario simulations to a JSON file.

Parameters:

Name Type Description Default
simulations ScenarioSimulations

The scenario simulations to write.

 required
path Path | str

Destination path for the JSON file.

 required

Returns:

Type Description
Path

The resolved path that was written to.

Raises:

Type Description
FileNotFoundError

If the parent directory of path does not exist.
Source code in src/avenir_goals_scenario/scenarios.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def write_simulations(simulations: ScenarioSimulations, path: Path | str) -> Path:
    """Write scenario simulations to a JSON file.

    Args:
        simulations: The scenario simulations to write.
        path: Destination path for the JSON file.

    Returns:
        The resolved path that was written to.

    Raises:
        FileNotFoundError: If the parent directory of ``path`` does not exist.
    """
    path = Path(path).expanduser().resolve()
    if not path.parent.exists():
        msg = f"Destination directory does not exist: {path.parent}"
        raise FileNotFoundError(msg)
    path.write_text(simulations.model_dump_json(indent=2))
    return path

avenir_goals_scenario.read_simulations(path)

Read scenario simulations from a JSON file.

Parameters:

Name Type Description Default
path Path | str

Path to the scenario simulations JSON file.

 required

Returns:

Type Description
ScenarioSimulations

The loaded avenir_goals_scenario.models.ScenarioSimulations.

Raises:

Type Description
FileNotFoundError

If path does not exist.
ValueError

If the file contents fail schema validation.
Source code in src/avenir_goals_scenario/scenarios.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def read_simulations(path: Path | str) -> ScenarioSimulations:
    """Read scenario simulations from a JSON file.

    Args:
        path: Path to the scenario simulations JSON file.

    Returns:
        The loaded ``avenir_goals_scenario.models.ScenarioSimulations``.

    Raises:
        FileNotFoundError: If ``path`` does not exist.
        ValueError: If the file contents fail schema validation.
    """
    path = Path(path).expanduser().resolve()
    if not path.exists():
        msg = f"Simulations file does not exist: {path}"
        raise FileNotFoundError(msg)
    return ScenarioSimulations.model_validate_json(path.read_bytes())

avenir_goals_scenario.run_scenario_analysis(config, simulations)

Run scenario analysis across a directory of PJNZ files.

Converts each PJNZ to leapfrog params once in the main process, dumps them to a temporary file using pickle.dump, then distributes (PJNZ, scenario) work units across worker processes. Workers load params via pickle.load.

Results are written to Parquet files under config.output_dir, one file per PJNZ/scenario combination at {output_dir}/{pjnz_stem}/scenario_{id}.parquet. Each file contains one dataset per indicator with shape (n_simulations, *indicator_dims).

Parameters:

Name Type Description Default
config RunConfig

Validated run configuration.

 required
simulations ScenarioSimulations

Scenario simulations to run. Use avenir_goals_scenario.draw_simulations to generate them or avenir_goals_scenario.read_simulations to load from a file.

 required

Raises:

Type Description
FileNotFoundError

If no PJNZ files are found in config.pjnz_dir.
ValueError

If any output indicator is not present in the Goals output, or if a PJNZ file cannot be parsed.
Source code in src/avenir_goals_scenario/runner.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def run_scenario_analysis(config: RunConfig, simulations: ScenarioSimulations) -> Path:
    """Run scenario analysis across a directory of PJNZ files.

    Converts each PJNZ to leapfrog params once in the main process, dumps them
    to a temporary file using ``pickle.dump``, then distributes
    ``(PJNZ, scenario)`` work units across worker processes. Workers load
    params via ``pickle.load``.

    Results are written to Parquet files under ``config.output_dir``, one file
    per PJNZ/scenario combination at
    ``{output_dir}/{pjnz_stem}/scenario_{id}.parquet``. Each file contains one
    dataset per indicator with shape ``(n_simulations, *indicator_dims)``.

    Args:
        config: Validated run configuration.
        simulations: Scenario simulations to run. Use
            `avenir_goals_scenario.draw_simulations` to generate them or
            `avenir_goals_scenario.read_simulations` to load from a file.

    Raises:
        FileNotFoundError: If no PJNZ files are found in ``config.pjnz_dir``.
        ValueError: If any output indicator is not present in the Goals output,
            or if a PJNZ file cannot be parsed.
    """
    return _run_scenario_analysis(config, simulations, RunCallbacks())

avenir_goals_scenario.RunConfig

Bases: BaseModel

Configuration for avenir_goals_scenario.run_scenario_analysis.

Field names are case-insensitive: PJNZ_Dir, pjnz_dir, and PJNZ_DIR are all accepted.

output_dir is created if it does not exist, but only one level deep - its parent must already exist.

Attributes:

Name Type Description
pjnz_dir Path

Directory containing .PJNZ files.
definition_path Path | None

Path to the scenario definition CSV file. Required for the draw command and for run when draws are not pre-generated. Either definition_path, scenario_path, or both must be supplied for run.
scenario_path Path | None

Path to a scenario simulations JSON file. For draw, this is where draws are written. For run, if supplied and the file exists the draws are loaded from it rather than regenerated.
output_dir Path

Directory where per-PJNZ result subdirectories are written. Created automatically if absent (parent must exist).
base_year int

First year of the output projection range.
output_indicators list[str]

Names of the Goals output indicators to write. Each name must be a key in the dict returned by run_goals.
n_simulations int

Number of simulations drawn per scenario (default 100). Ignored when loading draws from an existing scenario_path.
seed int | None

Optional RNG seed for reproducible draws. None (the default) uses a random seed.
n_workers int

Number of parallel worker processes. Follows joblib conventions: -1 uses all available CPUs, 1 runs sequentially, and any positive integer sets an explicit worker count. Zero is not valid. Uses 4 by default or the number of available CPUs if fewer than 4.
Source code in src/avenir_goals_scenario/models/run_config.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class RunConfig(BaseModel):
    """Configuration for `avenir_goals_scenario.run_scenario_analysis`.

    Field names are case-insensitive: ``PJNZ_Dir``, ``pjnz_dir``, and
    ``PJNZ_DIR`` are all accepted.

    ``output_dir`` is created if it does not exist, but only one level deep -
    its parent must already exist.

    Attributes:
        pjnz_dir (Path): Directory containing ``.PJNZ`` files.
        definition_path (Path | None): Path to the scenario definition CSV file.
            Required for the ``draw`` command and for ``run`` when draws are not
            pre-generated. Either ``definition_path``, ``scenario_path``, or
            both must be supplied for ``run``.
        scenario_path (Path | None): Path to a scenario simulations JSON file.
            For ``draw``, this is where draws are written. For ``run``, if
            supplied and the file exists the draws are loaded from it rather
            than regenerated.
        output_dir (Path): Directory where per-PJNZ result subdirectories are
            written.  Created automatically if absent (parent must exist).
        base_year (int): First year of the output projection range.
        output_indicators (list[str]): Names of the Goals output indicators to
            write. Each name must be a key in the dict returned by ``run_goals``.
        n_simulations (int): Number of simulations drawn per scenario (default
            100). Ignored when loading draws from an existing ``scenario_path``.
        seed (int | None): Optional RNG seed for reproducible draws. ``None``
            (the default) uses a random seed.
        n_workers (int): Number of parallel worker processes. Follows joblib
            conventions: ``-1`` uses all available CPUs, ``1`` runs
            sequentially, and any positive integer sets an explicit worker
            count. Zero is not valid. Uses 4 by default or the number of
            available CPUs if fewer than 4.
    """

    model_config = ConfigDict(extra="forbid")

    pjnz_dir: Path
    definition_path: Path | None = None
    scenario_path: Path | None = None
    output_dir: Path
    base_year: int
    output_indicators: list[str]
    n_simulations: int = 100
    seed: int | None = None
    n_workers: int = Field(default_factory=lambda: min(os.cpu_count() or 1, 4))

    @field_validator("n_workers")
    @classmethod
    def _n_workers_must_be_nonzero(cls, v: int) -> int:
        if v == 0:
            msg = "n_workers must be non-zero (-1 for all CPUs, or a positive integer)"
            raise ValueError(msg)
        return v

    @field_validator("n_simulations")
    @classmethod
    def _n_simulations_must_be_positive(cls, v: int) -> int:
        if v < 1:
            msg = "n_simulations must be a positive integer"
            raise ValueError(msg)
        return v

    @model_validator(mode="before")
    @classmethod
    def _lowercase_keys(cls, data: Any) -> Any:
        if isinstance(data, dict):
            return {k.lower(): v for k, v in data.items()}
        return data

    @field_validator("pjnz_dir")
    @classmethod
    def _pjnz_dir_must_be_directory(cls, v: Path) -> Path:
        path = v.expanduser().resolve()
        if not path.exists():
            msg = f"PJNZ directory does not exist: {path}"
            raise ValueError(msg)
        if not path.is_dir():
            msg = f"pjnz_dir is not a directory: {path}"
            raise ValueError(msg)
        return path

    @field_validator("definition_path")
    @classmethod
    def _definition_path_must_be_csv(cls, v: Path | None) -> Path | None:
        if v is None:
            return None
        path = v.expanduser().resolve()
        if not path.exists():
            msg = f"Scenario definition file does not exist: {path}"
            raise ValueError(msg)
        if not path.is_file():
            msg = f"definition_path is not a file: {path}"
            raise ValueError(msg)
        if path.suffix.lower() != ".csv":
            msg = f"definition_path must be a .csv file, got: {path.suffix or '(no extension)'}"
            raise ValueError(msg)
        return path

    @field_validator("scenario_path")
    @classmethod
    def _scenario_path_must_not_be_directory(cls, v: Path | None) -> Path | None:
        if v is None:
            return None
        path = v.expanduser().resolve()
        if path.exists() and not path.is_file():
            msg = f"scenario_path is not a file: {path}"
            raise ValueError(msg)
        return path

    @field_validator("output_dir")
    @classmethod
    def _validate_output_dir(cls, v: Path) -> Path:
        path = v.expanduser().resolve()
        if path.exists() and not path.is_dir():
            msg = f"output_dir exists but is not a directory: {path}"
            raise ValueError(msg)
        if not path.exists() and not path.parent.exists():
            msg = f"Cannot create output_dir {path}: parent directory does not exist: {path.parent}"
            raise ValueError(msg)
        return path