From a5545c591cb67bfbc4948f078a38f2e7aec4d7a5 Mon Sep 17 00:00:00 2001 From: pesap Date: Fri, 30 Aug 2024 14:27:02 -0600 Subject: [PATCH 1/3] Updating Sienna exporter to match latests model change --- src/r2x/exporter/sienna.py | 44 ++++++++++++++++++++++---------------- 1 file changed, 25 insertions(+), 19 deletions(-) diff --git a/src/r2x/exporter/sienna.py b/src/r2x/exporter/sienna.py index 7d15d55f..fdf798b9 100644 --- a/src/r2x/exporter/sienna.py +++ b/src/r2x/exporter/sienna.py @@ -12,7 +12,7 @@ from infrasys.time_series_models import SingleTimeSeries from pint import Quantity from r2x.exporter.handler import BaseExporter -from r2x.model import ( +from r2x.models import ( ACBranch, Bus, DCBranch, @@ -78,7 +78,6 @@ def run(self, *args, path=None, **kwargs) -> "SiennaExporter": self.process_storage_data() self.export_data() self.create_timeseries_pointers() - self.create_extra_data_json() return self def process_bus_data(self, fname: str = "bus.csv") -> None: @@ -101,7 +100,7 @@ def process_bus_data(self, fname: str = "bus.csv") -> None: Bus, fpath=self.output_folder / fname, fields=output_fields, - key_mapping={"id": "bus_id", "load_zone": "zone"}, + key_mapping={"number": "bus_id", "load_zone": "zone"}, restval="NA", ) @@ -127,7 +126,7 @@ def process_load_data(self, fname: str = "load.csv") -> None: PowerLoad, fpath=self.output_folder / fname, fields=output_fields, - unnest_key="id", + unnest_key="number", key_mapping={"bus": "bus_id"}, restval="0.0", ) @@ -164,12 +163,12 @@ def process_branch_data(self, fname: str = "branch.csv") -> None: ACBranch, fpath=self.output_folder / fname, fields=output_fields, - unnest_key="id", + unnest_key="number", key_mapping={ "from_bus": "connection_points_from", "to_bus": "connection_points_to", "class_type": "branch_type", - "rating_up": "rate", + "rating": "rate", "b": "primary_shunt", }, # restval=0.0, @@ -203,7 +202,7 @@ def process_dc_branch_data(self, fname="dc_branch.csv") -> None: DCBranch, fpath=self.output_folder / fname, fields=output_fields, - unnest_key="id", + unnest_key="number", key_mapping={ "from_bus": "connection_points_from", "to_bus": "connection_points_to", @@ -229,10 +228,9 @@ def process_gen_data(self, fname="gen.csv"): "prime_mover_type", "bus_id", "fuel", + "rating", + "unit_type", "base_power", - "fuel_price", - "heat_rate", - "vom_price", # "active_power_limits_max", # "active_power_limits_min", "min_rated_capacity", @@ -243,10 +241,10 @@ def process_gen_data(self, fname="gen.csv"): "planned_outage_rate", "ramp_up", "ramp_down", - "startup_cost", "category", "must_run", "pump_load", + "operation_cost", ] key_mapping = {"bus": "bus_id"} @@ -255,7 +253,7 @@ def process_gen_data(self, fname="gen.csv"): fpath=self.output_folder / fname, fields=output_fields, key_mapping=key_mapping, - unnest_key="id", + unnest_key="number", restval="NA", ) logger.info(f"File {fname} created.") @@ -292,8 +290,8 @@ def process_reserves_data(self, fname="reserves.csv") -> None: if len(reserve_map_list) > 1: logger.warning("We do not support multiple reserve maps per system") return - reserve_map: dict = reserve_map_list[0] + reserves: list[dict[str, Any]] = list(self.system.to_records(Reserve)) output_data = [] @@ -345,6 +343,7 @@ def process_storage_data(self, fname="storage.csv") -> None: "input_active_power_limit_min", "output_active_power_limit_max", "output_active_power_limit_min", + "unit_type", ] generic_storage = self.get_valid_records_properties( @@ -371,8 +370,8 @@ def process_storage_data(self, fname="storage.csv") -> None: output_dict["input_active_power_limit_min"] = 0 # output_dict["base_power"] output_dict["output_active_power_limit_min"] = 0 # output_dict["base_power"] output_dict["base_power"] = output_dict["base_power"] - output_dict["bus_id"] = getattr(self.system.get_component_by_label(output_dict["bus"]), "id") - output_dict["rating"] = 1 + output_dict["bus_id"] = getattr(self.system.get_component_by_label(output_dict["bus"]), "number") + output_dict["rating"] = output_dict["rating"] # NOTE: For pumped hydro storage we create a head and a tail # representation that keeps track of the upper and down reservoir @@ -393,14 +392,20 @@ def process_storage_data(self, fname="storage.csv") -> None: fpath=self.output_folder / fname, fields=output_fields, key_mapping=key_mapping, - unnest_key="id", + unnest_key="numer", restval="NA", ) logger.info("File storage.csv created.") def create_timeseries_pointers(self) -> None: - """Create timeseries_pointers.json file.""" + """Create timeseries_pointers.json file. + + Parameters + ---------- + fname : str + Name of the file to be created + """ ts_pointers_list = [] for component_type, time_series in self.time_series_objects.items(): @@ -412,12 +417,13 @@ def create_timeseries_pointers(self) -> None: ts_instance = time_series[i] resolution = ts_instance.resolution.seconds variable_name = self.property_map.get(ts_instance.variable_name, ts_instance.variable_name) - + # TODO(pedro): check if the time series data is pre normalized + # https://github.nrel.gov/PCM/R2X/issues/417 ts_pointers = { "category": component_type.split("_", maxsplit=1)[0], # Component_name is the first "component_name": component_name, "data_file": str(csv_fpath), - "normalization_factor": "MAX", + "normalization_factor": 1.0, "resolution": resolution, "name": variable_name, "scaling_factor_multiplier_module": "PowerSystems", From 683f390a22d154b0479de5740b0d251170400f09 Mon Sep 17 00:00:00 2001 From: pesap Date: Fri, 30 Aug 2024 15:02:02 -0600 Subject: [PATCH 2/3] Updates from internal --- pyproject.toml | 1 + src/r2x/__init__.py | 2 +- src/r2x/api.py | 157 +++--- src/r2x/exporter/plexos.py | 71 ++- src/r2x/model.py | 645 ----------------------- src/r2x/models/__init__.py | 24 + src/r2x/models/branch.py | 122 +++++ src/r2x/models/core.py | 59 +++ src/r2x/models/costs.py | 50 ++ src/r2x/models/generators.py | 312 +++++++++++ src/r2x/models/load.py | 85 +++ src/r2x/models/services.py | 105 ++++ src/r2x/models/topology.py | 97 ++++ src/r2x/models/utils.py | 31 ++ src/r2x/parser/plexos.py | 6 +- src/r2x/plugins/break_gens.py | 26 +- src/r2x/runner.py | 2 +- src/r2x/units.py | 8 +- src/r2x/utils.py | 135 ++++- tests/models/{systems.py => ieee5bus.py} | 87 ++- tests/test_api.py | 10 +- tests/test_break_gens.py | 6 +- tests/test_models.py | 12 +- tests/test_plexos_parser.py | 2 +- tests/test_sienna_exporter.py | 4 +- 25 files changed, 1226 insertions(+), 833 deletions(-) delete mode 100644 src/r2x/model.py create mode 100644 src/r2x/models/__init__.py create mode 100644 src/r2x/models/branch.py create mode 100644 src/r2x/models/core.py create mode 100644 src/r2x/models/costs.py create mode 100644 src/r2x/models/generators.py create mode 100644 src/r2x/models/load.py create mode 100644 src/r2x/models/services.py create mode 100644 src/r2x/models/topology.py create mode 100644 src/r2x/models/utils.py rename tests/models/{systems.py => ieee5bus.py} (67%) diff --git a/pyproject.toml b/pyproject.toml index 3474bd70..2495aa45 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/src/r2x/__init__.py b/src/r2x/__init__.py index 5eae8a49..ff2ef3c0 100644 --- a/src/r2x/__init__.py +++ b/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__ diff --git a/src/r2x/api.py b/src/r2x/api.py index 42790fd3..edc6c27e 100644 --- a/src/r2x/api.py +++ b/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 = { diff --git a/src/r2x/exporter/plexos.py b/src/r2x/exporter/plexos.py index dcfd2b15..7dbdef12 100644 --- a/src/r2x/exporter/plexos.py +++ b/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 diff --git a/src/r2x/model.py b/src/r2x/model.py deleted file mode 100644 index 3ab04ed8..00000000 --- a/src/r2x/model.py +++ /dev/null @@ -1,645 +0,0 @@ -"""R2X data model. - -This script contains the Sienna data model for power systems relevant to CEM to PCM translations. -It uses `infrasys.py` to store the components and pydantic for validating the fields. - -Mapping for models: - Zonal model: - - Buses = ReEDS BA - - LoadZone = ReEDS transmission region or Plexos Zones - - Area = States - Nodal model: - - Buses = Nodes - - LoadZone = ReEDS BA or Sienna Load Zones or Plexos Regions - - Area = Plexos Zones or Sienna Areas -""" -# ruff: noqa: D102 - -from collections import defaultdict -from typing import Annotated, DefaultDict # noqa: UP035 - -from infrasys.component import Component -from pydantic import ( - Field, - NonNegativeFloat, - NonPositiveFloat, - PositiveFloat, - PositiveInt, - confloat, -) - -from r2x.units import ( - ActivePower, - EmissionRate, - Energy, - FuelPrice, - HeatRate, - Percentage, - PowerRate, - Time, - ureg, - Voltage, -) - -from .enums import ACBusTypes, PrimeMoversType, ReserveDirection, ReserveType - -unit_size = confloat(ge=0, le=1) - -NonNegativeFloatType = NonNegativeFloat -NonPositiveFloatType = NonPositiveFloat -PositiveFloatType = PositiveFloat - - -class BaseComponent(Component): - """Infrasys base component with additional fields for R2X.""" - - ext: dict = Field(default_factory=dict, description="Additional information of the component.") - available: Annotated[bool, Field(description="If the component is available.")] = True - category: Annotated[str, Field(description="Category that this component belongs to.")] | None = None - - @property - def class_type(self) -> str: - """Create attribute that holds the class name.""" - return type(self).__name__ - - -class Service(BaseComponent): - """Base class for Service attached to components.""" - - -class Device(BaseComponent): - """Abstract class for devices.""" - - services: ( - Annotated[ - list[Service], - Field(description="Services that this component contributes to.", default_factory=list), - ] - | None - ) = None - - -class Topology(BaseComponent): - """Abstract type to represent the structure and interconnectedness of the system.""" - - services: ( - Annotated[ - list[Service], - Field(description="Services that this component contributes to.", default_factory=list), - ] - | None - ) = None - - -class AggregationTopology(Topology): - """Base class for area-type components.""" - - -class Area(AggregationTopology): - """Collection of buses in a given region.""" - - @classmethod - def example(cls) -> "Area": - return Area(name="New York") - - -class LoadZone(AggregationTopology): - """Collection of buses for electricity price analysis.""" - - @classmethod - def example(cls) -> "LoadZone": - return LoadZone(name="ExampleLoadZone") - - -class Bus(Topology): - """Power-system Bus abstract class.""" - - id: Annotated[PositiveInt, Field(description="ID/number associated to the bus.")] - load_zone: Annotated[LoadZone, Field(description="the load zone containing the DC bus.")] | None = None - area: Annotated[Area, Field(description="Area containing the bus.")] | None = None - lpf: Annotated[float, Field(description="Load participation factor of the bus.", ge=0, le=1)] | None = ( - None - ) - base_voltage: Annotated[Voltage, Field(gt=0, description="Base voltage in kV.")] | None = None - magnitude: ( - Annotated[PositiveFloatType, Field(description="Voltage as a multiple of base_voltage.")] | None - ) = None - bus_type: Annotated[ACBusTypes, Field(description="Type of category of bus")] | None = None - - @classmethod - def example(cls) -> "Bus": - return Bus( - id=1, - name="ExampleBus", - load_zone=LoadZone.example(), - area=Area.example(), - lpf=1, - ) - - -class DCBus(Bus): - """Power-system DC Bus.""" - - @classmethod - def example(cls) -> "DCBus": - return DCBus( - name="ExampleDCBus", - id=1, - load_zone=LoadZone.example(), - area=Area.example(), - base_voltage=100 * ureg.kV, - lpf=1, - ) - - -class ACBus(Bus): - """Power-system AC bus.""" - - @classmethod - def example(cls) -> "ACBus": - return ACBus( - name="ExampleACBus", - id=1, - load_zone=LoadZone.example(), - area=Area.example(), - base_voltage=100 * ureg.kV, - ) - - -class PowerLoad(BaseComponent): - """Class representing a Load object.""" - - bus: Bus | LoadZone = Field(description="Point of injection.") - max_active_power: Annotated[ActivePower, Field(gt=0, description="Max Load at the bus in MW")] | None = ( - None - ) - - @classmethod - def example(cls) -> "PowerLoad": - return PowerLoad(name="ExampleLoad", bus=Bus.example()) - - -class FixedLoad(PowerLoad): - """A static PowerLoad that is not interruptible.""" - - @classmethod - def example(cls) -> "PowerLoad": - return PowerLoad(name="ExampleLoad", bus=Bus.example(), max_active_power=100 * ureg.MW) - - -class InterruptiblePowerLoad(PowerLoad): - """A static interruptible power load.""" - - base_power: Annotated[ActivePower, Field(gt=0, description="Active power of the load type.")] - operation_cost: float | None = None - - -class TransmissionInterface(Service): - """Collection of branches that make up an interfece or corridor for the transfer of power.""" - - max_power_flow: Annotated[ActivePower, Field(ge=0, description="Maximum allowed flow.")] - min_power_flow: Annotated[ActivePower, Field(le=0, description="Minimum allowed flow.")] - ramp_up: ( - Annotated[PowerRate, Field(ge=0, description="Maximum ramp allowed on the positve direction.")] | None - ) = None - ramp_down: ( - Annotated[PowerRate, Field(ge=0, description="Minimum ramp allowed on the negative direction.")] - | None - ) = None - - @classmethod - def example(cls) -> "TransmissionInterface": - return TransmissionInterface( - name="ExampleTransmissionInterface", - max_power_flow=ActivePower(100, "MW"), - min_power_flow=ActivePower(-100, "MW"), - ) - - -class TransmissionInterfaceMap(BaseComponent): # noqa: D101 - mapping: DefaultDict[str, list] = defaultdict(list) # noqa: UP006, RUF012 - - -class Reserve(Service): - """Class representing a reserve contribution.""" - - time_frame: Annotated[ - PositiveFloatType, - Field(description="Timeframe in which the reserve is required in seconds"), - ] = 1e30 - region: ( - Annotated[ - LoadZone, - Field(description="LoadZone where reserve requirement is required."), - ] - | None - ) = None - vors: Annotated[ - float, - Field(description="Value of reserve shortage in $/MW. Any positive value as as soft constraint."), - ] = -1 - duration: ( - Annotated[ - PositiveFloatType, - Field(description="Time over which the required response must be maintained in seconds."), - ] - | None - ) = None - reserve_type: ReserveType - load_risk: ( - Annotated[ - NonNegativeFloatType, - Field( - description="Proportion of Load that contributes to the requirement.", - ), - ] - | None - ) = None - # ramp_rate: float | None = None # NOTE: Maybe we do not need this. - max_requirement: float = 0 # Should we specify which variable is the time series for? - direction: ReserveDirection - - @classmethod - def example(cls) -> "Reserve": - return Reserve( - name="ExampleReserve", - region=LoadZone.example(), - direction=ReserveDirection.Up, - reserve_type=ReserveType.Regulation, - ) - - -class ReserveMap(BaseComponent): # noqa: D101 - mapping: DefaultDict[str, list] = defaultdict(list) # noqa: UP006, RUF012 - - -class Branch(Device): - """Class representing a connection between components.""" - - # arc: Annotated[Arc, Field(description="The branch's connections.")] - from_bus: Annotated[Bus, Field(description="Bus connected upstream from the arc.")] - to_bus: Annotated[Bus, Field(description="Bus connected downstream from the arc.")] - - -class ACBranch(Branch): - """Class representing an AC connection between components.""" - - r: Annotated[float, Field(description=("Resistance of the branch"))] = 0 - x: Annotated[float, Field(description=("Reactance of the branch"))] = 0 - b: Annotated[float, Field(description=("Shunt susceptance of the branch"))] = 0 - - -class MonitoredLine(ACBranch): - """Class representing an AC transmission line.""" - - rating_up: Annotated[ActivePower, Field(ge=0, description="Forward rating of the line.")] | None = None - rating_down: Annotated[ActivePower, Field(le=0, description="Reverse rating of the line.")] | None = None - losses: Annotated[Percentage, Field(description="Power losses on the line.")] | None = None - - @classmethod - def example(cls) -> "MonitoredLine": - return MonitoredLine( - name="ExampleLine", - from_bus=Bus.example(), - to_bus=Bus.example(), - losses=Percentage(10, "%"), - rating_up=ActivePower(100, "MW"), - rating_down=ActivePower(-100, "MW"), - ) - - -class Transformer2W(ACBranch): - """Class representing a 2-W transformer.""" - - rate: Annotated[NonNegativeFloatType, Field(description="Rating of the transformer.")] - - @classmethod - def example(cls) -> "Transformer2W": - return Transformer2W( - name="Example2WTransformer", - rate=100, - from_bus=Bus.example(), - to_bus=Bus.example(), - ) - - -class DCBranch(Branch): - """Class representing a DC connection between components.""" - - -class TModelHVDCLine(DCBranch): - """Class representing a DC transmission line.""" - - rating_up: Annotated[NonNegativeFloatType, Field(description="Forward rating of the line.")] | None = None - rating_down: Annotated[NonPositiveFloatType, Field(description="Reverse rating of the line.")] | None = ( - None - ) - losses: Annotated[NonNegativeFloatType, Field(description="Power losses on the line.")] = 0 - resistance: ( - Annotated[NonNegativeFloatType, Field(description="Resistance of the line in p.u.")] | None - ) = 0 - inductance: ( - Annotated[NonNegativeFloatType, Field(description="Inductance of the line in p.u.")] | None - ) = 0 - capacitance: ( - Annotated[NonNegativeFloatType, Field(description="Capacitance of the line in p.u.")] | None - ) = 0 - - @classmethod - def example(cls) -> "TModelHVDCLine": - return TModelHVDCLine( - name="ExampleDCLine", - from_bus=Bus.example(), - to_bus=Bus.example(), - rating_up=100, - rating_down=80, - ) - - -class Emission(Service): - """Class representing an emission object that is attached to generators.""" - - rate: Annotated[EmissionRate, Field(description="Amount of emission produced in kg/MWh.")] - emission_type: Annotated[str, Field(description="Type of emission. E.g., CO2, NOx.")] - generator_name: Annotated[str, Field(description="Generator emitting.")] - - @classmethod - def example(cls) -> "Emission": - return Emission( - name="ExampleEmission", - generator_name="gen1", - rate=EmissionRate(105, "kg/MWh"), - emission_type="CO2", - ) - - -class Generator(Device): - """Abstract generator class.""" - - bus: Annotated[ACBus, Field(description="Bus where the generator is connected.")] | None = None - base_power: Annotated[ActivePower, Field(description="Active power generation in MW.")] - must_run: Annotated[int | None, Field(description="If we need to force the dispatch of the device.")] = ( - None - ) - vom_price: Annotated[FuelPrice, Field(description="Variable operational price $/MWh.")] | None = None - prime_mover_type: ( - Annotated[PrimeMoversType, Field(description="Prime mover technology according to EIA 923.")] | None - ) = None - min_rated_capacity: Annotated[ActivePower, Field(description="Minimum rated power generation.")] = ( - 0 * ureg.MW - ) - ramp_up: ( - Annotated[ - PowerRate, - Field(description="Ramping rate on the positve direction."), - ] - | None - ) = None - ramp_down: ( - Annotated[ - PowerRate, - Field(description="Ramping rate on the negative direction."), - ] - | None - ) = None - min_up_time: ( - Annotated[ - Time, - Field(ge=0, description="Minimum up time in hours for UC decision."), - ] - | None - ) = None - min_down_time: ( - Annotated[ - Time, - Field(ge=0, description="Minimum down time in hours for UC decision."), - ] - | None - ) = None - mean_time_to_repair: ( - Annotated[ - Time, - Field(gt=0, description="Total hours to repair after outage occur."), - ] - | None - ) = None - forced_outage_rate: ( - Annotated[ - Percentage, - Field(description="Expected level of unplanned outages in percent."), - ] - | None - ) = None - planned_outage_rate: ( - Annotated[ - Percentage, - Field(description="Expected level of planned outages in percent."), - ] - | None - ) = None - startup_cost: ( - Annotated[NonNegativeFloatType, Field(description="Cost in $ of starting a unit.")] | None - ) = None - - @classmethod - def example(cls) -> "Generator": - return Generator(name="gen01", base_power=100.0 * ureg.MW, prime_mover_type=PrimeMoversType.PV) - - -class RenewableGen(Generator): - """Abstract class for renewable generators.""" - - -class RenewableDispatch(RenewableGen): - """Curtailable renewable generator. - - This type of generator have a hourly capacity factor profile. - """ - - -class RenewableFix(RenewableGen): - """Non-curtailable renewable generator. - - Renewable technologies w/o operational cost. - """ - - -class HydroGen(Generator): - """Hydroelectric generator.""" - - -class HydroDispatch(HydroGen): - """Class representing flexible hydro generators.""" - - ramp_up: ( - Annotated[ - PowerRate, - Field(ge=0, description="Ramping rate on the positve direction."), - ] - | None - ) = None - ramp_down: ( - Annotated[ - PowerRate, - Field(ge=0, description="Ramping rate on the negative direction."), - ] - | None - ) = None - - -class HydroFix(HydroGen): - """Class representing unflexible hydro.""" - - -class HydroEnergyReservoir(HydroGen): - """Class representing hydro system with reservoirs.""" - - initial_energy: ( - Annotated[NonNegativeFloatType, Field(description="Initial water volume or percentage.")] | None - ) = 0 - storage_capacity: ( - Annotated[ - Energy, - Field(description="Total water volume or percentage."), - ] - | None - ) = None - min_storage_capacity: ( - Annotated[ - Energy, - Field(description="Minimum water volume or percentage."), - ] - | None - ) = None - storage_target: ( - Annotated[ - Energy, - Field(description="Maximum energy limit."), - ] - | None - ) = None - - -class HydroPumpedStorage(HydroGen): - """Class representing pumped hydro generators.""" - - storage_duration: ( - Annotated[ - Time, - Field(description="Storage duration in hours."), - ] - | None - ) = None - initial_volume: ( - Annotated[Energy, Field(gt=0, description="Initial water volume or percentage.")] | None - ) = None - storage_capacity: Annotated[ - Energy, - Field(gt=0, description="Total water volume or percentage."), - ] - min_storage_capacity: ( - Annotated[ - Energy, - Field(description="Minimum water volume or percentage."), - ] - | None - ) = None - pump_efficiency: Annotated[Percentage, Field(ge=0, le=1, description="Pumping efficiency.")] | None = None - pump_load: ( - Annotated[ - ActivePower, - Field(description="Load related to the usage of the pump."), - ] - | None - ) = None - - @classmethod - def example(cls) -> "HydroPumpedStorage": - return HydroPumpedStorage( - name="HydroStorage", - base_power=ActivePower(100, "MW"), - pump_load=ActivePower(100, "MW"), - bus=ACBus.example(), - prime_mover_type=PrimeMoversType.PS, - storage_duration=Time(10, "h"), - storage_capacity=Energy(1000, "MWh"), - min_storage_capacity=Energy(10, "MWh"), - pump_efficiency=Percentage(85, "%"), - initial_volume=Energy(500, "MWh"), - ext={"description": "Pumped hydro unit with 10 hour of duration"}, - ) - - -class ThermalGen(Generator): - """Class representing fuel based thermal generator.""" - - fuel: Annotated[str, Field(description="Fuel category")] | None = None - fuel_price: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( - 0.0, "usd/MWh" - ) - heat_rate: Annotated[HeatRate | None, Field(description="Heat rate")] = None - - @classmethod - def example(cls) -> "ThermalGen": - return ThermalGen( - name="ThermalGen", - bus=ACBus.example(), - fuel="gas", - base_power=100 * ureg.MW, - fuel_price=FuelPrice(5, "usd/MWh"), - ext={"Additional data": "Additional value"}, - ) - - -class ThermalStandard(ThermalGen): - """Standard representation of thermal device.""" - - -class ThermalMultiStart(ThermalGen): - """We will fill this class once we have the need for it.""" - - -class Storage(Generator): - """Default Storage class.""" - - storage_duration: ( - Annotated[ - Time, - Field(description="Storage duration in hours."), - ] - | None - ) = None - storage_capacity: Annotated[ - Energy, - Field(description="Maximum allowed volume or state of charge."), - ] - initial_energy: Annotated[Percentage, Field(description="Initial state of charge.")] | None = None - min_storage_capacity: Annotated[Percentage, Field(description="Minimum state of charge")] = Percentage( - 0, "%" - ) - max_storage_capacity: Annotated[Percentage, Field(description="Minimum state of charge")] = Percentage( - 100, "%" - ) - - -class GenericBattery(Storage): - """Battery energy storage model.""" - - charge_efficiency: Annotated[Percentage, Field(ge=0, description="Charge efficiency.")] | None = None - discharge_efficiency: Annotated[Percentage, Field(ge=0, description="Discharge efficiency.")] | None = ( - None - ) - - -class HybridSystem(Device): - """Representation of hybrid system with renewable generation, load, thermal generation and storage. - - This class is just a link between two components. - For the implementation see: - - https://github.com/NREL-Sienna/PowerSystems.jl/blob/main/src/models/HybridSystem.jl - """ - - storage_unit: Storage | None = None - renewable_unit: RenewableGen | None = None - thermal_unit: ThermalGen | None = None - electric_load: PowerLoad | None = None diff --git a/src/r2x/models/__init__.py b/src/r2x/models/__init__.py new file mode 100644 index 00000000..e190f494 --- /dev/null +++ b/src/r2x/models/__init__.py @@ -0,0 +1,24 @@ +# Models +# ruff: noqa +from .branch import Branch, ACBranch, DCBranch, MonitoredLine, Transformer2W +from .core import ReserveMap, TransmissionInterfaceMap +from .generators import ( + Generator, + ThermalGen, + Storage, + RenewableGen, + GenericBattery, + HybridSystem, + HydroGen, + HydroPumpedStorage, + HydroDispatch, + HydroEnergyReservoir, + RenewableDispatch, + RenewableNonDispatch, + ThermalStandard, + Storage, +) +from .costs import HydroGenerationCost, StorageCost, ThermalGenerationCost, RenewableGenerationCost +from .load import PowerLoad, InterruptiblePowerLoad +from .services import Emission, Reserve, TransmissionInterface +from .topology import ACBus, Area, Bus, LoadZone diff --git a/src/r2x/models/branch.py b/src/r2x/models/branch.py new file mode 100644 index 00000000..cabf6189 --- /dev/null +++ b/src/r2x/models/branch.py @@ -0,0 +1,122 @@ +"""Model related to branches.""" + +from r2x.models.core import Device +from r2x.models.topology import ACBus, DCBus, Area +from typing import Annotated +from pydantic import Field, NonNegativeFloat, NonPositiveFloat +from r2x.units import ActivePower, Percentage + + +class Branch(Device): + """Class representing a connection between components.""" + + @classmethod + def example(cls) -> "Branch": + return Branch(name="ExampleBranch") + + +class ACBranch(Branch): + """Class representing an AC connection between components.""" + + # arc: Annotated[Arc, Field(description="The branch's connections.")] + from_bus: Annotated[ACBus, Field(description="Bus connected upstream from the arc.")] + to_bus: Annotated[ACBus, Field(description="Bus connected downstream from the arc.")] + r: Annotated[float, Field(description=("Resistance of the branch"))] = 0 + x: Annotated[float, Field(description=("Reactance of the branch"))] = 0 + b: Annotated[float, Field(description=("Shunt susceptance of the branch"))] = 0 + rating: Annotated[ActivePower, Field(ge=0, description="Thermal rating of the line.")] | None = None + + +class MonitoredLine(ACBranch): + """Class representing an AC transmission line.""" + + rating_up: Annotated[ActivePower, Field(ge=0, description="Forward rating of the line.")] | None = None + rating_down: Annotated[ActivePower, Field(le=0, description="Reverse rating of the line.")] | None = None + losses: Annotated[Percentage, Field(description="Power losses on the line.")] | None = None + + @classmethod + def example(cls) -> "MonitoredLine": + return MonitoredLine( + name="ExampleMonitoredLine", + from_bus=ACBus.example(), + to_bus=ACBus.example(), + losses=Percentage(10, "%"), + rating_up=ActivePower(100, "MW"), + rating_down=ActivePower(-100, "MW"), + rating=ActivePower(100, "MW"), + ) + + +class Line(ACBranch): + """Class representing an AC transmission line.""" + + @classmethod + def example(cls) -> "Line": + return Line( + name="ExampleLine", + from_bus=ACBus.example(), + to_bus=ACBus.example(), + rating=ActivePower(100, "MW"), + ) + + +class Transformer2W(ACBranch): + """Class representing a 2-W transformer.""" + + rate: Annotated[NonNegativeFloat, Field(description="Rating of the transformer.")] + + @classmethod + def example(cls) -> "Transformer2W": + return Transformer2W( + name="Example2WTransformer", + rate=100, + from_bus=ACBus.example(), + to_bus=ACBus.example(), + ) + + +class DCBranch(Branch): + """Class representing a DC connection between components.""" + + from_bus: Annotated[DCBus, Field(description="Bus connected upstream from the arc.")] + to_bus: Annotated[DCBus, Field(description="Bus connected downstream from the arc.")] + + +class AreaInterchange(Branch): + """Collection of branches that make up an interfece or corridor for the transfer of power.""" + + max_power_flow: Annotated[ActivePower, Field(ge=0, description="Maximum allowed flow.")] + min_power_flow: Annotated[ActivePower, Field(le=0, description="Minimum allowed flow.")] + from_area: Annotated[Area, Field(description="Area containing the bus.")] | None = None + to_area: Annotated[Area, Field(description="Area containing the bus.")] | None = None + + @classmethod + def example(cls) -> "AreaInterchange": + return AreaInterchange( + name="ExampleAreaInterchange", + max_power_flow=ActivePower(100, "MW"), + min_power_flow=ActivePower(-100, "MW"), + from_area=Area.example(), + to_area=Area.example(), + ) + + +class TModelHVDCLine(DCBranch): + """Class representing a DC transmission line.""" + + rating_up: Annotated[NonNegativeFloat, Field(description="Forward rating of the line.")] | None = None + rating_down: Annotated[NonPositiveFloat, Field(description="Reverse rating of the line.")] | None = None + losses: Annotated[NonNegativeFloat, Field(description="Power losses on the line.")] = 0 + resistance: Annotated[NonNegativeFloat, Field(description="Resistance of the line in p.u.")] | None = 0 + inductance: Annotated[NonNegativeFloat, Field(description="Inductance of the line in p.u.")] | None = 0 + capacitance: Annotated[NonNegativeFloat, Field(description="Capacitance of the line in p.u.")] | None = 0 + + @classmethod + def example(cls) -> "TModelHVDCLine": + return TModelHVDCLine( + name="ExampleDCLine", + from_bus=DCBus.example(), + to_bus=DCBus.example(), + rating_up=100, + rating_down=80, + ) diff --git a/src/r2x/models/core.py b/src/r2x/models/core.py new file mode 100644 index 00000000..76489df0 --- /dev/null +++ b/src/r2x/models/core.py @@ -0,0 +1,59 @@ +"""Core models for R2X.""" + +from collections import defaultdict, namedtuple + +from infrasys.component import Component +from typing import Annotated +from pydantic import Field, field_serializer +from r2x.units import ureg + + +class BaseComponent(Component): + """Infrasys base component with additional fields for R2X.""" + + available: Annotated[bool, Field(description="If the component is available.")] = True + category: Annotated[str, Field(description="Category that this component belongs to.")] | None = None + ext: dict = Field(default_factory=dict, description="Additional information of the component.") + + @property + def class_type(self) -> str: + """Create attribute that holds the class name.""" + return type(self).__name__ + + @field_serializer("ext", when_used="json") + def serialize_ext(ext: dict): # type:ignore # noqa: N805 + for key, value in ext.items(): + if isinstance(value, ureg.Quantity): + ext[key] = value.magnitude + return ext + + +MinMax = namedtuple("MinMax", ["min", "max"]) + + +class Service(BaseComponent): + """Abstract class for services attached to components.""" + + +class Device(BaseComponent): + """Abstract class for devices.""" + + services: ( + Annotated[ + list[Service], + Field(description="Services that this component contributes to.", default_factory=list), + ] + | None + ) = None + + +class StaticInjection(Device): + """Supertype for all static injection devices.""" + + +class TransmissionInterfaceMap(BaseComponent): + mapping: defaultdict[str, list] = defaultdict(list) # noqa: RUF012 + + +class ReserveMap(BaseComponent): + mapping: defaultdict[str, list] = defaultdict(list) # noqa: RUF012 diff --git a/src/r2x/models/costs.py b/src/r2x/models/costs.py new file mode 100644 index 00000000..0c5d6735 --- /dev/null +++ b/src/r2x/models/costs.py @@ -0,0 +1,50 @@ +"""Cost related functions.""" + +from infrasys import Component +from typing import Annotated +from pydantic import Field +from infrasys.cost_curves import ProductionVariableCostCurve +from r2x.units import FuelPrice + + +class OperationalCost(Component): + name: Annotated[str, Field(frozen=True)] = "" + + +class RenewableGenerationCost(OperationalCost): + curtailment_cost: ProductionVariableCostCurve | None = None + variable: ProductionVariableCostCurve | None = None + + +class HydroGenerationCost(OperationalCost): + fixed: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice(0.0, "usd/MWh") + variable: ProductionVariableCostCurve | None = None + + +class ThermalGenerationCost(OperationalCost): + start_up: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + fixed: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice(0.0, "usd/MWh") + shut_down: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + variable: ProductionVariableCostCurve | None = None + + +class StorageCost(OperationalCost): + start_up: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + fixed: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice(0.0, "usd/MWh") + shut_down: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + energy_surplus_cost: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + energy_storage_cost: Annotated[FuelPrice, Field(description="Cost of using fuel in $/MWh.")] = FuelPrice( + 0.0, "usd/MWh" + ) + charge_variable_cost: ProductionVariableCostCurve | None = None + discharge_variable_cost: ProductionVariableCostCurve | None = None diff --git a/src/r2x/models/generators.py b/src/r2x/models/generators.py new file mode 100644 index 00000000..5402874a --- /dev/null +++ b/src/r2x/models/generators.py @@ -0,0 +1,312 @@ +"""Models for generator devices.""" + +from typing import Annotated + +from pydantic import Field, NonNegativeFloat + +from r2x.models.core import Device +from r2x.models.costs import HydroGenerationCost, RenewableGenerationCost, ThermalGenerationCost, StorageCost +from r2x.models.topology import ACBus +from r2x.models.load import PowerLoad +from r2x.enums import PrimeMoversType +from r2x.units import ( + ActivePower, + FuelPrice, + Percentage, + PowerRate, + ApparentPower, + ureg, + Time, + Energy, +) + + +class Generator(Device): + """Abstract generator class.""" + + bus: Annotated[ACBus, Field(description="Bus where the generator is connected.")] | None = None + rating: Annotated[ + ApparentPower | None, + Field(gt=0, description="Maximum output power rating of the unit (MVA)."), + ] = None + active_power: Annotated[ + ActivePower, + Field( + description=( + "Initial active power set point of the unit in MW. For power flow, this is the steady " + "state operating point of the system." + ), + ), + ] + operation_cost: ( + ThermalGenerationCost | RenewableGenerationCost | HydroGenerationCost | StorageCost | None + ) = None + base_power: Annotated[ + ApparentPower | None, + Field( + gt=0, + description="Base power of the unit (MVA) for per unitization.", + ), + ] = None + must_run: Annotated[int | None, Field(description="If we need to force the dispatch of the device.")] = ( + None + ) + vom_price: Annotated[FuelPrice, Field(description="Variable operational price $/MWh.")] | None = None + prime_mover_type: ( + Annotated[PrimeMoversType, Field(description="Prime mover technology according to EIA 923.")] | None + ) = None + unit_type: Annotated[ + PrimeMoversType | None, Field(description="Prime mover technology according to EIA 923.") + ] = None + min_rated_capacity: Annotated[ActivePower, Field(description="Minimum rated power generation.")] = ( + 0 * ureg.MW + ) + ramp_up: ( + Annotated[ + PowerRate, + Field(description="Ramping rate on the positve direction."), + ] + | None + ) = None + ramp_down: ( + Annotated[ + PowerRate, + Field(description="Ramping rate on the negative direction."), + ] + | None + ) = None + min_up_time: ( + Annotated[ + Time, + Field(ge=0, description="Minimum up time in hours for UC decision."), + ] + | None + ) = None + min_down_time: ( + Annotated[ + Time, + Field(ge=0, description="Minimum down time in hours for UC decision."), + ] + | None + ) = None + mean_time_to_repair: ( + Annotated[ + Time, + Field(gt=0, description="Total hours to repair after outage occur."), + ] + | None + ) = None + forced_outage_rate: ( + Annotated[ + Percentage, + Field(description="Expected level of unplanned outages in percent."), + ] + | None + ) = None + planned_outage_rate: ( + Annotated[ + Percentage, + Field(description="Expected level of planned outages in percent."), + ] + | None + ) = None + startup_cost: Annotated[NonNegativeFloat, Field(description="Cost in $ of starting a unit.")] | None = ( + None + ) + shutdown_cost: ( + Annotated[NonNegativeFloat, Field(description="Cost in $ of shuting down a unit.")] | None + ) = None + + +class RenewableGen(Generator): + """Abstract class for renewable generators.""" + + +class RenewableDispatch(RenewableGen): + """Curtailable renewable generator. + + This type of generator have a hourly capacity factor profile. + """ + + +class RenewableNonDispatch(RenewableGen): + """Non-curtailable renewable generator. + + Renewable technologies w/o operational cost. + """ + + +class HydroGen(Generator): + """Hydroelectric generator.""" + + +class HydroDispatch(HydroGen): + """Class representing flexible hydro generators.""" + + ramp_up: ( + Annotated[ + PowerRate, + Field(ge=0, description="Ramping rate on the positve direction."), + ] + | None + ) = None + ramp_down: ( + Annotated[ + PowerRate, + Field(ge=0, description="Ramping rate on the negative direction."), + ] + | None + ) = None + + +class HydroFix(HydroGen): + """Class representing unflexible hydro.""" + + +class HydroEnergyReservoir(HydroGen): + """Class representing hydro system with reservoirs.""" + + initial_energy: ( + Annotated[NonNegativeFloat, Field(description="Initial water volume or percentage.")] | None + ) = 0 + storage_capacity: ( + Annotated[ + Energy, + Field(description="Total water volume or percentage."), + ] + | None + ) = None + min_storage_capacity: ( + Annotated[ + Energy, + Field(description="Minimum water volume or percentage."), + ] + | None + ) = None + storage_target: ( + Annotated[ + Energy, + Field(description="Maximum energy limit."), + ] + | None + ) = None + + +class HydroPumpedStorage(HydroGen): + """Class representing pumped hydro generators.""" + + storage_duration: ( + Annotated[ + Time, + Field(description="Storage duration in hours."), + ] + | None + ) = None + initial_volume: ( + Annotated[Energy, Field(gt=0, description="Initial water volume or percentage.")] | None + ) = None + storage_capacity: Annotated[ + Energy, + Field(gt=0, description="Total water volume or percentage."), + ] + min_storage_capacity: ( + Annotated[ + Energy, + Field(description="Minimum water volume or percentage."), + ] + | None + ) = None + pump_efficiency: Annotated[Percentage, Field(ge=0, le=1, description="Pumping efficiency.")] | None = None + pump_load: ( + Annotated[ + ActivePower, + Field(description="Load related to the usage of the pump."), + ] + | None + ) = None + + @classmethod + def example(cls) -> "HydroPumpedStorage": + return HydroPumpedStorage( + name="HydroPumpedStorage", + active_power=ActivePower(100, "MW"), + pump_load=ActivePower(100, "MW"), + bus=ACBus.example(), + prime_mover_type=PrimeMoversType.PS, + storage_duration=Time(10, "h"), + storage_capacity=Energy(1000, "MWh"), + min_storage_capacity=Energy(10, "MWh"), + pump_efficiency=Percentage(85, "%"), + initial_volume=Energy(500, "MWh"), + ext={"description": "Pumped hydro unit with 10 hour of duration"}, + ) + + +class ThermalGen(Generator): + """Class representing fuel based thermal generator.""" + + fuel: Annotated[str, Field(description="Fuel category")] | None = None + + +class ThermalStandard(ThermalGen): + """Standard representation of thermal device.""" + + @classmethod + def example(cls) -> "ThermalStandard": + return ThermalStandard( + name="ThermalStandard", + bus=ACBus.example(), + fuel="gas", + active_power=100.0 * ureg.MW, + ext={"Additional data": "Additional value"}, + ) + + +class ThermalMultiStart(ThermalGen): + """We will fill this class once we have the need for it.""" + + +class Storage(Generator): + """Default Storage class.""" + + storage_duration: ( + Annotated[ + Time, + Field(description="Storage duration in hours."), + ] + | None + ) = None + storage_capacity: Annotated[ + Energy, + Field(description="Maximum allowed volume or state of charge."), + ] + initial_energy: Annotated[Percentage, Field(description="Initial state of charge.")] | None = None + min_storage_capacity: Annotated[Percentage, Field(description="Minimum state of charge")] = Percentage( + 0, "%" + ) + max_storage_capacity: Annotated[Percentage, Field(description="Minimum state of charge")] = Percentage( + 100, "%" + ) + + +class GenericBattery(Storage): + """Battery energy storage model.""" + + charge_efficiency: Annotated[Percentage, Field(ge=0, description="Charge efficiency.")] | None = None + discharge_efficiency: Annotated[Percentage, Field(ge=0, description="Discharge efficiency.")] | None = ( + None + ) + + +class HybridSystem(Device): + """Representation of hybrid system with renewable generation, load, thermal generation and storage. + + This class is just a link between two components. + For the implementation see: + - https://github.com/NREL-Sienna/PowerSystems.jl/blob/main/src/models/HybridSystem.jl + """ + + storage_unit: Storage | None = None + renewable_unit: RenewableGen | None = None + thermal_unit: ThermalGen | None = None + electric_load: PowerLoad | None = None diff --git a/src/r2x/models/load.py b/src/r2x/models/load.py new file mode 100644 index 00000000..da34fe17 --- /dev/null +++ b/src/r2x/models/load.py @@ -0,0 +1,85 @@ +"""Electric load related models.""" + +from typing import Annotated + +from pydantic import Field + +from r2x.models.core import StaticInjection +from r2x.models.topology import ACBus, Bus +from r2x.units import ActivePower, ApparentPower + + +class ElectricLoad(StaticInjection): + """Supertype for all electric loads.""" + + bus: Bus = Field(description="Point of injection.") + + +class StaticLoad(ElectricLoad): + """Supertype for static loads.""" + + +class ControllableLoad(ElectricLoad): + """Abstract class for controllable loads.""" + + +class PowerLoad(StaticLoad): + """Class representing a Load object.""" + + active_power: ( + Annotated[ + ActivePower, + Field(gt=0, description="Initial steady-state active power demand."), + ] + | None + ) = None + reactive_power: ( + Annotated[float, Field(gt=0, description="Reactive Power of Load at the bus in MW.")] | None + ) = None + max_active_power: Annotated[ActivePower, Field(gt=0, description="Max Load at the bus in MW.")] | None = ( + None + ) + max_reactive_power: ( + Annotated[ActivePower, Field(gt=0, description=" Initial steady-state reactive power demand.")] | None + ) = None + base_power: Annotated[ + ApparentPower | None, + Field( + gt=0, + description="Base power of the unit (MVA) for per unitization.", + ), + ] = None + operation_cost: float | None = None + + @classmethod + def example(cls) -> "PowerLoad": + return PowerLoad(name="ExampleLoad", bus=ACBus.example(), active_power=ActivePower(1000, "MW")) + + +class InterruptiblePowerLoad(ControllableLoad): + """A static interruptible power load.""" + + base_power: Annotated[ + ApparentPower | None, + Field( + gt=0, + description="Base power of the unit (MVA) for per unitization.", + ), + ] = None + active_power: ( + Annotated[ + ActivePower, + Field(gt=0, description="Initial steady-state active power demand."), + ] + | None + ) = None + reactive_power: ( + Annotated[float, Field(gt=0, description="Reactive Power of Load at the bus in MW.")] | None + ) = None + max_active_power: Annotated[ActivePower, Field(ge=0, description="Max Load at the bus in MW.")] | None = ( + None + ) + max_reactive_power: ( + Annotated[ActivePower, Field(gt=0, description=" Initial steady-state reactive power demand.")] | None + ) = None + operation_cost: float | None = None diff --git a/src/r2x/models/services.py b/src/r2x/models/services.py new file mode 100644 index 00000000..83df94dc --- /dev/null +++ b/src/r2x/models/services.py @@ -0,0 +1,105 @@ +"""Models related to services.""" + +from typing import Annotated + +from pydantic import Field, NonNegativeFloat, PositiveFloat + +from r2x.enums import ReserveDirection, ReserveType +from r2x.models.core import MinMax, Service +from r2x.models.topology import LoadZone +from r2x.units import EmissionRate + + +class Reserve(Service): + """Class representing a reserve contribution.""" + + time_frame: Annotated[ + PositiveFloat, + Field(description="Timeframe in which the reserve is required in seconds"), + ] = 1e30 + region: ( + Annotated[ + LoadZone, + Field(description="LoadZone where reserve requirement is required."), + ] + | None + ) = None + vors: Annotated[ + float, + Field(description="Value of reserve shortage in $/MW. Any positive value as as soft constraint."), + ] = -1 + duration: ( + Annotated[ + PositiveFloat, + Field(description="Time over which the required response must be maintained in seconds."), + ] + | None + ) = None + reserve_type: ReserveType + load_risk: ( + Annotated[ + NonNegativeFloat, + Field( + description="Proportion of Load that contributes to the requirement.", + ), + ] + | None + ) = None + # ramp_rate: float | None = None # NOTE: Maybe we do not need this. + max_requirement: float = 0 # Should we specify which variable is the time series for? + direction: ReserveDirection + + @classmethod + def example(cls) -> "Reserve": + return Reserve( + name="ExampleReserve", + region=LoadZone.example(), + direction=ReserveDirection.Up, + reserve_type=ReserveType.Regulation, + ) + + +class Emission(Service): + """Class representing an emission object that is attached to generators.""" + + rate: Annotated[EmissionRate, Field(description="Amount of emission produced in kg/MWh.")] + emission_type: Annotated[str, Field(description="Type of emission. E.g., CO2, NOx.")] + generator_name: Annotated[str, Field(description="Generator emitting.")] + + @classmethod + def example(cls) -> "Emission": + return Emission( + name="ExampleEmission", + generator_name="gen1", + rate=EmissionRate(105, "kg/MWh"), + emission_type="CO2", + ) + + +class TransmissionInterface(Service): + """A collection of branches that make up an interface or corridor for the transfer of power + such as between different :class:Area or :class:LoadZone. + + The interface can be used to constrain the power flow across it + """ + + active_power_flow_limits: Annotated[ + MinMax, Field(description="Minimum and maximum active power flow limits on the interface (MW)") + ] + direction_mapping: Annotated[ + dict[str, int], + Field( + description=( + "Dictionary of the line names in the interface and their direction of flow (1 or -1) " + "relative to the flow of the interface" + ) + ), + ] + + @classmethod + def example(cls) -> "TransmissionInterface": + return TransmissionInterface( + name="ExampleTransmissionInterface", + active_power_flow_limits=MinMax(-100, 100), + direction_mapping={"line-01": 1, "line-02": -2}, + ) diff --git a/src/r2x/models/topology.py b/src/r2x/models/topology.py new file mode 100644 index 00000000..f6178d4e --- /dev/null +++ b/src/r2x/models/topology.py @@ -0,0 +1,97 @@ +"""Models that capture topology types.""" + +from typing import Annotated + +from pydantic import Field, NonNegativeFloat, PositiveFloat, PositiveInt + +from r2x.enums import ACBusTypes +from r2x.models.core import BaseComponent +from r2x.units import Voltage, ureg + + +class Topology(BaseComponent): + """Abstract type to represent the structure and interconnectedness of the system.""" + + +class AggregationTopology(Topology): + """Base class for area-type components.""" + + +class Area(AggregationTopology): + """Collection of buses in a given region.""" + + peak_active_power: Annotated[NonNegativeFloat, Field(description="Peak active power in the area")] = 0.0 + peak_reactive_power: Annotated[NonNegativeFloat, Field(description="Peak reactive power in the area")] = ( + 0.0 + ) + load_response: Annotated[ + float, + Field( + description=( + "Load-frequency damping parameter modeling how much the load in the area changes " + "due to changes in frequency (MW/Hz)." + ) + ), + ] = 0.0 + + @classmethod + def example(cls) -> "Area": + return Area(name="New York") + + +class LoadZone(AggregationTopology): + """Collection of buses for electricity price analysis.""" + + @classmethod + def example(cls) -> "LoadZone": + return LoadZone(name="ExampleLoadZone") + + +class Bus(Topology): + """Abstract class for a bus.""" + + number: Annotated[PositiveInt, Field(description="A unique bus identification number.")] + bus_type: Annotated[ACBusTypes, Field(description="Type of category of bus,")] | None = None + area: Annotated[Area, Field(description="Area containing the bus.")] | None = None + load_zone: Annotated[LoadZone, Field(description="the load zone containing the DC bus.")] | None = None + base_voltage: ( + Annotated[Voltage, Field(gt=0, description="Base voltage in kV. Unit compatible with voltage.")] + | None + ) = None + magnitude: ( + Annotated[PositiveFloat, Field(description="Voltage as a multiple of base_voltage.")] | None + ) = None + + +class DCBus(Bus): + """Power-system DC Bus.""" + + @classmethod + def example(cls) -> "DCBus": + return DCBus( + number=1, + name="ExampleBus", + load_zone=LoadZone.example(), + area=Area.example(), + base_voltage=20 * ureg.kV, + bus_type=ACBusTypes.PV, + ) + + +class ACBus(Bus): + """Power-system AC bus.""" + + angle: Annotated[float | None, Field(description="Angle of the bus in radians.", gt=-1.571, lt=1.571)] = ( + None + ) + + @classmethod + def example(cls) -> "ACBus": + return ACBus( + number=1, + name="ExampleBus", + load_zone=LoadZone.example(), + area=Area.example(), + base_voltage=13 * ureg.kV, + bus_type=ACBusTypes.PV, + ) diff --git a/src/r2x/models/utils.py b/src/r2x/models/utils.py new file mode 100644 index 00000000..5c05fad0 --- /dev/null +++ b/src/r2x/models/utils.py @@ -0,0 +1,31 @@ +"""Useful function for models.""" + +from .generators import Generator, ThermalGen, HydroGen, Storage, RenewableGen +from .costs import ( + OperationalCost, + ThermalGenerationCost, + StorageCost, + HydroGenerationCost, + RenewableGenerationCost, +) +from loguru import logger + + +def get_operational_cost(model: type["Generator"]) -> type["OperationalCost"] | None: + """Return operational cost for the type of generator model.""" + match model: + case _ if issubclass(model, ThermalGen): + return ThermalGenerationCost + case _ if issubclass(model, HydroGen): + return HydroGenerationCost + case _ if issubclass(model, RenewableGen): + return RenewableGenerationCost + case _ if issubclass(model, Storage): + return StorageCost + case _: + msg = ( + f"{model=} does not have an operational cost. " + "Check that a operational cost exist for the model." + ) + logger.warning(msg) + return None diff --git a/src/r2x/parser/plexos.py b/src/r2x/parser/plexos.py index 1417a257..f2d17471 100644 --- a/src/r2x/parser/plexos.py +++ b/src/r2x/parser/plexos.py @@ -26,7 +26,7 @@ from r2x.exceptions import ModelError, ParserError from plexosdb import PlexosSQLite from plexosdb.enums import ClassEnum, CollectionEnum -from r2x.model import ( +from r2x.models import ( ACBus, Generator, GenericBattery, @@ -44,9 +44,9 @@ from .handler import PCMParser from .parser_helpers import handle_leap_year_adjustment, fill_missing_timestamps, resample_data_to_hourly -models = importlib.import_module("r2x.model") +models = importlib.import_module("r2x.models") -R2X_MODELS = importlib.import_module("r2x.model") +R2X_MODELS = importlib.import_module("r2x.models") BASE_WEATHER_YEAR = 2007 XML_FILE_KEY = "xml_file" PROPERTY_SV_COLUMNS_BASIC = ["name", "value"] diff --git a/src/r2x/plugins/break_gens.py b/src/r2x/plugins/break_gens.py index a280aef3..37aef945 100644 --- a/src/r2x/plugins/break_gens.py +++ b/src/r2x/plugins/break_gens.py @@ -13,7 +13,7 @@ import pandas as pd from r2x.api import System -from r2x.model import Emission, Generator +from r2x.models import Emission, Generator from r2x.config import Scenario from r2x.parser.handler import BaseParser from r2x.units import ureg, ActivePower @@ -116,16 +116,16 @@ def break_generators( # noqa: C901 continue logger.trace("Average_capacity: {}", avg_capacity) reference_base_power = ( - component.base_power.magnitude - if isinstance(component.base_power, BaseQuantity) - else component.base_power + component.active_power.magnitude + if isinstance(component.active_power, BaseQuantity) + else component.active_power ) no_splits = int(reference_base_power // avg_capacity) remainder = reference_base_power % avg_capacity if no_splits > 1: split_no = 1 logger.trace( - "Breaking generator {} with base_power {} into {} generators of {} capacity", + "Breaking generator {} with active_power {} into {} generators of {} capacity", component.name, reference_base_power, no_splits, @@ -135,19 +135,19 @@ def break_generators( # noqa: C901 component_name = component.name + f"_{split_no:02}" new_component = system.copy_component(component, name=component_name, attach=True) new_base_power = ( - ActivePower(avg_capacity, component.base_power.units) - if isinstance(component.base_power, BaseQuantity) + ActivePower(avg_capacity, component.active_power.units) + if isinstance(component.active_power, BaseQuantity) else avg_capacity * ureg.MW ) - new_component.base_power = new_base_power + new_component.active_power = new_base_power proportion = ( avg_capacity / reference_base_power - ) # Required to recalculate properties that depend on base_power + ) # Required to recalculate properties that depend on active_power for property in PROPERTIES_TO_BREAK: if attr := getattr(new_component, property, None): new_component.ext[f"{property}_original"] = attr setattr(new_component, property, attr * proportion) - new_component.ext["original_capacity"] = component.base_power + new_component.ext["original_capacity"] = component.active_power new_component.ext["original_name"] = component.name new_component.ext["broken"] = True @@ -172,15 +172,15 @@ def break_generators( # noqa: C901 if remainder > capacity_threshold: component_name = component.name + f"_{split_no:02}" new_component = system.copy_component(component, name=component_name, attach=True) - new_component.base_power = remainder * ureg.MW + new_component.active_power = remainder * ureg.MW proportion = ( remainder / reference_base_power - ) # Required to recalculate properties that depend on base_power + ) # Required to recalculate properties that depend on active_power for property in PROPERTIES_TO_BREAK: if attr := getattr(new_component, property, None): new_component.ext[f"{property}_original"] = attr setattr(new_component, property, attr * proportion) - new_component.ext["original_capacity"] = component.base_power + new_component.ext["original_capacity"] = component.active_power new_component.ext["original_name"] = component.name new_component.ext["broken"] = True # NOTE: This will be migrated once we implement the SQLite for the components. diff --git a/src/r2x/runner.py b/src/r2x/runner.py index 9cdda89f..8e51b35a 100644 --- a/src/r2x/runner.py +++ b/src/r2x/runner.py @@ -10,7 +10,7 @@ from infrasys.system import System from r2x.exporter.handler import get_exporter -# from r2x.model import * +# from r2x.models import * # Module level imports from .config import Configuration, Scenario from .exporter import exporter_list diff --git a/src/r2x/units.py b/src/r2x/units.py index cd9c0a45..dc0adbb7 100644 --- a/src/r2x/units.py +++ b/src/r2x/units.py @@ -14,7 +14,7 @@ class Distance(BaseQuantity): class Voltage(BaseQuantity): - __base_unit__ = "volt" + __base_unit__ = "kilovolt" class Current(BaseQuantity): @@ -26,7 +26,11 @@ class Angle(BaseQuantity): class ActivePower(BaseQuantity): - __base_unit__ = "watt" + __base_unit__ = "megawatt" + + +class ApparentPower(BaseQuantity): + __base_unit__ = "volt_ampere" class Time(BaseQuantity): diff --git a/src/r2x/utils.py b/src/r2x/utils.py index 8e9f2c45..6487cf1d 100644 --- a/src/r2x/utils.py +++ b/src/r2x/utils.py @@ -28,6 +28,9 @@ import yaml from jsonschema import validate from loguru import logger +import pint +from infrasys.base_quantity import BaseQuantity +from r2x.models import Generator DEFAULT_OUTPUT_FOLDER: str = "r2x_export" @@ -57,10 +60,24 @@ def match_input_model(input_model: str) -> dict: fmap = {} case "reeds-US": fmap = read_fmap("r2x/defaults/reeds_us_mapping.json") + case "reeds-India": + fmap = read_fmap("r2x/defaults/india_mapping.json") case "sienna": fmap = read_fmap("r2x/defaults/sienna_mapping.json") case "plexos": fmap = read_fmap("r2x/defaults/plexos_mapping.json") + case "nodal-plexos": + fmap = ( + read_fmap("r2x/defaults/nodal_mapping.json") + | read_fmap("r2x/defaults/plexos_mapping.json") + | read_fmap("r2x/defaults/reeds_us_mapping.json") + ) + case "nodal-sienna": + fmap = ( + read_fmap("r2x/defaults/sienna_mapping.json") + | read_fmap("r2x/defaults/nodal_mapping.json") + | read_fmap("r2x/defaults/reeds_us_mapping.json") + ) case _: logger.error("Input model {} not recognized", input_model) raise KeyError(f"Input model {input_model=} not valid") @@ -147,7 +164,6 @@ 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): @@ -437,6 +453,74 @@ def get_csv(fpath: str, fname: str, fmap: dict[str, str | dict | list] = {}, **k return data +# NOTE: This is not actually used in the process but is kept here for documentation +# purposes. +def create_nodal_database( + plexos_node_to_reeds_region_fpath: str, + node_objects_fpath: str, + reeds_shapefiles_fpath: str, +) -> pd.DataFrame: + """Un-used function, but creates the nodes objects for the Nodal dataset.""" + import geopandas as gpd # type: ignore + + nodes_to_ba = pd.read_csv(plexos_node_to_reeds_region_fpath) + nodes_to_ba["node_id"] = nodes_to_ba["node"].str.split("_", n=1).str[0] + nodes_to_ba = nodes_to_ba.loc[nodes_to_ba["node_id"].str.isnumeric()] + nodes_to_ba = nodes_to_ba.groupby("node_id").agg({"reeds_region": list}) + nodes_to_ba["num_bas"] = nodes_to_ba["reeds_region"].str.len() + nodes_to_ba = nodes_to_ba.loc[nodes_to_ba["num_bas"] == 1] + nodes_to_ba["reeds_region"] = nodes_to_ba["reeds_region"].str[0] + nodes_to_ba = nodes_to_ba.reset_index()[["node_id", "reeds_region"]] + + node_objects = node_objects_fpath + usecols = ["Latitude", "Longitude", "Node ID", "kV", "Node", "Load Factor"] + rename_dict = { + "Node ID": "node_id", + "kV": "voltage", + "Load Factor": "load_participation_factor", + } + dtype = { + "Node ID": "str", + "Latitude": "float64", + "Longitude": "float64", + "kV": "float64", + "Load Factor": "float64", + } + nodes = ( + pd.read_csv(node_objects, low_memory=False, dtype=dtype, usecols=usecols) # type: ignore + .rename(columns=rename_dict) + .rename(columns=str.lower) + ) + + # Convert it to geopandas dataframe to assign the ReEDs region from the shapefile + # NOTE: This step is not necessary if the node csv has the ReEDS region. + reeds_shp = gpd.read_file(reeds_shapefiles_fpath) + nodes = gpd.GeoDataFrame(nodes, geometry=gpd.points_from_xy(nodes.longitude, nodes.latitude)).set_crs( + "epsg:4326" + ) + nodes = gpd.sjoin(nodes, reeds_shp, predicate="within") + + nodes_to_bas = nodes.merge(nodes_to_ba, on="node_id", how="outer") + nodes_to_bas.loc[nodes_to_bas["reeds_region"].isna(), "reeds_region"] = nodes_to_bas["pca"] + nodes_to_bas.loc[ + (nodes_to_bas["reeds_region"] != nodes_to_bas["pca"]) & (~nodes_to_bas["pca"].isna()), + "reeds_region", + ] = nodes_to_bas["pca"] + + node_data = nodes_to_bas[ + [ + "node_id", + "latitude", + "longitude", + "reeds_region", + "voltage", + "load_participation_factor", + ] + ] + node_data["plexos_id"] = node_data["node_id"] + return node_data + + def invert_dict(d: dict[str, list]) -> dict[str, str]: """Inverse dictionary with list values. @@ -483,16 +567,43 @@ def get_defaults( logger.debug("Returning base defaults") return config_dict - # There is 3 paths for this to go: + # There is 4 paths for this to go: # 1. Zonal translations should always go throught "reeds-US" (as far as we only support one CEM"), # 2. If you want to translate a Plexos <-> Sienna model, # 3. If you want to read an existing infrasys system, + # 4. If you want to run zonal to nodal. + # 4.1 Plexos + # 4.2 Sienna match input_model: case "infrasys": logger.debug("Returning infrasys defaults") case "reeds-US": config_dict = config_dict | read_json("r2x/defaults/reeds_input.json") logger.debug("Returning reeds defaults") + case "reeds-India": + config_dict = config_dict | read_json("r2x/defaults/nodal_defaults.json") + logger.debug("Returning nodal defaults") + case "nodal-sienna": + config_dict = ( + config_dict + | read_json("r2x/defaults/nodal_defaults.json") + | read_json("r2x/defaults/pcm_defaults.json") + | read_json("r2x/defaults/sienna_config.json") + | read_json("r2x/defaults/reeds_input.json") + ) + logger.debug("Returning nodal-sienna defaults") + case "nodal-plexos": + config_dict = ( + config_dict + | read_json("r2x/defaults/nodal_defaults.json") + | read_json("r2x/defaults/pcm_defaults.json") + | read_json("r2x/defaults/plexos_input.json") + | read_json("r2x/defaults/reeds_input.json") + ) + logger.debug("Returning nodal-plexos defaults") + case "sienna": + config_dict = config_dict | read_json("r2x/defaults/sienna_config.json") + logger.debug("Returning sienna defaults") case "plexos": config_dict = config_dict | read_json("r2x/defaults/plexos_input.json") logger.debug("Returning input_model {} defaults", input_model) @@ -578,5 +689,25 @@ def batched(lst, n): return iter(lambda: tuple(islice(it, n)), ()) +def get_property_magnitude(property_value, to_unit: str | None = None) -> float: + """Return magnitude with the given units for a pint Quantity. + + Parameters + ---------- + property_name + + property_value + pint.Quantity to extract magnitude from + to_unit + String that contains the unit conversion desired. Unit must be compatible. + """ + if not isinstance(property_value, pint.Quantity | BaseQuantity): + return property_value + if to_unit: + unit = to_unit.replace("$", "usd") # Dollars are named usd on pint + property_value = property_value.to(unit) + return property_value.magnitude + + DEFAULT_COLUMN_MAP = read_json("r2x/defaults/config.json").get("default_column_mapping") mapping_schema = json.loads(files("r2x.defaults").joinpath("mapping_schema.json").read_text()) diff --git a/tests/models/systems.py b/tests/models/ieee5bus.py similarity index 67% rename from tests/models/systems.py rename to tests/models/ieee5bus.py index 3d9b2a8d..e53a0754 100644 --- a/tests/models/systems.py +++ b/tests/models/ieee5bus.py @@ -1,22 +1,24 @@ """Script that creates simple systems for testing.""" from datetime import datetime, timedelta + +from infrasys.cost_curves import FuelCurve +from infrasys.function_data import LinearFunctionData from infrasys.time_series_models import SingleTimeSeries -import pytest +from infrasys.value_curves import InputOutputCurve from r2x.api import System -from r2x.model import ( +from r2x.enums import PrimeMoversType +from r2x.models import ( + ACBus, Area, GenericBattery, - MonitoredLine, - ACBus, LoadZone, + MonitoredLine, RenewableDispatch, - Reserve, - ReserveMap, ThermalStandard, ) -from r2x.enums import PrimeMoversType, ReserveDirection, ReserveType -from r2x.units import Energy, Time, ureg +from r2x.models.costs import ThermalGenerationCost +from r2x.units import Energy, Percentage, Time, ureg def ieee5bus_system() -> System: @@ -25,11 +27,11 @@ def ieee5bus_system() -> System: area_1 = Area(name="region1") load_zone_1 = LoadZone(name="LoadZone1") - bus_1 = ACBus(id=1, name="node_a", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) - bus_2 = ACBus(id=2, name="node_b", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) - bus_3 = ACBus(id=3, name="node_c", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) - bus_4 = ACBus(id=4, name="node_d", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) - bus_5 = ACBus(id=5, name="node_e", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) + bus_1 = ACBus(number=1, name="node_a", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) + bus_2 = ACBus(number=2, name="node_b", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) + bus_3 = ACBus(number=3, name="node_c", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) + bus_4 = ACBus(number=4, name="node_d", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) + bus_5 = ACBus(number=5, name="node_e", base_voltage=100 * ureg.volt, area=area_1, load_zone=load_zone_1) # Append buses generator for bus in [bus_1, bus_2, bus_3, bus_4, bus_5]: system.add_component(bus) @@ -46,14 +48,14 @@ def ieee5bus_system() -> System: name="SolarPV1", bus=bus_3, prime_mover_type=PrimeMoversType.PV, - base_power=384 * ureg.MW, + active_power=384 * ureg.MW, category="solar", ) solar_pv_02 = RenewableDispatch( name="SolarPV2", bus=bus_4, prime_mover_type=PrimeMoversType.PV, - base_power=384 * ureg.MW, + active_power=384 * ureg.MW, category="solar", ) system.add_component(solar_pv_01) @@ -65,8 +67,8 @@ def ieee5bus_system() -> System: name="Battery1", bus=bus_2, prime_mover_type=PrimeMoversType.BA, - base_power=200 * ureg.MW, - charge_efficiency=0.85 * ureg.Unit(""), + active_power=200 * ureg.MW, + charge_efficiency=Percentage(85, "%"), storage_capacity=Energy(800, "MWh"), storage_duration=Time(4, "h"), category="storage", @@ -78,9 +80,8 @@ def ieee5bus_system() -> System: name="Alta", fuel="gas", prime_mover_type=PrimeMoversType.CC, - base_power=40 * ureg.MW, + active_power=40 * ureg.MW, min_rated_capacity=10 * ureg.MW, - fuel_price=10 * ureg.Unit("usd/MWh"), bus=bus_1, category="thermal", ) @@ -89,9 +90,8 @@ def ieee5bus_system() -> System: name="Brighton", fuel="Gas", prime_mover_type=PrimeMoversType.CC, - base_power=600 * ureg.MW, + active_power=600 * ureg.MW, min_rated_capacity=150 * ureg.MW, - fuel_price=10 * ureg.Unit("usd/MWh"), bus=bus_5, category="thermal", ) @@ -100,9 +100,16 @@ def ieee5bus_system() -> System: name="Park City", fuel="Gas", prime_mover_type=PrimeMoversType.CC, - base_power=170 * ureg.MW, + active_power=170 * ureg.MW, min_rated_capacity=20 * ureg.MW, - fuel_price=10 * ureg.Unit("usd/MWh"), + operation_cost=ThermalGenerationCost( + variable=FuelCurve( + value_curve=InputOutputCurve( + function_data=LinearFunctionData(proportional_term=10, constant_term=0) + ), + fuel_cost=15, + ) + ), bus=bus_1, category="thermal", ) @@ -111,9 +118,15 @@ def ieee5bus_system() -> System: name="Solitude", fuel="Gas", prime_mover_type=PrimeMoversType.CC, - base_power=520 * ureg.MW, - min_rated_capacity=100 * ureg.MW, - fuel_price=10 * ureg.Unit("usd/MWh"), + active_power=520 * ureg.MW, + operation_cost=ThermalGenerationCost( + variable=FuelCurve( + value_curve=InputOutputCurve( + function_data=LinearFunctionData(proportional_term=10, constant_term=0) + ), + fuel_cost=15, + ) + ), bus=bus_3, category="thermal", ) @@ -122,7 +135,7 @@ def ieee5bus_system() -> System: name="Sundance", fuel="Gas", prime_mover_type=PrimeMoversType.CC, - base_power=400 * ureg.MW, + active_power=400 * ureg.MW, min_rated_capacity=80 * ureg.MW, bus=bus_4, category="thermal", @@ -148,24 +161,4 @@ def ieee5bus_system() -> System: branch_ed = MonitoredLine(name="line_ed", from_bus=bus_5, to_bus=bus_4, rating_up=240 * ureg.MW) system.add_component(branch_ed) - # Adding reserves - reserve = Reserve( - name="reg_down", - max_requirement=100, # MW - reserve_type=ReserveType.Regulation, - direction=ReserveDirection.Down, - ) - system.add_component(reserve) - reserve_map = ReserveMap(name="System reserves") - system.add_component(reserve_map) - - # Adding contributing devices - reserve_map.mapping[reserve.name].append(solitude.name) - reserve_map.mapping[reserve.name].append(sundance.name) - return system - - -@pytest.fixture -def ieee_5bus_test() -> System: - return ieee5bus_system() diff --git a/tests/test_api.py b/tests/test_api.py index 45f4ec11..8fe1a7f9 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,7 +1,7 @@ import pytest from r2x.__version__ import __data_model_version__ from r2x.api import System -from r2x.model import ACBus, Generator +from r2x.models import Area, ACBus, Generator from r2x.units import ureg @@ -20,9 +20,9 @@ def test_system_data_model_version(empty_system): def test_add_single_component(empty_system): - generator = Generator.example() - empty_system.add_component(generator) - assert isinstance(empty_system.get_component(Generator, generator.name), Generator) + area = Area.example() + empty_system.add_component(area) + assert isinstance(empty_system.get_component(Area, area.name), Area) def test_add_composed_component(): @@ -30,7 +30,7 @@ def test_add_composed_component(): # Simple scenario of Generator with a bus attached bus = ACBus.example() - generator = Generator(name="TestGen", base_power=100 * ureg.MW, bus=bus) + generator = Generator(name="TestGen", active_power=100 * ureg.MW, bus=bus) system.add_component(generator) assert system.get_component(Generator, "TestGen") == generator diff --git a/tests/test_break_gens.py b/tests/test_break_gens.py index 40b2d19e..70e8d9ba 100644 --- a/tests/test_break_gens.py +++ b/tests/test_break_gens.py @@ -1,7 +1,7 @@ from r2x.enums import PrimeMoversType -from r2x.model import Generator +from r2x.models import Generator from r2x.plugins.break_gens import break_generators -from .models.systems import ieee5bus_system +from .models.ieee5bus import ieee5bus_system def test_break_generators(): @@ -18,7 +18,7 @@ def test_break_generators(): Generator, filter_func=lambda x: x.prime_mover_type == PrimeMoversType.BA ) for generator in new_generators: - assert generator.base_power.magnitude == 100 + assert generator.active_power.magnitude == 100 assert generator.ext["original_capacity"].magnitude == 200 assert generator.ext["broken"] diff --git a/tests/test_models.py b/tests/test_models.py index 11e21d84..9193472f 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,12 +1,12 @@ from r2x.enums import PrimeMoversType -from r2x.model import Generator, ACBus, Emission, HydroPumpedStorage +from r2x.models import Generator, ACBus, Emission, HydroPumpedStorage, ThermalStandard from r2x.units import EmissionRate, ureg def test_generator_model(): - generator = Generator.example() - assert isinstance(generator, Generator) - assert isinstance(generator.base_power.magnitude, float) + generator = ThermalStandard.example() + assert isinstance(generator, ThermalStandard) + assert isinstance(generator.active_power.magnitude, float) def test_emission_objects(): @@ -21,12 +21,12 @@ def test_emission_objects(): def test_bus_model(): bus = ACBus.example() assert isinstance(bus, ACBus) - assert isinstance(bus.id, int) + assert isinstance(bus.number, int) def test_generator_objects(): bus = ACBus.example() - generator = Generator(name="GEN01", base_power=100 * ureg.MW, bus=bus) + generator = Generator(name="GEN01", active_power=100 * ureg.MW, bus=bus) assert isinstance(generator.bus, ACBus) diff --git a/tests/test_plexos_parser.py b/tests/test_plexos_parser.py index 735133c1..53c80246 100644 --- a/tests/test_plexos_parser.py +++ b/tests/test_plexos_parser.py @@ -39,7 +39,7 @@ def test_plexos_parser_instance(plexos_parser_instance): assert isinstance(plexos_parser_instance.db, PlexosSQLite) -@pytest.mark.skip +@pytest.mark.skip(reason="We need a better test model") def test_build_system(plexos_parser_instance): system = plexos_parser_instance.build_system() assert isinstance(system, System) diff --git a/tests/test_sienna_exporter.py b/tests/test_sienna_exporter.py index 10cd824e..d6d9c866 100644 --- a/tests/test_sienna_exporter.py +++ b/tests/test_sienna_exporter.py @@ -1,7 +1,7 @@ import pytest from r2x.config import Scenario from r2x.exporter.sienna import SiennaExporter -from .models.systems import ieee5bus_system +from .models.ieee5bus import ieee5bus_system @pytest.fixture @@ -38,7 +38,7 @@ def test_sienna_exporter_run(sienna_exporter, tmp_folder): "bus.csv", "timeseries_pointers.json", "storage.csv", - "reserves.csv", # Reserve could be optional + # "reserves.csv", # Reserve could be optional "dc_branch.csv", "branch.csv", ] From 936284077ee0d6071341da41716783d723e544f0 Mon Sep 17 00:00:00 2001 From: pesap Date: Tue, 3 Sep 2024 12:00:49 -0600 Subject: [PATCH 3/3] Incorporating Kodi comments --- src/r2x/exporter/plexos.py | 24 ------------------------ src/r2x/exporter/sienna.py | 2 +- 2 files changed, 1 insertion(+), 25 deletions(-) diff --git a/src/r2x/exporter/plexos.py b/src/r2x/exporter/plexos.py index 7dbdef12..e421d0ee 100644 --- a/src/r2x/exporter/plexos.py +++ b/src/r2x/exporter/plexos.py @@ -4,12 +4,10 @@ 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 @@ -838,25 +836,3 @@ def get_valid_component_properties( 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 diff --git a/src/r2x/exporter/sienna.py b/src/r2x/exporter/sienna.py index fdf798b9..64973e09 100644 --- a/src/r2x/exporter/sienna.py +++ b/src/r2x/exporter/sienna.py @@ -392,7 +392,7 @@ def process_storage_data(self, fname="storage.csv") -> None: fpath=self.output_folder / fname, fields=output_fields, key_mapping=key_mapping, - unnest_key="numer", + unnest_key="number", restval="NA", )