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

premise/battery.py

@@ -246,6 +246,7 @@ def adjust_battery_mass(self) -> None:
 
                 for exc in ws.technosphere(ds, ws.equals("unit", "kilogram")):
                     exc["amount"] *= scaling_factor
+                    exc["loc"] *= scaling_factor
                     exc["minimum"] *= scaling_factor_min
                     exc["maximum"] *= scaling_factor_max
 

premise/data/utils/logging/logconfig.yaml

@@ -80,6 +80,14 @@ handlers:
     formatter: simple
     mode: a
 
+  file_wind_turbine:
+    class: logging.FileHandler
+    level: INFO
+    filename: "export/logs/premise_wind_turbine.log"
+    encoding: utf8
+    formatter: simple
+    mode: a
+
   file_emissions:
     class: logging.FileHandler
     level: INFO
@@ -145,6 +153,11 @@ loggers:
     handlers: [ file_battery ]
     propagate: False
 
+  wind_turbine:
+    level: INFO
+    handlers: [ file_wind_turbine ]
+    propagate: False
+
   heat:
     level: INFO
     handlers: [ file_heat ]

premise/data/utils/logging/reporting.yaml

@@ -549,6 +549,38 @@ premise_battery:
         unit: 0-1
     tab: Battery
 
+premise_wind_turbine:
+    columns:
+      timestamp:
+        name: timestamp
+        description: Timestamp of the log entry
+      module:
+        name: module
+        description: Module name
+      level:
+        name: level
+        description: Log level
+      status:
+        name: status
+        description: Status of the dataset
+      model:
+        name: model
+        description: IAM model name
+      pathway:
+        name: pathway
+        description: Pathway name
+      year:
+        name: year
+        description: Year
+      dataset:
+        name: dataset
+        description: Dataset name
+      region:
+        name: region
+        description: Region name
+    tab: Wind Turbine
+
+
 premise_emissions:
     columns:
       timestamp:

premise/new_database.py

@@ -55,6 +55,7 @@
     print_version,
     warning_about_biogenic_co2,
 )
+from .wind_turbines import _update_wind_turbines
 
 logger = logging.getLogger("module")
 
@@ -870,6 +871,10 @@ def update(self, sectors: [str, list, None] = None) -> None:
             },
             "steel": {"func": _update_steel, "args": (self.version, self.system_model)},
             "fuels": {"func": _update_fuels, "args": (self.version, self.system_model)},
+            "wind": {
+                "func": _update_wind_turbines,
+                "args": (self.version, self.system_model),
+            },
             "metals": {
                 "func": _update_metals,
                 "args": (self.version, self.system_model),

premise/report.py

@@ -633,6 +633,7 @@ def generate_change_report(source, version, source_type, system_model):
         "premise_fuel",
         "premise_heat",
         "premise_battery",
+        "premise_wind_turbine",
         "premise_transport",
         "premise_steel",
         "premise_metal",

premise/wind_turbines.py

@@ -0,0 +1,117 @@
+"""
+module to create a variant of the onshore and offshore wind
+turbine datasets to represent the direct-drive technology.
+
+"""
+
+import copy
+import uuid
+
+from .activity_maps import InventorySet
+from .logger import create_logger
+from .transformation import BaseTransformation, IAMDataCollection, List, np, ws
+
+logger = create_logger("wind_turbine")
+
+
+def _update_wind_turbines(scenario, version, system_model):
+    wind_turbine = WindTurbine(
+        database=scenario["database"],
+        iam_data=scenario["iam data"],
+        model=scenario["model"],
+        pathway=scenario["pathway"],
+        year=scenario["year"],
+        version=version,
+        system_model=system_model,
+        cache=scenario.get("cache"),
+        index=scenario.get("index"),
+    )
+
+    wind_turbine.create_direct_drive_turbines()
+
+    scenario["database"] = wind_turbine.database
+    scenario["index"] = wind_turbine.index
+    scenario["cache"] = wind_turbine.cache
+
+    return scenario
+
+
+class WindTurbine(BaseTransformation):
+    """
+    Class that create additional wind turbine datasets
+    to represent the direct-drive technology.
+
+    """
+
+    def __init__(
+        self,
+        database: List[dict],
+        iam_data: IAMDataCollection,
+        model: str,
+        pathway: str,
+        year: int,
+        version: str,
+        system_model: str,
+        cache: dict = None,
+        index: dict = None,
+    ) -> None:
+        super().__init__(
+            database,
+            iam_data,
+            model,
+            pathway,
+            year,
+            version,
+            system_model,
+            cache,
+            index,
+        )
+        self.system_model = system_model
+        mapping = InventorySet(database=database, version=version, model=model)
+        self.powerplant_map = mapping.generate_powerplant_map()
+
+    def create_direct_drive_turbines(self):
+        """
+        Create direct-drive wind turbine datasets.
+        """
+
+        wind_turbine_technologies = ["Wind Onshore", "Wind Offshore"]
+
+        for technology in wind_turbine_technologies:
+            for dataset in ws.get_many(
+                self.database,
+                ws.either(
+                    *[
+                        ws.equals("name", tech)
+                        for tech in self.powerplant_map[technology]
+                    ]
+                ),
+            ):
+                # 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."
+                )
+
+                for exc in ws.production(dataset_copy):
+                    exc["name"] = dataset_copy["name"]
+                    if "input" in exc:
+                        del exc["input"]
+
+                self.database.append(dataset_copy)
+
+                self.write_log(dataset_copy)
+
+    def write_log(self, dataset, status="created"):
+        """
+        Write log file.
+        """
+
+        logger.info(
+            f"{status}|{self.model}|{self.scenario}|{self.year}|"
+            f"{dataset['name']}|{dataset['location']}"
+        )