feat: add build_period axis to investment once-effect matrices

pyproject.toml

@@ -116,6 +116,7 @@ extend-fixable = ["B", "SIM", "RUF", "C4", "UP"]
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = ["S101"]
 "scripts/**/*.py" = ["T201", "RUF003"]
+"src/fluxopt/types.py" = ["UP040"]  # TypeAlias for mkdocstrings compatibility
 "docs/notebooks/*.ipynb" = ["T201", "RUF001", "RUF003"]
 
 [tool.ruff.format]

src/fluxopt/__init__.py

@@ -7,9 +7,13 @@
 from fluxopt.model_data import Dims, ModelData
 from fluxopt.results import Result
 from fluxopt.types import (
+    ArrayInput,
     IdList,
+    OnceEffectInput,
+    PeriodicInput,
+    TemporalInput,
+    TemporalPeriodicInput,
     TimeIndex,
-    TimeSeries,
     Timesteps,
     as_dataarray,
 )
@@ -63,6 +67,7 @@ def optimize(
 
 __all__ = [
     'PENALTY_EFFECT_ID',
+    'ArrayInput',
     'Carrier',
     'Converter',
     'Dims',
@@ -72,13 +77,16 @@ def optimize(
     'IdList',
     'Investment',
     'ModelData',
+    'OnceEffectInput',
+    'PeriodicInput',
     'Port',
     'Result',
     'Sizing',
     'Status',
     'Storage',
+    'TemporalInput',
+    'TemporalPeriodicInput',
     'TimeIndex',
-    'TimeSeries',
     'Timesteps',
     'as_dataarray',
     'optimize',

src/fluxopt/components.py

@@ -8,7 +8,7 @@
 
 if TYPE_CHECKING:
     from fluxopt.elements import Flow
-    from fluxopt.types import TimeSeries
+    from fluxopt.types import TemporalInput
 
 
 def _qualify_flows(component_id: str, flows: list[Flow]) -> IdList[Flow]:
@@ -49,7 +49,7 @@ class Converter:
     id: str
     inputs: list[Flow] | IdList[Flow]
     outputs: list[Flow] | IdList[Flow]
-    conversion_factors: list[dict[str, TimeSeries]] = field(default_factory=list)  # a_f
+    conversion_factors: list[dict[str, TemporalInput]] = field(default_factory=list)  # a_f
     _short_to_id: dict[str, str] = field(init=False, default_factory=dict)
 
     def __post_init__(self) -> None:
@@ -59,7 +59,7 @@ def __post_init__(self) -> None:
         self._short_to_id = {f.short_id: f.id for f in (*self.inputs, *self.outputs)}
 
     @classmethod
-    def _single_io(cls, id: str, coefficient: TimeSeries, input_flow: Flow, output_flow: Flow) -> Converter:
+    def _single_io(cls, id: str, coefficient: TemporalInput, input_flow: Flow, output_flow: Flow) -> Converter:
         """Create a single-input/single-output converter: input * coefficient = output."""
         return cls(
             id,
@@ -69,7 +69,7 @@ def _single_io(cls, id: str, coefficient: TimeSeries, input_flow: Flow, output_f
         )
 
     @classmethod
-    def boiler(cls, id: str, thermal_efficiency: TimeSeries, fuel_flow: Flow, thermal_flow: Flow) -> Converter:
+    def boiler(cls, id: str, thermal_efficiency: TemporalInput, fuel_flow: Flow, thermal_flow: Flow) -> Converter:
         """Create a boiler converter: fuel * eta = thermal.
 
         Args:
@@ -84,7 +84,7 @@ def boiler(cls, id: str, thermal_efficiency: TimeSeries, fuel_flow: Flow, therma
     def heat_pump(
         cls,
         id: str,
-        cop: TimeSeries,
+        cop: TemporalInput,
         electrical_flow: Flow,
         source_flow: Flow,
         thermal_flow: Flow,
@@ -113,7 +113,7 @@ def heat_pump(
         )
 
     @classmethod
-    def power2heat(cls, id: str, efficiency: TimeSeries, electrical_flow: Flow, thermal_flow: Flow) -> Converter:
+    def power2heat(cls, id: str, efficiency: TemporalInput, electrical_flow: Flow, thermal_flow: Flow) -> Converter:
         """Create an electric resistance heater: electrical * eta = thermal.
 
         Args:
@@ -128,8 +128,8 @@ def power2heat(cls, id: str, efficiency: TimeSeries, electrical_flow: Flow, ther
     def chp(
         cls,
         id: str,
-        eta_el: TimeSeries,
-        eta_th: TimeSeries,
+        eta_el: TemporalInput,
+        eta_th: TemporalInput,
         fuel_flow: Flow,
         electrical_flow: Flow,
         thermal_flow: Flow,

src/fluxopt/elements.py

@@ -4,7 +4,12 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from fluxopt.types import TimeSeries
+    from fluxopt.types import (
+        OnceEffectInput,
+        PeriodicInput,
+        TemporalInput,
+        TemporalPeriodicInput,
+    )
 
 PENALTY_EFFECT_ID = 'penalty'
 
@@ -55,8 +60,8 @@ class Sizing:
     min_size: float
     max_size: float
     mandatory: bool = True
-    effects_per_size: dict[str, TimeSeries] = field(default_factory=dict)
-    effects_fixed: dict[str, TimeSeries] = field(default_factory=dict)
+    effects_per_size: dict[str, PeriodicInput] = field(default_factory=dict)
+    effects_fixed: dict[str, PeriodicInput] = field(default_factory=dict)
 
 
 @dataclass
@@ -73,8 +78,10 @@ class Investment:
         mandatory: If True, must build exactly once; if False, may build at most once.
         lifetime: Periods active after build; None = forever.
         prior_size: Pre-existing capacity available from period 0.
-        effects_per_size: One-time per-MW costs charged in the build period.
-        effects_fixed: One-time fixed costs charged in the build period.
+        effects_per_size: One-time per-MW costs. Scalar or 1D ``(build_period,)``
+            → diagonal (cost lands in the build period). 2D ``(period, build_period)``
+            → as-is (e.g. installment plans, learning curves).
+        effects_fixed: One-time fixed costs. Same expansion rules as effects_per_size.
         effects_per_size_periodic: Recurring per-MW costs charged every active period.
         effects_fixed_periodic: Recurring fixed costs charged every active period.
     """
@@ -84,10 +91,10 @@ class Investment:
     mandatory: bool = True
     lifetime: int | None = None
     prior_size: float = 0.0
-    effects_per_size: dict[str, TimeSeries] = field(default_factory=dict)
-    effects_fixed: dict[str, TimeSeries] = field(default_factory=dict)
-    effects_per_size_periodic: dict[str, TimeSeries] = field(default_factory=dict)
-    effects_fixed_periodic: dict[str, TimeSeries] = field(default_factory=dict)
+    effects_per_size: dict[str, OnceEffectInput] = field(default_factory=dict)
+    effects_fixed: dict[str, OnceEffectInput] = field(default_factory=dict)
+    effects_per_size_periodic: dict[str, PeriodicInput] = field(default_factory=dict)
+    effects_fixed_periodic: dict[str, PeriodicInput] = field(default_factory=dict)
 
 
 @dataclass
@@ -102,8 +109,8 @@ class Status:
     max_uptime: float | None = None  # [h]
     min_downtime: float | None = None  # [h]
     max_downtime: float | None = None  # [h]
-    effects_per_running_hour: dict[str, TimeSeries] = field(default_factory=dict)
-    effects_per_startup: dict[str, TimeSeries] = field(default_factory=dict)
+    effects_per_running_hour: dict[str, TemporalPeriodicInput] = field(default_factory=dict)
+    effects_per_startup: dict[str, PeriodicInput] = field(default_factory=dict)
 
 
 @dataclass(eq=False)
@@ -129,10 +136,10 @@ class Flow:
     id: str = field(init=False, default='')
     node: str | None = None
     size: float | Sizing | Investment | None = None  # P̄_f  [MW]
-    relative_minimum: TimeSeries = 0.0  # p̲_f  [-]
-    relative_maximum: TimeSeries = 1.0  # p̄_f  [-]
-    fixed_relative_profile: TimeSeries | None = None  # π_f  [-]
-    effects_per_flow_hour: dict[str, TimeSeries] = field(default_factory=dict)  # c_{f,k}  [varies]
+    relative_minimum: TemporalInput = 0.0  # p̲_f  [-]
+    relative_maximum: TemporalInput = 1.0  # p̄_f  [-]
+    fixed_relative_profile: TemporalInput | None = None  # π_f  [-]
+    effects_per_flow_hour: dict[str, TemporalPeriodicInput] = field(default_factory=dict)  # c_{f,k}  [varies]
     status: Status | None = None
     prior_rates: list[float] | None = None  # flow rates before horizon [MW]
 
@@ -156,10 +163,10 @@ class Effect:
     is_objective: bool = False
     maximum_total: float | None = None  # Φ̄_k  [unit]
     minimum_total: float | None = None  # Φ̲_k  [unit]
-    maximum_per_hour: TimeSeries | None = None  # Φ̄_{k,t}  [unit]
-    minimum_per_hour: TimeSeries | None = None  # Φ̲_{k,t}  [unit]
-    contribution_from: dict[str, TimeSeries] = field(default_factory=dict)
-    contribution_from_per_hour: dict[str, TimeSeries] = field(default_factory=dict)
+    maximum_per_hour: TemporalInput | None = None  # Φ̄_{k,t}  [unit]
+    minimum_per_hour: TemporalInput | None = None  # Φ̲_{k,t}  [unit]
+    contribution_from: dict[str, PeriodicInput] = field(default_factory=dict)
+    contribution_from_per_hour: dict[str, TemporalPeriodicInput] = field(default_factory=dict)
     period_weights_periodic: list[float] | None = None  # ω_periodic[p] — scales temporal+periodic
     period_weights_once: list[float] | None = None  # ω_once[p] — scales once
 
@@ -183,13 +190,13 @@ class Storage:
     charging: Flow
     discharging: Flow
     capacity: float | Sizing | Investment | None = None  # Ē_s  [MWh]
-    eta_charge: TimeSeries = 1.0  # η^c_s  [-]
-    eta_discharge: TimeSeries = 1.0  # η^d_s  [-]
-    relative_loss_per_hour: TimeSeries = 0.0  # δ_s  [1/h]
+    eta_charge: TemporalInput = 1.0  # η^c_s  [-]
+    eta_discharge: TemporalInput = 1.0  # η^d_s  [-]
+    relative_loss_per_hour: TemporalInput = 0.0  # δ_s  [1/h]
     prior_level: float | None = None  # E_{s,0}  [MWh]
     cyclic: bool = True  # E_{s,first} == E_{s,last}
-    relative_minimum_level: TimeSeries = 0.0  # e̲_s  [-]
-    relative_maximum_level: TimeSeries = 1.0  # ē_s  [-]
+    relative_minimum_level: TemporalInput = 0.0  # e̲_s  [-]
+    relative_maximum_level: TemporalInput = 1.0  # ē_s  [-]
 
     def __post_init__(self) -> None:
         """Validate carrier match, rename colliding flow ids, and qualify."""

src/fluxopt/model.py

@@ -877,17 +877,20 @@ def _create_effects(self) -> None:
 
         once_direct: Any = 0
 
-        # Investment: one-time per-size costs (charged in build period)
+        # Investment: one-time per-size costs — (flow, effect, period, build_period)
+        # Multiply by invest_size_at_build renamed to build_period, sum over both.
         if self.invest_size_at_build is not None and d.flows.invest_effects_per_size is not None:
             eps_once = d.flows.invest_effects_per_size.rename({'invest_flow': 'flow'})
             if (eps_once != 0).any():
-                once_direct = once_direct + (eps_once * self.invest_size_at_build).sum('flow')
+                sab_bp = self.invest_size_at_build.rename({'period': 'build_period'})
+                once_direct = once_direct + (eps_once * sab_bp).sum(['flow', 'build_period'])
 
-        # Investment: one-time fixed costs (charged in build period)
+        # Investment: one-time fixed costs — (flow, effect, period, build_period)
         if self.invest_build is not None and d.flows.invest_effects_fixed is not None:
             ef_once = d.flows.invest_effects_fixed.rename({'invest_flow': 'flow'})
             if (ef_once != 0).any():
-                once_direct = once_direct + (ef_once * self.invest_build).sum('flow')
+                bld_bp = self.invest_build.rename({'period': 'build_period'})
+                once_direct = once_direct + (ef_once * bld_bp).sum(['flow', 'build_period'])
 
         self.m.add_constraints(self.effect_once == once_direct, name='effect_once_eq')