notebooks/How to use ecoinvent.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "2ac8fe12-3fc8-4cf7-a7eb-c9cb5d83c607",
   "metadata": {},
   "outputs": [],
   "source": [
    "import ecoinvent_interface as ei\n",
    "settings = ei.Settings(username=\"XXX\", password=\"XXX\")\n",
    "release = ei.EcoinventRelease(settings)\n",
    "release.list_versions()\n",
    "release.list_system_models('3.7.1')\n",
    "RELEASE_PATH = release.get_release(\n",
    "    version='3.7.1',\n",
    "    system_model='apos',\n",
    "    release_type=ei.ReleaseType.ecospold\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "e2c4620a-b7a6-4613-9414-b5cc75ff87b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "release.list_extra_files('3.7.1')\n",
    "LCIA_PATH = release.get_extra(version='3.7.1', filename='ecoinvent 3.7.1_LCIA_implementation.7z')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d9e9600c-ca7e-4fd0-954d-38a598e69dae",
   "metadata": {},
   "outputs": [],
   "source": [
    "from lxml import objectify\n",
    "\n",
    "NS = \"{http://www.EcoInvent.org/EcoSpold02}\"\n",
    "\n",
    "ACTIVITIES_FP = RELEASE_PATH / \"MasterData\" / \"ActivityIndex.xml\"\n",
    "GEOGRAPHIES_FP = RELEASE_PATH / \"MasterData\" / \"Geographies.xml\"\n",
    "ACTIVITY_NAME_FP = RELEASE_PATH / \"MasterData\" / \"ActivityNames.xml\"\n",
    "\n",
    "SPECIAL_ACTIVITY_TYPE_MAP: dict[int, str] = {\n",
    "    0: \"ordinary transforming activity (default)\",\n",
    "    1: \"market activity\",\n",
    "    2: \"IO activity\",\n",
    "    3: \"Residual activity\",\n",
    "    4: \"production mix\",\n",
    "    5: \"import activity\",\n",
    "    6: \"supply mix\",\n",
    "    7: \"export activity\",\n",
    "    8: \"re-export activity\",\n",
    "    9: \"correction activity\",\n",
    "    10: \"market group\",\n",
    "}\n",
    "\n",
    "def maybe_missing(element: objectify.ObjectifiedElement, attribute: str) -> str | None:\n",
    "    try:\n",
    "        return getattr(element, attribute).text\n",
    "    except AttributeError:\n",
    "        return None\n",
    "\n",
    "geographies_mapping = {\n",
    "    elem.get(\"id\"): elem.name.text\n",
    "    for elem in objectify.parse(open(GEOGRAPHIES_FP)).getroot().iterchildren(NS + \"geography\")\n",
    "}\n",
    "activity_names_mapping = {\n",
    "    elem.get(\"id\"): elem.name.text\n",
    "    for elem in objectify.parse(open(ACTIVITY_NAME_FP)).getroot().iterchildren(NS + \"activityName\")\n",
    "}\n",
    "\n",
    "activity_mapping = {\n",
    "    elem.get(\"id\"): {\n",
    "        \"name\": activity_names_mapping[elem.get(\"activityNameId\")],\n",
    "        \"geography\": geographies_mapping[elem.get(\"geographyId\")],\n",
    "        \"start\": elem.get(\"startDate\"),\n",
    "        \"end\": elem.get(\"endDate\"),\n",
    "        \"type\": SPECIAL_ACTIVITY_TYPE_MAP[int(elem.get(\"specialActivityType\"))],\n",
    "    }\n",
    "    for elem in objectify.parse(open(ACTIVITIES_FP)).getroot().iterchildren(NS + \"activityIndexEntry\")\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e99cf79b-49e9-4c2c-8ec6-f2b9a207af7f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'name': '2-butanol production by hydration of butene',\n",
       " 'geography': 'Global',\n",
       " 'start': '1981-01-01',\n",
       " 'end': '2005-12-31',\n",
       " 'type': 'ordinary transforming activity (default)'}"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "activity_mapping[\"f7e93a25-56e4-4268-a603-3bfd57c79eff\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "35e40e74-2a39-4b0e-8b34-74bbd855c045",
   "metadata": {},
   "outputs": [],
   "source": [
    "PRODUCTS_FP = RELEASE_PATH / \"MasterData\" / \"IntermediateExchanges.xml\"\n",
    "\n",
    "def maybe_missing(\n",
    "    element: objectify.ObjectifiedElement, attribute: str, pi: bool | None = False\n",
    "):\n",
    "    try:\n",
    "        if pi:\n",
    "            return element.productInformation.find(NS + \"text\")\n",
    "        else:\n",
    "            return getattr(element, attribute).text\n",
    "    except AttributeError:\n",
    "        return \"\"\n",
    "\n",
    "\n",
    "product_mapping = {\n",
    "    elem.get(\"id\"): {\n",
    "        \"name\": elem.name.text,\n",
    "        \"unit\": elem.unitName.text,\n",
    "        \"comment\": maybe_missing(elem, \"comment\"),\n",
    "        \"product_information\": maybe_missing(elem, \"productInformation\", True),\n",
    "        \"classifications\": dict(\n",
    "            [\n",
    "                (c.classificationSystem.text, c.classificationValue.text)\n",
    "                for c in elem.iterchildren(NS + \"classification\")\n",
    "            ]\n",
    "        ),\n",
    "    }\n",
    "    for elem in objectify.parse(open(PRODUCTS_FP)).getroot().iterchildren()\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "84f092f4-a895-44f6-9677-df53d7050ac5",
   "metadata": {},
   "outputs": [],
   "source": [
    "FLOWS_FP = RELEASE_PATH / \"MasterData\" / \"ElementaryExchanges.xml\"\n",
    "\n",
    "ecosphere_flows_mapping = {\n",
    "    elem.get(\"id\"): {\n",
    "        \"name\": elem.name.text,\n",
    "        \"unit\": elem.unitName.text,\n",
    "        \"chemical_formula\": elem.get(\"formula\") or None,\n",
    "        \"CAS\": elem.get(\"casNumber\") or None,\n",
    "        \"compartments\": [\n",
    "            elem.compartment.compartment.text, \n",
    "            elem.compartment.subcompartment.text\n",
    "        ],\n",
    "        \"synonyms\": [obj.text for obj in elem.iterchildren(NS + \"synonym\")]\n",
    "    }\n",
    "    for elem in objectify.parse(open(FLOWS_FP)).getroot().iterchildren(NS + \"elementaryExchange\")\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "efc82b6b-a7bd-4c52-9a6e-f95148e8a17e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('9aac0778-3c9c-4ca6-b3dd-0be8226231e1', 20),\n",
       " ('2b856090-9c59-4de8-819c-eaf92a8575aa', 19),\n",
       " ('50116c55-67c9-489f-b2f9-ce04f0d62a8b', 17),\n",
       " ('3a06bdf5-24c4-43c6-8bfb-b8e4ec829916', 10),\n",
       " ('e92d2e87-3ff5-4bc8-9a46-affcc4e0b068', 10),\n",
       " ('562af63a-2c99-4896-ad9c-dfddaa86e36d', 9),\n",
       " ('f761c9ab-3ffc-479d-8f51-b4e33fd8d6a5', 9),\n",
       " ('8980bfa0-a957-4a1f-9a63-8c3a26a04cce', 9),\n",
       " ('35aad4e8-0882-4d98-8377-8c9bddd31d3f', 8),\n",
       " ('7e7169e9-ad99-443c-942f-b36e28868b45', 7)]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "product_list, process_list = [], []\n",
    "\n",
    "for filepath in (RELEASE_PATH / \"datasets\").iterdir():\n",
    "    if \".spold\" in filepath.name:\n",
    "        process, product = filepath.name.replace(\".spold\", \"\").split(\"_\")\n",
    "        process_list.append(process)\n",
    "        product_list.append(product)\n",
    "\n",
    "from collections import Counter\n",
    "Counter(process_list).most_common(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2eadc9e9-6dcd-451d-badb-197aac5385de",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'name': 'primary zinc production from concentrate',\n",
       " 'geography': 'Rest-of-World',\n",
       " 'start': '2015-01-01',\n",
       " 'end': '2020-12-31',\n",
       " 'type': 'ordinary transforming activity (default)'}"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "activity_mapping['9aac0778-3c9c-4ca6-b3dd-0be8226231e1']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "1f54d836-05b6-49cc-863e-c63eb6308dfc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ammonium sulfate',\n",
       " 'cadmium',\n",
       " 'cadmium sludge from zinc electrolysis',\n",
       " 'cobalt',\n",
       " 'copper concentrate, sulfide ore',\n",
       " 'copper sulfate',\n",
       " 'copper, cathode',\n",
       " 'gold',\n",
       " 'gypsum, mineral',\n",
       " 'heat, from steam, in chemical industry',\n",
       " 'indium rich leaching residues, from zinc production',\n",
       " 'iron scrap, unsorted',\n",
       " 'lead',\n",
       " 'lead concentrate',\n",
       " 'silver',\n",
       " 'sulfur',\n",
       " 'sulfur dioxide, liquid',\n",
       " 'sulfuric acid',\n",
       " 'zinc',\n",
       " 'zinc monosulfate']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "zinc_products = []\n",
    "\n",
    "for filepath in (RELEASE_PATH / \"datasets\").iterdir():\n",
    "    if \".spold\" in filepath.name:\n",
    "        process, product = filepath.name.replace(\".spold\", \"\").split(\"_\")\n",
    "        if process == '9aac0778-3c9c-4ca6-b3dd-0be8226231e1':\n",
    "            zinc_products.append(product_mapping[product]['name'])\n",
    "\n",
    "\n",
    "sorted(zinc_products)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "4b662cc0-f54e-494c-8c15-db7da8a22172",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'name': 'electricity, high voltage',\n",
       " 'unit': 'kWh',\n",
       " 'comment': '',\n",
       " 'product_information': '',\n",
       " 'classifications': {'By-product classification': 'allocatable product',\n",
       "  'CPC': '17100: Electrical energy'}}"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "product_mapping['66c93e71-f32b-4591-901c-55395db5c132']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "0a52ede9-078d-4e18-815b-7782e4324229",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('66c93e71-f32b-4591-901c-55395db5c132', 2172),\n",
       " ('d69294d7-8d64-4915-a896-9996a014c410', 674),\n",
       " ('759b89bd-3aa6-42ad-b767-5bb9ef5d331d', 410),\n",
       " ('71e2f1db-a2c5-44d0-8337-dfff15be974d', 349),\n",
       " ('9b9edcf3-0539-4642-9516-0df642a5c41a', 228),\n",
       " ('4c2bb0f0-422d-4fb3-bcfe-a36ef1b55874', 197),\n",
       " ('d7432632-40dc-4af8-8125-cb70dd9742c5', 169),\n",
       " ('45fbbc41-7ae9-46cc-bb31-abfa11e69de0', 159),\n",
       " ('637ee275-a239-4dcb-b084-abfa110dd65b', 155),\n",
       " ('dd80f0f2-f4d5-40f0-9035-09c1a7f3f07b', 137)]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "Counter(product_list).most_common(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "56e90b7e-2ace-407f-8894-7a15d7b50fd8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2172"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "Counter(product_list)['66c93e71-f32b-4591-901c-55395db5c132']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "46e266db-0269-4004-826f-e9bedb59c813",
   "metadata": {},
   "outputs": [],
   "source": [
    "import hashlib\n",
    "\n",
    "_ = lambda str: str.encode(\"utf-8\")\n",
    "\n",
    "\n",
    "def unique_identifier(process_dict: dict, product_dict: dict, type: str) -> str:\n",
    "    return hashlib.md5(\n",
    "        _(process_dict[\"name\"])\n",
    "        + _(product_dict[\"name\"])\n",
    "        + _(product_dict[\"unit\"])\n",
    "        + _(process_dict[\"geography\"])\n",
    "        + _(type)\n",
    "    ).hexdigest()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "261c2e3c-be68-4a57-ad5f-90a907086236",
   "metadata": {},
   "outputs": [],
   "source": [
    "from dataclasses import dataclass\n",
    "\n",
    "@dataclass\n",
    "class TechnosphereEdge:\n",
    "    source: str  # Our unique identifier\n",
    "    target: str  # Our unique identifier\n",
    "    amount: float\n",
    "    positive: bool = True\n",
    "\n",
    "@dataclass\n",
    "class EcosphereEdge:\n",
    "    flow: str     # ecoinvent UUID\n",
    "    process: str  # Our unique identifier\n",
    "    amount: float"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "c3070cd5-e9a5-49e7-b1aa-63032b363ffa",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyecospold\n",
    "from pyecospold.model_v2 import IntermediateExchange, Activity, FlowData\n",
    "from tqdm import tqdm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "4086d064-7c0b-4714-a321-cf67ddb4e78b",
   "metadata": {},
   "outputs": [],
   "source": [
    "process_nodes, product_nodes = {}, {}\n",
    "technosphere_edges, ecosphere_edges = [], []\n",
    "INPUTS = (\"Materials/Fuels\", \"Electricity/Heat\", \"Services\", \"From Technosphere (unspecified)\")\n",
    "\n",
    "\n",
    "def get_process_id(edge: IntermediateExchange, activity: Activity) -> str:\n",
    "    return edge.activityLinkId or activity.id\n",
    "\n",
    "\n",
    "def reference_product(flows: FlowData) -> str:\n",
    "    candidates = [\n",
    "        edge for edge in flows.intermediateExchanges\n",
    "        if edge.groupStr == \"ReferenceProduct\"\n",
    "        and edge.amount != 0\n",
    "    ]\n",
    "    if not len(candidates) == 1:\n",
    "        raise ValueError(\"Can't find reference product\")\n",
    "    return candidates[0].intermediateExchangeId"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f393e6e9-d5fd-4467-a3d5-6c1e2d5e2aef",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "19271it [02:12, 145.85it/s]\n"
     ]
    }
   ],
   "source": [
    "for filepath in tqdm((RELEASE_PATH / \"datasets\").iterdir()):\n",
    "    if not filepath.name.endswith(\".spold\"):\n",
    "        continue\n",
    "    ecospold = pyecospold.parse_file_v2(filepath)\n",
    "    activity = ecospold.activityDataset.activityDescription.activity[0]\n",
    "    \n",
    "    this_process = activity_mapping[activity.id]\n",
    "    this_product = product_mapping[reference_product(ecospold.activityDataset.flowData)]\n",
    "\n",
    "    this_process_id = unique_identifier(this_process, this_product, \"process\")\n",
    "    this_product_id = unique_identifier(this_process, this_product, \"product\")\n",
    "\n",
    "    process_nodes[this_process_id] = (this_process, this_product)\n",
    "    product_nodes[this_product_id] = (this_process, this_product)\n",
    "    \n",
    "    for edge in ecospold.activityDataset.flowData.intermediateExchanges:\n",
    "        other_process = activity_mapping[get_process_id(edge=edge, activity=activity)]\n",
    "        other_product = product_mapping[edge.intermediateExchangeId]\n",
    "        other_product_id = unique_identifier(other_process, other_product, \"product\")\n",
    "        \n",
    "        is_input_edge = edge.groupStr in INPUTS\n",
    "        if is_input_edge:\n",
    "            technosphere_edges.append(TechnosphereEdge(\n",
    "                source=other_product_id,\n",
    "                target=this_process_id,\n",
    "                amount=edge.amount,\n",
    "                positive=False\n",
    "            ))\n",
    "        else:\n",
    "            technosphere_edges.append(TechnosphereEdge(\n",
    "                source=this_process_id,\n",
    "                target=other_product_id,\n",
    "                amount=edge.amount,\n",
    "                positive=True\n",
    "            ))\n",
    "\n",
    "    for edge in ecospold.activityDataset.flowData.elementaryExchanges:\n",
    "        ecosphere_edges.append(EcosphereEdge(\n",
    "            flow=edge.elementaryExchangeId,\n",
    "            process=this_process_id,\n",
    "            amount=edge.amount\n",
    "        ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "2cd26e8c-e13c-4a5e-85ed-5163d3b135ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas\n",
    "\n",
    "characterization_factors = pandas.read_excel(\n",
    "    LCIA_PATH / \"LCIA_implementation_3.7.1.xlsx\", sheet_name=\"CFs\"\n",
    ")\n",
    "characterization_units = pandas.read_excel(\n",
    "    LCIA_PATH / \"LCIA_implementation_3.7.1.xlsx\", sheet_name=\"units\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "c99d0455-83f8-4fd9-a7c9-13b301d740f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass\n",
    "class CharacterizationFactor:\n",
    "    flow: str\n",
    "    amount: float\n",
    "\n",
    "\n",
    "lcia_reverse_mapping = {\n",
    "    (v['name'],) + tuple(v[\"compartments\"]): k\n",
    "    for k, v in ecosphere_flows_mapping.items()\n",
    "}\n",
    "\n",
    "impact_categories = {\n",
    "    tuple(obj[:3]): {\n",
    "        'cfs': []\n",
    "    }\n",
    "    for obj in characterization_factors.values.tolist()\n",
    "}\n",
    "\n",
    "for obj in characterization_factors.values.tolist():\n",
    "    impact_categories[tuple(obj[:3])]['cfs'].append(\n",
    "        CharacterizationFactor(\n",
    "            flow=lcia_reverse_mapping[tuple(obj[3:6])],\n",
    "            amount=obj[6]\n",
    "        )\n",
    "    )\n",
    "\n",
    "for obj in characterization_factors.values.tolist():\n",
    "    impact_categories[tuple(obj[:3])]['unit'] = obj[3]"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}