smif.data_layer package

Data access modules for loading system-of-systems model configuration

class smif.data_layer.DataArray(spec: ~smif.metadata.spec.Spec, data: <MagicMock id='135493239504048'>)[source]

Bases: object

DataArray provides access to input/parameter/results data, with conversions to common python data libraries (for example: numpy, pandas, xarray).

spec
Type:

smif.metadata.spec.Spec

data
Type:

numpy.ndarray

as_df() → <MagicMock id='135493245860944'>[source]

Access DataArray as a pandas.DataFrame

as_dict()[source]
as_ndarray() → <MagicMock id='135493240406144'>[source]

Access as a numpy.ndarray

as_xarray()[source]

Access DataArray as a xarray.DataArray

property coords

Coordinate labels for each dimension.

property description

A human-friendly description

dim_coords(dim)[source]

Coordinates for a given dimension

dim_elements(dim)[source]

Coordinate elements for a given dimension

dim_names(dim)[source]

Coordinate names for a given dimension

property dims

Names for each dimension

classmethod from_df(spec, dataframe)[source]

Create a DataArray from a pandas.DataFrame

classmethod from_xarray(spec, xr_data_array)[source]

Create a DataArray from a xarray.DataArray

property name

The name of the data that this spec describes.

property shape

The shape of the data array

property unit

The unit for all data points.

update(other)[source]

Update data values with any from other which are non-null

validate_as_full()[source]

Check that the data array contains no NaN values

class smif.data_layer.DataHandle(store: Store, modelrun_name, current_timestep, timesteps, model, decision_iteration=None)[source]

Bases: object

Get/set model parameters and data

property base_timestep

Base timestep

property current_timestep

Current timestep

property decision_iteration
derive_for(model)[source]

Derive a new DataHandle configured for the given Model

Parameters:

model (Model) – Model which will use this DataHandle

get_base_timestep_data(input_name)[source]

Get data from the base timestep as required for model inputs

Parameters:

input_name (str)

Return type:

smif.data_layer.data_array.DataArray

get_current_interventions()[source]

Get the interventions that exist in the current state

Returns:

A dict of intervention dicts with build_year attribute keyed by name

Return type:

dict of dicts

get_data(input_name: str, timestep=None) → DataArray[source]

Get data required for model inputs

Parameters:
Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

Raises:

SmifDataError – If any data reading error occurs below this method, the error is handled and reraised within the context of the current call

get_parameter(parameter_name)[source]

Get the value for a parameter

Parameters:

parameter_name (str)

Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

get_parameters()[source]

Get all parameter values

Returns:

parameters – Read-only view of parameters (like a read-only dict)

Return type:

MappingProxyType

get_previous_timestep_data(input_name)[source]

Get data from the previous timestep as required for model inputs

Parameters:

input_name (str)

Return type:

smif.data_layer.data_array.DataArray

get_results(output_name, decision_iteration=None, timestep=None)[source]

Get results values for model outputs

Parameters:

  • output_name (str) – The name of an output for model_name

  • decision_iteration (int, default=None)

  • timestep (int or RelativeTimestep, default=None)

Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

Notes

Access to model results is only granted to models contained within self._model if self._model is a smif.model.model.CompositeModel

get_state()[source]

The current state of the model

If the DataHandle instance has a timestep, then state is established from the state file.

Returns:

A list of (intervention name, build_year) installed in the current timestep

Return type:

list of tuple

Raises:

ValueError – If self._current_timestep is None an error is raised.

property previous_timestep

Previous timestep

read_coefficients(source_dim: str, destination_dim: str) → <MagicMock id='135493267455824'>[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

Return type:

numpy.ndarray

read_unit_definitions() → List[str][source]

Read unit definitions

Return type:

list[str]

set_results(output_name, data)[source]

Set results values for model outputs

Parameters:
property timesteps

All timesteps (as tuple)

write_coefficients(source_dim: str, destination_dim: str, data: <MagicMock id='135493236816448'>)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

  • data (numpy.ndarray)

class smif.data_layer.Results(store: Store | dict)[source]

Bases: object

Common interface to access results from model runs.

Parameters:

store (Store or dict) – pre-created Store object or dictionary of the form {‘interface’: <interface>, ‘dir’: <dir>} where <interface> is either ‘local_csv’ or ‘local_parquet’, and <dir> is the model base directory

available_results(model_run_name)[source]

Return the results available for a given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

A nested dictionary data structure of the results available for the given model run

get_units(output_name: str)[source]

Return the units of a given output.

Parameters:

output_name (str)

Return type:

str

list_model_runs()[source]

Return a list of model run names.

Return type:

List of model run names

list_outputs(sector_model_name: str)[source]

Return a list of model run names.

Parameters:

sector_model_name (str the requested sector model)

Return type:

List of outputs for the given sector model

list_scenario_outputs(scenario_name: str)[source]

Return a list of outputs of a given scenario.

Parameters:

scenario_name (str the requested scenario)

Return type:

List of outputs for the requested scenario

list_scenarios(model_run_name: str)[source]

Return a dictionary of scenarios for given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

Dictionary of (scenario name, variant) for the given model run.

list_sector_models(model_run_name: str)[source]

Return a list of sector models for given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

List of sector models for the given model run

read_results(model_run_names: list, model_names: list, output_names: list, timesteps: list = None, decisions: list = None, time_decision_tuples: list = None)[source]

Return results from the store as a formatted pandas data frame. There are a number of ways of requesting specific timesteps/decisions. You can specify either:

a list of (timestep, decision) tuples

in which case data for all of those tuples matching the available results will be returned

or:
a list of timesteps

in which case data for all of those timesteps (and any decision iterations) matching the available results will be returned

or:
a list of decision iterations

in which case data for all of those decision iterations (and any timesteps) matching the available results will be returned

or:
a list of timesteps and a list of decision iterations

in which case data for the Cartesian product of those timesteps and those decision iterations matching the available results will be returned

or:
nothing

in which case all available results will be returned

Parameters:

  • model_run_names (list) – the requested model run names

  • model_names (list) – the requested sector model names (exactly one required)

  • output_names (list) – the requested output names (output specs must all match)

  • timesteps (list) – the requested timesteps

  • decisions (list) – the requested decision iterations

  • time_decision_tuples (list) – a list of requested (timestep, decision) tuples

Raises:

  • SmifDataNotFoundError – If data cannot be found in the store when try to read from the store

  • SmifDataMismatchError – Data presented to read, write and update methods is in the incorrect format or of wrong dimensions to that expected

  • SmifDataReadError – When unable to read data e.g. unable to handle file type or connect to database

Return type:

pandas.DataFrame

read_scenario_data(scenario_name: str, variant_name: str, variable_name: str, timesteps: list) → <MagicMock id='135493236657712'>[source]

Return scenario variant data from the store as a formatted pandas data frame.

Parameters:

  • scenario_name (str) – the requested scenario name

  • variant_name (str) – the requested scenario variant name

  • variable_name (str) – the requested output variable name that the requested scenario provides

  • timesteps (list) – the requested timesteps

Raises:

  • SmifDataNotFoundError – If data cannot be found in the store when try to read from the store

  • SmifDataMismatchError – Data presented to read, write and update methods is in the incorrect format or of wrong dimensions to that expected

  • SmifDataReadError – When unable to read data e.g. unable to handle file type or connect to database

Return type:

pandas.DataFrame

validate_names(model_run_names, sec_model_names, output_names)[source]
class smif.data_layer.Store(config_store, metadata_store: MetadataStore, data_store: DataStore, model_base_folder='.')[source]

Bases: object

Common interface to data store, composed of config, metadata and data store implementations.

Parameters:
available_results(model_run_name)[source]

List available results from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (timestep, decision_iteration, model_name, output_name)

Return type:

list[tuple]

canonical_available_results(model_run_name)[source]

List the results that are available from a model run, collapsing all decision iterations.

This is the unique items from calling available_results, with all decision iterations set to 0.

This method is used to determine whether a model run is complete, given that it is impossible to know how many decision iterations to expect: we simply check that each expected timestep has been completed.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing available results

canonical_expected_results(model_run_name)[source]

List the results that are expected from a model run, collapsing all decision iterations.

For a complete model run, this would coincide with the unique list returned from available_results, where all decision iterations are set to 0.

This method is used to determine whether a model run is complete, given that it is impossible to know how many decision iterations to expect: we simply check that each expected timestep has been completed.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing expected results

canonical_missing_results(model_run_name)[source]

List the results that are missing from a model run, collapsing all decision iterations.

For a complete model run, this is what is left after removing canonical_available_results from canonical_expected_results.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing missing results

clear_results(model_run_name)[source]

Clear all results from a single model run

Parameters:

model_run_name (str)

completed_jobs(model_run_name)[source]

List completed jobs from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (timestep, decision_iteration, model_name)

Return type:

list[tuple]

convert_initial_conditions_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_interventions_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_model_parameter_default_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_narrative_data(sos_model_name, tgt_store, noclobber=False)[source]
convert_scenario_data(model_run_name, tgt_store, noclobber=False)[source]
convert_strategies_data(model_run_name, tgt_store, noclobber=False)[source]
delete_dimension(dimension_name)[source]

Delete dimension

Parameters:

dimension_name (str)

delete_model(model_name)[source]

Delete a model

Parameters:

model_name (str)

delete_model_run(model_run_name)[source]

Delete a system-of-system model run

Parameters:

model_run_name (str)

delete_results(model_run_name, model_name, output_name, timestep=None, decision_iteration=None)[source]

Delete results for a single timestep/iteration of a model output in a model run

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

delete_scenario(scenario_name)[source]

Delete scenario from project configuration

Parameters:

scenario_name (str)

delete_scenario_variant(scenario_name, variant_name)[source]

Delete scenario from project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

delete_sos_model(sos_model_name)[source]

Delete a system-of-system model

Parameters:

sos_model_name (str)

expected_model_outputs(model_run_name)[source]

List expected model outputs from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (model_name, output_name)

Return type:

list[tuple]

static filter_complete_available_results(available_results, expected_model_outputs)[source]

Filter available results from a model run to include only complete timestep/decision iteration combinations

Parameters:

  • available_results (list[tuple]) – List of (timestep, decision_iteration, model_name, output_name)

  • expected_model_outputs (list[tuple]) – List or set of (model_name, output_name)

Returns:

Each tuple is (timestep, decision_iteration, model_name)

Return type:

list[tuple]

classmethod from_dict(config)[source]

Create Store from configuration dict

get_result_darray(model_run_name, model_name, output_name, timesteps=None, decision_iterations=None, time_decision_tuples=None)[source]

Return data for multiple timesteps and decision iterations for a given output from a given sector model in a specific model run.

You can specify either:
a list of (timestep, decision) tuples

in which case data for all of those tuples matching the available results will be returned

or:
a list of timesteps

in which case data for all of those timesteps (and any decision iterations) matching the available results will be returned

or:
a list of decision iterations

in which case data for all of those decision iterations (and any timesteps) matching the available results will be returned

or:
a list of timesteps and a list of decision iterations

in which case data for the Cartesian product of those timesteps and those decision iterations matching the available results will be returned

or:
nothing

in which case all available results will be returned

Then, for each tuple, the data array from the corresponding read_results call is stacked, and together with the new spec this information is returned as a new DataArray.

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timesteps (optional list of timesteps)

  • decision_iterations (optional list of decision iterations)

  • time_decision_tuples (optional list of unique (timestep, decision) tuples)

Return type:

DataArray with expanded spec and the data requested

get_results(model_run_names: list, model_name: str, output_names: list, timesteps: list = None, decisions: list = None, time_decision_tuples: list = None)[source]

Return data for multiple timesteps and decision iterations for a given output from a given sector model for multiple model runs.

Parameters:

  • model_run_names (list[str]) – the requested model run names

  • model_name (str) – the requested sector model name

  • output_names (list[str]) – the requested output names (output specs must all match)

  • timesteps (list[int]) – the requested timesteps

  • decisions (list[int]) – the requested decision iterations

  • time_decision_tuples (list[tuple]) – a list of requested (timestep, decision) tuples

Returns:

Nested dictionary of DataArray objects, keyed on model run name and output name. Returned DataArrays include one extra (timestep, decision_iteration) dimension.

Return type:

dict

prepare_model_runs(model_run_name, scenario_name, first_var, last_var)[source]

Write multiple model run config files corresponding to multiple scenario variants of {scenario_name}, based on template {model_run_name}

Write batchfile containing each of the generated model runs

Parameters:

  • model_run_name (str)

  • scenario_name (str)

  • first_var (int - between 0 and number of variants-1)

  • last_var (int - between first_var and number of variants-1)

prepare_scenario(scenario_name, list_of_variants)[source]

Modify {scenario_name} config file to include multiple scenario variants.

Parameters:

  • scenario_name (str)

  • list_of_variants (list[int] - indices of scenario variants)

prepare_warm_start(model_run_name)[source]

Copy the results from the previous model_run if available

The method allows a previous unsuccessful model_run to ‘warm start’ a new model run from a later timestep. Model results are recovered from the timestep that the previous model_run was run until, and the new model run runs from the returned timestep

Parameters:

model_run_name (str)

Return type:

int The timestep to which the data store was recovered

Notes

Called from smif.controller.execute

read_all_initial_conditions(model_run_name) → List[Dict][source]

A list of all historical interventions

Return type:

list[dict]

read_coefficients(source_dim: str, destination_dim: str) → <MagicMock id='135493238206208'>[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

read_dimension(dimension_name, skip_coords=False)[source]

Return dimension

Parameters:

dimension_name (str)

Returns:

A dimension definition (including elements)

Return type:

Coords

read_dimensions(skip_coords=False)[source]

Read dimensions

Return type:

list[Coords]

read_initial_conditions(model_name) → List[Dict][source]

Read historical interventions for model_name

Returns:

A list of historical interventions, with keys ‘name’ and ‘build_year’

Return type:

list[dict]

read_initial_conditions_file(model_name, string_id, assert_exists=False)[source]
read_interventions(model_name)[source]

Read interventions data for model_name

Returns:

A dict of intervention dictionaries containing intervention attributes keyed by intervention name

Return type:

dict[str, dict]

read_interventions_file(model_name, string_id, assert_exists=False)[source]
read_model(model_name, skip_coords=False)[source]

Read a model

Parameters:

model_name (str)

Return type:

Model

read_model_parameter_default(model_name, parameter_name, assert_exists=False)[source]

Read default data for a sector model parameter

Parameters:

  • model_name (str)

  • parameter_name (str)

Return type:

DataArray

read_model_run(model_run_name)[source]

Read a system-of-system model run

Parameters:

model_run_name (str)

Return type:

ModelRun

read_model_runs()[source]

Read all system-of-system model runs

Return type:

list[ModelRun]

read_models(skip_coords=False)[source]

Read all models

Return type:

list[Model]

read_narrative(sos_model_name, narrative_name)[source]

Read narrative from sos_model

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

read_narrative_variant_data(sos_model_name, narrative_name, variant_name, parameter_name, timestep=None, assert_exists=False)[source]

Read narrative data file

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

  • variant_name (str)

  • parameter_name (str)

  • timestep (int (optional)) – If None, read data for all timesteps

Return type:

DataArray

read_results(model_run_name: str, model_name: str, output_spec: Spec, timestep: int | None = None, decision_iteration: int | None = None) → DataArray[source]

Return results of a model_name in model_run_name for a given output_name

Parameters:
Return type:

DataArray

read_scenario(scenario_name, skip_coords=False)[source]

Read a scenario

Parameters:

scenario_name (str)

Return type:

ScenarioModel

read_scenario_variant(scenario_name, variant_name)[source]

Read a scenario variant

Parameters:

  • scenario_name (str)

  • variant_name (str)

Return type:

dict

read_scenario_variant_data(scenario_name: str, variant_name: str, variable: str, timestep: int | None = None, timesteps: List[int] | None = None, assert_exists: bool = False) → DataArray | bool[source]

Read scenario data file

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variable (str)

  • timestep (int)

Returns:

data

Return type:

DataArray

read_scenario_variants(scenario_name)[source]

Read variants of a given scenario

Parameters:

scenario_name (str)

Return type:

list[dict]

read_scenarios(skip_coords=False)[source]

Read scenarios

Return type:

list[ScenarioModel]

read_sos_model(sos_model_name)[source]

Read a specific system-of-system model

Parameters:

sos_model_name (str)

Return type:

SosModel

read_sos_models()[source]

Read all system-of-system models

Return type:

list[SosModel]

read_state(model_run_name, timestep, decision_iteration=None) → List[Dict][source]

Read list of (name, build_year) for a given model_run, timestep, decision

Parameters:

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

Return type:

list[dict]

read_strategies(model_run_name)[source]

Read strategies for a given model run

Parameters:

model_run_name (str)

Return type:

list[dict]

read_strategy_interventions(strategy, assert_exists=False)[source]

Read interventions as defined in a model run strategy

read_unit_definitions() → List[str][source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

update_dimension(dimension_name, dimension)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

update_model(model_name, model)[source]

Update a model

Parameters:
update_model_run(model_run_name, model_run)[source]

Update system-of-system model run

Parameters:
update_scenario(scenario_name, scenario)[source]

Update scenario

Parameters:
update_scenario_variant(scenario_name, variant_name, variant)[source]

Update scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variant (dict)

update_sos_model(sos_model_name, sos_model)[source]

Update system-of-system model

Parameters:
write_coefficients(source_dim: str, destination_dim: str, data: <MagicMock id='135493237137168'>)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

  • data (numpy.ndarray)

Notes

To be called from Adaptor implementations.

write_dimension(dimension)[source]

Write dimension to project configuration

Parameters:

dimension (Coords)

write_initial_conditions(model_name, initial_conditions)[source]

Write historical interventions for a model

Parameters:

list[dict] – A list of historical interventions, with keys ‘name’ and ‘build_year’

write_initial_conditions_file(model_name, string_id, initial_conditions)[source]
write_interventions(model_name, interventions)[source]

Write interventions data for a model

Parameters:

  • dict[str – A dict of intervention dictionaries containing intervention attributes keyed by intervention name

  • dict] – A dict of intervention dictionaries containing intervention attributes keyed by intervention name

write_interventions_file(model_name, string_id, interventions)[source]
write_model(model)[source]

Write a model

Parameters:

model (Model)

write_model_parameter_default(model_name, parameter_name, data)[source]

Write default data for a sector model parameter

Parameters:
write_model_run(model_run)[source]

Write system-of-system model run

Parameters:

model_run (ModelRun)

write_narrative_variant_data(sos_model_name, narrative_name, variant_name, data)[source]

Read narrative data file

Parameters:
write_results(data_array, model_run_name, model_name, timestep=None, decision_iteration=None)[source]

Write results of a model_name in model_run_name for a given output_name

Parameters:

  • data_array (DataArray)

  • model_run_id (str)

  • model_name (str)

  • timestep (int, optional)

  • decision_iteration (int, optional)

write_scenario(scenario)[source]

Write scenario

Parameters:

scenario (ScenarioModel)

write_scenario_variant(scenario_name, variant)[source]

Write scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant (dict)

write_scenario_variant_data(scenario_name, variant_name, data)[source]

Write scenario data file

Parameters:
write_sos_model(sos_model)[source]

Write system-of-system model

Parameters:

sos_model (SosModel)

write_state(state, model_run_name, timestep, decision_iteration=None)[source]

State is a list of decisions with name and build_year.

State is output from the DecisionManager

Parameters:

  • state (list[dict])

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

write_strategies(model_run_name, strategies)[source]

Write strategies for a given model_run

Parameters:
write_strategy_interventions(strategy, data)[source]
Parameters:

list[dicts]

write_unit_definitions(definitions)[source]

Reads custom unit definitions

Parameters:

list[str] – Pint-compatible unit definitions

Subpackages

Submodules

smif.data_layer.abstract_config_store module

A config store holds the configuration data for running system-of-systems models with smif: - model runs - system-of-systems models - model definitions - strategies - scenarios and scenario variants - narratives

class smif.data_layer.abstract_config_store.ConfigStore[source]

Bases: object

A ConfigStore must implement each of the abstract methods defined in this interface

abstractmethod delete_model(model_name)[source]

Delete a model

Parameters:

model_name (str)

abstractmethod delete_model_run(model_run_name)[source]

Delete a system-of-system model run

Parameters:

model_run_name (str)

abstractmethod delete_scenario(scenario_name)[source]

Delete scenario from project configuration

Parameters:

scenario_name (str)

abstractmethod delete_scenario_variant(scenario_name, variant_name)[source]

Delete scenario from project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

abstractmethod delete_sos_model(sos_model_name)[source]

Delete a system-of-system model

Parameters:

sos_model_name (str)

abstractmethod read_model(model_name)[source]

Read a model

Parameters:

model_name (str)

Return type:

Model

abstractmethod read_model_run(model_run_name)[source]

Read a system-of-system model run

Parameters:

model_run_name (str)

Return type:

ModelRun

abstractmethod read_model_runs()[source]

Read all system-of-system model runs

Return type:

list[ModelRun]

abstractmethod read_models()[source]

Read all models

Return type:

list[Model]

abstractmethod read_narrative(sos_model_name, narrative_name)[source]

Read narrative from sos_model

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

abstractmethod read_scenario(scenario_name)[source]

Read a scenario

Parameters:

scenario_name (str)

Return type:

ScenarioModel

abstractmethod read_scenario_variant(scenario_name, variant_name)[source]

Read a scenario variant

Parameters:

  • scenario_name (str)

  • variant_name (str)

Return type:

dict

abstractmethod read_scenario_variants(scenario_name)[source]

Read variants of a given scenario

Parameters:

scenario_name (str)

Return type:

list[dict]

abstractmethod read_scenarios()[source]

Read scenarios

Return type:

list[ScenarioModel]

abstractmethod read_sos_model(sos_model_name)[source]

Read a specific system-of-system model

Parameters:

sos_model_name (str)

Return type:

SosModel

abstractmethod read_sos_models()[source]

Read all system-of-system models

Return type:

list[SosModel]

abstractmethod read_strategies(modelrun_name)[source]

Read strategies for a given model run

Parameters:

model_run_name (str)

Return type:

list[dict]

abstractmethod update_model(model_name, model)[source]

Update a model

Parameters:
abstractmethod update_model_run(model_run_name, model_run)[source]

Update system-of-system model run

Parameters:
abstractmethod update_scenario(scenario_name, scenario)[source]

Update scenario

Parameters:
abstractmethod update_scenario_variant(scenario_name, variant_name, variant)[source]

Update scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variant (dict)

abstractmethod update_sos_model(sos_model_name, sos_model)[source]

Update system-of-system model

Parameters:
abstractmethod write_model(model)[source]

Write a model

Parameters:

model (Model)

abstractmethod write_model_run(model_run)[source]

Write system-of-system model run

Parameters:

model_run (ModelRun)

abstractmethod write_scenario(scenario)[source]

Write scenario

Parameters:

scenario (ScenarioModel)

abstractmethod write_scenario_variant(scenario_name, variant)[source]

Write scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant (dict)

abstractmethod write_sos_model(sos_model)[source]

Write system-of-system model

Parameters:

sos_model (SosModel)

abstractmethod write_strategies(modelrun_name, strategies)[source]

Write strategies for a given model_run

Parameters:

smif.data_layer.abstract_data_store module

A data store holds the bulk of model setup, intermediate and output data: - scenario variant and narrative data (including parameter defaults) - model interventions, initial conditions and state - conversion coefficients - results

class smif.data_layer.abstract_data_store.DataStore[source]

Bases: object

A DataStore must implement each of the abstract methods defined in this interface

abstractmethod available_results(modelrun_name)[source]

List available results from a model run

Returns:

Each tuple is (timestep, decision_iteration, model_name, output_name)

Return type:

list[tuple]

static dataframe_to_data_array(dataframe, spec, path)[source]
abstractmethod delete_results(model_run_name, model_name, output_name, timestep=None, decision_iteration=None)[source]

Delete results for a single timestep/iteration of a model output in a model run

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

classmethod filter_on_timesteps(dataframe, spec, path, timestep=None, timesteps=None)[source]

Filter dataframe by timestep

The ‘timestep’ dimension is treated as follows:

If a single timestep is specified, the spec MAY include ‘timestep’ as a dimension. If so, the returned DataArray’s spec will match the timestep requested. Otherwise, the DataArray will not include timestep as a dimension.

If multiple timesteps are specified, the returned DataArray’s spec will include a ‘timestep’ dimension to match the timesteps requested.

If timestep and timesteps are None, and the stored data has a timestep column, read all available timesteps. The returned DataArray’s spec ‘timestep’ dimension will match the timesteps requested. If the stored data does not have a timestep column, ignore and pass through unchanged.

abstractmethod read_coefficients(source_dim, destination_dim)[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between a single pair of dimensions.

Parameters:

  • source_dim (str) – dimension name

  • destination_dim (str) – dimension name

Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

abstractmethod read_initial_conditions(key) → List[Dict][source]

Read historical interventions for key

Parameters:

key (str)

Return type:

list[dict]

abstractmethod read_interventions(key)[source]

Read interventions data for key

Parameters:

key (str)

Returns:

A dict of intervention dictionaries containing intervention attributes keyed by intervention name

Return type:

dict[str, dict]

abstractmethod read_model_parameter_default(key, spec)[source]

Read data array

Parameters:
Returns:

data_array

Return type:

DataArray

abstractmethod read_narrative_variant_data(key, spec, timestep=None)[source]

Read data array

Parameters:

  • key (str)

  • spec (Spec)

  • timestep (int (optional)) – If None, read data for all timesteps

Returns:

data_array

Return type:

DataArray

abstractmethod read_results(modelrun_name, model_name, output_spec, timestep=None, decision_iteration=None) → DataArray[source]

Return results of a model from a model_run for a given output at a timestep and decision iteration

Parameters:

  • model_run_id (str)

  • model_name (str)

  • output_spec (Spec)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

Return type:

DataArray

abstractmethod read_scenario_variant_data(key, spec, timestep=None, timesteps=None) → DataArray[source]

Read scenario variant data array.

If a single timestep is specified, the spec MAY include ‘timestep’ as a dimension, which should match the timestep specified.

If multiple timesteps are specified, the spec MUST include ‘timestep’ as a dimension, which should match the timesteps specified.

If timestep and timesteps are None, read all available timesteps. Whether or not the spec includes ‘timestep’ as a dimension, the returned DataArray will include a ‘timestep’ dimension with all available timesteps included.

Parameters:

  • key (str)

  • spec (Spec)

  • timestep (int (optional)) – If set, read data for single timestep

  • timesteps (list[int] (optional)) – If set, read data for specified timesteps

Returns:

data_array

Return type:

DataArray

abstractmethod read_state(modelrun_name, timestep, decision_iteration=None) → List[Dict][source]

Read list of (name, build_year) for a given model_run, timestep, decision

Parameters:

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

Return type:

list[dict]

abstractmethod scenario_variant_data_exists(key) → bool[source]

Test if scenario variant data exists

Parameters:

key (str)

Return type:

bool

abstractmethod write_coefficients(source_dim, destination_dim, data)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between a single pair of dimensions.

Parameters:

  • source_dim (str) – dimension name

  • destination_dim (str) – dimension name

  • data (numpy.ndarray)

Notes

To be called from Adaptor implementations.

abstractmethod write_initial_conditions(key, initial_conditions)[source]

Write historical interventions for key

Parameters:
abstractmethod write_interventions(key, interventions)[source]

Write interventions data for key

Parameters:
abstractmethod write_model_parameter_default(key, data_array)[source]

Read data array

Parameters:
Returns:

data_array

Return type:

DataArray

abstractmethod write_narrative_variant_data(key, data_array)[source]

Write data array

Parameters:
abstractmethod write_results(data, modelrun_name, model_name, timestep=None, decision_iteration=None)[source]

Write results of a model_name in model_run_name for a given output_name

Parameters:

  • data_array (DataArray)

  • model_run_id (str)

  • model_name (str)

  • timestep (int, optional)

  • decision_iteration (int, optional)

abstractmethod write_scenario_variant_data(key, data_array)[source]

Write data array

Parameters:
abstractmethod write_state(state: List[Dict], modelrun_name: str, timestep: int, decision_iteration=None)[source]

State is a list of decisions with name and build_year.

State is output from the DecisionManager

Parameters:

  • state (list[dict])

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

smif.data_layer.abstract_metadata_store module

A metadata store holds shared metadata for smif model scenarios, inputs, parameters and outputs: - units - dimension definitions

class smif.data_layer.abstract_metadata_store.MetadataStore[source]

Bases: object

A MetaDataStore must implement each of the abstract methods defined in this interface

abstractmethod delete_dimension(dimension_name)[source]

Delete dimension

Parameters:

dimension_name (str)

abstractmethod read_dimension(dimension_name, skip_coords=False)[source]

Return dimension

Parameters:

  • dimension_name (str)

  • skip_coords (bool, default False) – If True, skip reading dimension elements (names and metadata)

Returns:

A dimension definition (including elements)

Return type:

Coords

abstractmethod read_dimensions(skip_coords=False)[source]

Read dimensions

Parameters:

skip_coords (bool, default False) – If True, skip reading dimension elements (names and metadata)

Return type:

list[Coords]

abstractmethod read_unit_definitions() → List[str][source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

abstractmethod update_dimension(dimension_name, dimension)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

abstractmethod write_dimension(dimension)[source]

Write dimension to project configuration

Parameters:

dimension (Coords)

write_unit_definitions(definitions: List[str])[source]

Reads custom unit definitions

Parameters:

list[str] – Pint-compatible unit definitions

smif.data_layer.data_array module

DataArray provides a thin wrapper around multidimensional arrays and metadata

class smif.data_layer.data_array.DataArray(spec: ~smif.metadata.spec.Spec, data: <MagicMock id='135493239504048'>)[source]

Bases: object

DataArray provides access to input/parameter/results data, with conversions to common python data libraries (for example: numpy, pandas, xarray).

spec
Type:

smif.metadata.spec.Spec

data
Type:

numpy.ndarray

as_df() → <MagicMock id='135493245860944'>[source]

Access DataArray as a pandas.DataFrame

as_dict()[source]
as_ndarray() → <MagicMock id='135493240406144'>[source]

Access as a numpy.ndarray

as_xarray()[source]

Access DataArray as a xarray.DataArray

property coords

Coordinate labels for each dimension.

property description

A human-friendly description

dim_coords(dim)[source]

Coordinates for a given dimension

dim_elements(dim)[source]

Coordinate elements for a given dimension

dim_names(dim)[source]

Coordinate names for a given dimension

property dims

Names for each dimension

classmethod from_df(spec, dataframe)[source]

Create a DataArray from a pandas.DataFrame

classmethod from_xarray(spec, xr_data_array)[source]

Create a DataArray from a xarray.DataArray

property name

The name of the data that this spec describes.

property shape

The shape of the data array

property unit

The unit for all data points.

update(other)[source]

Update data values with any from other which are non-null

validate_as_full()[source]

Check that the data array contains no NaN values

smif.data_layer.data_array.find_duplicate_indices(dataframe)[source]

Find duplicate indices in a DataFrame

Return type:

list[dict]

smif.data_layer.data_array.show_null(dataframe) → <MagicMock id='135493238438800'>[source]

Shows missing data

Return type:

pandas.DataFrame

smif.data_layer.data_handle module

Provide an interface to data, parameters and results

A DataHandle is passed in to a Model at runtime, to provide transparent access to the relevant data and parameters for the current ModelRun and iteration. It gives read access to parameters and input data (at any computed or pre-computed timestep) and write access to output data (at the current timestep).

class smif.data_layer.data_handle.DataHandle(store: Store, modelrun_name, current_timestep, timesteps, model, decision_iteration=None)[source]

Bases: object

Get/set model parameters and data

property base_timestep

Base timestep

property current_timestep

Current timestep

property decision_iteration
derive_for(model)[source]

Derive a new DataHandle configured for the given Model

Parameters:

model (Model) – Model which will use this DataHandle

get_base_timestep_data(input_name)[source]

Get data from the base timestep as required for model inputs

Parameters:

input_name (str)

Return type:

smif.data_layer.data_array.DataArray

get_current_interventions()[source]

Get the interventions that exist in the current state

Returns:

A dict of intervention dicts with build_year attribute keyed by name

Return type:

dict of dicts

get_data(input_name: str, timestep=None) → DataArray[source]

Get data required for model inputs

Parameters:
Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

Raises:

SmifDataError – If any data reading error occurs below this method, the error is handled and reraised within the context of the current call

get_parameter(parameter_name)[source]

Get the value for a parameter

Parameters:

parameter_name (str)

Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

get_parameters()[source]

Get all parameter values

Returns:

parameters – Read-only view of parameters (like a read-only dict)

Return type:

MappingProxyType

get_previous_timestep_data(input_name)[source]

Get data from the previous timestep as required for model inputs

Parameters:

input_name (str)

Return type:

smif.data_layer.data_array.DataArray

get_results(output_name, decision_iteration=None, timestep=None)[source]

Get results values for model outputs

Parameters:

  • output_name (str) – The name of an output for model_name

  • decision_iteration (int, default=None)

  • timestep (int or RelativeTimestep, default=None)

Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

Notes

Access to model results is only granted to models contained within self._model if self._model is a smif.model.model.CompositeModel

get_state()[source]

The current state of the model

If the DataHandle instance has a timestep, then state is established from the state file.

Returns:

A list of (intervention name, build_year) installed in the current timestep

Return type:

list of tuple

Raises:

ValueError – If self._current_timestep is None an error is raised.

property previous_timestep

Previous timestep

read_coefficients(source_dim: str, destination_dim: str) → <MagicMock id='135493267455824'>[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

Return type:

numpy.ndarray

read_unit_definitions() → List[str][source]

Read unit definitions

Return type:

list[str]

set_results(output_name, data)[source]

Set results values for model outputs

Parameters:
property timesteps

All timesteps (as tuple)

write_coefficients(source_dim: str, destination_dim: str, data: <MagicMock id='135493236816448'>)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

  • data (numpy.ndarray)

class smif.data_layer.data_handle.ResultsHandle(store: Store, modelrun_name: str, sos_model, current_timestep: int, timesteps: List[int] | None = None, decision_iteration: int | None = None)[source]

Bases: object

Results access for decision modules

property base_timestep: int
property current_timestep: int
property decision_iteration: int
get_results(model_name: str, output_name: str, timestep: int | RelativeTimestep, decision_iteration: int) → DataArray[source]

Access model results

Parameters:
Returns:

Contains data annotated with the metadata and provides utility methods to access the data in different ways

Return type:

smif.data_layer.data_array.DataArray

get_state(timestep: int, decision_iteration: int) → List[Dict][source]

Retrieve the pre-decision state of the model

If the DataHandle instance has a timestep, then state is established from the state file.

Returns:

A list of {‘name’, ‘build_year’} dictionaries showing the history of decisions made up to this point

Return type:

List[Dict]

property previous_timestep: None | int

smif.data_layer.database_interface module

smif.data_layer.memory_interface module

Memory-backed store implementations

class smif.data_layer.memory_interface.MemoryConfigStore[source]

Bases: ConfigStore

Config store in memory

delete_model(model_name)[source]

Delete a model

Parameters:

model_name (str)

delete_model_run(model_run_name)[source]

Delete a system-of-system model run

Parameters:

model_run_name (str)

delete_scenario(scenario_name)[source]

Delete scenario from project configuration

Parameters:

scenario_name (str)

delete_scenario_variant(scenario_name, variant_name)[source]

Delete scenario from project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

delete_sos_model(sos_model_name)[source]

Delete a system-of-system model

Parameters:

sos_model_name (str)

read_interventions_index(model_name, index_name, ext)[source]
read_model(model_name)[source]

Read a model

Parameters:

model_name (str)

Return type:

Model

read_model_run(model_run_name)[source]

Read a system-of-system model run

Parameters:

model_run_name (str)

Return type:

ModelRun

read_model_runs()[source]

Read all system-of-system model runs

Return type:

list[ModelRun]

read_models()[source]

Read all models

Return type:

list[Model]

read_narrative(sos_model_name, narrative_name)[source]

Read narrative from sos_model

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

read_scenario(scenario_name)[source]

Read a scenario

Parameters:

scenario_name (str)

Return type:

ScenarioModel

read_scenario_variant(scenario_name, variant_name)[source]

Read a scenario variant

Parameters:

  • scenario_name (str)

  • variant_name (str)

Return type:

dict

read_scenario_variants(scenario_name)[source]

Read variants of a given scenario

Parameters:

scenario_name (str)

Return type:

list[dict]

read_scenarios()[source]

Read scenarios

Return type:

list[ScenarioModel]

read_sos_model(sos_model_name)[source]

Read a specific system-of-system model

Parameters:

sos_model_name (str)

Return type:

SosModel

read_sos_models()[source]

Read all system-of-system models

Return type:

list[SosModel]

read_strategies(modelrun_name)[source]

Read strategies for a given model run

Parameters:

model_run_name (str)

Return type:

list[dict]

update_interventions_index(model_name, index_name, int_file, ext)[source]
update_model(model_name, model)[source]

Update a model

Parameters:
update_model_run(model_run_name, model_run)[source]

Update system-of-system model run

Parameters:
update_scenario(scenario_name, scenario)[source]

Update scenario

Parameters:
update_scenario_variant(scenario_name, variant_name, variant)[source]

Update scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variant (dict)

update_sos_model(sos_model_name, sos_model)[source]

Update system-of-system model

Parameters:
write_model(model)[source]

Write a model

Parameters:

model (Model)

write_model_run(model_run)[source]

Write system-of-system model run

Parameters:

model_run (ModelRun)

write_scenario(scenario)[source]

Write scenario

Parameters:

scenario (ScenarioModel)

write_scenario_variant(scenario_name, variant)[source]

Write scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant (dict)

write_sos_model(sos_model)[source]

Write system-of-system model

Parameters:

sos_model (SosModel)

write_strategies(modelrun_name, strategies)[source]

Write strategies for a given model_run

Parameters:
class smif.data_layer.memory_interface.MemoryDataStore[source]

Bases: DataStore

Store data in-memory

available_results(model_run_name)[source]

List available results from a model run

Returns:

Each tuple is (timestep, decision_iteration, model_name, output_name)

Return type:

list[tuple]

delete_results(model_run_name, model_name, output_name, timestep=None, decision_iteration=None)[source]

Delete results for a single timestep/iteration of a model output in a model run

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

initial_conditions_data_exists(key)[source]
interventions_data_exists(key)[source]
model_parameter_default_data_exists(key)[source]
narrative_variant_data_exists(key)[source]
read_coefficients(source_dim, destination_dim)[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between a single pair of dimensions.

Parameters:

  • source_dim (str) – dimension name

  • destination_dim (str) – dimension name

Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

read_initial_conditions(keys)[source]

Read historical interventions for key

Parameters:

key (str)

Return type:

list[dict]

read_interventions(keys)[source]

Read interventions data for key

Parameters:

key (str)

Returns:

A dict of intervention dictionaries containing intervention attributes keyed by intervention name

Return type:

dict[str, dict]

read_model_parameter_default(key, spec)[source]

Read data array

Parameters:
Returns:

data_array

Return type:

DataArray

read_narrative_variant_data(key, spec, timestep=None)[source]

Read data array

Parameters:

  • key (str)

  • spec (Spec)

  • timestep (int (optional)) – If None, read data for all timesteps

Returns:

data_array

Return type:

DataArray

read_results(modelrun_name, model_name, output_spec, timestep=None, decision_iteration=None)[source]

Return results of a model from a model_run for a given output at a timestep and decision iteration

Parameters:

  • model_run_id (str)

  • model_name (str)

  • output_spec (Spec)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

Return type:

DataArray

read_scenario_variant_data(key, spec, timestep=None, timesteps=None)[source]

Read scenario variant data array.

If a single timestep is specified, the spec MAY include ‘timestep’ as a dimension, which should match the timestep specified.

If multiple timesteps are specified, the spec MUST include ‘timestep’ as a dimension, which should match the timesteps specified.

If timestep and timesteps are None, read all available timesteps. Whether or not the spec includes ‘timestep’ as a dimension, the returned DataArray will include a ‘timestep’ dimension with all available timesteps included.

Parameters:

  • key (str)

  • spec (Spec)

  • timestep (int (optional)) – If set, read data for single timestep

  • timesteps (list[int] (optional)) – If set, read data for specified timesteps

Returns:

data_array

Return type:

DataArray

read_state(modelrun_name, timestep=None, decision_iteration=None)[source]

Read list of (name, build_year) for a given model_run, timestep, decision

Parameters:

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

Return type:

list[dict]

read_strategy_interventions(strategy)[source]
scenario_variant_data_exists(key)[source]

Test if scenario variant data exists

Parameters:

key (str)

Return type:

bool

strategy_data_exists(strategy)[source]
write_coefficients(source_dim, destination_dim, data)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between a single pair of dimensions.

Parameters:

  • source_dim (str) – dimension name

  • destination_dim (str) – dimension name

  • data (numpy.ndarray)

Notes

To be called from Adaptor implementations.

write_initial_conditions(key, initial_conditions)[source]

Write historical interventions for key

Parameters:
write_interventions(key, interventions)[source]

Write interventions data for key

Parameters:
write_model_parameter_default(key, data)[source]

Read data array

Parameters:
Returns:

data_array

Return type:

DataArray

write_narrative_variant_data(key, data)[source]

Write data array

Parameters:
write_results(data_array, modelrun_name, model_name, timestep=None, decision_iteration=None)[source]

Write results of a model_name in model_run_name for a given output_name

Parameters:

  • data_array (DataArray)

  • model_run_id (str)

  • model_name (str)

  • timestep (int, optional)

  • decision_iteration (int, optional)

write_scenario_variant_data(key, data)[source]

Write data array

Parameters:
write_state(state, modelrun_name, timestep=None, decision_iteration=None)[source]

State is a list of decisions with name and build_year.

State is output from the DecisionManager

Parameters:

  • state (list[dict])

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

write_strategy_interventions(strategy, data)[source]
class smif.data_layer.memory_interface.MemoryMetadataStore[source]

Bases: MetadataStore

Store metadata in-memory

delete_dimension(dimension_name)[source]

Delete dimension

Parameters:

dimension_name (str)

read_dimension(dimension_name, skip_coords=False)[source]

Return dimension

Parameters:

  • dimension_name (str)

  • skip_coords (bool, default False) – If True, skip reading dimension elements (names and metadata)

Returns:

A dimension definition (including elements)

Return type:

Coords

read_dimensions(skip_coords=False)[source]

Read dimensions

Parameters:

skip_coords (bool, default False) – If True, skip reading dimension elements (names and metadata)

Return type:

list[Coords]

read_unit_definitions()[source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

update_dimension(dimension_name, dimension)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

write_dimension(dimension)[source]

Write dimension to project configuration

Parameters:

dimension (Coords)

write_unit_definitions(units)[source]

Reads custom unit definitions

Parameters:

list[str] – Pint-compatible unit definitions

smif.data_layer.model_loader module

ModelLoader reads python modules as specified at runtime, loading and instantiating objects.

class smif.data_layer.model_loader.ModelLoader[source]

Bases: object

Load Model from config

Examples

Call ModelLoader.load() to create, load and return a Model object.

>>> loader = ModelLoader()
>>> sector_model = loader.load(sector_model_config)
>>> conversion_model = loader.load(conversion_model_config)
load(config)[source]

Loads the model class specified by the config, returns an instance of that class using the Model.from_dict method.

Parameters:

config (dict) –

The model configuration data. Must include:

  • name (name for smif internal use)

  • path (absolute path to python module file)

  • classname (name of Model implementation class)

  • anything required by the Model.from_dict classmethod

Return type:

Model

load_model_class(model_name, model_path, classname)[source]

Dynamically load model class

Parameters:

  • model_name (str) – The name used internally to identify the SectorModel

  • model_path (str) – The path to the python module which contains the SectorModel implementation

  • classname (str) – The name of the class of the SectorModel implementation

Returns:

The SectorModel implementation

Return type:

class

smif.data_layer.results module

Results provides a common interface to access results from model runs.

class smif.data_layer.results.Results(store: Store | dict)[source]

Bases: object

Common interface to access results from model runs.

Parameters:

store (Store or dict) – pre-created Store object or dictionary of the form {‘interface’: <interface>, ‘dir’: <dir>} where <interface> is either ‘local_csv’ or ‘local_parquet’, and <dir> is the model base directory

available_results(model_run_name)[source]

Return the results available for a given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

A nested dictionary data structure of the results available for the given model run

get_units(output_name: str)[source]

Return the units of a given output.

Parameters:

output_name (str)

Return type:

str

list_model_runs()[source]

Return a list of model run names.

Return type:

List of model run names

list_outputs(sector_model_name: str)[source]

Return a list of model run names.

Parameters:

sector_model_name (str the requested sector model)

Return type:

List of outputs for the given sector model

list_scenario_outputs(scenario_name: str)[source]

Return a list of outputs of a given scenario.

Parameters:

scenario_name (str the requested scenario)

Return type:

List of outputs for the requested scenario

list_scenarios(model_run_name: str)[source]

Return a dictionary of scenarios for given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

Dictionary of (scenario name, variant) for the given model run.

list_sector_models(model_run_name: str)[source]

Return a list of sector models for given model run.

Parameters:

model_run_name (str the requested model run)

Return type:

List of sector models for the given model run

read_results(model_run_names: list, model_names: list, output_names: list, timesteps: list = None, decisions: list = None, time_decision_tuples: list = None)[source]

Return results from the store as a formatted pandas data frame. There are a number of ways of requesting specific timesteps/decisions. You can specify either:

a list of (timestep, decision) tuples

in which case data for all of those tuples matching the available results will be returned

or:
a list of timesteps

in which case data for all of those timesteps (and any decision iterations) matching the available results will be returned

or:
a list of decision iterations

in which case data for all of those decision iterations (and any timesteps) matching the available results will be returned

or:
a list of timesteps and a list of decision iterations

in which case data for the Cartesian product of those timesteps and those decision iterations matching the available results will be returned

or:
nothing

in which case all available results will be returned

Parameters:

  • model_run_names (list) – the requested model run names

  • model_names (list) – the requested sector model names (exactly one required)

  • output_names (list) – the requested output names (output specs must all match)

  • timesteps (list) – the requested timesteps

  • decisions (list) – the requested decision iterations

  • time_decision_tuples (list) – a list of requested (timestep, decision) tuples

Raises:

  • SmifDataNotFoundError – If data cannot be found in the store when try to read from the store

  • SmifDataMismatchError – Data presented to read, write and update methods is in the incorrect format or of wrong dimensions to that expected

  • SmifDataReadError – When unable to read data e.g. unable to handle file type or connect to database

Return type:

pandas.DataFrame

read_scenario_data(scenario_name: str, variant_name: str, variable_name: str, timesteps: list) → <MagicMock id='135493236657712'>[source]

Return scenario variant data from the store as a formatted pandas data frame.

Parameters:

  • scenario_name (str) – the requested scenario name

  • variant_name (str) – the requested scenario variant name

  • variable_name (str) – the requested output variable name that the requested scenario provides

  • timesteps (list) – the requested timesteps

Raises:

  • SmifDataNotFoundError – If data cannot be found in the store when try to read from the store

  • SmifDataMismatchError – Data presented to read, write and update methods is in the incorrect format or of wrong dimensions to that expected

  • SmifDataReadError – When unable to read data e.g. unable to handle file type or connect to database

Return type:

pandas.DataFrame

validate_names(model_run_names, sec_model_names, output_names)[source]

smif.data_layer.store module

The store provides a common data interface to smif configuration, data and metadata.

raises SmifDataNotFoundError:

If data cannot be found in the store when try to read from the store

raises SmifDataExistsError:

If data already exists in the store when trying to write to the store (use an update method instead)

raises SmifDataMismatchError:

Data presented to read, write and update methods is in the incorrect format or of wrong dimensions to that expected

raises SmifDataReadError:

When unable to read data e.g. unable to handle file type or connect to database

class smif.data_layer.store.Store(config_store, metadata_store: MetadataStore, data_store: DataStore, model_base_folder='.')[source]

Bases: object

Common interface to data store, composed of config, metadata and data store implementations.

Parameters:
available_results(model_run_name)[source]

List available results from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (timestep, decision_iteration, model_name, output_name)

Return type:

list[tuple]

canonical_available_results(model_run_name)[source]

List the results that are available from a model run, collapsing all decision iterations.

This is the unique items from calling available_results, with all decision iterations set to 0.

This method is used to determine whether a model run is complete, given that it is impossible to know how many decision iterations to expect: we simply check that each expected timestep has been completed.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing available results

canonical_expected_results(model_run_name)[source]

List the results that are expected from a model run, collapsing all decision iterations.

For a complete model run, this would coincide with the unique list returned from available_results, where all decision iterations are set to 0.

This method is used to determine whether a model run is complete, given that it is impossible to know how many decision iterations to expect: we simply check that each expected timestep has been completed.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing expected results

canonical_missing_results(model_run_name)[source]

List the results that are missing from a model run, collapsing all decision iterations.

For a complete model run, this is what is left after removing canonical_available_results from canonical_expected_results.

Parameters:

model_run_name (str)

Return type:

set Set of tuples representing missing results

clear_results(model_run_name)[source]

Clear all results from a single model run

Parameters:

model_run_name (str)

completed_jobs(model_run_name)[source]

List completed jobs from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (timestep, decision_iteration, model_name)

Return type:

list[tuple]

convert_initial_conditions_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_interventions_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_model_parameter_default_data(sector_model_name, tgt_store, noclobber=False)[source]
convert_narrative_data(sos_model_name, tgt_store, noclobber=False)[source]
convert_scenario_data(model_run_name, tgt_store, noclobber=False)[source]
convert_strategies_data(model_run_name, tgt_store, noclobber=False)[source]
delete_dimension(dimension_name)[source]

Delete dimension

Parameters:

dimension_name (str)

delete_model(model_name)[source]

Delete a model

Parameters:

model_name (str)

delete_model_run(model_run_name)[source]

Delete a system-of-system model run

Parameters:

model_run_name (str)

delete_results(model_run_name, model_name, output_name, timestep=None, decision_iteration=None)[source]

Delete results for a single timestep/iteration of a model output in a model run

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timestep (int, default=None)

  • decision_iteration (int, default=None)

delete_scenario(scenario_name)[source]

Delete scenario from project configuration

Parameters:

scenario_name (str)

delete_scenario_variant(scenario_name, variant_name)[source]

Delete scenario from project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

delete_sos_model(sos_model_name)[source]

Delete a system-of-system model

Parameters:

sos_model_name (str)

expected_model_outputs(model_run_name)[source]

List expected model outputs from a model run

Parameters:

model_run_name (str)

Returns:

Each tuple is (model_name, output_name)

Return type:

list[tuple]

static filter_complete_available_results(available_results, expected_model_outputs)[source]

Filter available results from a model run to include only complete timestep/decision iteration combinations

Parameters:

  • available_results (list[tuple]) – List of (timestep, decision_iteration, model_name, output_name)

  • expected_model_outputs (list[tuple]) – List or set of (model_name, output_name)

Returns:

Each tuple is (timestep, decision_iteration, model_name)

Return type:

list[tuple]

classmethod from_dict(config)[source]

Create Store from configuration dict

get_result_darray(model_run_name, model_name, output_name, timesteps=None, decision_iterations=None, time_decision_tuples=None)[source]

Return data for multiple timesteps and decision iterations for a given output from a given sector model in a specific model run.

You can specify either:
a list of (timestep, decision) tuples

in which case data for all of those tuples matching the available results will be returned

or:
a list of timesteps

in which case data for all of those timesteps (and any decision iterations) matching the available results will be returned

or:
a list of decision iterations

in which case data for all of those decision iterations (and any timesteps) matching the available results will be returned

or:
a list of timesteps and a list of decision iterations

in which case data for the Cartesian product of those timesteps and those decision iterations matching the available results will be returned

or:
nothing

in which case all available results will be returned

Then, for each tuple, the data array from the corresponding read_results call is stacked, and together with the new spec this information is returned as a new DataArray.

Parameters:

  • model_run_name (str)

  • model_name (str)

  • output_name (str)

  • timesteps (optional list of timesteps)

  • decision_iterations (optional list of decision iterations)

  • time_decision_tuples (optional list of unique (timestep, decision) tuples)

Return type:

DataArray with expanded spec and the data requested

get_results(model_run_names: list, model_name: str, output_names: list, timesteps: list = None, decisions: list = None, time_decision_tuples: list = None)[source]

Return data for multiple timesteps and decision iterations for a given output from a given sector model for multiple model runs.

Parameters:

  • model_run_names (list[str]) – the requested model run names

  • model_name (str) – the requested sector model name

  • output_names (list[str]) – the requested output names (output specs must all match)

  • timesteps (list[int]) – the requested timesteps

  • decisions (list[int]) – the requested decision iterations

  • time_decision_tuples (list[tuple]) – a list of requested (timestep, decision) tuples

Returns:

Nested dictionary of DataArray objects, keyed on model run name and output name. Returned DataArrays include one extra (timestep, decision_iteration) dimension.

Return type:

dict

prepare_model_runs(model_run_name, scenario_name, first_var, last_var)[source]

Write multiple model run config files corresponding to multiple scenario variants of {scenario_name}, based on template {model_run_name}

Write batchfile containing each of the generated model runs

Parameters:

  • model_run_name (str)

  • scenario_name (str)

  • first_var (int - between 0 and number of variants-1)

  • last_var (int - between first_var and number of variants-1)

prepare_scenario(scenario_name, list_of_variants)[source]

Modify {scenario_name} config file to include multiple scenario variants.

Parameters:

  • scenario_name (str)

  • list_of_variants (list[int] - indices of scenario variants)

prepare_warm_start(model_run_name)[source]

Copy the results from the previous model_run if available

The method allows a previous unsuccessful model_run to ‘warm start’ a new model run from a later timestep. Model results are recovered from the timestep that the previous model_run was run until, and the new model run runs from the returned timestep

Parameters:

model_run_name (str)

Return type:

int The timestep to which the data store was recovered

Notes

Called from smif.controller.execute

read_all_initial_conditions(model_run_name) → List[Dict][source]

A list of all historical interventions

Return type:

list[dict]

read_coefficients(source_dim: str, destination_dim: str) → <MagicMock id='135493238206208'>[source]

Reads coefficients from the store

Coefficients are uniquely identified by their source/destination dimensions. This method and write_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

read_dimension(dimension_name, skip_coords=False)[source]

Return dimension

Parameters:

dimension_name (str)

Returns:

A dimension definition (including elements)

Return type:

Coords

read_dimensions(skip_coords=False)[source]

Read dimensions

Return type:

list[Coords]

read_initial_conditions(model_name) → List[Dict][source]

Read historical interventions for model_name

Returns:

A list of historical interventions, with keys ‘name’ and ‘build_year’

Return type:

list[dict]

read_initial_conditions_file(model_name, string_id, assert_exists=False)[source]
read_interventions(model_name)[source]

Read interventions data for model_name

Returns:

A dict of intervention dictionaries containing intervention attributes keyed by intervention name

Return type:

dict[str, dict]

read_interventions_file(model_name, string_id, assert_exists=False)[source]
read_model(model_name, skip_coords=False)[source]

Read a model

Parameters:

model_name (str)

Return type:

Model

read_model_parameter_default(model_name, parameter_name, assert_exists=False)[source]

Read default data for a sector model parameter

Parameters:

  • model_name (str)

  • parameter_name (str)

Return type:

DataArray

read_model_run(model_run_name)[source]

Read a system-of-system model run

Parameters:

model_run_name (str)

Return type:

ModelRun

read_model_runs()[source]

Read all system-of-system model runs

Return type:

list[ModelRun]

read_models(skip_coords=False)[source]

Read all models

Return type:

list[Model]

read_narrative(sos_model_name, narrative_name)[source]

Read narrative from sos_model

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

read_narrative_variant_data(sos_model_name, narrative_name, variant_name, parameter_name, timestep=None, assert_exists=False)[source]

Read narrative data file

Parameters:

  • sos_model_name (str)

  • narrative_name (str)

  • variant_name (str)

  • parameter_name (str)

  • timestep (int (optional)) – If None, read data for all timesteps

Return type:

DataArray

read_results(model_run_name: str, model_name: str, output_spec: Spec, timestep: int | None = None, decision_iteration: int | None = None) → DataArray[source]

Return results of a model_name in model_run_name for a given output_name

Parameters:
Return type:

DataArray

read_scenario(scenario_name, skip_coords=False)[source]

Read a scenario

Parameters:

scenario_name (str)

Return type:

ScenarioModel

read_scenario_variant(scenario_name, variant_name)[source]

Read a scenario variant

Parameters:

  • scenario_name (str)

  • variant_name (str)

Return type:

dict

read_scenario_variant_data(scenario_name: str, variant_name: str, variable: str, timestep: int | None = None, timesteps: List[int] | None = None, assert_exists: bool = False) → DataArray | bool[source]

Read scenario data file

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variable (str)

  • timestep (int)

Returns:

data

Return type:

DataArray

read_scenario_variants(scenario_name)[source]

Read variants of a given scenario

Parameters:

scenario_name (str)

Return type:

list[dict]

read_scenarios(skip_coords=False)[source]

Read scenarios

Return type:

list[ScenarioModel]

read_sos_model(sos_model_name)[source]

Read a specific system-of-system model

Parameters:

sos_model_name (str)

Return type:

SosModel

read_sos_models()[source]

Read all system-of-system models

Return type:

list[SosModel]

read_state(model_run_name, timestep, decision_iteration=None) → List[Dict][source]

Read list of (name, build_year) for a given model_run, timestep, decision

Parameters:

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

Return type:

list[dict]

read_strategies(model_run_name)[source]

Read strategies for a given model run

Parameters:

model_run_name (str)

Return type:

list[dict]

read_strategy_interventions(strategy, assert_exists=False)[source]

Read interventions as defined in a model run strategy

read_unit_definitions() → List[str][source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

update_dimension(dimension_name, dimension)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

update_model(model_name, model)[source]

Update a model

Parameters:
update_model_run(model_run_name, model_run)[source]

Update system-of-system model run

Parameters:
update_scenario(scenario_name, scenario)[source]

Update scenario

Parameters:
update_scenario_variant(scenario_name, variant_name, variant)[source]

Update scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant_name (str)

  • variant (dict)

update_sos_model(sos_model_name, sos_model)[source]

Update system-of-system model

Parameters:
write_coefficients(source_dim: str, destination_dim: str, data: <MagicMock id='135493237137168'>)[source]

Writes coefficients to the store

Coefficients are uniquely identified by their source/destination dimensions. This method and read_coefficients implement caching of conversion coefficients between dimensions.

Parameters:

  • source_dim (str) – Dimension name

  • destination_dim (str) – Dimension name

  • data (numpy.ndarray)

Notes

To be called from Adaptor implementations.

write_dimension(dimension)[source]

Write dimension to project configuration

Parameters:

dimension (Coords)

write_initial_conditions(model_name, initial_conditions)[source]

Write historical interventions for a model

Parameters:

list[dict] – A list of historical interventions, with keys ‘name’ and ‘build_year’

write_initial_conditions_file(model_name, string_id, initial_conditions)[source]
write_interventions(model_name, interventions)[source]

Write interventions data for a model

Parameters:

  • dict[str – A dict of intervention dictionaries containing intervention attributes keyed by intervention name

  • dict] – A dict of intervention dictionaries containing intervention attributes keyed by intervention name

write_interventions_file(model_name, string_id, interventions)[source]
write_model(model)[source]

Write a model

Parameters:

model (Model)

write_model_parameter_default(model_name, parameter_name, data)[source]

Write default data for a sector model parameter

Parameters:
write_model_run(model_run)[source]

Write system-of-system model run

Parameters:

model_run (ModelRun)

write_narrative_variant_data(sos_model_name, narrative_name, variant_name, data)[source]

Read narrative data file

Parameters:
write_results(data_array, model_run_name, model_name, timestep=None, decision_iteration=None)[source]

Write results of a model_name in model_run_name for a given output_name

Parameters:

  • data_array (DataArray)

  • model_run_id (str)

  • model_name (str)

  • timestep (int, optional)

  • decision_iteration (int, optional)

write_scenario(scenario)[source]

Write scenario

Parameters:

scenario (ScenarioModel)

write_scenario_variant(scenario_name, variant)[source]

Write scenario to project configuration

Parameters:

  • scenario_name (str)

  • variant (dict)

write_scenario_variant_data(scenario_name, variant_name, data)[source]

Write scenario data file

Parameters:
write_sos_model(sos_model)[source]

Write system-of-system model

Parameters:

sos_model (SosModel)

write_state(state, model_run_name, timestep, decision_iteration=None)[source]

State is a list of decisions with name and build_year.

State is output from the DecisionManager

Parameters:

  • state (list[dict])

  • model_run_name (str)

  • timestep (int)

  • decision_iteration (int, optional)

write_strategies(model_run_name, strategies)[source]

Write strategies for a given model_run

Parameters:
write_strategy_interventions(strategy, data)[source]
Parameters:

list[dicts]

write_unit_definitions(definitions)[source]

Reads custom unit definitions

Parameters:

list[str] – Pint-compatible unit definitions

smif.data_layer.validate module

Validate the correct format and presence of the config data for the system-of-systems model

smif.data_layer.validate.validate_dependency(dep)[source]

Check a dependency specification

smif.data_layer.validate.validate_dependency_spec(input_spec, model_name)[source]

Check the input specification for a single sector model

smif.data_layer.validate.validate_initial_condition(datum, file_path)[source]

Check a single initial condition datum

smif.data_layer.validate.validate_initial_conditions(data, file_path)[source]

Check a list of initial condition observations

smif.data_layer.validate.validate_interval_sets_config(interval_sets)[source]

Check interval sets

smif.data_layer.validate.validate_interventions(data, path)[source]

Validate the loaded data as required for model interventions

smif.data_layer.validate.validate_path_to_timesteps(timesteps)[source]

Check timesteps is a path to timesteps file

smif.data_layer.validate.validate_planning_config(planning)[source]

Check planning options

smif.data_layer.validate.validate_region_sets_config(region_sets)[source]

Check regions sets

smif.data_layer.validate.validate_scenario(scenario)[source]

Check a single scenario specification

smif.data_layer.validate.validate_scenario_data(data, file_path)[source]

Check a list of scenario observations

smif.data_layer.validate.validate_scenario_data_config(scenario_data)[source]

Check scenario data

smif.data_layer.validate.validate_scenario_datum(datum, file_path)[source]

Check a single scenario datum

smif.data_layer.validate.validate_sector_model_initial_config(sector_model_config)[source]

Check a single sector model initial configuration

smif.data_layer.validate.validate_sector_models_initial_config(sector_models)[source]

Check list of sector models initial configuration

smif.data_layer.validate.validate_sos_model_config(sos_model, sector_models, scenarios)[source]

Check expected values for data loaded from master config file

smif.data_layer.validate.validate_sos_model_format(sos_model)[source]
smif.data_layer.validate.validate_time_interval(interval)[source]

Check a single time interval

smif.data_layer.validate.validate_time_intervals(intervals, file_path)[source]

Check time intervals

smif.data_layer.validate.validate_timesteps(timesteps, file_path)[source]

Check timesteps is a list of integers