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
• SmifDataExistsError – If data already exists in the store when trying to write to the store (use an update method instead)
• 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

Summary

Data:

Store

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

Reference

class smif.data_layer.store.Store(config_store, metadata_store: smif.data_layer.abstract_metadata_store.MetadataStore, data_store: smif.data_layer.abstract_data_store.DataStore, model_base_folder='.')

Bases: object

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

Parameters:

• config_store (ConfigStore) –
• metadata_store (MetadataStore) –
• data_store (DataStore) –

classmethod from_dict(config)

Create Store from configuration dict

read_model_runs()

Read all system-of-system model runs

Returns: Return type:list[ModelRun]

read_model_run(model_run_name)

Read a system-of-system model run

Parameters:model_run_name (str) – Returns: Return type:ModelRun

write_model_run(model_run)

Write system-of-system model run

Parameters:model_run (ModelRun) –

update_model_run(model_run_name, model_run)

Update system-of-system model run

Parameters:

• model_run_name (str) –
• model_run (ModelRun) –

delete_model_run(model_run_name)

Delete a system-of-system model run

Parameters:model_run_name (str) –

read_sos_models()

Read all system-of-system models

Returns: Return type:list[SosModel]

read_sos_model(sos_model_name)

Read a specific system-of-system model

Parameters:sos_model_name (str) – Returns: Return type:SosModel

write_sos_model(sos_model)

Write system-of-system model

Parameters:sos_model (SosModel) –

update_sos_model(sos_model_name, sos_model)

Update system-of-system model

Parameters:

• sos_model_name (str) –
• sos_model (SosModel) –

delete_sos_model(sos_model_name)

Delete a system-of-system model

Parameters:sos_model_name (str) –

read_models(skip_coords=False)

Read all models

Returns: Return type:list[Model]

read_model(model_name, skip_coords=False)

Read a model

Parameters:model_name (str) – Returns: Return type:Model

write_model(model)

Write a model

Parameters:model (Model) –

update_model(model_name, model)

Update a model

Parameters:

• model_name (str) –
• model (Model) –

delete_model(model_name)

Delete a model

Parameters:model_name (str) –

read_scenarios(skip_coords=False)

Read scenarios

Returns: Return type:list[ScenarioModel]

read_scenario(scenario_name, skip_coords=False)

Read a scenario

Parameters:scenario_name (str) – Returns: Return type:ScenarioModel

write_scenario(scenario)

Write scenario

Parameters:scenario (ScenarioModel) –

update_scenario(scenario_name, scenario)

Update scenario

Parameters:

• scenario_name (str) –
• scenario (ScenarioModel) –

delete_scenario(scenario_name)

Delete scenario from project configuration

Parameters:scenario_name (str) –

prepare_scenario(scenario_name, list_of_variants)

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

Parameters:

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

prepare_model_runs(model_run_name, scenario_name, first_var, last_var)

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) –

read_scenario_variants(scenario_name)

Read variants of a given scenario

Parameters:scenario_name (str) – Returns: Return type:list[dict]

read_scenario_variant(scenario_name, variant_name)

Read a scenario variant

Parameters:

• scenario_name (str) –
• variant_name (str) –

Returns:

Return type:

dict

write_scenario_variant(scenario_name, variant)

Write scenario to project configuration

Parameters:

• scenario_name (str) –
• variant (dict) –

update_scenario_variant(scenario_name, variant_name, variant)

Update scenario to project configuration

Parameters:

• scenario_name (str) –
• variant_name (str) –
• variant (dict) –

delete_scenario_variant(scenario_name, variant_name)

Delete scenario from project configuration

Parameters:

• scenario_name (str) –
• variant_name (str) –

read_narrative(sos_model_name, narrative_name)

Read narrative from sos_model

Parameters:

• sos_model_name (str) –
• narrative_name (str) –

read_strategies(model_run_name)

Read strategies for a given model run

Parameters:model_run_name (str) – Returns: Return type:list[dict]

write_strategies(model_run_name, strategies)

Write strategies for a given model_run

Parameters:

• model_run_name (str) –
• strategies (list[dict]) –

convert_strategies_data(model_run_name, tgt_store, noclobber=False)

read_unit_definitions() → List[str]

Reads custom unit definitions

Returns:Pint-compatible unit definitions Return type:list[str]

write_unit_definitions(definitions)

Reads custom unit definitions

Parameters:list[str] – Pint-compatible unit definitions

read_dimensions(skip_coords=False)

Read dimensions

Returns: Return type:list[Coords]

read_dimension(dimension_name, skip_coords=False)

Return dimension

Parameters:dimension_name (str) – Returns:A dimension definition (including elements) Return type:Coords

write_dimension(dimension)

Write dimension to project configuration

Parameters:dimension (Coords) –

update_dimension(dimension_name, dimension)

Update dimension

Parameters:

• dimension_name (str) –
• dimension (Coords) –

delete_dimension(dimension_name)

Delete dimension

Parameters:dimension_name (str) –

read_scenario_variant_data(scenario_name: str, variant_name: str, variable: str, timestep: Optional[int] = None, timesteps: Optional[List[int]] = None, assert_exists: bool = False) → Union[smif.data_layer.data_array.DataArray, bool]

Read scenario data file

Parameters:

• scenario_name (str) –
• variant_name (str) –
• variable (str) –
• timestep (int) –

Returns:

data

Return type:

DataArray

write_scenario_variant_data(scenario_name, variant_name, data)

Write scenario data file

Parameters:

• scenario_name (str) –
• variant_name (str) –
• data (DataArray) –

convert_scenario_data(model_run_name, tgt_store, noclobber=False)

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

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

Returns:

Return type:

DataArray

write_narrative_variant_data(sos_model_name, narrative_name, variant_name, data)

Read narrative data file

Parameters:

• sos_model_name (str) –
• narrative_name (str) –
• variant_name (str) –
• data (DataArray) –

convert_narrative_data(sos_model_name, tgt_store, noclobber=False)

read_model_parameter_default(model_name, parameter_name, assert_exists=False)

Read default data for a sector model parameter

Parameters:

• model_name (str) –
• parameter_name (str) –

Returns:

Return type:

DataArray

write_model_parameter_default(model_name, parameter_name, data)

Write default data for a sector model parameter

Parameters:

• model_name (str) –
• parameter_name (str) –
• data (DataArray) –

convert_model_parameter_default_data(sector_model_name, tgt_store, noclobber=False)

read_interventions(model_name)

Read interventions data for model_name

Returns:A dict of intervention dictionaries containing intervention attributes keyed by intervention name Return type:dict[str, dict]

write_interventions(model_name, interventions)

Write interventions data for a model

Parameters:dict] (dict[str,) – A dict of intervention dictionaries containing intervention attributes keyed by intervention name

write_interventions_file(model_name, string_id, interventions)

read_interventions_file(model_name, string_id, assert_exists=False)

convert_interventions_data(sector_model_name, tgt_store, noclobber=False)

read_strategy_interventions(strategy, assert_exists=False)

Read interventions as defined in a model run strategy

write_strategy_interventions(strategy, data)

Parameters:list[dicts] –

read_initial_conditions(model_name) → List[Dict[KT, VT]]

Read historical interventions for model_name

Returns:A list of historical interventions, with keys ‘name’ and ‘build_year’ Return type:list[dict]

write_initial_conditions(model_name, initial_conditions)

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)

read_initial_conditions_file(model_name, string_id, assert_exists=False)

convert_initial_conditions_data(sector_model_name, tgt_store, noclobber=False)

read_all_initial_conditions(model_run_name) → List[Dict[KT, VT]]

A list of all historical interventions

Returns: Return type:list[dict]

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

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

Parameters:

• model_run_name (str) –
• timestep (int) –
• decision_iteration (int, optional) –

Returns:

Return type:

list[dict]

write_state(state, model_run_name, timestep, decision_iteration=None)

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) –

read_coefficients(source_dim: str, destination_dim: str) → <MagicMock id='139764768849104'>

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

Returns:

Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

write_coefficients(source_dim: str, destination_dim: str, data: <MagicMock id='139764768991408'>)

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.

read_results(model_run_name: str, model_name: str, output_spec: smif.metadata.spec.Spec, timestep: Optional[int] = None, decision_iteration: Optional[int] = None) → smif.data_layer.data_array.DataArray

Return results of a model_name in model_run_name for a given output_name

Parameters:

• model_run_name (str) –
• model_name (str) –
• output_spec (smif.metadata.Spec) –
• timestep (int, default=None) –
• decision_iteration (int, default=None) –

Returns:

Return type:

DataArray

write_results(data_array, model_run_name, model_name, timestep=None, decision_iteration=None)

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) –

delete_results(model_run_name, model_name, output_name, timestep=None, decision_iteration=None)

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) –

clear_results(model_run_name)

Clear all results from a single model run

Parameters:model_run_name (str) –

available_results(model_run_name)

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]

completed_jobs(model_run_name)

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]

static filter_complete_available_results(available_results, expected_model_outputs)

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]

expected_model_outputs(model_run_name)

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]

prepare_warm_start(model_run_name)

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) – Returns: Return type:int The timestep to which the data store was recovered

Notes

Called from smif.controller.execute

canonical_available_results(model_run_name)

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) – Returns: Return type:set Set of tuples representing available results

canonical_expected_results(model_run_name)

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) – Returns: Return type:set Set of tuples representing expected results

canonical_missing_results(model_run_name)

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) – Returns: Return type:set Set of tuples representing missing results

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

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) –

Returns:

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)

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