bw2io.importers.base_lci

Attributes

DEFAULT_TARGET_FIELDS
EXCHANGE_SPECIFIC_KEYS

Classes

LCIImporter

Base class for format-specific importers.

Functions

_reformat_biosphere_exc_as_new_node(→ dict)

Module Contents

class bw2io.importers.base_lci.LCIImporter(db_name: str)

Bases: bw2io.importers.base.ImportBase

Base class for format-specific importers.

Defines workflow for applying strategies.

Takes a database name (string) as initialization parameter.

Initialize the ImportBase object.

Parameters:

• *args – Variable length argument list.

• **kwargs – Arbitrary keyword arguments.

Raises:

NotImplemented : – This class should be subclassed.

_prepare_activity_parameters(data: List[dict] = None, delete_existing: bool = True) → List[dict]

_write_activity_parameters(activity_parameters: List[dict]) → None

add_unlinked_activities() → None

Add technosphere flows to self.data.

add_unlinked_flows_to_biosphere_database(biosphere_name: str | None = None, fields: Set[str] = {'name', 'unit', 'categories'}) → None

all_source_fields_in_unlinked_data() → Set[str]

Return set of all field labels (dict keys) in unlinked edges.

create_new_biosphere(biosphere_name: str) → None

Create new biosphere database from unlinked biosphere flows in self.data

create_new_database_for_flows_with_missing_top_level_context(target_db_name: str, placeholder_db_name: str, fields: List[str] = ['name', 'unit', 'categories'], kinds: List[str] = labels.biosphere_edge_types) → None

Create proxy datasets for flows who have corresponding flows in another database, but not with the given top-level context.

In other words, if we are trying to match {‘name’: ‘foo’, ‘categories’: [‘foo’]}, and our corresponding database only has {‘name’: ‘foo’, ‘categories’: [‘bar’]}, then we can create a placeholder dataset in a new database, as no amount of category manipulation will result in a match in the given target database.

create_randonneur_excel_template_for_unlinked(target_fields: List[str] = DEFAULT_TARGET_FIELDS, source_fields: List[str] | None = None, edge_filter: Callable | None = None, filename: str | None = None, output_dir: pathlib.Path | None = None, replace_existing: bool = False) → pathlib.Path

Create Excel template with source data in the randonneur format for unlinked exchanges.

Intended to be used with read_randonneur_excel_template to create a migration file, which can then be applied to resolve unlinked data.

Should only use string values - no conversion for numbers, booleans, etc. if made in either direction.

target_fields is a list of labels for the target fields, which must be filled by the practitioner. Defaults to [“name”, “location”, “reference product”, “unit”]

source_fields is a list of string labels to include when defining the matchings. Defaults to all available fields except for fields in EXCHANGE_SPECIFIC_KEYS. Use .all_source_fields_in_unlinked_data() to get a list of fields to select from.

edge_filter: Optional function to reduce the number of unlinked edges to write to the template. Takes the unlinked edge as input argument.

output_dir: Where to write the template file. Defaults to bw2data.projects.output_dir.

Returns the Path of the created file.

database_class(db_name: str, requested_backend: str = 'sqlite') → bw2data.data_store.ProcessedDataStore

drop_unlinked(i_am_reckless: bool = False) → None

match_database(db_name: str | None = None, fields: List[str] | None = None, ignore_categories: bool = False, relink: bool = False, kind: List[str] | str | None = None, edge_kinds: List[str] | None = None, this_node_kinds: List[str] | None = None, other_node_kinds: List[str] | None = None, processes_to_products: bool = False) → None

Match current database against itself or another database.

If db_name is None, match against current data. Otherwise, db_name should be the name of an existing Database.

fields is a list of fields to use for matching. Field values are case-insensitive, but otherwise must match exactly for a link to be valid. If fields is None, use the default fields of ‘name’, ‘categories’, ‘unit’, ‘reference product’, and ‘location’.

If ignore_categories, link based only on name, unit and location. ignore_categories conflicts with fields.

If relink, relink exchanges even if a link is already present.

kind can be a string or a list of strings. Common values are “technosphere”, “biosphere”, “production”, and “substitution”.

Nothing is returned, but self.data is changed.

match_database_against_only_available_in_given_context_tree(other_db_name: str, fields: List[str] = ['name', 'unit', 'categories'], kinds: List[str] = labels.biosphere_edge_types) → None

For unlinked edges with a categories context (‘a’, ‘b’, …), try to match against flows in other_db_name with categories context (‘a’, ‘c’) if that flow is the only one available in other_db_name within the context tree (‘a’,).

Parameters:

• other_db_name (str) – The name of the database with flows to link to.

• fields (list[str]) – List of field names to use when determining if there is a match. Default is [“name”, “unit”, “categories”].

• kinds (list[str]) – Try to match exchanges with these type values. Default is bw2data.labels.biosphere_edge_types.

match_database_against_top_level_context(other_db_name: str, fields: List[str] = ['name', 'unit', 'categories'], kinds: List[str] = labels.biosphere_edge_types) → None

For unlinked edges with a categories context (‘a’, ‘b’, …), try to match against flows in db_name with categories context (‘a’,).

Parameters:

• other_db_name (str) – The name of the database with flows to link to.

• fields (list[str]) – List of field names to use when determining if there is a match. Default is [“name”, “unit”, “categories”].

• kinds (list[str]) – Try to match exchanges with these type values. Default is bw2data.labels.biosphere_edge_types.

migrate(migration_name: str) → None

randonneur(label: str | None = None, data_registry_path: pathlib.Path | None = None, datapackage: randonneur.Datapackage | None = None, fields: list | None = None, mapping: dict | None = None, node_filter: Callable | None = None, edge_filter: Callable | None = None, verbose: bool = False, case_sensitive: bool = False, add_extra_attributes: bool = True, verbs: List[str] | None = rn.utils.SAFE_VERBS, migrate_edges: bool = True, migrate_nodes: bool = False) → None

Apply a stored transformation from randonneur_data. See the randonneur [README](brightway-lca/randonneur) and the [current registry](brightway-lca/randonneur_data) for more information.

label: Label for the transformation if loading from the randonneur_data registry.

data_registry_path: Filepath for randonneur_data data registry. Default to the library data.

datapackage: In-memory randonneur.Datapackage object if not using the randonneur_data registry.

fields: A list of object keys as strings, used when checking if the given transformation matches the node or edge under consideration. In other words, only use the fields in fields when checking the source values in each transformation for a match. Each field in fields doesn’t have to be in each transformation.

If you changed labels in mapping, use the changed labels, not the original key labels.

mapping: Change the labels in the migrations data to match your data schema. mapping can change the labels in the migration source and target sections. The mapping input should be a dict with keys “source” and “target”, and have values of {old_label: new_label} pairs:

node_filter: A callable which determines whether or not the given node should be modified. Applies to both verbs and edges, with the exception of node creation - it doesn’t make sense to filter existing nodes as we are creating new objects.

node_filter needs to be a callable which takes a node object and returns a boolean which tells if the node should be modified. In this example, the filter returns False and the node isn’t modified:

edge_filter: A callable which determines whether or not the given edge should be modified. Applies only to edge transformations, and does not apply to edge creation, as this function is always called on the edge to modified, not on the transformation object. Returns

edge_filter needs to be a callable which takes an edge object and returns a boolean which indicates if the edge should be modified.

verbose: Display progress bars and more logging messages.

case_sensitive: Flag indicating whether to do case sensitive matching of transformations to nodes or edges in the graph. Default is false, as practical experience has shown us that cases get commonly changed by software developers or users. Only applies to string values.

add_extra_attributes: Flag indicating whether to include additional attributes when doing replace, update, and disaggregate changes. Extra attributes are defined outside the “source” and “target” transformation keys. Note that keys in randonneur.utils.EXCLUDED_ATTRS are never added.

verbs: The list of transformation types from migrations to apply. Transformations are run in the order as given in verbs, and in some complicated cases you may want to keep the same verbs but change their order to get the desired output state. In general, such complicated transformations should be broken down to smaller discrete and independent transformations whenever possible, and logs checked carefully after their application.

The default value of verbs are the “safe” transformations - replace, update, and disaggregate. To get create and delete you need to specify them in the configuration.

Only the verbs create, disaggregate, replace, update, and delete are used in our functions, regardless of what is given in verbs, as we don’t know how to handle custom verbs. We need to write custom functions for each verb as they have difference behaviour.

migrate_edges: Flag on whether to apply this transformation to edges, if allowed by the transformation metadata.

migrate_nodes: Flag on whether to apply this transformation to nodes, if allowed by the transformation metadata.

statistics(print_stats: bool = True) → Tuple[int, int, int, int]

write_database(data: dict | None = None, delete_existing: bool = True, backend: str | None = None, activate_parameters: bool = False, db_name: str | None = None, searchable: bool = True, check_typos: bool = True, **kwargs) → bw2data.data_store.ProcessedDataStore

Write data to a Database.

All arguments are optional, and are normally not specified.

delete_existing effects both the existing database (it will be emptied prior to writing if True, which is the default), and, if activate_parameters is True, existing database and activity parameters. Database parameters will only be deleted if the import data specifies a new set of database parameters (i.e. database_parameters is not None) - the same is true for activity parameters. If you need finer-grained control, please use the DatabaseParameter, etc. objects directly.

Parameters:

• data (*) – The data to write to the Database. Default is self.data.

• delete_existing (*) – See above.

• activate_parameters (*)

• backend (*) – Storage backend to use when creating Database. Default is the default backend.

Returns:

Database instance.

write_database_parameters(activate_parameters: bool = False, delete_existing: bool = True) → None

write_excel(only_unlinked: bool = False, only_names: bool = False) → pathlib.Path

Write database information to a spreadsheet.

If only_unlinked, then only write unlinked exchanges.

If only_names, then write only activity names, no exchange data.

Returns the filepath to the spreadsheet file.

write_project_parameters(data: List[dict] = None, delete_existing: bool = True) → None

Write global parameters to ProjectParameter database table.

delete_existing controls whether new parameters will delete_existing existing parameters, or just update values. The name field is used to determine if a parameter exists.

data should be a list of dictionaries (self.project_parameters is used by default):

[{
    'name': name of variable (unique),
    'amount': numeric value of variable (optional),
    'formula': formula in Python as string (optional),
    optional keys like uncertainty, etc. (no limitations)
}]

property all_linked: bool

database_parameters = None

db_name

format = 'Generic LCIImporter'

metadata

property needs_multifunctional_database: bool

project_parameters = None

strategies

bw2io.importers.base_lci._reformat_biosphere_exc_as_new_node(exc: dict, db_name: str) → dict

bw2io.importers.base_lci.DEFAULT_TARGET_FIELDS = ('name', 'location', 'reference product', 'unit')

bw2io.importers.base_lci.EXCHANGE_SPECIFIC_KEYS = ('amount', 'functional', 'loc', 'maximum', 'minimum', 'output', 'scale', 'shape',...