fluxopt.elements

Classes:

Name	Description
Carrier	Physical energy medium (electricity, heat, gas, …).
Sizing	Capacity optimization parameters.
Investment	Singular discrete build-timing optimization.
Status	Binary on/off behavior parameters.
Flow	A single energy flow on a carrier.
Effect	A tracked quantity across the optimization horizon (cost, CO₂, …).
Storage	Energy storage with level dynamics.
PiecewiseConversion	Piecewise-linear conversion linking N flows.

Functions:

Name	Description
qualified_id	Format a qualified flow id: component(flow).
node_id	Format a carrier-node id: carrier:node.

Carrier dataclass

Carrier(
    id: str,
    nodes: list[str] = list(),
    unit: str = 'MWh',
    color: str | None = None,
    description: str = '',
)

Physical energy medium (electricity, heat, gas, …).

Parameters:

Name	Type	Description	Default
id	str	Unique identifier used as xarray coordinate.	required
nodes	list[str]	Sub-nodes for multi-node balancing. Empty means single-node.	list()
unit	str	Energy unit label.	'MWh'
color	str | None	Optional color for plotting.	None
description	str	Human-readable description.	''

Sizing dataclass

Sizing(
    min_size: float,
    max_size: float,
    mandatory: bool = True,
    effects_per_size: dict[str, Variate] = dict(),
    effects_fixed: dict[str, Variate] = dict(),
)

Capacity optimization parameters.

The solver decides the optimal size within [min_size, max_size].

• mandatory=True: continuous, size in [min, max], no binary.
• mandatory=False: binary indicator gates size: 0 or [min, max].
• min_size == max_size with mandatory=False: binary invest at exact size (yes/no).

See: docs/math/sizing.md

Parameters:

Name	Type	Description	Default
min_size	float	Minimum capacity if invested.	required
max_size	float	Maximum capacity.	required
mandatory	bool	If True, must be built (no binary indicator).	True
effects_per_size	dict[str, Variate]	Effect cost per unit size (e.g. €/MW).	dict()
effects_fixed	dict[str, Variate]	Fixed effect cost if built (optional only).	dict()

Investment dataclass

Investment(
    min_size: float,
    max_size: float,
    mandatory: bool = True,
    lifetime: int | None = None,
    prior_size: float = 0.0,
    effects_per_size_at_build: dict[str, Variate] = dict(),
    effects_fixed_at_build: dict[str, Variate] = dict(),
    effects_per_size_recurring: dict[str, Variate] = dict(),
    effects_fixed_recurring: dict[str, Variate] = dict(),
)

Singular discrete build-timing optimization.

The solver decides WHEN to build (which period) and at what size. Once built, capacity is available for lifetime periods. Size is decided once — no growth or partial retirement.

Parameters:

Name	Type	Description	Default
min_size	float	Minimum capacity if built.	required
max_size	float	Maximum capacity.	required
mandatory	bool	If True, must build exactly once; if False, may build at most once.	True
lifetime	int | None	Periods active after build; None = forever.	None
prior_size	float	Pre-existing capacity available from period 0.	0.0
effects_per_size_at_build	dict[str, Variate]	One-time per-MW costs charged in the build period.	dict()
effects_fixed_at_build	dict[str, Variate]	One-time fixed costs charged in the build period.	dict()
effects_per_size_recurring	dict[str, Variate]	Recurring per-MW costs charged every active period.	dict()
effects_fixed_recurring	dict[str, Variate]	Recurring fixed costs charged every active period.	dict()

Status dataclass

Status(
    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, Variate] = dict(),
    effects_per_startup: dict[str, Variate] = dict(),
)

Binary on/off behavior parameters.

Together with relative bounds, gives semi-continuous behavior: {0} U [min, max] * size.

See: docs/math/status.md

Parameters:

Name	Type	Description	Default
min_uptime	float | None	Minimum consecutive on-hours.	None
max_uptime	float | None	Maximum consecutive on-hours.	None
min_downtime	float | None	Minimum consecutive off-hours.	None
max_downtime	float | None	Maximum consecutive off-hours.	None
effects_per_running_hour	dict[str, Variate]	Effect cost per running hour.	dict()
effects_per_startup	dict[str, Variate]	Effect cost per startup event.	dict()

Flow dataclass

Flow(
    carrier: str,
    short_id: str = '',
    node: str | None = None,
    size: float | Sizing | Investment | None = None,
    relative_minimum: Variate = 0.0,
    relative_maximum: Variate = 1.0,
    fixed_relative_profile: Variate | None = None,
    effects_per_flow_hour: dict[str, Variate] = dict(),
    status: Status | None = None,
    prior_rates: list[float] | None = None,
)

A single energy flow on a carrier.

short_id defaults to carrier (or carrier:node when node is set). Set explicitly to disambiguate multiple flows on the same carrier::

Flow('elec')  # short_id='elec'
Flow('heat', node='A')  # short_id='heat:A'
Flow('elec', short_id='base')  # short_id='base'

short_id must be unique within a component. Storage renames colliding short_ids to charge / discharge before qualification. id is the qualified form set by the parent component: component(short_id).

See: docs/math/flows.md

Parameters:

Name	Type	Description	Default
carrier	str	Carrier this flow connects to.	required
short_id	str	Component-local identifier; defaults to carrier (or carrier:node). The qualified form component(short_id) is stored in id.	''
node	str | None	Sub-node for multi-node carrier balancing.	None
size	float | Sizing | Investment | None	Nominal capacity [MW], Sizing for investment optimization, or None (unsized / unbounded).	None
relative_minimum	Variate	Lower bound as fraction of size.	0.0
relative_maximum	Variate	Upper bound as fraction of size.	1.0
fixed_relative_profile	Variate | None	Fixed profile as fraction of size; sets both lower and upper bounds equal to the profile value.	None
effects_per_flow_hour	dict[str, Variate]	Effect coefficients per flow-hour (e.g. €/MWh).	dict()
status	Status | None	On/off behavior (semi-continuous, startup costs, durations).	None
prior_rates	list[float] | None	Flow rates [MW] before the horizon, used for status initial conditions.	None

Methods:

Name	Description
__post_init__	Default short_id from carrier/node, set id = short_id.

__post_init__

__post_init__() -> None

Default short_id from carrier/node, set id = short_id.

Source code in src/fluxopt/elements.py

def __post_init__(self) -> None:
    """Default short_id from carrier/node, set id = short_id."""
    if not self.short_id:
        self.short_id = node_id(self.carrier, self.node) if self.node else self.carrier
    self.id = self.short_id
    if self.status is not None and isinstance(self.relative_minimum, (int, float)) and self.relative_minimum <= 0:
        msg = (
            f'Flow {self.short_id!r}: relative_minimum must be > 0 when status is set, '
            f'otherwise on/off is indistinguishable (got {self.relative_minimum})'
        )
        raise ValueError(msg)

Effect dataclass

Effect(
    id: str,
    unit: str = '',
    maximum: float | None = None,
    minimum: float | None = None,
    maximum_per_period: float | None = None,
    minimum_per_period: float | None = None,
    maximum_per_hour: Variate | None = None,
    minimum_per_hour: Variate | None = None,
    contribution_from: dict[str, Variate] = dict(),
    period_weights: list[float] | None = None,
)

A tracked quantity across the optimization horizon (cost, CO₂, …).

One effect is designated as the objective to minimize via the objective_effects argument of optimize(). Others can be bounded to enforce budgets (e.g. emission caps).

Effects accumulate contributions from two domains:

• Temporal — per-timestep flow costs, running costs, startup costs.
• Lump — one-time sizing costs and fixed costs.

Cross-effect chains (e.g. CO₂ → cost) are supported via contribution_from.

See: docs/math/effects.md

Parameters:

Name	Type	Description	Default
id	str	Unique identifier.	required
unit	str	Unit label (e.g. '€', 'kg').	''
maximum	float | None	Upper bound on weighted total across all periods.	None
minimum	float | None	Lower bound on weighted total across all periods.	None
maximum_per_period	float | None	Upper bound applied to each period independently.	None
minimum_per_period	float | None	Lower bound applied to each period independently.	None
maximum_per_hour	Variate | None	Upper bound rate [unit/h], scaled by Δt.	None
minimum_per_hour	Variate | None	Lower bound rate [unit/h], scaled by Δt.	None
contribution_from	dict[str, Variate]	Cross-effect factors {source_effect: factor}. Scalar factors apply identically to both domains; time-varying factors are averaged for the lump domain.	dict()
period_weights	list[float] | None	Per-period weights ω for total aggregation; overrides global period_weights.	None

Storage dataclass

Storage(
    id: str,
    charging: Flow,
    discharging: Flow,
    capacity: float | Sizing | Investment | None = None,
    eta_charge: Variate = 1.0,
    eta_discharge: Variate = 1.0,
    relative_loss_per_hour: Variate = 0.0,
    prior_level: float | None = None,
    cyclic: bool = True,
    relative_minimum_level: Variate = 0.0,
    relative_maximum_level: Variate = 1.0,
    status: Status | None = None,
)

Energy storage with level dynamics.

Flow ids are qualified as storage(flow). When both flows connect to the same carrier, they are renamed to charge / discharge::

Storage('bat', Flow('elec'), Flow('elec'))  # bat(charge), bat(discharge)
Storage('bat', Flow('elec'), Flow('heat'))  # bat(elec), bat(heat)

Level balance::

E_{s,t+1} = E_{s,t} (1 - δ)^Δt + P^c η^c Δt - P^d / η^d Δt

See: docs/math/storage.md

Parameters:

Name	Type	Description	Default
id	str	Storage identifier.	required
charging	Flow	Charging flow.	required
discharging	Flow	Discharging flow.	required
capacity	float | Sizing | Investment | None	Maximum stored energy [MWh], Sizing for investment optimization, or None.	None
eta_charge	Variate	Charging efficiency.	1.0
eta_discharge	Variate	Discharging efficiency.	1.0
relative_loss_per_hour	Variate	Self-discharge rate [1/h].	0.0
prior_level	float | None	Initial energy level [MWh]; None = unconstrained.	None
cyclic	bool	If True, end level must equal start level.	True
relative_minimum_level	Variate	Min SOC as fraction of capacity.	0.0
relative_maximum_level	Variate	Max SOC as fraction of capacity.	1.0
status	Status | None	Component-level on/off behavior gating both charging and discharging. Forbids flow-level status on the child flows (the two switches would have no defined precedence).	None

Methods:

Name	Description
__post_init__	Validate carrier match, rename colliding flow ids, and qualify.

__post_init__

__post_init__() -> None

Validate carrier match, rename colliding flow ids, and qualify.

Source code in src/fluxopt/elements.py

def __post_init__(self) -> None:
    """Validate carrier match, rename colliding flow ids, and qualify."""
    if self.charging.carrier != self.discharging.carrier:
        msg = (
            f'Storage {self.id!r}: charging carrier {self.charging.carrier!r} '
            f'!= discharging carrier {self.discharging.carrier!r}'
        )
        raise ValueError(msg)
    if self.charging.short_id == self.discharging.short_id:
        self.charging.short_id = 'charge'
        self.discharging.short_id = 'discharge'
    self.charging.id = qualified_id(self.id, self.charging.short_id)
    self.discharging.id = qualified_id(self.id, self.discharging.short_id)
    if self.status is not None:
        for f in (self.charging, self.discharging):
            if f.status is not None:
                msg = (
                    f'Storage {self.id!r}: flow {f.short_id!r} cannot have flow-level '
                    f'status when Storage.status is set; the component status already gates both flows'
                )
                raise ValueError(msg)
            if f.size is None:
                msg = (
                    f'Storage {self.id!r}: flow {f.short_id!r} must have a size when '
                    f'Storage.status is set — without it, the on/off binary cannot gate '
                    f'the rate (no upper bound to scale)'
                )
                raise ValueError(msg)

PiecewiseConversion dataclass

PiecewiseConversion(
    points: dict[str, list[Variate]] | list[_CurveTuple],
    method: PiecewiseMethod = 'auto',
    status: Status | None = None,
    availability: Variate = 1.0,
)

Piecewise-linear conversion linking N flows.

Wraps :func:linopy.piecewise.add_piecewise_formulation. All flows share interpolation weights — every operating point lies on the same piece of the curve.

Two input forms:

• Dict — equality-only, terse for the common case::

  PiecewiseConversion({'fuel': [0, 50, 100], 'Heat': [0, 45, 70]})

• List of tuples — supports per-flow inequality bounds::

  PiecewiseConversion( [ ('fuel', [0, 50, 100]), ('Heat', [0, 45, 70], '>='), ] )

See: docs/math/converters.md

Parameters:

Name	Type	Description	Default
points	dict[str, list[Variate]] | list[_CurveTuple]	Per-flow breakpoints. Either {flow: [bp...]} (equality only) or a list of (flow, [bp...]) / (flow, [bp...], '<='|'>=') tuples. Need >=2 flows; all breakpoint lists must share the same length (>=2). At most one tuple may carry a non-equality bound, and only when exactly two flows are present.	required
method	PiecewiseMethod	Formulation. "auto" picks LP (2 flows + bounded + matching convexity), else incremental (monotonic) or sos2. Override with "sos2" / "incremental" / "lp".	'auto'
status	Status | None	Component-level on/off behavior gating the curve.	None
availability	Variate	Time-varying scaling of the upper breakpoint.	1.0

Methods:

Name	Description
__post_init__	Validate normalized breakpoints and bound combinations.

__post_init__

__post_init__() -> None

Validate normalized breakpoints and bound combinations.

Source code in src/fluxopt/elements.py

def __post_init__(self) -> None:
    """Validate normalized breakpoints and bound combinations."""
    flows_pts_bounds = list(self._iter_normalized())

    if len(flows_pts_bounds) < 2:
        msg = f'PiecewiseConversion needs >=2 flows, got {len(flows_pts_bounds)}'
        raise ValueError(msg)

    n = len(flows_pts_bounds[0][1])
    if n < 2:
        msg = f'PiecewiseConversion needs >=2 breakpoints per flow, got {n}'
        raise ValueError(msg)

    if any(len(pts) != n for _, pts, _ in flows_pts_bounds):
        lengths = {flow: len(pts) for flow, pts, _ in flows_pts_bounds}
        msg = f'PiecewiseConversion breakpoint lists must all have the same length, got {lengths}'
        raise ValueError(msg)

    flows = [flow for flow, _, _ in flows_pts_bounds]
    if len(set(flows)) != len(flows):
        dupes = [f for f in flows if flows.count(f) > 1]
        msg = f'PiecewiseConversion has duplicate flow ids: {sorted(set(dupes))}'
        raise ValueError(msg)

    nonequal = [b for _, _, b in flows_pts_bounds if b != '==']
    if len(nonequal) > 1:
        msg = f'At most one bounded flow per PiecewiseConversion, got {len(nonequal)}'
        raise ValueError(msg)
    if nonequal and len(flows_pts_bounds) > 2:
        msg = f'Inequality bounds require exactly 2 flows, got {len(flows_pts_bounds)}'
        raise ValueError(msg)
    if self.method == 'lp' and not nonequal:
        msg = "method='lp' requires one flow with bound '<=' or '>='"
        raise ValueError(msg)

qualified_id

qualified_id(component: str, flow: str) -> str

Format a qualified flow id: component(flow).

Source code in src/fluxopt/elements.py

def qualified_id(component: str, flow: str) -> str:
    """Format a qualified flow id: ``component(flow)``."""
    return QUAL_FMT.format(component=component, flow=flow)

node_id

node_id(carrier: str, node: str) -> str

Format a carrier-node id: carrier:node.

Source code in src/fluxopt/elements.py

def node_id(carrier: str, node: str) -> str:
    """Format a carrier-node id: ``carrier:node``."""
    return f'{carrier}{NODE_SEP}{node}'