fix: Cleaning configuration file for plexos and adding more testing (#11

)

src/r2x/defaults/config.json

@@ -30,6 +30,7 @@
     "Flexibility",
     "Regulation"
   ],
+  "device_inference_string": {},
   "distribution_losses": 1,
   "generator_map": {},
   "heatrate_fits_file": "heatrate_generic_fits.csv",

src/r2x/defaults/plexos_input.json

@@ -1,5 +1,6 @@
 {
   "plexos_device_map": {},
+  "plexos_fuel_map": {},
   "plexos_property_map": {
     "Charge Efficiency": "charge_efficiency",
     "Commit": "must_run",

src/r2x/exceptions.py

@@ -24,3 +24,7 @@ class ModelError(Exception):
 
 class MultipleFilesError(Exception):
     pass
+
+
+class ParserError(Exception):
+    pass

src/r2x/parser/plexos.py

@@ -19,7 +19,7 @@
 from r2x.api import System
 from r2x.config import Scenario
 from r2x.enums import ACBusTypes, ReserveDirection, ReserveType, PrimeMoversType
-from r2x.exceptions import ModelError
+from r2x.exceptions import ModelError, ParserError
 from plexosdb import PlexosSQLite
 from plexosdb.enums import ClassEnum, CollectionEnum
 from r2x.model import (
@@ -126,12 +126,22 @@ class PlexosParser(PCMParser):
     def __init__(self, *args, xml_file: str | None = None, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         assert self.config.run_folder
-
         self.run_folder = Path(self.config.run_folder)
         self.system = System(name=self.config.name)
         self.property_map = self.config.defaults["plexos_property_map"]
         self.device_map = self.config.defaults["plexos_device_map"]
-        self.prime_mover_map = self.config.defaults["tech_fuel_pm_map"]
+        self.fuel_map = self.config.defaults["plexos_fuel_map"]
+        self.device_match_string = self.config.defaults["device_inference_string"]
+
+        # TODO(pesap): Rename exceptions to include R2X
+        # https://github.com/NREL/R2X/issues/5
+        # R2X needs at least one of this maps defined to correctly work.
+        if not self.fuel_map and not self.device_map and not self.device_match_string:
+            msg = (
+                "Neither `plexos_fuel_map` or `plexos_device_map` or `device_match_string` was provided. "
+                "To fix, provide any of the mappings."
+            )
+            raise ParserError(msg)
 
         # Populate databse from XML file.
         xml_file = xml_file or self.run_folder / self.config.fmap["xml_file"]["fname"]
@@ -183,7 +193,6 @@ def _collect_horizon_data(self, model_name: str) -> dict[str, float]:
     def build_system(self) -> System:
         """Create infrasys system."""
         logger.info("Building infrasys system using {}", self.__class__.__name__)
-        # self.append_to_db = self.config.defaults.get("append_to_existing_database", False)
 
         # If we decide to change the engine for handling the data we can do it here.
         object_data = self._plexos_table_data()
@@ -478,10 +487,16 @@ def _construct_generators(self):
             generator_name = generator_name[0]
             generator_fuel_type = generator_fuel_map.get(generator_name)
             logger.trace("Parsing generator = {} with fuel type = {}", generator_name, generator_fuel_type)
+
+            # Get prime mover map from fuel
+            generator_prime_mover = self.fuel_map.get(generator_fuel_type, "")
+
+            # First we check if there is a mapping one the name, if not, we try to use the prime
+            # mover map for the fuel and if not we infer the model by the name and a string.
             model_map = (
-                self.config.device_map.get(generator_name, "")
-                or self.config.fuel_map.get(generator_fuel_type, "")
-                or self._infer_model_type(generator_name)
+                self.device_map.get(generator_name, "") or generator_prime_mover["device"]
+                if generator_prime_mover
+                else None or self._infer_model_type(generator_name)
             )
 
             if getattr(R2X_MODELS, model_map, None) is None:
@@ -519,15 +534,15 @@ def _construct_generators(self):
 
             # Add prime mover mapping
             mapped_records["prime_mover_type"] = (
-                self.prime_mover_map[generator_fuel_type].get("type")
-                if generator_fuel_type in self.prime_mover_map.keys()
-                else self.prime_mover_map["default"].get("type")
+                self.fuel_map[generator_fuel_type].get("type")
+                if generator_fuel_type in self.fuel_map.keys()
+                else self.fuel_map["default"].get("type")
             )
             mapped_records["prime_mover_type"] = PrimeMoversType[mapped_records["prime_mover_type"]]
             mapped_records["fuel"] = (
-                self.prime_mover_map[generator_fuel_type].get("fuel")
-                if generator_fuel_type in self.prime_mover_map.keys()
-                else self.prime_mover_map["default"].get("fuel")
+                self.fuel_map[generator_fuel_type].get("fuel")
+                if generator_fuel_type in self.fuel_map.keys()
+                else self.fuel_map["default"].get("fuel")
             )
 
             match model_map:
@@ -696,6 +711,10 @@ def _construct_batteries(self):
 
     def _add_buses_to_batteries(self):
         batteries = [battery["name"] for battery in self.system.to_records(GenericBattery)]
+        if not batteries:
+            msg = "No battery objects found on the system. Skipping adding membership to buses."
+            logger.warning(msg)
+            return
         generator_memberships = self.db.get_memberships(
             *batteries,
             object_class=ClassEnum.Battery,
@@ -720,6 +739,10 @@ def _add_buses_to_batteries(self):
     def _add_battery_reserves(self):
         reserve_map = self.system.get_component(ReserveMap, name="contributing_generators")
         batteries = [battery["name"] for battery in self.system.to_records(GenericBattery)]
+        if not batteries:
+            msg = "No battery objects found on the system. Skipping adding membership to buses."
+            logger.warning(msg)
+            return
         generator_memberships = self.db.get_memberships(
             *batteries,
             object_class=ClassEnum.Battery,
@@ -823,7 +846,7 @@ def _construct_interfaces(self, default_model=TransmissionInterface):
         return
 
     def _select_model_name(self):
-        # TODO(pesap): Add fail mechanism
+        # TODO(pesap): Handle exception if no model name found
         # https://github.com/NREL/R2X/issues/10
         query = f"""
         select obj.name

src/r2x/utils.py

@@ -147,6 +147,7 @@ def update_dict(base_dict: dict, override_dict: ChainMap | dict | None = None) -
         "model_map",
         "tech_fuel_pm_map",
         "device_map",
+        "plexos_fuel_map",
     ]
     for key, value in override_dict.items():
         if key in base_dict and all(replace_key not in key for replace_key in _replace_keys):

tests/test_plexos_parser.py

@@ -1,6 +1,9 @@
+from plexosdb.sqlite import PlexosSQLite
 import pytest
 from plexosdb import XMLHandler
+from r2x.api import System
 from r2x.config import Scenario
+from r2x.exceptions import ParserError
 from r2x.parser.handler import get_parser_data
 from r2x.parser.plexos import PlexosParser
 
@@ -24,10 +27,34 @@ def plexos_scenario(tmp_path, data_folder):
 
 @pytest.fixture
 def plexos_parser_instance(plexos_scenario):
+    plexos_device_map = {"SolarPV_01": "RenewableFix", "ThermalCC": "ThermalStandard"}
+    plexos_scenario.defaults["plexos_device_map"] = plexos_device_map
     return get_parser_data(plexos_scenario, parser_class=PlexosParser)
 
 
 def test_plexos_parser_instance(plexos_parser_instance):
     assert isinstance(plexos_parser_instance, PlexosParser)
     assert len(plexos_parser_instance.data) == 1  # Plexos parser just parses a single file
     assert isinstance(plexos_parser_instance.data["xml_file"], XMLHandler)
+    assert isinstance(plexos_parser_instance.db, PlexosSQLite)
+
+
+@pytest.mark.skip
+def test_build_system(plexos_parser_instance):
+    system = plexos_parser_instance.build_system()
+    assert isinstance(system, System)
+
+
+def test_raise_if_no_map_provided(tmp_path, data_folder):
+    scenario = Scenario.from_kwargs(
+        name="plexos_test",
+        input_model="plexos",
+        run_folder=data_folder,
+        output_folder=tmp_path,
+        solve_year=2035,
+        model=MODEL_NAME,
+        weather_year=2012,
+        fmap={"xml_file": {"fname": DB_NAME}},
+    )
+    with pytest.raises(ParserError):
+        _ = get_parser_data(scenario, parser_class=PlexosParser)