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

Classes:

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, data_store, model_base_folder='.')[source]

Bases: object

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

Parameters:
read_model_runs()[source]

Read all system-of-system model runs

Returns:
Return type:list[ModelRun]
read_model_run(model_run_name)[source]

Read a system-of-system model run

Parameters:model_run_name (str) –
Returns:
Return type:ModelRun
write_model_run(model_run)[source]

Write system-of-system model run

Parameters:model_run (ModelRun) –
update_model_run(model_run_name, model_run)[source]

Update system-of-system model run

Parameters:
delete_model_run(model_run_name)[source]

Delete a system-of-system model run

Parameters:model_run_name (str) –
read_sos_models()[source]

Read all system-of-system models

Returns:
Return type:list[SosModel]
read_sos_model(sos_model_name)[source]

Read a specific system-of-system model

Parameters:sos_model_name (str) –
Returns:
Return type:SosModel
write_sos_model(sos_model)[source]

Write system-of-system model

Parameters:sos_model (SosModel) –
update_sos_model(sos_model_name, sos_model)[source]

Update system-of-system model

Parameters:
delete_sos_model(sos_model_name)[source]

Delete a system-of-system model

Parameters:sos_model_name (str) –
read_models(skip_coords=False)[source]

Read all models

Returns:
Return type:list[Model]
read_model(model_name, skip_coords=False)[source]

Read a model

Parameters:model_name (str) –
Returns:
Return type:Model
write_model(model)[source]

Write a model

Parameters:model (Model) –
update_model(model_name, model)[source]

Update a model

Parameters:
  • model_name (str) –
  • model (Model) –
delete_model(model_name)[source]

Delete a model

Parameters:model_name (str) –
read_scenarios(skip_coords=False)[source]

Read scenarios

Returns:
Return type:list[ScenarioModel]
read_scenario(scenario_name, skip_coords=False)[source]

Read a scenario

Parameters:scenario_name (str) –
Returns:
Return type:ScenarioModel
write_scenario(scenario)[source]

Write scenario

Parameters:scenario (ScenarioModel) –
update_scenario(scenario_name, scenario)[source]

Update scenario

Parameters:
  • scenario_name (str) –
  • scenario (ScenarioModel) –
delete_scenario(scenario_name)[source]

Delete scenario from project configuration

Parameters:scenario_name (str) –
read_scenario_variants(scenario_name)[source]

Read variants of a given scenario

Parameters:scenario_name (str) –
Returns:
Return type:list[dict]
read_scenario_variant(scenario_name, variant_name)[source]

Read a scenario variant

Parameters:
  • scenario_name (str) –
  • variant_name (str) –
Returns:
Return type:

dict
write_scenario_variant(scenario_name, variant)[source]

Write scenario to project configuration

Parameters:
  • scenario_name (str) –
  • variant (dict) –
update_scenario_variant(scenario_name, variant_name, variant)[source]

Update scenario to project configuration

Parameters:
  • scenario_name (str) –
  • variant_name (str) –
  • variant (dict) –
delete_scenario_variant(scenario_name, variant_name)[source]

Delete scenario from project configuration

Parameters:
  • scenario_name (str) –
  • variant_name (str) –
read_narrative(sos_model_name, narrative_name)[source]

Read narrative from sos_model

Parameters:
  • sos_model_name (str) –
  • narrative_name (str) –
read_strategies(model_run_name)[source]

Read strategies for a given model run

Parameters:model_run_name (str) –
Returns:
Return type:list[dict]
write_strategies(model_run_name, strategies)[source]

Write strategies for a given model_run

Parameters:
read_unit_definitions()[source]

Reads custom unit definitions

Returns:Pint-compatible unit definitions
Return type:list[str]
write_unit_definitions(definitions)[source]

Reads custom unit definitions

Parameters:list[str] – Pint-compatible unit definitions
read_dimensions(skip_coords=False)[source]

Read dimensions

Returns:
Return type:list[Coords]
read_dimension(dimension_name, skip_coords=False)[source]

Return dimension

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

Write dimension to project configuration

Parameters:dimension (Coords) –
update_dimension(dimension_name, dimension)[source]

Update dimension

Parameters:
  • dimension_name (str) –
  • dimension (Coords) –
delete_dimension(dimension_name)[source]

Delete dimension

Parameters:dimension_name (str) –
read_scenario_variant_data(scenario_name, variant_name, variable, timestep=None) → smif.data_layer.data_array.DataArray[source]

Read scenario data file

Parameters:
  • scenario_name (str) –
  • variant_name (str) –
  • variable (str) –
  • timestep (int (optional)) – If None, read data for all timesteps
Returns:

data
Return type:

DataArray
write_scenario_variant_data(scenario_name, variant_name, data, timestep=None)[source]

Write scenario data file

Parameters:
  • scenario_name (str) –
  • variant_name (str) –
  • data (DataArray) –
  • timestep (int (optional)) – If None, write data for all timesteps
read_narrative_variant_data(sos_model_name, narrative_name, variant_name, parameter_name, timestep=None)[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
Returns:
Return type:

DataArray
write_narrative_variant_data(sos_model_name, narrative_name, variant_name, data, timestep=None)[source]

Read narrative data file

Parameters:
  • sos_model_name (str) –
  • narrative_name (str) –
  • variant_name (str) –
  • data (DataArray) –
  • timestep (int (optional)) – If None, write data for all timesteps
read_model_parameter_default(model_name, parameter_name)[source]

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)[source]

Write default data for a sector model parameter

Parameters:
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]
write_interventions(model_name, interventions)[source]

Write interventions data for a model

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

Read interventions as defined in a model run strategy

read_initial_conditions(model_name)[source]

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)[source]

Write historical interventions for a model

Parameters:list[dict] – A list of historical interventions, with keys ‘name’ and ‘build_year’
read_all_initial_conditions(model_run_name)[source]

A list of all historical interventions

Returns:
Return type:list[dict]
read_state(model_run_name, timestep, 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) –
Returns:
Return type:

list[dict]
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) –
read_coefficients(source_dim: str, destination_dim: str) → numpy.ndarray[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
Returns:
Return type:

numpy.ndarray

Notes

To be called from Adaptor implementations.

write_coefficients(source_dim: str, destination_dim: str, data: numpy.ndarray)[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.

read_results(model_run_name, model_name, output_spec, timestep=None, decision_iteration=None) → smif.data_layer.data_array.DataArray[source]

Return results of a model_name in model_run_name for a given output_name

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

Notes

Called from smif.controller.execute