smif.data_layer.file package

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

class smif.data_layer.file.CSVDataStore(base_folder)[source]

Bases: FileDataStore

CSV text file data store

class smif.data_layer.file.FileMetadataStore(base_folder)[source]

Bases: MetadataStore

File-based metadata store (supports YAML, CSV, or GDAL-compatible files)

delete_dimension(dimension_name: str)[source]

Delete dimension

Parameters:

dimension_name (str)

read_dimension(dimension_name: str, 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) → List[dict][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() → List[str][source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

update_dimension(dimension_name: str, dimension: Dict)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

write_dimension(dimension: Dict)[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

class smif.data_layer.file.ParquetDataStore(base_folder)[source]

Bases: FileDataStore

Binary file data store

class smif.data_layer.file.YamlConfigStore(base_folder, validation=False)[source]

Bases: ConfigStore

Config backend saving to YAML configuration files.

Parameters:

base_folder (str) – The path to the configuration and data files

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

Read the project configuration

Returns:

The project configuration

Return type:

dict

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_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:

Submodules

smif.data_layer.file.file_config_store module

File-backed config store

class smif.data_layer.file.file_config_store.YamlConfigStore(base_folder, validation=False)[source]

Bases: ConfigStore

Config backend saving to YAML configuration files.

Parameters:

base_folder (str) – The path to the configuration and data files

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

Read the project configuration

Returns:

The project configuration

Return type:

dict

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_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:

smif.data_layer.file.file_data_store module

File-backed data store

class smif.data_layer.file.file_data_store.CSVDataStore(base_folder)[source]

Bases: FileDataStore

CSV text file data store

class smif.data_layer.file.file_data_store.FileDataStore(base_folder)[source]

Bases: DataStore

Abstract file data store

available_results(modelrun_name)[source]

List available results for a given model run

See _get_results_path for path construction.

On the pattern of:

results/<modelrun_name>/<model_name>/ decision_<id>/ output_<output_name>_timestep_<timestep>.csv

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)

get_timesteps_from_data(key, spec_dict)[source]
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_id, model_name, output_spec, timestep, 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, 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_id, 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.file.file_data_store.ParquetDataStore(base_folder)[source]

Bases: FileDataStore

Binary file data store

smif.data_layer.file.file_metadata_store module

File-backed metadata store

class smif.data_layer.file.file_metadata_store.FileMetadataStore(base_folder)[source]

Bases: MetadataStore

File-based metadata store (supports YAML, CSV, or GDAL-compatible files)

delete_dimension(dimension_name: str)[source]

Delete dimension

Parameters:

dimension_name (str)

read_dimension(dimension_name: str, 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) → List[dict][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() → List[str][source]

Reads custom unit definitions

Returns:

Pint-compatible unit definitions

Return type:

list[str]

update_dimension(dimension_name: str, dimension: Dict)[source]

Update dimension

Parameters:

  • dimension_name (str)

  • dimension (Coords)

write_dimension(dimension: Dict)[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