Aggregated Elementary Flows for Brightway

aggregated_flows.py

"""Add aggregated LCIA CFs as elementary flows

Usage:
  aggregated_flows.py add <project_name> <lcia-stem> [--dryrun] [--aggregated-flow-db=<db-name>]
  aggregated_flows.py list-stems <project_name>
  aggregated_flows.py (-h | --help)
  aggregated_flows.py --version

Options:
  -h --help                       Show this screen.
  --version                       Show version.
  --dryrun                        Print what would be done but don't make any changes
  --aggregated-flow-db=<db-name>  Name of new database for aggregated biosphere flows

"""


class Colors:
    """ ANSI color codes """
    RED = "\033[0;31m"
    BLUE = "\033[0;34m"
    PURPLE = "\033[0;35m"
    LIGHT_RED = "\033[1;31m"
    LIGHT_GREEN = "\033[1;32m"
    LIGHT_BLUE = "\033[1;34m"
    LIGHT_PURPLE = "\033[1;35m"
    LIGHT_WHITE = "\033[1;37m"
    BOLD = "\033[1m"
    END = "\033[0m"
    # cancel SGR codes if we don't write to a terminal
    if not __import__("sys").stdout.isatty():
        for _ in dir():
            if isinstance(_, str) and _[0] != "_":
                locals()[_] = ""
    else:
        # set Windows console in VT mode
        if __import__("platform").system() == "Windows":
            kernel32 = __import__("ctypes").windll.kernel32
            kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7)
            del kernel32


c = Colors()


import sys
import uuid
try:
    from docopt import docopt
except ImportError:
    print(f"\nThis script requires {c.BLUE}{c.BOLD}docopt{c.END}.\nPlease install it with one of the following and run the script again:\n\t{c.PURPLE}conda install -c conda-forge docopt{c.END}\n\t{c.LIGHT_PURPLE}pip install docopt{c.END}\n")
    sys.exit(1)
try:
    import bw2data as bd
except ImportError:
    print(f"\nCan't import {c.LIGHT_RED}{c.BOLD}bw2data{c.END}.\nPlease make sure you have activated the correct environment.\n")
    sys.exit(1)


if __name__ == '__main__':
    arguments = docopt(__doc__, version='Aggregated Elementary Flows for Brightway 1.0')
    project_name = arguments['<project_name>']

    if project_name not in bd.projects:
        projects = "\n\t".join(f"{c.LIGHT_BLUE}{p.name}{c.END}" for p in sorted(bd.projects, key=lambda x: x.name.lower()))
        print(f"\nProject name `{c.RED}{project_name}{c.END}` not found. Please correct and run again.\nHere are the available projects:\n\t{projects}\n")
        sys.exit(1)

    bd.projects.set_current(project_name)
    print(f"\nUsing project {c.LIGHT_RED}{project_name}{c.END}\n")

    if arguments['list-stems']:
        stems = "\n\t".join([f"{c.LIGHT_BLUE}\"{stem}\"{c.END}" for stem in sorted({x[0] for x in bd.methods})])
        print(f"Found the following LCIA impact category stems:\n\t{stems}\n")
        sys.exit(0)
    else:
        db_name = arguments['--aggregated-flow-db'] or "aggregated-flows-db"

        if db_name in bd.databases:
            print(f"Using existing database for aggregated flows: {c.RED}{db_name}{c.END}\n")
        else:
            print(f"Creating new database for aggregated flows: {c.LIGHT_RED}{db_name}{c.END}\n")
        db = bd.Database(db_name)
        if not arguments['--dryrun']:
            db.register(
                created_by="Aggregated Elementary Flows for Brightway 1.0",
                lcia_stems=[]
            )
            if arguments['<lcia-stem>'] in bd.databases[db_name]["lcia_stems"]:
                print("This LCIA method has already had aggregated flows added.")
                sys.exit(1)

        print(f"Creating new elementary flows for aggregated LCIA scores.\n")

        for method in sorted(m for m in bd.methods if m[0] == arguments['<lcia-stem>']):
            name = "-".join(method)
            unit = bd.methods[method].get('unit', 'Unknown')
            print(f"""Creating new flow for aggregated LCIA results:
    Method: {c.LIGHT_BLUE}{method[1:]}{c.END}
    Name: {c.LIGHT_RED}{name}{c.END}
    Unit: {c.LIGHT_PURPLE}{unit}{c.END}
            """)
            if not arguments['--dryrun']:
                act = db.new_activity(
                    code=uuid.uuid4().hex,
                    name=name,
                    unit=unit,
                    categories=("aggregated scores", )
                )
                act.save()
                method_data = bd.Method(method).load()
                for row in method_data:
                    if row[0] == act.key:
                        raise ValueError("This flow is already chararcterized")
                bd.Method(method).write(method_data + [(act.key, 1.)])

        if not arguments['--dryrun']:
            bd.databases[db_name]["lcia_stems"].append(arguments['<lcia-stem>'])
            bd.databases.flush()