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 scenario definitions file (.json).

 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 supported file type 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 scenario definitions file (``.json``).
        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 supported file type 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, batch) work units — each a batch of config.scenarios_per_file scenarios — across worker processes. Workers load params via pickle.load.

Results are written to Parquet files under config.output_dir in long format, one file per (indicator, PJNZ, batch) at {output_dir}/{indicator}/pjnz_name={pjnz_stem}/{part_name}.parquet. Each file holds every scenario in the batch, identified by a scenario_id column.

Scenarios whose simulations fail are logged and skipped rather than aborting the run; the rest still complete. Failed scenarios are recorded in the returned :class:RunResult and written to a failures.json manifest under config.output_dir for re-running. PJNZ import failures remain fatal and raise.

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

Returns:

Name Type Description
A RunResult

class:RunResult with the output directory and any failed units.

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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def run_scenario_analysis(config: RunConfig, simulations: ScenarioSimulations) -> RunResult:
    """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, batch)``
    work units — each a batch of ``config.scenarios_per_file`` scenarios — across
    worker processes. Workers load params via ``pickle.load``.

    Results are written to Parquet files under ``config.output_dir`` in
    long format, one file per ``(indicator, PJNZ, batch)`` at
    ``{output_dir}/{indicator}/pjnz_name={pjnz_stem}/{part_name}.parquet``. Each
    file holds every scenario in the batch, identified by a ``scenario_id``
    column.

    Scenarios whose simulations fail are logged and skipped rather than aborting
    the run; the rest still complete. Failed scenarios are recorded in the
    returned :class:`RunResult` and written to a ``failures.json`` manifest under
    ``config.output_dir`` for re-running. PJNZ *import* failures remain fatal and
    raise.

    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.

    Returns:
        A :class:`RunResult` with the output directory and any failed units.

    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 JSON 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. -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.
scenarios_per_file int

Number of scenarios written to each output Parquet file, which is also the unit of parallel work (default 128). The run parallelises over (PJNZ, batch) units, so the total unit count is ceil(n_scenarios / scenarios_per_file) * n_pjnz — keep it at least the core count or workers sit idle. A larger value means bigger, better-compressed files and fewer of them, but more memory: one batch is held in flight per worker, so peak memory is roughly n_workers * scenarios_per_file * n_simulations * 3.5MB.
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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 JSON 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. ``-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.
        scenarios_per_file (int): Number of scenarios written to each output
            Parquet file, which is also the unit of parallel work (default 128).
            The run parallelises over ``(PJNZ, batch)`` units, so the total unit
            count is ``ceil(n_scenarios / scenarios_per_file) * n_pjnz`` — keep
            it at least the core count or workers sit idle. A larger value means
            bigger, better-compressed files and fewer of them, but more memory:
            one batch is held in flight per worker, so peak memory is roughly
            ``n_workers * scenarios_per_file * n_simulations * 3.5MB``.
    """

    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))
    scenarios_per_file: int = 128

    @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

    @field_validator("scenarios_per_file")
    @classmethod
    def _scenarios_per_file_must_be_positive(cls, v: int) -> int:
        if v < 1:
            msg = "scenarios_per_file 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_json(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() != ".json":
            msg = f"definition_path must be a .json 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