Merge remote-tracking branch 'origin/pathways' into pathways

premise/data/metals/activities_mapping.yml

@@ -15,26 +15,23 @@ EV:
 Wind-DDPM:
   ecoinvent_aliases:
     fltr:
-      - wind power plant
-      - wind turbine
+      - direct drive
     mask:
       - fixed
-      - network
-      - concrete
-      - onshore
-      - 800kW
       - market
+      - electricity
 Wind-Gearbox:
   ecoinvent_aliases:
     fltr:
-      - wind power plant
-      - wind turbine
+      - wind power plant construction
+      - wind turbine construction
     mask:
       - fixed
-      - network
-      - concrete
-      - offshore
       - market
+      - electricity
+      - direct drive
+      - precast
+      - small-scale
 #LMO:
 #  ecoinvent_aliases:
 #    fltr:
@@ -196,6 +193,7 @@ c-Si:
       - laminate
       - panel
       - slanted-roof
+      - perovskite
 a-Si:
   ecoinvent_aliases:
     fltr:
@@ -214,6 +212,7 @@ CdTe:
       - photovoltaic laminate, CdTe
     mask:
       - market
+      - Series
 perovskite:
   ecoinvent_aliases:
     fltr:

premise/data/renewables/efficiency_solar_PV.csv

@@ -8,6 +8,8 @@ CdTe;2010;0.1
 micro-Si;2020;0.119
 single-Si;2020;0.179
 multi-Si;2020;0.168
+perovskite;2020;0.25
+GaAs;2020;0.28
 CIGS;2020;0.14
 CIS;2020;0.14
 CdTe;2020;0.168
@@ -17,3 +19,5 @@ multi-Si;2050;0.244
 CIGS;2050;0.234
 CIS;2050;0.234
 CdTe;2050;0.21
+perovskite;2050;0.25
+GaAs;2050;0.28

premise/electricity.py

@@ -1402,6 +1402,8 @@ def update_efficiency_of_solar_pv(self) -> None:
             "CIGS",
             "CIS",
             "CdTe",
+            "perovskite",
+            "GaAs",
         ]
 
         # TODO: check if IAM data provides efficiencies for PV panels and use them instead
@@ -1423,7 +1425,12 @@ def update_efficiency_of_solar_pv(self) -> None:
         )
 
         for dataset in datasets:
-            power = float(re.findall(r"[-+]?\d*\.\d+|\d+", dataset["name"])[0])
+            numbers = re.findall(r"[-+]?\d*\.\d+|\d+", dataset["name"])
+            if not numbers:
+                print(f"No numerical value found in dataset name: {dataset['name']}")
+                continue
+
+            power = float(numbers[0])
 
             if "mwp" in dataset["name"].lower():
                 power *= 1000

premise/wind_turbines.py

@@ -87,24 +87,99 @@ def create_direct_drive_turbines(self):
                     ]
                 ),
             ):
-                # create a true copy of the dataset
-                dataset_copy = copy.deepcopy(dataset)
-
-                dataset_copy["name"] += ", direct drive"
-                dataset_copy["code"] = str(uuid.uuid4().hex)
-                dataset_copy["comment"] = (
-                    "This dataset represents the direct-drive technology "
-                    "variant of the ecoinvent dataset."
+                dataset_copy = self.create_dataset_copy(dataset, "direct drive")
+                self.database.append(dataset_copy)
+
+                construction_datasets = self.create_construction_datasets()
+                self.database.extend(construction_datasets)
+
+                market_datasets = self.create_market_datasets(
+                    dataset_copy, construction_datasets
                 )
+                self.database.extend(market_datasets)
 
-                for exc in ws.production(dataset_copy):
-                    exc["name"] = dataset_copy["name"]
-                    if "input" in exc:
-                        del exc["input"]
+                self.update_production_dataset_links(dataset_copy, market_datasets)
 
-                self.database.append(dataset_copy)
+    def create_dataset_copy(self, dataset, suffix):
+        dataset_copy = copy.deepcopy(dataset)
+        dataset_copy["name"] += f", {suffix}"
+        dataset_copy["code"] = str(uuid.uuid4().hex)
+        dataset_copy["comment"] = (
+            f"This dataset represents the {suffix} technology "
+            "variant of the ecoinvent dataset."
+        )
 
-                self.write_log(dataset_copy)
+        for exc in ws.production(dataset_copy):
+            exc["name"] = dataset_copy["name"]
+            if "input" in exc:
+                del exc["input"]
+
+        self.write_log(dataset_copy)
+        return dataset_copy
+
+    def create_construction_datasets(self):
+        construction_names = [
+            "wind power plant construction, 800kW, moving parts",
+            "wind power plant construction, 2MW, offshore, moving parts",
+            "wind turbine construction, 4.5MW, onshore",
+            "wind turbine construction, 2MW, onshore",
+        ]
+
+        construction_datasets = []
+        for name in construction_names:
+            construction_dataset = ws.get_one(self.database, ws.equals("name", name))
+            construction_dataset_copy = self.create_dataset_copy(
+                construction_dataset, "direct drive"
+            )
+            construction_datasets.append(construction_dataset_copy)
+
+        return construction_datasets
+
+    def create_market_datasets(self, parent_dataset, construction_datasets):
+        market_names = [
+            "market for wind power plant, 800kW, moving parts",
+            "market for wind power plant, 2MW, offshore, moving parts",
+            "market for wind turbine, 4.5MW, onshore",
+            "market for wind turbine, 2MW, onshore",
+        ]
+
+        product_names = [
+            "wind power plant, 800kW, moving parts",
+            "wind power plant, 2MW, offshore, moving parts",
+            "wind turbine, 4.5MW, onshore",
+            "wind turbine, 2MW, onshore",
+        ]
+
+        market_datasets = []
+        for market_name, product_name, construction_dataset in zip(
+            market_names, product_names, construction_datasets
+        ):
+            market_dataset = ws.get_one(self.database, ws.equals("name", market_name))
+            market_dataset_copy = self.create_dataset_copy(
+                market_dataset, "direct drive"
+            )
+
+            for exc in ws.technosphere(market_dataset_copy):
+                if exc["product"] == product_name:
+                    exc["name"] = construction_dataset["name"]
+                    exc["product"] = product_name
+
+            self.write_log(market_dataset_copy)
+            market_datasets.append(market_dataset_copy)
+
+        return market_datasets
+
+    @staticmethod
+    def update_production_dataset_links(production_dataset, market_datasets):
+        for market_dataset in market_datasets:
+            market_product = [exc for exc in ws.production(market_dataset)][0][
+                "product"
+            ]
+            market_name = market_dataset["name"]
+            for exc in ws.technosphere(production_dataset):
+                if exc["product"] == market_product:
+                    exc["name"] = market_name
+                    exc["product"] = market_product
 
     def write_log(self, dataset, status="created"):
         """