How to use bw2io - 10 common examples

def run_import(self, import_dir):
        try:
            importer = SingleOutputEcospold2Importer(
                import_dir,
                self.db_name,
                extractor=ActivityBrowserExtractor,
                signal=import_signals.strategy_progress
            )
            importer.apply_strategies()
            importer.write_database(backend='activitybrowser')
            if not import_signals.cancel_sentinel:
                import_signals.finished.emit()
            else:
                self.delete_canceled_db()
        except ImportCanceledError:
            self.delete_canceled_db()
        except InvalidExchange:
            # Likely caused by new version of ecoinvent not finding required
            # biosphere flows.

def run_local_import(self):
        try:
            import_signals.db_progress.emit(0, 0)
            result = ABPackage.import_file(self.archive_path)
            if not import_signals.cancel_sentinel:
                db = next(iter(result))
                if db.name != self.db_name:
                    db.rename(self.db_name)
                import_signals.db_progress.emit(1, 1)
                import_signals.finished.emit()
            else:
                self.delete_canceled_db()
        except InvalidPackage as e:
            self.delete_canceled_db()
            import_signals.import_failure.emit(
                ("Missing databases", str(e))
            )
        except ImportCanceledError:
            self.delete_canceled_db()
        except InvalidExchange:
            self.delete_canceled_db()
            import_signals.import_failure.emit(
                ("Missing exchanges", "The import has failed, likely due missing exchanges.")
            )
        except UnknownObject as e:
            self.delete_canceled_db()
            import_signals.import_failure.emit(
                ("Unknown object", str(e))
            )

def finished(self, result: int = None) -> None:
        outcome = result or 0
        self.thread.exit(outcome)
        self.setMaximum(1)
        self.setValue(1)
        signals.database_changed.emit(bw.config.biosphere)
        signals.databases_changed.emit()

    @Slot(int)
    def update_progress(self, current: int):
        self.setValue(current)


class UpdateBiosphereThread(QtCore.QThread):
    PATCHES = (
        add_ecoinvent_33_biosphere_flows,
        add_ecoinvent_34_biosphere_flows,
        add_ecoinvent_35_biosphere_flows,
        add_ecoinvent_36_biosphere_flows,
    )
    progress = Signal(int)

    def __init__(self, parent=None):
        super().__init__(parent)
        self.total_patches = len(self.PATCHES)

    def run(self):
        try:
            for i, patch in enumerate(self.PATCHES):
                self.progress.emit(i)
                patch()
        except ValidityError as e:

outcome = result or 0
        self.thread.exit(outcome)
        self.setMaximum(1)
        self.setValue(1)
        signals.database_changed.emit(bw.config.biosphere)
        signals.databases_changed.emit()

    @Slot(int)
    def update_progress(self, current: int):
        self.setValue(current)


class UpdateBiosphereThread(QtCore.QThread):
    PATCHES = (
        add_ecoinvent_33_biosphere_flows,
        add_ecoinvent_34_biosphere_flows,
        add_ecoinvent_35_biosphere_flows,
        add_ecoinvent_36_biosphere_flows,
    )
    progress = Signal(int)

    def __init__(self, parent=None):
        super().__init__(parent)
        self.total_patches = len(self.PATCHES)

    def run(self):
        try:
            for i, patch in enumerate(self.PATCHES):
                self.progress.emit(i)
                patch()
        except ValidityError as e:
            print("Could not patch biosphere: {}".format(str(e)))

self.thread.exit(outcome)
        self.setMaximum(1)
        self.setValue(1)
        signals.database_changed.emit(bw.config.biosphere)
        signals.databases_changed.emit()

    @Slot(int)
    def update_progress(self, current: int):
        self.setValue(current)


class UpdateBiosphereThread(QtCore.QThread):
    PATCHES = (
        add_ecoinvent_33_biosphere_flows,
        add_ecoinvent_34_biosphere_flows,
        add_ecoinvent_35_biosphere_flows,
        add_ecoinvent_36_biosphere_flows,
    )
    progress = Signal(int)

    def __init__(self, parent=None):
        super().__init__(parent)
        self.total_patches = len(self.PATCHES)

    def run(self):
        try:
            for i, patch in enumerate(self.PATCHES):
                self.progress.emit(i)
                patch()
        except ValidityError as e:
            print("Could not patch biosphere: {}".format(str(e)))
            self.exit(1)

self.setMaximum(1)
        self.setValue(1)
        signals.database_changed.emit(bw.config.biosphere)
        signals.databases_changed.emit()

    @Slot(int)
    def update_progress(self, current: int):
        self.setValue(current)


class UpdateBiosphereThread(QtCore.QThread):
    PATCHES = (
        add_ecoinvent_33_biosphere_flows,
        add_ecoinvent_34_biosphere_flows,
        add_ecoinvent_35_biosphere_flows,
        add_ecoinvent_36_biosphere_flows,
    )
    progress = Signal(int)

    def __init__(self, parent=None):
        super().__init__(parent)
        self.total_patches = len(self.PATCHES)

    def run(self):
        try:
            for i, patch in enumerate(self.PATCHES):
                self.progress.emit(i)
                patch()
        except ValidityError as e:
            print("Could not patch biosphere: {}".format(str(e)))
            self.exit(1)

valid = False
        if exists:
            base, ext = os.path.splitext(self.path.text())
            valid = True if ext.lower() == ".bw2package" else False
            if not valid:
                import_signals.import_failure.emit(
                    ("Invalid extension", "Expecting 'local' import database file to have '.bw2package' extension")
                )
        self.complete = all([exists, valid])
        self.completeChanged.emit()

    def isComplete(self):
        return self.complete


class ActivityBrowserExtractor(Ecospold2DataExtractor):
    """
    - modified from bw2io
    - qt and python multiprocessing don't like each other on windows
    - need to display progress in gui
    """
    @classmethod
    def extract(cls, dirpath, db_name, *args, **kwargs):
        assert os.path.exists(dirpath), dirpath
        if os.path.isdir(dirpath):
            filelist = [filename for filename in os.listdir(dirpath)
                        if os.path.isfile(os.path.join(dirpath, filename))
                        and filename.split(".")[-1].lower() == "spold"
                        ]
        elif os.path.isfile(dirpath):
            filelist = [dirpath]
        else:

def evaluate_metadata(cls, metadata: dict):
        """ Take the given metadata dictionary and test it against realities
        of the current brightway project.
        """
        if "depends" in metadata:
            missing = set(metadata["depends"]).difference(bw.databases)
            if missing:
                raise InvalidPackage("Package data links to database names that do not exist: {}".format(missing))

def create_LcoptModel_from_BW2Package(import_filename, autosetup=True):
    
    import_data = BW2Package.load_file(import_filename)
    orig_db = import_data[0]['data']
    db_name = import_data[0]['name']
    model = LcoptModel(db_name, autosetup=autosetup)
    db = deepcopy(orig_db)

    temp_param_set = []
    temp_production_param_set = []

    for k, v in db.items():
        exchanges = []
        production_amount = v.get('production amount', 1)

        if production_amount != 1:
            print("NOTE: Production amount for {} is not 1 unit ({})".format(v['name'], production_amount, production_amount))

        temp_production_param_set.append({'of': v['name'], 'amount': production_amount})

Arguments:

        * ``data`` (list): Database output from Ocelot system model
        * ``database_name`` (str): Name of new database. Should not already exist in the current project.

    """
    if not bw2:
        raise ImportError("Brightway2 not found")
    assert isinstance(database_name, str), "Database name must be a string"
    print("Creating database `{}` in project `{}`".format(
        database_name, bw2.projects.current)
    )
    assert database_name not in bw2.databases
    # Don't store two copies in memory
    data[:] = list(Brightway2Converter.convert_to_brightway2(data, database_name))
    importer = LCIImporter(database_name)

    # Already have products
    del importer.strategies[2]

    importer.data = data
    importer.apply_strategies()
    importer.match_database("biosphere3")
    stats = importer.statistics()
    if not stats[2]:
        print("Writing database")
        importer.write_database()
        return bw2.Database(database_name)
    else:
        print("Unlinked exchanges; not writing database")
        print(importer.statistics())
        return importer