fix: Updating codebase to match internal

pyproject.toml

@@ -134,6 +134,7 @@ docstring-code-line-length = "dynamic"
 "__init__.py" = ["E402", "F401", "F403", "D104"]
 "__main__.py" = ["E402", "F401", "D104"]
 "**/{tests,docs,tools}/*" = ["D100", "D103", "E402"]
+"src/r2x/models/*" = ["D"]
 
 [tool.pytest.ini_options]
 pythonpath = [

src/r2x/__init__.py

@@ -1,4 +1,4 @@
 from .core import *
-from .model import ACBus, Generator
+from .models import ACBus, Generator
 from .__version__ import __version__
 from .__version__ import __data_model_version__

src/r2x/api.py

@@ -6,21 +6,13 @@
 from pathlib import Path
 from itertools import chain
 from collections.abc import Iterable
-import pandas as pd
-import polars as pl
 from loguru import logger
 
 from infrasys.component import Component
 from infrasys.system import System as ISSystem
 from .__version__ import __data_model_version__
-from .model import (
-    Branch,
-    Bus,
-    Generator,
-    LoadZone,
-    Area,
-)
-from .utils import unnest_all
+import uuid
+import infrasys.cost_curves
 
 
 class System(ISSystem):
@@ -41,79 +33,6 @@ def version(self):
         """The version property."""
         return __data_model_version__
 
-    def get_generators(self, attributes: list | None = None) -> pl.DataFrame | pd.DataFrame:
-        """Return the list of generators in the system as a DataFrame."""
-        generator_list = []
-
-        generator_components = [
-            component_class
-            for component_class in self.get_component_types()
-            if Generator in component_class.__mro__
-        ]
-        for model in generator_components:
-            generator_list.extend(
-                list(
-                    map(
-                        lambda component: component.model_dump(),
-                        self.get_components(model),
-                    )
-                )
-            )
-
-        generators = pl.from_pandas(
-            pd.json_normalize(generator_list).drop(columns="services", errors="ignore")
-        )
-        if attributes:
-            generators = generators.select(pl.col(attributes))
-
-        # NOTE: This can work in the short term. In a near future we might want to patch it.
-        # We only incldue one nested attribut which is the bus
-        generators = unnest_all(generators)
-
-        return generators
-
-    def get_load_zones(self, attributes: list | None = None) -> pl.DataFrame:
-        """Return all LoadZone objects in the system."""
-        load_zones = pl.DataFrame(
-            map(lambda component: component.model_dump(), self.get_components(LoadZone))
-        )
-
-        if attributes:
-            load_zones = load_zones.select(pl.col(attributes))
-
-        load_zones = unnest_all(load_zones)
-        return load_zones
-
-    def get_areas(self, attributes: list | None = None) -> pl.DataFrame:
-        """Return all Area objects in the system."""
-        areas = pl.DataFrame(map(lambda component: component.model_dump(), self.get_components(Area)))
-
-        if attributes:
-            areas = areas.select(pl.col(attributes))
-
-        areas = unnest_all(areas)
-        return areas
-
-    def get_buses(self, attributes: list | None = None) -> pl.DataFrame:
-        """Return all Bus objects in the system."""
-        buses = pl.DataFrame(map(lambda component: component.model_dump(), self.get_components(Bus)))
-
-        if attributes:
-            buses = buses.select(pl.col(attributes))
-
-        buses = unnest_all(buses)
-        return buses
-
-    def get_branches(self, attributes: list | None = None) -> pl.DataFrame:
-        """Get Branch objects in the system."""
-        branches = pl.DataFrame(map(lambda component: component.model_dump(), self.get_components(Branch)))
-
-        if attributes:
-            branches = branches.select(pl.col(attributes))
-
-        branches = unnest_all(branches)
-        return branches
-
     def export_component_to_csv(
         self,
         component: type[Component],
@@ -148,6 +67,73 @@ def export_component_to_csv(
             **dict_writer_kwargs,
         )
 
+    def _add_operation_cost_data(  # noqa: C901
+        self,
+        data: Iterable[dict],
+        fields: list | None = None,
+    ):
+        operation_cost_fields = set()
+        for sub_dict in data:
+            if "operation_cost" not in sub_dict.keys():
+                continue
+
+            operation_cost = sub_dict["operation_cost"]
+            for cost_field_key, cost_field_value in operation_cost.items():
+                if isinstance(cost_field_value, dict):
+                    assert (
+                        "uuid" in cost_field_value.keys()
+                    ), f"Operation cost field {cost_field_key} was assumed to be a component but is not."
+                    variable_cost = self.get_component_by_uuid(uuid.UUID(cost_field_value["uuid"]))
+                    sub_dict["variable_cost"] = variable_cost.vom_units.function_data.proportional_term
+                    if "fuel_cost" in variable_cost.model_fields:
+                        # Note: We multiply the fuel price by 1000 to offset the division
+                        # done by Sienna when it parses .csv files
+                        sub_dict["fuel_price"] = variable_cost.fuel_cost * 1000
+                        operation_cost_fields.add("fuel_price")
+
+                    function_data = variable_cost.value_curve.function_data
+                    if "constant_term" in function_data.model_fields:
+                        sub_dict["heat_rate_a0"] = function_data.constant_term
+                        operation_cost_fields.add("heat_rate_a0")
+                    if "proportional_term" in function_data.model_fields:
+                        sub_dict["heat_rate_a1"] = function_data.proportional_term
+                        operation_cost_fields.add("heat_rate_a1")
+                    if "quadratic_term" in function_data.model_fields:
+                        sub_dict["heat_rate_a2"] = function_data.quadratic_term
+                        operation_cost_fields.add("heat_rate_a2")
+                    if "x_coords" in function_data.model_fields:
+                        x_y_coords = dict(zip(function_data.x_coords, function_data.y_coords))
+                        match type(variable_cost):
+                            case infrasys.cost_curves.CostCurve:
+                                for i, (x_coord, y_coord) in enumerate(x_y_coords.items()):
+                                    output_point_col = f"output_point_{i}"
+                                    sub_dict[output_point_col] = x_coord
+                                    operation_cost_fields.add(output_point_col)
+
+                                    cost_point_col = f"cost_point_{i}"
+                                    sub_dict[cost_point_col] = y_coord
+                                    operation_cost_fields.add(cost_point_col)
+
+                            case infrasys.cost_curves.FuelCurve:
+                                for i, (x_coord, y_coord) in enumerate(x_y_coords.items()):
+                                    output_point_col = f"output_point_{i}"
+                                    sub_dict[output_point_col] = x_coord
+                                    operation_cost_fields.add(output_point_col)
+
+                                    heat_rate_col = "heat_rate_avg_0" if i == 0 else f"heat_rate_incr_{i}"
+                                    sub_dict[heat_rate_col] = y_coord
+                                    operation_cost_fields.add(heat_rate_col)
+                elif cost_field_key not in sub_dict.keys():
+                    sub_dict[cost_field_key] = cost_field_value
+                    operation_cost_fields.add(cost_field_key)
+                else:
+                    pass
+
+        fields.remove("operation_cost")  # type: ignore
+        fields.extend(list(operation_cost_fields))  # type: ignore
+
+        return data, fields
+
     def _export_dict_to_csv(
         self,
         data: Iterable[dict],
@@ -169,8 +155,11 @@ def _export_dict_to_csv(
         if fields is None:
             fields = list(set(chain.from_iterable(data)))
 
+        if "operation_cost" in fields:
+            data, fields = self._add_operation_cost_data(data, fields)
+
         with open(str(fpath), "w", newline="") as csvfile:
-            writer = csv.DictWriter(csvfile, fieldnames=fields, extrasaction="ignore", **dict_writer_kwargs)
+            writer = csv.DictWriter(csvfile, fieldnames=fields, extrasaction="ignore", **dict_writer_kwargs)  # type: ignore
             writer.writeheader()
             for row in data:
                 filter_row = {

src/r2x/exporter/plexos.py

@@ -2,19 +2,22 @@
 
 from typing import Any
 import uuid
+import string
 from collections.abc import Callable
+from pathlib import Path
 
 from infrasys.component import Component
 from loguru import logger
 
+from r2x.config import Scenario
 from r2x.enums import ReserveType
 from r2x.exporter.handler import BaseExporter
 from plexosdb import PlexosSQLite
 from plexosdb.enums import ClassEnum, CollectionEnum
-from r2x.model import (
+from r2x.models import (
     ACBus,
     Emission,
-    FixedLoad,
+    InterruptiblePowerLoad,
     Generator,
     GenericBattery,
     HydroDispatch,
@@ -24,13 +27,14 @@
     MonitoredLine,
     PowerLoad,
     RenewableDispatch,
-    RenewableFix,
+    RenewableNonDispatch,
     Reserve,
+    ThermalStandard,
     Transformer2W,
     TransmissionInterface,
 )
 from r2x.units import get_magnitude
-from r2x.utils import custom_attrgetter, get_enum_from_string, read_json
+from r2x.utils import custom_attrgetter, get_enum_from_string, read_json, get_property_magnitude
 
 NESTED_ATTRIBUTES = ["ext", "bus", "services"]
 TIME_SERIES_PROPERTIES = ["Min Provision", "Static Risk"]
@@ -43,7 +47,6 @@ def __init__(
         self,
         *args,
         plexos_scenario: str = "default",
-        plexos_model: str | None = None,
         database_manager=None,
         xml_fname: str | None = None,
         **kwargs,
@@ -83,27 +86,32 @@ def run(self, *args, new_database: bool = True, **kwargs) -> "PlexosExporter":
 
         return self
 
-    def get_time_series_properties(self, component):  # noqa: C901
+    def _get_time_series_properties(self, component):  # noqa: C901
         """Add time series object to certain plexos properties."""
         if not self.system.has_time_series(component):
             return
-        component_name = component.__class__.__name__
-        csv_fname = (
-            f"{component_name}_{self.config.name}_{self.config.solve_year}{self.config.weather_year}.csv"
-        )
+
+        ts_metadata = self.system.get_time_series(component)
+
+        config_dict = self.config.__dict__
+        config_dict["name"] = self.config.name
+        csv_fname = config_dict.get("time_series_fname", "${component_type}_${name}_${weather_year}.csv")
+        string_template = string.Template(csv_fname)
+        config_dict["component_type"] = f"{component.__class__.__name__}_{ts_metadata.variable_name}"
+        csv_fname = string_template.safe_substitute(config_dict)
         csv_fpath = self.ts_directory / csv_fname
         time_series_property: dict[str, Any] = {"Data File": str(csv_fpath)}
 
         # Property with time series can change. Validate this option based on the component type.
         match component:
-            case FixedLoad():
+            case InterruptiblePowerLoad():
                 time_series_property["Fixed Load"] = "0"
             case PowerLoad():
                 time_series_property["Load"] = "0"
             case RenewableDispatch():
                 time_series_property["Rating"] = "0"
                 time_series_property["Load Subtracter"] = "0"
-            case RenewableFix():
+            case RenewableNonDispatch():
                 time_series_property["Rating"] = "0"
                 time_series_property["Load Subtracter"] = "0"
             case Reserve():
@@ -120,6 +128,13 @@ def get_time_series_properties(self, component):  # noqa: C901
                 time_series_property["Max Energy"] = "0"
             case HydroEnergyReservoir():
                 time_series_property["Fixed Load"] = "0"
+            case ThermalStandard():
+                variable_name = self.system.get_time_series(component).variable_name
+                if not variable_name:
+                    return None
+                property_name = self.property_map.get(variable_name, None)
+                if property_name:
+                    time_series_property[property_name] = "0"
             case _:
                 raise NotImplementedError(f"Time Series for {component.label} not supported yet.")
         return time_series_property
@@ -329,7 +344,7 @@ def add_topology(self) -> None:
         for component in self.system.get_components(
             PowerLoad, filter_func=lambda component: self.system.has_time_series(component)
         ):
-            time_series_properties = self.get_time_series_properties(component)
+            time_series_properties = self._get_time_series_properties(component)
             if time_series_properties:
                 text = time_series_properties.pop("Data File")
                 for property_name, property_value in time_series_properties.items():
@@ -477,7 +492,7 @@ def add_reserves(self) -> None:
                     collection=CollectionEnum.SystemReserves,
                     scenario=self.plexos_scenario,
                 )
-            time_series_properties = self.get_time_series_properties(reserve)
+            time_series_properties = self._get_time_series_properties(reserve)
             if time_series_properties:
                 text = time_series_properties.pop("Data File")
                 for property_name, property_value in time_series_properties.items():
@@ -559,7 +574,7 @@ def exclude_battery(component):
                 child_class=ClassEnum.Node,
                 collection=CollectionEnum.GeneratorNodes,
             )
-            properties = self.get_time_series_properties(generator)
+            properties = self._get_time_series_properties(generator)
             if properties:
                 text = properties.pop("Data File")
                 for property_name, property_value in properties.items():
@@ -820,8 +835,28 @@ def get_valid_component_properties(
         valid_properties = [key[0] for key in collection_properties]
         for property_name, property_value in component_dict_mapped.items():
             if property_name in valid_properties:
-                property_value = self.get_property_magnitude(
-                    property_value, to_unit=unit_map.get(property_name)
-                )
+                property_value = get_property_magnitude(property_value, to_unit=unit_map.get(property_name))
                 valid_component_properties[property_name] = property_value
         return valid_component_properties
+
+
+if __name__ == "__main__":
+    run_folder = Path("tests/data/pacific/")
+    # Functions relative to the parser.
+    from tests.models.systems import ieee5bus_system
+
+    config = Scenario.from_kwargs(
+        name="PlexosExportTest",
+        input_model="reeds-US",
+        output_model="plexos",
+        run_folder=run_folder,
+        solve_year=2035,
+        weather_year=2012,
+    )
+    system = ieee5bus_system()
+
+    # fpath = "/Users/psanchez/downloads/test.xml"
+    fpath = "/Volumes/r2x/test_models/test.xml"
+    exporter = PlexosExporter(config=config, system=system)
+    exporter.run()
+    # handler = exporter.xml