Source code for bw2io.migrations
import os
from bw2data import projects
from bw2data.data_store import DataStore
from bw2data.serialization import JsonWrapper, SerializedDict
from .data import (
get_biosphere_2_3_category_migration_data,
get_biosphere_2_3_name_migration_data,
get_ecoinvent_pre35_migration_data,
get_exiobase_biosphere_migration_data,
get_simapro_ecoinvent_3_migration_data,
get_simapro_water_migration_data,
get_us_lci_migration_data,
)
from .units import get_default_units_migration_data, get_unusual_units_migration_data
[docs]
class _Migrations(SerializedDict):
[docs]
filename = "migrations.json"
[docs]
migrations = _Migrations()
[docs]
class Migration(DataStore):
"""
A migration is a set of data that can be used to modify a database.
Migrations are stored in the ``migrations`` directory of the project. They
are stored as JSON files, and are registered in the ``migrations.json`` file.
Methods
-------
write(data, description)
Write migration data. Requires a description.
load()
Load migration data.
validate()
Validate migration data.
"""
def __init__(self, *args, **kwargs):
super(Migration, self).__init__(*args, **kwargs)
@property
[docs]
def description(self):
return self.metadata["description"]
[docs]
def validate(self, *args, **kwargs):
return
[docs]
def write(self, data, description):
"""
Write migration data. Requires a description.
Parameters
----------
data : dict
Migration data.
description : str
Description of the migration.
"""
try:
self.register()
migrations[self.name]["description"] = description
except:
self.register(description=description)
filepath = os.path.join(self._intermediate_dir, self.filename + ".json")
JsonWrapper.dump(data, filepath)
[docs]
def load(self):
self.register()
filepath = os.path.join(self._intermediate_dir, self.filename + ".json")
return JsonWrapper.load(filepath)
[docs]
def create_core_migrations():
"""
Add pre-defined core migrations data files.
"""
Migration("biosphere-2-3-categories").write(
get_biosphere_2_3_category_migration_data(),
"Change biosphere category and subcategory labels to ecoinvent version 3",
)
Migration("biosphere-2-3-names").write(
get_biosphere_2_3_name_migration_data(),
"Change biosphere flow names to ecoinvent version 3",
)
Migration("simapro-ecoinvent-3.1").write(
get_simapro_ecoinvent_3_migration_data("3.1"),
"Change SimaPro names from ecoinvent 3.1 to ecoinvent names",
)
Migration("simapro-ecoinvent-3.2").write(
get_simapro_ecoinvent_3_migration_data("3.2"),
"Change SimaPro names from ecoinvent 3.2 to ecoinvent names",
)
Migration("simapro-ecoinvent-3.3").write(
get_simapro_ecoinvent_3_migration_data("3.3"),
"Change SimaPro names from ecoinvent 3.3 to ecoinvent names",
)
Migration("simapro-ecoinvent-3.4").write(
get_simapro_ecoinvent_3_migration_data("3.4"),
"Change SimaPro names from ecoinvent 3.4 to ecoinvent names",
)
Migration("simapro-ecoinvent-3.5").write(
get_simapro_ecoinvent_3_migration_data("3.5"),
"Change SimaPro names from ecoinvent 3.5 to ecoinvent names",
)
Migration("simapro-water").write(
get_simapro_water_migration_data(),
"Change SimaPro water flows to more standard names",
)
Migration("us-lci").write(
get_us_lci_migration_data(), "Fix names in US LCI database"
)
Migration("default-units").write(
get_default_units_migration_data(), "Convert to default units"
)
Migration("unusual-units").write(
get_unusual_units_migration_data(), "Convert non-Ecoinvent units"
)
Migration("exiobase-biosphere").write(
get_exiobase_biosphere_migration_data(),
"Change biosphere flow names to ecoinvent version 3",
)
Migration("fix-ecoinvent-flows-pre-35").write(
get_ecoinvent_pre35_migration_data(),
"Update new biosphere UUIDs in Consequential 3.4",
)
