feat: final element & component design

[FBumann]

Final Element & Component Design

Complete API design for all elements and components in fluxopt. This defines the user-facing dataclass layer that feeds into ModelData → Model.

---

Core Elements (unchanged)

Bus

@dataclass
class Bus:
    id: str
    carrier: str | None = None
    imbalance_penalty: float | None = None

Effect

@dataclass
class Effect:
    id: str
    unit: str = ''
    is_objective: bool = False
    maximum_total: float | None = None
    minimum_total: float | None = None
    maximum_per_hour: TimeSeries | None = None
    minimum_per_hour: TimeSeries | None = None
    contribution_from: dict[str, float] = field(default_factory=dict)
    contribution_from_per_hour: dict[str, TimeSeries] = field(default_factory=dict)

Sizing

@dataclass
class Sizing:
    """Continuous capacity optimization within bounds."""
    min_size: float
    max_size: float
    mandatory: bool = True
    effects_per_size: dict[str, float] = field(default_factory=dict)
    effects_fixed: dict[str, float] = field(default_factory=dict)

Status

@dataclass
class Status:
    """Binary on/off operational behavior."""
    min_uptime: float | None = None
    max_uptime: float | None = None
    min_downtime: float | None = None
    max_downtime: float | None = None
    effects_per_running_hour: dict[str, TimeSeries] = field(default_factory=dict)
    effects_per_startup: dict[str, TimeSeries] = field(default_factory=dict)

---

Flow

One universal Flow class used everywhere. Parent components validate what makes sense in their context.

@dataclass(eq=False)
class Flow:
    bus: str
    id: str = ''
    size: float | Sizing | None = None
    relative_minimum: TimeSeries = 0.0
    relative_maximum: TimeSeries = 1.0
    fixed_relative_profile: TimeSeries | None = None
    effects_per_flow_hour: dict[str, TimeSeries] = field(default_factory=dict)
    status: Status | None = None
    prior_rates: list[float] | None = None

What each parent allows/forbids

Field	Port	Converter (factors)	Converter (curve)	Storage
bus	✓	✓	✓	✓
size	✓	✓	✗ (from curve)	✓
status	✓	✓	✗ (on curve)	✗ (on Storage)
fixed_relative_profile	✓	✓	✗	✗
effects_per_flow_hour	✓	✓	✓	✓

Forbidden combinations raise ValueError in the parent's __post_init__.

---

Conversion Concepts

ConversionCurve — variable efficiency

Carries its own operational parameters because the curve defines what on/off and built/not-built mean.

@dataclass
class ConversionCurve:
    breakpoints: dict[str, list[TimeSeries]]
    status: Status | None = None
    mandatory: bool = True
    effects_fixed: dict[str, float] = field(default_factory=dict)
    availability: TimeSeries = 1.0

Method selection (automatic, invisible to user):

• All flows strictly monotonic → incremental formulation (pure LP, 0 extra binaries)
• Any flow non-monotonic → SOS2 fallback

Availability constraint — applied to first input flow rate variable:

$$P_{fuel,t} \leq a_t \cdot b^K_{fuel,t} \cdot u_t$$

conversion as list[dict] — fixed ratio equations

Each dict is one linear equation. No wrapper class needed — the structure is self-explanatory.

# One equation — boiler
conversion=[
    {'gas': 0.9, 'heat': -1},
]

# Two equations — CHP (simultaneous)
conversion=[
    {'gas': 0.4,  'power': -1},
    {'gas': 0.45, 'heat':  -1},
]

# Co-firing — multiple inputs in one equation
conversion=[
    {'gas': 0.9, 'biomass': 0.8, 'heat': -1},
]

• Positive coefficients = inputs, negative = outputs
• Multiple inputs in one equation = co-firing/blending (LP, no binaries)
• Multiple equations = simultaneous constraints (CHP, trigeneration)

What it cannot express:

• Variable efficiency → use ConversionCurve
• Exclusive fuel switching (binary either/or) → future exclusive_inputs, not in scope

---

Components

Converter

@dataclass
class Converter:
    id: str
    inputs: list[Flow]
    outputs: list[Flow]
    conversion: list[dict[str, TimeSeries]] | ConversionCurve | None = None

    def __post_init__(self) -> None:
        # Qualify flow ids
        for f in [*self.inputs, *self.outputs]:
            f.id = f'{self.id}({f.id})'

        # Curve-specific flow validation
        if isinstance(self.conversion, ConversionCurve):
            for f in [*self.inputs, *self.outputs]:
                if f.size is not None:
                    raise ValueError(
                        f'Flow {f.id!r}: size not allowed with ConversionCurve. '
                        f'Operating range is defined by breakpoints.'
                    )
                if f.status is not None:
                    raise ValueError(
                        f'Flow {f.id!r}: status not allowed with ConversionCurve. '
                        f'Use ConversionCurve.status instead.'
                    )

Factory methods — primary API for common cases:

    @classmethod
    def boiler(cls, id, eta, fuel, heat): ...

    @classmethod
    def chp(cls, id, eta_el, eta_th, fuel, power, heat): ...

    @classmethod
    def heat_pump(cls, id, cop, electrical, source, thermal): ...

    @classmethod
    def power2heat(cls, id, eta, electrical, thermal): ...

Port

@dataclass
class Port:
    id: str
    imports: list[Flow] = field(default_factory=list)
    exports: list[Flow] = field(default_factory=list)

    def __post_init__(self) -> None:
        for f in [*self.imports, *self.exports]:
            f.id = f'{self.id}({f.id})'

Storage

Component-level status gates both directions. Individual flow concerns (size, bounds, effects, availability via relative_maximum) stay on the Flow objects.

No availability parameter on Storage. Unlike ConversionCurve (where flows are coupled through shared segment variables and there's no other hook), storage charging/discharging are independent variables with independent upper bounds. Use relative_maximum on each flow instead — it's more flexible and already does the job.

@dataclass
class Storage:
    id: str
    charging: Flow
    discharging: Flow
    capacity: float | Sizing | None = None
    eta_charge: TimeSeries = 1.0
    eta_discharge: TimeSeries = 1.0
    relative_loss_per_hour: TimeSeries = 0.0
    status: Status | None = None          # gates both charging and discharging
    prior_level: float | None = None
    cyclic: bool = True
    relative_minimum_level: TimeSeries = 0.0
    relative_maximum_level: TimeSeries = 1.0

    def __post_init__(self) -> None:
        # Qualify flow ids
        if self.charging.id == self.discharging.id:
            self.charging.id = 'charge'
            self.discharging.id = 'discharge'
        self.charging.id = f'{self.id}({self.charging.id})'
        self.discharging.id = f'{self.id}({self.discharging.id})'

        # Validate forbidden flow fields
        for flow, direction in [(self.charging, 'charging'), (self.discharging, 'discharging')]:
            if flow.status is not None:
                raise ValueError(
                    f'Storage {self.id!r}: status on {direction} flow is not supported. '
                    f'Use Storage.status instead.'
                )
            if flow.fixed_relative_profile is not None:
                raise ValueError(
                    f'Storage {self.id!r}: fixed_relative_profile on {direction} flow '
                    f'is not supported. Storage dispatch is always optimized.'
                )

---

Full Examples

Gas CHP with Variable Efficiency

chp = Converter(
    id='gas_chp',
    inputs=[
        Flow(bus='gas', effects_per_flow_hour={'co2': 0.2}),
    ],
    outputs=[
        Flow(bus='electricity'),
        Flow(bus='heat'),
    ],
    conversion=ConversionCurve(
        breakpoints={
            'gas':         [150, 300, 500],
            'electricity': [ 67, 135, 250],
            'heat':        [ 45,  90, 125],
        },
        status=Status(
            min_uptime=2,
            min_downtime=1,
            effects_per_startup={'cost': 500},
        ),
        mandatory=False,
        effects_fixed={'cost': 500_000},
        availability=maintenance_schedule,
    ),
)

Battery Storage

battery = Storage(
    id='battery',
    charging=Flow(bus='electricity', size=Sizing(min_size=0, max_size=100),
                  relative_maximum=availability_series),
    discharging=Flow(bus='electricity', size=Sizing(min_size=0, max_size=100),
                     relative_maximum=availability_series),
    capacity=Sizing(min_size=0, max_size=500, effects_per_size={'cost': 400}),
    eta_charge=0.95,
    eta_discharge=0.95,
    relative_loss_per_hour=0.001,
    status=Status(min_downtime=1),
)

Co-firing Boiler

boiler = Converter(
    id='boiler',
    inputs=[
        Flow(bus='gas',     size=300, relative_minimum=0.1,
             status=Status(min_uptime=2)),
        Flow(bus='biomass', size=200),
    ],
    outputs=[
        Flow(bus='heat'),
    ],
    conversion=[
        {'gas': 0.9, 'biomass': 0.85, 'heat': -1},
    ],
)

---

Design Principles

1. One Flow class — universal, used everywhere. Parents validate context.
2. Validation over types — forbidden combinations raise clear ValueError in __post_init__ rather than being prevented by a separate class.
3. Component-level concerns — Status lives on the component (ConversionCurve, Storage) when flows are coupled. availability lives on ConversionCurve (flows are coupled, no other hook) but not on Storage (use relative_maximum on flows instead). Status lives on Flow only when the flow is genuinely independent (Port, linear Converter).
4. conversion as plain list or ConversionCurve — one parameter, type discriminates. List of dicts for fixed ratios. ConversionCurve justifies a class because it carries real operational state beyond the data.
5. Factory methods as primary API — Converter.boiler, Converter.chp etc. hide conversion list construction for common cases. Raw parameters are the escape hatch for configurations no factory covers.

Out of Scope (future work)

• Exclusive fuel switching (exclusive_inputs) — binary either/or between fuels
• Gaps / forbidden operating zones — disjunctive, separate abstraction needed
• Continuous sizing with ConversionCurve — bilinear, fundamentally incompatible
• Multi-period investment (Investment) — separate concept from Sizing