feat: Add ValueCurves and cost functions from InfrastructureSystem.jl…

… to infrasys (#38)

- Added classes for InputOutputCurve, IncrementalCurve, and AverageRateCurve that call function_data classes to build generic cost curves. These classes are analogous to the structs found in PowerSystems
- Added functions to convert between the different value_curves classes (ex: InputOutputToIncremental creates an IncrementalCurve from an existing InputOutputCurve)
- Added functions to function_data to perform calculations for value_curve conversion
- Added additional serialization tests to test_function_data
---------

Co-authored-by: pesap <[email protected]>

docs/reference/api/function_data.md

@@ -0,0 +1,9 @@
+```{eval-rst}
+.. _function-data-api:
+```
+#  Function data
+
+```{eval-rst}
+.. automodule:: infrasys.function_data
+   :members:
+```

docs/reference/api/index.md

@@ -13,4 +13,5 @@
     time_series
     location
     quantities
+    function_data
 ```

pyproject.toml

@@ -83,12 +83,13 @@ target-version = "py311"
 [tool.ruff.lint]
 # Enable Pyflakes (`F`) and a subset of the pycodestyle (`E`)  codes by default.
 select = [
-    "C901", # McCabe complexity
-    "E4",   # Subset of pycodestyle (E)
+    "C901",   # McCabe complexity
+    "E4",     # Subset of pycodestyle (E)
     "E7",
     "E9",
-    "F",    # Pyflakes
-    "W",    # pycodestyle warnings
+    "EM",     # string formatting in an exception message
+    "F",      # Pyflakes
+    "W",      # pycodestyle warnings
 ]
 ignore = []
 

src/infrasys/base_quantity.py

@@ -24,7 +24,8 @@ def __new__(cls: Type["BaseQuantity"], value, units=None):
 
     def __init_subclass__(cls, **kwargs):
         if not cls.__base_unit__:
-            raise TypeError("__base_unit__ should be defined")
+            msg = "__base_unit__ should be defined"
+            raise TypeError(msg)
         super().__init_subclass__(**kwargs)
 
     # Required for pydantic validation

src/infrasys/component_manager.py

@@ -256,7 +256,8 @@ def deepcopy(self, component: Component) -> Component:
 
     def change_uuid(self, component: Component) -> None:
         """Change the component UUID."""
-        raise NotImplementedError("change_component_uuid")
+        msg = "change_component_uuid"
+        raise NotImplementedError(msg)
 
     def update(
         self,

src/infrasys/cost_curves.py

@@ -0,0 +1,59 @@
+from typing_extensions import Annotated
+from infrasys.component import Component
+from pydantic import Field
+from infrasys.value_curves import InputOutputCurve, IncrementalCurve, AverageRateCurve
+from infrasys.function_data import LinearFunctionData
+
+
+class ProductionVariableCostCurve(Component):
+    name: Annotated[str, Field(frozen=True)] = ""
+
+
+class CostCurve(ProductionVariableCostCurve):
+    """Direct representation of the variable operation cost of a power plant in currency.
+
+    Composed of a Value Curve that may represent input-output, incremental, or average rate
+    data. The default units for the x-axis are MW and can be specified with
+    `power_units`.
+    """
+
+    value_curve: Annotated[
+        InputOutputCurve | IncrementalCurve | AverageRateCurve,
+        Field(
+            description="The underlying `ValueCurve` representation of this `ProductionVariableCostCurve`"
+        ),
+    ]
+    vom_units: Annotated[
+        InputOutputCurve,
+        Field(description="(default: natural units (MW)) The units for the x-axis of the curve"),
+    ] = InputOutputCurve(
+        function_data=LinearFunctionData(proportional_term=0.0, constant_term=0.0)
+    )
+
+
+class FuelCurve(ProductionVariableCostCurve):
+    """Representation of the variable operation cost of a power plant in terms of fuel.
+
+    Fuel units (MBTU, liters, m^3, etc.) coupled with a conversion factor between fuel and currency.
+    Composed of a Value Curve that may represent input-output, incremental, or average rate data.
+    The default units for the x-axis are MW and can be specified with `power_units`.
+    """
+
+    value_curve: Annotated[
+        InputOutputCurve | IncrementalCurve | AverageRateCurve,
+        Field(
+            description="The underlying `ValueCurve` representation of this `ProductionVariableCostCurve`"
+        ),
+    ]
+    vom_units: Annotated[
+        InputOutputCurve,
+        Field(description="(default: natural units (MW)) The units for the x-axis of the curve"),
+    ] = InputOutputCurve(
+        function_data=LinearFunctionData(proportional_term=0.0, constant_term=0.0)
+    )
+    fuel_cost: Annotated[
+        float,
+        Field(
+            description="Either a fixed value for fuel cost or the key to a fuel cost time series"
+        ),
+    ]

src/infrasys/function_data.py

@@ -15,158 +15,189 @@ class XYCoords(NamedTuple):
     y: float
 
 
-class LinearFunctionData(Component):
+class FunctionData(Component):
+    """BaseClass of FunctionData"""
+
+    name: Annotated[str, Field(frozen=True)] = ""
+
+
+class LinearFunctionData(FunctionData):
     """Data representation for linear cost function.
 
-    Class to represent the underlying data of linear functions. Principally used for
-    the representation of cost functions `f(x) = proportional_term*x + constant_term`.
+    Used to represent linear cost functions of the form
+
+    .. math:: f(x) = mx + c,
+
+    where :math:`m` is the proportional term and :math:`c` is the constant term.
     """
 
-    name: Annotated[str, Field(frozen=True)] = ""
     proportional_term: Annotated[
-        float, Field(description="the proportional term in the represented function")
+        float, Field(description="the proportional term in the represented function.")
     ]
     constant_term: Annotated[
-        float, Field(description="the constant term in the represented function")
+        float, Field(description="the constant term in the represented function.")
     ]
 
 
-class QuadraticFunctionData(Component):
-    """Data representation for quadratic cost functions.
+class QuadraticFunctionData(FunctionData):
+    """Data representation for quadratic cost function.
+
+    Used to represent quadratic of cost functions of the form
+
+    .. math:: f(x) = ax^2 + bx + c,
 
-    Class to represent the underlying data of quadratic functions. Principally used for the
-    representation of cost functions
-    `f(x) = quadratic_term*x^2 + proportional_term*x + constant_term`.
+    where :math:`a` is the quadratic term, :math:`b` is the proportional term and :math:`c` is the
+    constant term.
     """
 
-    name: Annotated[str, Field(frozen=True)] = ""
     quadratic_term: Annotated[
-        float, Field(description="the quadratic term in the represented function")
+        float, Field(description="the quadratic term in the represented function.")
     ]
     proportional_term: Annotated[
-        float, Field(description="the proportional term in the represented function")
+        float, Field(description="the proportional term in the represented function.")
     ]
     constant_term: Annotated[
-        float, Field(description="the constant term in the represented function")
+        float, Field(description="the constant term in the represented function.")
     ]
 
 
 def validate_piecewise_linear_x(points: List[XYCoords]) -> List[XYCoords]:
     """Validates the x data for PiecewiseLinearData class
 
-    Function used to validate given x data for the PiecewiseLinearData class.
-    X data is checked to ensure there is at least two values of x,
-    which is the minimum required to generate a cost curve, and is
-    given in ascending order (e.g. [1, 2, 3], not [1, 3, 2]).
+    X data is checked to ensure there is at least two values of x, which is the minimum required to
+    generate a cost curve, and is given in ascending order (e.g. [1, 2, 3], not [1, 3, 2]).
 
     Parameters
     ----------
     points : List[XYCoords]
         List of named tuples of (x,y) coordinates for cost function
 
     Returns
-    ----------
+    -------
     points : List[XYCoords]
         List of (x,y) data for cost function after successful validation.
     """
 
     x_coords = [p.x for p in points]
 
     if len(x_coords) < 2:
-        raise ValueError("Must specify at least two x-coordinates")
+        msg = "Must specify at least two x-coordinates."
+        raise ValueError(msg)
     if not (
         x_coords == sorted(x_coords)
         or (np.isnan(x_coords[0]) and x_coords[1:] == sorted(x_coords[1:]))
     ):
-        raise ValueError(f"Piecewise x-coordinates must be ascending, got {x_coords}")
+        msg = f"Piecewise x-coordinates must be ascending, got {x_coords}."
+        raise ValueError(msg)
 
     return points
 
 
 def validate_piecewise_step_x(x_coords: List[float]) -> List[float]:
     """Validates the x data for PiecewiseStepData class
 
-    Function used to validate given x data for the PiecewiseStepData class.
-    X data is checked to ensure there is at least two values of x,
-    which is the minimum required to generate a cost curve, and is
-    given in ascending order (e.g. [1, 2, 3], not [1, 3, 2]).
+    X data is checked to ensure there is at least two values of x, which is the minimum required to
+    generate a cost curve, and is given in ascending order (e.g. [1, 2, 3], not [1, 3, 2]).
 
     Parameters
     ----------
     x_coords : List[float]
         List of x data for cost function.
 
     Returns
-    ----------
+    -------
     x_coords : List[float]
         List of x data for cost function after successful validation.
     """
 
     if len(x_coords) < 2:
-        raise ValueError("Must specify at least two x-coordinates")
+        msg = "Must specify at least two x-coordinates."
+        raise ValueError(msg)
     if not (
         x_coords == sorted(x_coords)
         or (np.isnan(x_coords[0]) and x_coords[1:] == sorted(x_coords[1:]))
     ):
-        raise ValueError(f"Piecewise x-coordinates must be ascending, got {x_coords}")
+        msg = f"Piecewise x-coordinates must be ascending, got {x_coords}."
+        raise ValueError(msg)
 
     return x_coords
 
 
-class PiecewiseLinearData(Component):
+class PiecewiseLinearData(FunctionData):
     """Data representation for piecewise linear cost function.
 
-    Class to represent piecewise linear data as a series of points: two points define one
-    segment, three points define two segments, etc. The curve starts at the first point given,
-    not the origin. Principally used for the representation of cost functions where the points
-    store quantities (x, y), such as (MW, USD/h).
+    Used to represent linear data as a series of points: two points define one segment, three
+    points define two segments, etc. The curve starts at the first point given, not the origin.
+    Principally used for the representation of cost functions where the points store quantities (x,
+    y), such as (MW, USD/h).
     """
 
-    name: Annotated[str, Field(frozen=True)] = ""
     points: Annotated[
         List[XYCoords],
         AfterValidator(validate_piecewise_linear_x),
-        Field(description="list of (x,y) points that define the function"),
+        Field(description="list of (x,y) points that define the function."),
     ]
 
 
-class PiecewiseStepData(Component):
+class PiecewiseStepData(FunctionData):
     """Data representation for piecewise step cost function.
 
-    Class to represent a step function as a series of endpoint x-coordinates and segment
+    Used to represent a step function as a series of endpoint x-coordinates and segment
     y-coordinates: two x-coordinates and one y-coordinate defines a single segment, three
-    x-coordinates and two y-coordinates define two segments, etc. This can be useful to
-    represent the derivative of a `PiecewiseLinearData`, where the y-coordinates of this
-    step function represent the slopes of that piecewise linear function.
-    Principally used for the representation of cost functions where the points store
-    quantities (x, dy/dx), such as (MW, USD/MWh).
+    x-coordinates and two y-coordinates define two segments, etc.
+
+    This can be useful to represent the derivative of a :class:`PiecewiseLinearData`, where the
+    y-coordinates of this step function represent the slopes of that piecewise linear function.
+    Principally used for the representation of cost functions where the points store quantities (x,
+    :math:`dy/dx`), such as (MW, USD/MWh).
     """
 
-    name: Annotated[str, Field(frozen=True)] = ""
     x_coords: Annotated[
         List[float],
-        Field(description="the x-coordinates of the endpoints of the segments"),
+        Field(description="the x-coordinates of the endpoints of the segments."),
     ]
     y_coords: Annotated[
         List[float],
         Field(
-            description="the y-coordinates of the segments: `y_coords[1]` is the y-value \
-                between `x_coords[0]` and `x_coords[1]`, etc. Must have one fewer elements than `x_coords`."
+            description=(
+                "The y-coordinates of the segments: `y_coords[1]` is the y-value between "
+                "`x_coords[0]` and `x_coords[1]`, etc. Must have one fewer elements than `x_coords`."
+            )
         ),
     ]
 
     @model_validator(mode="after")
     def validate_piecewise_xy(self):
         """Method to validate the x and y data for PiecewiseStepData class
 
-        Model validator used to validate given data for the PiecewiseStepData class.
-        Calls `validate_piecewise_step_x` to check if `x_coords` is valid, then checks if
-        the length of `y_coords` is exactly one less than `x_coords`, which is necessary
-        to define the cost functions correctly.
+        Model validator used to validate given data for the :class:`PiecewiseStepData`. Calls
+        `validate_piecewise_step_x` to check if `x_coords` is valid, then checks if the length of
+        `y_coords` is exactly one less than `x_coords`, which is necessary to define the cost
+        functions correctly.
         """
         validate_piecewise_step_x(self.x_coords)
 
         if len(self.y_coords) != len(self.x_coords) - 1:
-            raise ValueError("Must specify one fewer y-coordinates than x-coordinates")
+            msg = "Must specify one fewer y-coordinates than x-coordinates"
+            raise ValueError(msg)
 
         return self
+
+
+def get_x_lengths(x_coords: List[float]) -> List[float]:
+    return np.subtract(x_coords[1:], x_coords[:-1])
+
+
+def running_sum(data: PiecewiseStepData) -> List[XYCoords]:
+    points = []
+    slopes = data.y_coords
+    x_coords = data.x_coords
+    x_lengths = get_x_lengths(x_coords)
+    running_y = 0.0
+
+    points.append(XYCoords(x=x_coords[0], y=running_y))
+    for prev_slope, this_x, dx in zip(slopes, x_coords[1:], x_lengths):
+        running_y += prev_slope * dx
+        points.append(XYCoords(x=this_x, y=running_y))
+
+    return points

src/infrasys/parquet_time_series_storage.py

@@ -19,7 +19,8 @@ def get_time_series(
         start_time: datetime | None = None,
         length: int | None = None,
     ) -> TimeSeriesData:
-        raise NotImplementedError("ParquetTimeSeriesStorage.get_time_series")
+        msg = "ParquetTimeSeriesStorage.get_time_series"
+        raise NotImplementedError(msg)
 
     def remove_time_series(self, uuid: UUID) -> None:
         ...

src/infrasys/system.py

@@ -154,7 +154,8 @@ def to_json(self, filename: Path | str, overwrite=False, indent=None, data=None)
             data = system_data
         else:
             if "system" in data:
-                raise ISConflictingArguments("data contains the key 'system'")
+                msg = "data contains the key 'system'"
+                raise ISConflictingArguments(msg)
             data["system"] = system_data
         with open(filename, "w", encoding="utf-8") as f_out:
             json.dump(data, f_out, indent=indent)
@@ -337,9 +338,8 @@ def save(
             fpath = Path(fpath)
 
         if fpath.exists() and not overwrite:
-            raise FileExistsError(
-                f"{fpath} exists already. To overwrite the folder pass `overwrite=True`"
-            )
+            msg = f"{fpath} exists already. To overwrite the folder pass `overwrite=True`"
+            raise FileExistsError(msg)
 
         fpath.mkdir(parents=True, exist_ok=True)
         self.to_json(fpath / filename, overwrite=overwrite)
@@ -991,7 +991,8 @@ def handle_data_format_upgrade(
 
     def merge_system(self, other: "System") -> None:
         """Merge the contents of another system into this one."""
-        raise NotImplementedError("merge_system")
+        msg = "merge_system"
+        raise NotImplementedError(msg)
 
     # TODO: add delete methods that (1) don't raise if not found and (2) don't return anything?