Skip to content

fluxopt

Modules:

Name Description
benchmark

User-runnable benchmark: build a few realistic energy systems, report speed and memory.

components
contract

The cross-layer data contract.

contributions

Per-contributor effect breakdown, read off the program.

elements
flow_system

FlowSystem — the declarative top of the element layer.

leontief

Cross-effect propagation: the Leontief inverse of contribution_from.

math

fluxopt's math, and the engine that builds it.

model_data
results
schema

JSON Schema and dict/JSON round-trip for the element layer.

stats

Derived statistics from optimization results.

types
validation

System-level structural validation, shared by every entry path.

Type Aliases:

Name Description
Variate

Any input that varies over a subset of the model's variate dims (time,

Classes:

Name Description
Converter

Conversion between input and output flows.

Port

System boundary importing and exporting carriers.

Carrier

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

Effect

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

Flow

A single energy flow on a carrier.

Investment

Singular discrete build-timing optimization.

PiecewiseConversion

Piecewise-linear conversion linking N flows.

Sizing

Capacity optimization parameters.

Status

Binary on/off behavior parameters.

Storage

Energy storage with level dynamics.

FlowSystem

A declarative flow-system description (see module docstring).

Dims

The model's temporal axes: what a timestep is, and what a period is.

ModelData
Result

Optimization result with solution variables and model data.

ProfileRef

Reference to a time-series stored outside the model definition.

Functions:

Name Description
all_element_schemas

Return JSON Schemas for every element type, keyed by class name.

element_schema

Return the JSON Schema for one element type.

from_dict

Reconstruct an element from a dict produced by :func:to_dict.

to_dict

Serialize an element to a JSON-safe dict.

as_dataarray

Convert a Variate to a DataArray aligned to given coordinates.

optimize

Build data, build model, optimize, return results.

Variate

Any input that varies over a subset of the model's variate dims (time, optionally period, eventually scenario).

  • Scalar: broadcast to all variate dims.
  • 1-D (list/ndarray): matched to a coord by length (must be unambiguous).
  • 1-D (pd.Series): index name selects the dim if set; else matched by length.
  • 2-D (pd.DataFrame): index.name and columns.name must match target dims.
  • n-D (xr.DataArray): dims must be a subset of the target; coords must match exactly.

Per-field reach (which dims a particular field can vary over) is documented on the field itself; as_dataarray enforces that user input only uses dims the caller declared in coords.

Converter pydantic-model

Bases: Element

Conversion between input and output flows.

Two mutually exclusive modes:

  • Linear — conversion_factors=[{flow_short_id: a_f}, ...], one dict per equation; constraint sum_f(a_f * P_{f,t}) = 0.
  • Piecewise — conversion=PiecewiseConversion(...); the solver interpolates between breakpoints, optionally with on/off via PiecewiseConversion.status.
Show JSON schema:
{
  "$defs": {
    "Flow": {
      "description": "A single energy flow on a carrier.\n\n``short_id`` defaults to ``carrier`` (or ``carrier:node`` when ``node``\nis set).  Set explicitly to disambiguate multiple flows on the same\ncarrier::\n\n    Flow(carrier='elec')  # short_id='elec'\n    Flow(carrier='heat', node='A')  # short_id='heat:A'\n    Flow(carrier='elec', short_id='base')  # short_id='base'\n\n``short_id`` must be unique within a component.  The system-global\nqualified id ``component(short_id)`` is derived at build time and\nappears as the ``flow`` coordinate in ``ModelData`` and results; the\nflow itself is never modified. Storage resolves colliding short_ids\nto ``charge`` / ``discharge`` at that point.\n\nSee: docs/math/flows.md",
      "properties": {
        "carrier": {
          "title": "Carrier",
          "type": "string"
        },
        "short_id": {
          "default": "",
          "title": "Short Id",
          "type": "string"
        },
        "node": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Node"
        },
        "size": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "$ref": "#/$defs/Sizing"
            },
            {
              "$ref": "#/$defs/Investment"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "relative_rate_min": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "relative_rate_max": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "fixed_relative_profile": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "effects_per_flow_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Flow Hour",
          "type": "object"
        },
        "flow_hours_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Min"
        },
        "flow_hours_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Max"
        },
        "load_factor_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Min"
        },
        "load_factor_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Max"
        },
        "ramp_up_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "ramp_down_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "prior_rates": {
          "anyOf": [
            {
              "items": {
                "type": "number"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Prior Rates"
        }
      },
      "required": [
        "carrier"
      ],
      "title": "Flow",
      "type": "object"
    },
    "Investment": {
      "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "lifetime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lifetime"
        },
        "prior_size": {
          "default": 0.0,
          "minimum": 0,
          "title": "Prior Size",
          "type": "number"
        },
        "effects_per_size_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size At Build",
          "type": "object"
        },
        "effects_fixed_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed At Build",
          "type": "object"
        },
        "effects_per_size_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size Recurring",
          "type": "object"
        },
        "effects_fixed_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed Recurring",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Investment",
      "type": "object"
    },
    "PiecewiseConversion": {
      "description": "Piecewise-linear conversion linking N flows.\n\nWraps :func:`linopy.piecewise.add_piecewise_formulation`. All flows\nshare interpolation weights \u2014 every operating point lies on the same\npiece of the curve.\n\nTwo input forms:\n\n- **Dict** \u2014 equality-only, terse for the common case::\n\n      PiecewiseConversion(points={'fuel': [0, 50, 100], 'Heat': [0, 45, 70]})\n\n- **List of tuples** \u2014 supports per-flow inequality bounds::\n\n      PiecewiseConversion(\n          points=[\n              ('fuel', [0, 50, 100]),\n              ('Heat', [0, 45, 70], '>='),\n          ]\n      )\n\nSee: docs/math/converters.md",
      "properties": {
        "points": {
          "anyOf": [
            {
              "additionalProperties": {
                "items": {
                  "$ref": "#/$defs/Variate"
                },
                "type": "array"
              },
              "type": "object"
            },
            {
              "items": {
                "anyOf": [
                  {
                    "maxItems": 2,
                    "minItems": 2,
                    "prefixItems": [
                      {
                        "type": "string"
                      },
                      {
                        "items": {
                          "$ref": "#/$defs/Variate"
                        },
                        "type": "array"
                      }
                    ],
                    "type": "array"
                  },
                  {
                    "maxItems": 3,
                    "minItems": 3,
                    "prefixItems": [
                      {
                        "type": "string"
                      },
                      {
                        "items": {
                          "$ref": "#/$defs/Variate"
                        },
                        "type": "array"
                      },
                      {
                        "enum": [
                          "==",
                          "<=",
                          ">="
                        ],
                        "type": "string"
                      }
                    ],
                    "type": "array"
                  }
                ]
              },
              "type": "array"
            }
          ],
          "title": "Points"
        },
        "method": {
          "$ref": "#/$defs/PiecewiseMethod",
          "default": "auto"
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "availability": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        }
      },
      "required": [
        "points"
      ],
      "title": "PiecewiseConversion",
      "type": "object"
    },
    "PiecewiseMethod": {
      "enum": [
        "auto",
        "sos2",
        "incremental",
        "lp"
      ],
      "type": "string"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Sizing": {
      "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "effects_per_size": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size",
          "type": "object"
        },
        "effects_fixed": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Sizing",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Conversion between input and output flows.\n\nTwo mutually exclusive modes:\n\n- **Linear** \u2014 ``conversion_factors=[{flow_short_id: a_f}, ...]``,\n  one dict per equation; constraint ``sum_f(a_f * P_{f,t}) = 0``.\n- **Piecewise** \u2014 ``conversion=PiecewiseConversion(...)``; the solver\n  interpolates between breakpoints, optionally with on/off via\n  ``PiecewiseConversion.status``.",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "inputs": {
      "items": {
        "$ref": "#/$defs/Flow"
      },
      "title": "Inputs",
      "type": "array"
    },
    "outputs": {
      "items": {
        "$ref": "#/$defs/Flow"
      },
      "title": "Outputs",
      "type": "array"
    },
    "conversion_factors": {
      "items": {
        "additionalProperties": {
          "$ref": "#/$defs/Variate"
        },
        "type": "object"
      },
      "title": "Conversion Factors",
      "type": "array"
    },
    "conversion": {
      "anyOf": [
        {
          "$ref": "#/$defs/PiecewiseConversion"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "id",
    "inputs",
    "outputs"
  ],
  "title": "Converter",
  "type": "object"
}

Fields:

id pydantic-field

id: str

Converter id.

inputs pydantic-field

inputs: list[Flow]

Input flows.

outputs pydantic-field

outputs: list[Flow]

Output flows.

conversion_factors pydantic-field

conversion_factors: list[dict[str, Variate]]

Linear-mode equations. Empty when conversion is set.

conversion pydantic-field

conversion: PiecewiseConversion | None = None

Piecewise-mode curve. None for linear mode.

model_post_init

model_post_init(__context: Any) -> None

Validate short_id uniqueness, mode exclusivity, and factor references.

Source code in src/fluxopt/components.py
@override
def model_post_init(self, __context: Any) -> None:
    """Validate short_id uniqueness, mode exclusivity, and factor references."""
    flows = [*self.inputs, *self.outputs]
    _check_unique_short_ids(f'Converter {self.id!r}', flows)
    known = {f.short_id for f in flows}

    if self.conversion is not None:
        if self.conversion_factors:
            msg = (
                f'Converter {self.id!r}: cannot set both conversion_factors and conversion '
                f'(they are mutually exclusive linear vs piecewise modes)'
            )
            raise ValueError(msg)
        curve_flows = {flow for flow, _, _ in self.conversion._iter_normalized()}
        unknown = curve_flows - known
        if unknown:
            msg = (
                f'Converter {self.id!r}: PiecewiseConversion references unknown flow short_ids '
                f'{sorted(unknown)}; known: {sorted(known)}'
            )
            raise ValueError(msg)
        if self.conversion.status is not None:
            for f in flows:
                if f.status is not None:
                    msg = (
                        f'Converter {self.id!r}: flow {f.short_id!r} cannot have flow-level '
                        f'status when PiecewiseConversion.status is set'
                    )
                    raise ValueError(msg)
    else:
        for eq_i, equation in enumerate(self.conversion_factors):
            unknown = set(equation) - known
            if unknown:
                msg = (
                    f'Converter {self.id!r}: conversion_factors[{eq_i}] references unknown '
                    f'flow short_ids {sorted(unknown)}; known: {sorted(known)}'
                )
                raise ValueError(msg)

boiler classmethod

Create a boiler converter: fuel * eta = thermal.

Parameters:

Name Type Description Default

id

str

Converter id.

required

thermal_efficiency

Variate

Thermal efficiency eta.

required

fuel_flow

Flow

Input fuel flow.

required

thermal_flow

Flow

Output thermal flow.

required
Source code in src/fluxopt/components.py
@classmethod
def boiler(cls, id: str, thermal_efficiency: Variate, fuel_flow: Flow, thermal_flow: Flow) -> Converter:
    """Create a boiler converter: fuel * eta = thermal.

    Args:
        id: Converter id.
        thermal_efficiency: Thermal efficiency eta.
        fuel_flow: Input fuel flow.
        thermal_flow: Output thermal flow.
    """
    return cls._single_io(id, thermal_efficiency, fuel_flow, thermal_flow)

heat_pump classmethod

Create a heat pump converter with source heat.

Two conversion equations

electrical * COP = thermal electrical + source = thermal

Parameters:

Name Type Description Default

id

str

Converter id.

required

cop

Variate

Coefficient of performance.

required

electrical_flow

Flow

Input electrical flow.

required

source_flow

Flow

Input environmental heat flow (air, ground, water).

required

thermal_flow

Flow

Output thermal flow.

required
Source code in src/fluxopt/components.py
@classmethod
def heat_pump(
    cls,
    id: str,
    cop: Variate,
    electrical_flow: Flow,
    source_flow: Flow,
    thermal_flow: Flow,
) -> Converter:
    """Create a heat pump converter with source heat.

    Two conversion equations:
        electrical * COP = thermal
        electrical + source = thermal

    Args:
        id: Converter id.
        cop: Coefficient of performance.
        electrical_flow: Input electrical flow.
        source_flow: Input environmental heat flow (air, ground, water).
        thermal_flow: Output thermal flow.
    """
    return cls(
        id=id,
        inputs=[electrical_flow, source_flow],
        outputs=[thermal_flow],
        conversion_factors=[
            {electrical_flow.short_id: cop, thermal_flow.short_id: -1},
            {electrical_flow.short_id: 1, source_flow.short_id: 1, thermal_flow.short_id: -1},
        ],
    )

power2heat classmethod

Create an electric resistance heater: electrical * eta = thermal.

Parameters:

Name Type Description Default

id

str

Converter id.

required

efficiency

Variate

Electrical-to-thermal efficiency.

required

electrical_flow

Flow

Input electrical flow.

required

thermal_flow

Flow

Output thermal flow.

required
Source code in src/fluxopt/components.py
@classmethod
def power2heat(cls, id: str, efficiency: Variate, electrical_flow: Flow, thermal_flow: Flow) -> Converter:
    """Create an electric resistance heater: electrical * eta = thermal.

    Args:
        id: Converter id.
        efficiency: Electrical-to-thermal efficiency.
        electrical_flow: Input electrical flow.
        thermal_flow: Output thermal flow.
    """
    return cls._single_io(id, efficiency, electrical_flow, thermal_flow)

chp classmethod

Create a CHP converter with separate electrical and thermal outputs.

Parameters:

Name Type Description Default

id

str

Converter id.

required

eta_el

Variate

Electrical efficiency.

required

eta_th

Variate

Thermal efficiency.

required

fuel_flow

Flow

Input fuel flow.

required

electrical_flow

Flow

Output electrical flow.

required

thermal_flow

Flow

Output thermal flow.

required
Source code in src/fluxopt/components.py
@classmethod
def chp(
    cls,
    id: str,
    eta_el: Variate,
    eta_th: Variate,
    fuel_flow: Flow,
    electrical_flow: Flow,
    thermal_flow: Flow,
) -> Converter:
    """Create a CHP converter with separate electrical and thermal outputs.

    Args:
        id: Converter id.
        eta_el: Electrical efficiency.
        eta_th: Thermal efficiency.
        fuel_flow: Input fuel flow.
        electrical_flow: Output electrical flow.
        thermal_flow: Output thermal flow.
    """
    return cls(
        id=id,
        inputs=[fuel_flow],
        outputs=[electrical_flow, thermal_flow],
        conversion_factors=[
            {fuel_flow.short_id: eta_el, electrical_flow.short_id: -1},
            {fuel_flow.short_id: eta_th, thermal_flow.short_id: -1},
        ],
    )

Port pydantic-model

Bases: Element

System boundary importing and exporting carriers.

imports bring flow into the system (sources); exports send it out of the system (sinks).

Show JSON schema:
{
  "$defs": {
    "Flow": {
      "description": "A single energy flow on a carrier.\n\n``short_id`` defaults to ``carrier`` (or ``carrier:node`` when ``node``\nis set).  Set explicitly to disambiguate multiple flows on the same\ncarrier::\n\n    Flow(carrier='elec')  # short_id='elec'\n    Flow(carrier='heat', node='A')  # short_id='heat:A'\n    Flow(carrier='elec', short_id='base')  # short_id='base'\n\n``short_id`` must be unique within a component.  The system-global\nqualified id ``component(short_id)`` is derived at build time and\nappears as the ``flow`` coordinate in ``ModelData`` and results; the\nflow itself is never modified. Storage resolves colliding short_ids\nto ``charge`` / ``discharge`` at that point.\n\nSee: docs/math/flows.md",
      "properties": {
        "carrier": {
          "title": "Carrier",
          "type": "string"
        },
        "short_id": {
          "default": "",
          "title": "Short Id",
          "type": "string"
        },
        "node": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Node"
        },
        "size": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "$ref": "#/$defs/Sizing"
            },
            {
              "$ref": "#/$defs/Investment"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "relative_rate_min": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "relative_rate_max": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "fixed_relative_profile": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "effects_per_flow_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Flow Hour",
          "type": "object"
        },
        "flow_hours_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Min"
        },
        "flow_hours_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Max"
        },
        "load_factor_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Min"
        },
        "load_factor_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Max"
        },
        "ramp_up_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "ramp_down_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "prior_rates": {
          "anyOf": [
            {
              "items": {
                "type": "number"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Prior Rates"
        }
      },
      "required": [
        "carrier"
      ],
      "title": "Flow",
      "type": "object"
    },
    "Investment": {
      "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "lifetime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lifetime"
        },
        "prior_size": {
          "default": 0.0,
          "minimum": 0,
          "title": "Prior Size",
          "type": "number"
        },
        "effects_per_size_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size At Build",
          "type": "object"
        },
        "effects_fixed_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed At Build",
          "type": "object"
        },
        "effects_per_size_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size Recurring",
          "type": "object"
        },
        "effects_fixed_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed Recurring",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Investment",
      "type": "object"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Sizing": {
      "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "effects_per_size": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size",
          "type": "object"
        },
        "effects_fixed": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Sizing",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "System boundary importing and exporting carriers.\n\n``imports`` bring flow into the system (sources); ``exports`` send it out\nof the system (sinks).",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "imports": {
      "items": {
        "$ref": "#/$defs/Flow"
      },
      "title": "Imports",
      "type": "array"
    },
    "exports": {
      "items": {
        "$ref": "#/$defs/Flow"
      },
      "title": "Exports",
      "type": "array"
    }
  },
  "required": [
    "id"
  ],
  "title": "Port",
  "type": "object"
}

Config:

  • default: _PYDANTIC_CFG

Fields:

id pydantic-field

id: str

Port id (prefixes qualified flow ids).

imports pydantic-field

imports: list[Flow]

Flows bringing the carrier into the system (sources, e.g. grid supply).

exports pydantic-field

exports: list[Flow]

Flows sending the carrier out of the system (sinks, e.g. demand).

model_post_init

model_post_init(__context: Any) -> None

Reject duplicate short_ids across imports and exports.

Source code in src/fluxopt/components.py
@override
def model_post_init(self, __context: Any) -> None:
    """Reject duplicate short_ids across imports and exports."""
    _check_unique_short_ids(f'Port {self.id!r}', [*self.imports, *self.exports])

Carrier pydantic-model

Bases: Element

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

Show JSON schema:
{
  "description": "Physical energy medium (electricity, heat, gas, \u2026).",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "nodes": {
      "items": {
        "type": "string"
      },
      "title": "Nodes",
      "type": "array"
    },
    "unit": {
      "default": "MWh",
      "title": "Unit",
      "type": "string"
    },
    "color": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Color"
    },
    "description": {
      "default": "",
      "title": "Description",
      "type": "string"
    }
  },
  "required": [
    "id"
  ],
  "title": "Carrier",
  "type": "object"
}

Fields:

id pydantic-field

id: str

Unique identifier used as xarray coordinate.

nodes pydantic-field

nodes: list[str]

Sub-nodes for multi-node balancing. Empty means single-node.

unit pydantic-field

unit: str = 'MWh'

Energy unit label.

color pydantic-field

color: str | None = None

Optional color for plotting.

description pydantic-field

description: str = ''

Human-readable description.

Effect pydantic-model

Bases: Element

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

One effect is designated as the objective to minimize via the objective 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

Show JSON schema:
{
  "$defs": {
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "A tracked quantity across the optimization horizon (cost, CO\u2082, \u2026).\n\nOne effect is designated as the objective to minimize via the\n``objective`` argument of ``optimize()``. Others can be bounded\nto enforce budgets (e.g. emission caps).\n\nEffects accumulate contributions from two domains:\n\n- **Temporal** \u2014 per-timestep flow costs, running costs, startup costs.\n- **Lump** \u2014 one-time sizing costs and fixed costs.\n\nCross-effect chains (e.g. CO\u2082 \u2192 cost) are supported via\n``contribution_from``.\n\nSee: docs/math/effects.md",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "unit": {
      "default": "",
      "title": "Unit",
      "type": "string"
    },
    "total_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Total Max"
    },
    "total_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Total Min"
    },
    "periodic_max": {
      "anyOf": [
        {
          "$ref": "#/$defs/Variate"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "periodic_min": {
      "anyOf": [
        {
          "$ref": "#/$defs/Variate"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "contribution_from": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Contribution From",
      "type": "object"
    },
    "period_weights": {
      "anyOf": [
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Period Weights"
    }
  },
  "required": [
    "id"
  ],
  "title": "Effect",
  "type": "object"
}

Fields:

id pydantic-field

id: str

Unique identifier.

unit pydantic-field

unit: str = ''

Unit label (e.g. '€', 'kg').

total_max pydantic-field

total_max: float | None = None

Upper bound on weighted total across all periods.

total_min pydantic-field

total_min: float | None = None

Lower bound on weighted total across all periods.

periodic_max pydantic-field

periodic_max: Variate | None = None

Upper bound applied to each period independently. Scalar or per-period values (multi-period only).

periodic_min pydantic-field

periodic_min: Variate | None = None

Lower bound applied to each period independently. Scalar or per-period values (multi-period only).

contribution_from pydantic-field

contribution_from: dict[str, Variate]

Cross-effect factors {source_effect: factor}. Scalar factors apply identically to both domains. Time-varying factors apply per-timestep in the temporal domain and are rejected at build time when the source effect carries lump (sizing/fixed) contributions.

period_weights pydantic-field

period_weights: list[float] | None = None

Per-period weights ω for total aggregation; overrides global period_weights.

Flow pydantic-model

Bases: Element

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(carrier='elec')  # short_id='elec'
Flow(carrier='heat', node='A')  # short_id='heat:A'
Flow(carrier='elec', short_id='base')  # short_id='base'

short_id must be unique within a component. The system-global qualified id component(short_id) is derived at build time and appears as the flow coordinate in ModelData and results; the flow itself is never modified. Storage resolves colliding short_ids to charge / discharge at that point.

See: docs/math/flows.md

Show JSON schema:
{
  "$defs": {
    "Investment": {
      "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "lifetime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lifetime"
        },
        "prior_size": {
          "default": 0.0,
          "minimum": 0,
          "title": "Prior Size",
          "type": "number"
        },
        "effects_per_size_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size At Build",
          "type": "object"
        },
        "effects_fixed_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed At Build",
          "type": "object"
        },
        "effects_per_size_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size Recurring",
          "type": "object"
        },
        "effects_fixed_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed Recurring",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Investment",
      "type": "object"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Sizing": {
      "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "effects_per_size": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size",
          "type": "object"
        },
        "effects_fixed": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Sizing",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "A single energy flow on a carrier.\n\n``short_id`` defaults to ``carrier`` (or ``carrier:node`` when ``node``\nis set).  Set explicitly to disambiguate multiple flows on the same\ncarrier::\n\n    Flow(carrier='elec')  # short_id='elec'\n    Flow(carrier='heat', node='A')  # short_id='heat:A'\n    Flow(carrier='elec', short_id='base')  # short_id='base'\n\n``short_id`` must be unique within a component.  The system-global\nqualified id ``component(short_id)`` is derived at build time and\nappears as the ``flow`` coordinate in ``ModelData`` and results; the\nflow itself is never modified. Storage resolves colliding short_ids\nto ``charge`` / ``discharge`` at that point.\n\nSee: docs/math/flows.md",
  "properties": {
    "carrier": {
      "title": "Carrier",
      "type": "string"
    },
    "short_id": {
      "default": "",
      "title": "Short Id",
      "type": "string"
    },
    "node": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Node"
    },
    "size": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "$ref": "#/$defs/Sizing"
        },
        {
          "$ref": "#/$defs/Investment"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "relative_rate_min": {
      "$ref": "#/$defs/Variate",
      "default": 0.0
    },
    "relative_rate_max": {
      "$ref": "#/$defs/Variate",
      "default": 1.0
    },
    "fixed_relative_profile": {
      "anyOf": [
        {
          "$ref": "#/$defs/Variate"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "effects_per_flow_hour": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Flow Hour",
      "type": "object"
    },
    "flow_hours_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Flow Hours Min"
    },
    "flow_hours_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Flow Hours Max"
    },
    "load_factor_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Load Factor Min"
    },
    "load_factor_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Load Factor Max"
    },
    "ramp_up_per_hour": {
      "anyOf": [
        {
          "$ref": "#/$defs/Variate"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "ramp_down_per_hour": {
      "anyOf": [
        {
          "$ref": "#/$defs/Variate"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "status": {
      "anyOf": [
        {
          "$ref": "#/$defs/Status"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "prior_rates": {
      "anyOf": [
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Prior Rates"
    }
  },
  "required": [
    "carrier"
  ],
  "title": "Flow",
  "type": "object"
}

Fields:

carrier pydantic-field

carrier: str

Carrier this flow connects to.

short_id pydantic-field

short_id: str = ''

Component-local identifier; defaults to carrier (or carrier:node). The qualified form component(short_id) is derived at build time.

node pydantic-field

node: str | None = None

Sub-node for multi-node carrier balancing.

size pydantic-field

size: float | Sizing | Investment | None = None

Nominal capacity [MW], Sizing for investment optimization, or None (unsized / unbounded).

relative_rate_min pydantic-field

relative_rate_min: Variate = 0.0

Lower bound as fraction of size.

relative_rate_max pydantic-field

relative_rate_max: Variate = 1.0

Upper bound as fraction of size.

fixed_relative_profile pydantic-field

fixed_relative_profile: Variate | None = None

Fixed profile as fraction of size; sets both lower and upper bounds equal to the profile value.

effects_per_flow_hour pydantic-field

effects_per_flow_hour: dict[str, Variate]

Effect coefficients per flow-hour (e.g. €/MWh).

flow_hours_min pydantic-field

flow_hours_min: float | None = None

Lower bound on total flow-hours Σ P·Δt [MWh], applied to each period independently.

flow_hours_max pydantic-field

flow_hours_max: float | None = None

Upper bound on total flow-hours Σ P·Δt [MWh], applied to each period independently.

load_factor_min pydantic-field

load_factor_min: float | None = None

Lower bound on utilization Σ P·Δt / (size·T), applied to each period independently. Requires size.

load_factor_max pydantic-field

load_factor_max: float | None = None

Upper bound on utilization Σ P·Δt / (size·T), applied to each period independently. Requires size.

ramp_up_per_hour pydantic-field

ramp_up_per_hour: Variate | None = None

Maximum rate increase between consecutive timesteps, as fraction of size per hour. Requires size.

ramp_down_per_hour pydantic-field

ramp_down_per_hour: Variate | None = None

Maximum rate decrease between consecutive timesteps, as fraction of size per hour. Requires size.

status pydantic-field

status: Status | None = None

On/off behavior (semi-continuous, startup costs, durations).

prior_rates pydantic-field

prior_rates: list[float] | None = None

Flow rates [MW] before the horizon, used for status initial conditions.

model_post_init

model_post_init(__context: Any) -> None

Default short_id from carrier/node and validate field combinations.

Source code in src/fluxopt/elements.py
@override
def model_post_init(self, __context: Any) -> None:
    """Default short_id from carrier/node and validate field combinations."""
    if not self.short_id:
        self.short_id = node_id(self.carrier, self.node) if self.node else self.carrier
    if self.status is not None and isinstance(self.relative_rate_min, (int, float)) and self.relative_rate_min <= 0:
        msg = (
            f'Flow {self.short_id!r}: relative_rate_min must be > 0 when status is set, '
            f'otherwise on/off is indistinguishable (got {self.relative_rate_min})'
        )
        raise ValueError(msg)
    if (
        self.status is not None
        and self.fixed_relative_profile is not None
        and isinstance(self.size, (Sizing, Investment))
    ):
        msg = (
            f'Flow {self.short_id!r}: fixed_relative_profile is not supported together with '
            f'Status and Sizing/Investment — fix the size or drop the profile'
        )
        raise ValueError(msg)
    if (self.load_factor_min is not None or self.load_factor_max is not None) and self.size is None:
        msg = (
            f'Flow {self.short_id!r}: load_factor bounds require a size '
            f'(fixed or Sizing/Investment) — the load factor is relative to size'
        )
        raise ValueError(msg)
    if (self.ramp_up_per_hour is not None or self.ramp_down_per_hour is not None) and self.size is None:
        msg = (
            f'Flow {self.short_id!r}: ramp limits require a size '
            f'(fixed or Sizing/Investment) — ramps are relative to size'
        )
        raise ValueError(msg)

Investment pydantic-model

Bases: Element

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.

Show JSON schema:
{
  "$defs": {
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
  "properties": {
    "size_min": {
      "minimum": 0,
      "title": "Size Min",
      "type": "number"
    },
    "size_max": {
      "minimum": 0,
      "title": "Size Max",
      "type": "number"
    },
    "mandatory": {
      "default": true,
      "title": "Mandatory",
      "type": "boolean"
    },
    "lifetime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lifetime"
    },
    "prior_size": {
      "default": 0.0,
      "minimum": 0,
      "title": "Prior Size",
      "type": "number"
    },
    "effects_per_size_at_build": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Size At Build",
      "type": "object"
    },
    "effects_fixed_at_build": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Fixed At Build",
      "type": "object"
    },
    "effects_per_size_recurring": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Size Recurring",
      "type": "object"
    },
    "effects_fixed_recurring": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Fixed Recurring",
      "type": "object"
    }
  },
  "required": [
    "size_min",
    "size_max"
  ],
  "title": "Investment",
  "type": "object"
}

Fields:

Validators:

  • _bounds_are_ordered

size_min pydantic-field

size_min: float

Minimum capacity if built.

size_max pydantic-field

size_max: float

Maximum capacity.

mandatory pydantic-field

mandatory: bool = True

If True, must build exactly once; if False, may build at most once.

lifetime pydantic-field

lifetime: int | None = None

Periods active after build; None = forever.

prior_size pydantic-field

prior_size: float = 0.0

Pre-existing capacity available from period 0.

effects_per_size_at_build pydantic-field

effects_per_size_at_build: dict[str, Variate]

One-time per-MW costs charged in the build period.

effects_fixed_at_build pydantic-field

effects_fixed_at_build: dict[str, Variate]

One-time fixed costs charged in the build period.

effects_per_size_recurring pydantic-field

effects_per_size_recurring: dict[str, Variate]

Recurring per-MW costs charged every active period.

effects_fixed_recurring pydantic-field

effects_fixed_recurring: dict[str, Variate]

Recurring fixed costs charged every active period.

PiecewiseConversion pydantic-model

Bases: Element

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(points={'fuel': [0, 50, 100], 'Heat': [0, 45, 70]})

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

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

See: docs/math/converters.md

Show JSON schema:
{
  "$defs": {
    "PiecewiseMethod": {
      "enum": [
        "auto",
        "sos2",
        "incremental",
        "lp"
      ],
      "type": "string"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Piecewise-linear conversion linking N flows.\n\nWraps :func:`linopy.piecewise.add_piecewise_formulation`. All flows\nshare interpolation weights \u2014 every operating point lies on the same\npiece of the curve.\n\nTwo input forms:\n\n- **Dict** \u2014 equality-only, terse for the common case::\n\n      PiecewiseConversion(points={'fuel': [0, 50, 100], 'Heat': [0, 45, 70]})\n\n- **List of tuples** \u2014 supports per-flow inequality bounds::\n\n      PiecewiseConversion(\n          points=[\n              ('fuel', [0, 50, 100]),\n              ('Heat', [0, 45, 70], '>='),\n          ]\n      )\n\nSee: docs/math/converters.md",
  "properties": {
    "points": {
      "anyOf": [
        {
          "additionalProperties": {
            "items": {
              "$ref": "#/$defs/Variate"
            },
            "type": "array"
          },
          "type": "object"
        },
        {
          "items": {
            "anyOf": [
              {
                "maxItems": 2,
                "minItems": 2,
                "prefixItems": [
                  {
                    "type": "string"
                  },
                  {
                    "items": {
                      "$ref": "#/$defs/Variate"
                    },
                    "type": "array"
                  }
                ],
                "type": "array"
              },
              {
                "maxItems": 3,
                "minItems": 3,
                "prefixItems": [
                  {
                    "type": "string"
                  },
                  {
                    "items": {
                      "$ref": "#/$defs/Variate"
                    },
                    "type": "array"
                  },
                  {
                    "enum": [
                      "==",
                      "<=",
                      ">="
                    ],
                    "type": "string"
                  }
                ],
                "type": "array"
              }
            ]
          },
          "type": "array"
        }
      ],
      "title": "Points"
    },
    "method": {
      "$ref": "#/$defs/PiecewiseMethod",
      "default": "auto"
    },
    "status": {
      "anyOf": [
        {
          "$ref": "#/$defs/Status"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "availability": {
      "$ref": "#/$defs/Variate",
      "default": 1.0
    }
  },
  "required": [
    "points"
  ],
  "title": "PiecewiseConversion",
  "type": "object"
}

Fields:

points pydantic-field

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.

method pydantic-field

method: PiecewiseMethod = 'auto'

Formulation. "auto" picks LP (2 flows + bounded + matching convexity), else incremental (monotonic) or sos2. Override with "sos2" / "incremental" / "lp".

status pydantic-field

status: Status | None = None

Component-level on/off behavior gating the curve.

availability pydantic-field

availability: Variate = 1.0

Time-varying scaling of the upper breakpoint.

model_post_init

model_post_init(__context: Any) -> None

Validate normalized breakpoints and bound combinations.

Source code in src/fluxopt/elements.py
@override
def model_post_init(self, __context: Any) -> 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)

Sizing pydantic-model

Bases: Element

Capacity optimization parameters.

The solver decides the optimal size within [size_min, size_max].

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

See: docs/math/sizing.md

Show JSON schema:
{
  "$defs": {
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
  "properties": {
    "size_min": {
      "minimum": 0,
      "title": "Size Min",
      "type": "number"
    },
    "size_max": {
      "minimum": 0,
      "title": "Size Max",
      "type": "number"
    },
    "mandatory": {
      "default": true,
      "title": "Mandatory",
      "type": "boolean"
    },
    "effects_per_size": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Size",
      "type": "object"
    },
    "effects_fixed": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Fixed",
      "type": "object"
    }
  },
  "required": [
    "size_min",
    "size_max"
  ],
  "title": "Sizing",
  "type": "object"
}

Fields:

Validators:

  • _bounds_are_ordered

size_min pydantic-field

size_min: float

Minimum capacity if invested.

size_max pydantic-field

size_max: float

Maximum capacity.

mandatory pydantic-field

mandatory: bool = True

If True, must be built (no binary indicator).

effects_per_size pydantic-field

effects_per_size: dict[str, Variate]

Effect cost per unit size (e.g. €/MW).

effects_fixed pydantic-field

effects_fixed: dict[str, Variate]

Fixed effect cost if built (optional only).

Status pydantic-model

Bases: Element

Binary on/off behavior parameters.

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

See: docs/math/status.md

Show JSON schema:
{
  "$defs": {
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
  "properties": {
    "uptime_min": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uptime Min"
    },
    "uptime_max": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uptime Max"
    },
    "downtime_min": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Downtime Min"
    },
    "downtime_max": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Downtime Max"
    },
    "effects_per_running_hour": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Running Hour",
      "type": "object"
    },
    "effects_per_startup": {
      "additionalProperties": {
        "$ref": "#/$defs/Variate"
      },
      "title": "Effects Per Startup",
      "type": "object"
    }
  },
  "title": "Status",
  "type": "object"
}

Fields:

Validators:

  • _durations_are_ordered

uptime_min pydantic-field

uptime_min: float | None = None

Minimum consecutive on-hours.

uptime_max pydantic-field

uptime_max: float | None = None

Maximum consecutive on-hours.

downtime_min pydantic-field

downtime_min: float | None = None

Minimum consecutive off-hours.

downtime_max pydantic-field

downtime_max: float | None = None

Maximum consecutive off-hours.

effects_per_running_hour pydantic-field

effects_per_running_hour: dict[str, Variate]

Effect cost per running hour.

effects_per_startup pydantic-field

effects_per_startup: dict[str, Variate]

Effect cost per startup event.

Storage pydantic-model

Bases: Element

Energy storage with level dynamics.

Flow ids are qualified as storage(flow) at build time. When both flows share a short_id, the qualified ids use charge / discharge instead (the flows themselves are never modified)::

Storage(
    id='bat', charging=Flow(carrier='elec'), discharging=Flow(carrier='elec')
)  # bat(charge), bat(discharge)
Storage(
    id='bat',
    charging=Flow(carrier='elec', short_id='in'),
    discharging=Flow(carrier='elec', short_id='out'),
)  # bat(in), bat(out)

Level balance::

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

See: docs/math/storage.md

Show JSON schema:
{
  "$defs": {
    "Flow": {
      "description": "A single energy flow on a carrier.\n\n``short_id`` defaults to ``carrier`` (or ``carrier:node`` when ``node``\nis set).  Set explicitly to disambiguate multiple flows on the same\ncarrier::\n\n    Flow(carrier='elec')  # short_id='elec'\n    Flow(carrier='heat', node='A')  # short_id='heat:A'\n    Flow(carrier='elec', short_id='base')  # short_id='base'\n\n``short_id`` must be unique within a component.  The system-global\nqualified id ``component(short_id)`` is derived at build time and\nappears as the ``flow`` coordinate in ``ModelData`` and results; the\nflow itself is never modified. Storage resolves colliding short_ids\nto ``charge`` / ``discharge`` at that point.\n\nSee: docs/math/flows.md",
      "properties": {
        "carrier": {
          "title": "Carrier",
          "type": "string"
        },
        "short_id": {
          "default": "",
          "title": "Short Id",
          "type": "string"
        },
        "node": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Node"
        },
        "size": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "$ref": "#/$defs/Sizing"
            },
            {
              "$ref": "#/$defs/Investment"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "relative_rate_min": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "relative_rate_max": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "fixed_relative_profile": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "effects_per_flow_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Flow Hour",
          "type": "object"
        },
        "flow_hours_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Min"
        },
        "flow_hours_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Max"
        },
        "load_factor_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Min"
        },
        "load_factor_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Max"
        },
        "ramp_up_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "ramp_down_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "prior_rates": {
          "anyOf": [
            {
              "items": {
                "type": "number"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Prior Rates"
        }
      },
      "required": [
        "carrier"
      ],
      "title": "Flow",
      "type": "object"
    },
    "Investment": {
      "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "lifetime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lifetime"
        },
        "prior_size": {
          "default": 0.0,
          "minimum": 0,
          "title": "Prior Size",
          "type": "number"
        },
        "effects_per_size_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size At Build",
          "type": "object"
        },
        "effects_fixed_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed At Build",
          "type": "object"
        },
        "effects_per_size_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size Recurring",
          "type": "object"
        },
        "effects_fixed_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed Recurring",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Investment",
      "type": "object"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Sizing": {
      "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "effects_per_size": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size",
          "type": "object"
        },
        "effects_fixed": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Sizing",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "Energy storage with level dynamics.\n\nFlow ids are qualified as ``storage(flow)`` at build time. When both\nflows share a short_id, the qualified ids use ``charge`` /\n``discharge`` instead (the flows themselves are never modified)::\n\n    Storage(\n        id='bat', charging=Flow(carrier='elec'), discharging=Flow(carrier='elec')\n    )  # bat(charge), bat(discharge)\n    Storage(\n        id='bat',\n        charging=Flow(carrier='elec', short_id='in'),\n        discharging=Flow(carrier='elec', short_id='out'),\n    )  # bat(in), bat(out)\n\nLevel balance::\n\n    E_{s,t+1} = E_{s,t} (1 - \u03b4)^\u0394t + P^c \u03b7^c \u0394t - P^d / \u03b7^d \u0394t\n\nSee: docs/math/storage.md",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "charging": {
      "$ref": "#/$defs/Flow"
    },
    "discharging": {
      "$ref": "#/$defs/Flow"
    },
    "capacity": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "$ref": "#/$defs/Sizing"
        },
        {
          "$ref": "#/$defs/Investment"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Capacity"
    },
    "eta_charge": {
      "$ref": "#/$defs/Variate",
      "default": 1.0
    },
    "eta_discharge": {
      "$ref": "#/$defs/Variate",
      "default": 1.0
    },
    "relative_loss_per_hour": {
      "$ref": "#/$defs/Variate",
      "default": 0.0
    },
    "prior_level": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Prior Level"
    },
    "cyclic": {
      "default": true,
      "title": "Cyclic",
      "type": "boolean"
    },
    "relative_level_min": {
      "$ref": "#/$defs/Variate",
      "default": 0.0
    },
    "relative_level_max": {
      "$ref": "#/$defs/Variate",
      "default": 1.0
    },
    "final_level_min": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Final Level Min"
    },
    "final_level_max": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Final Level Max"
    },
    "prevent_simultaneous": {
      "default": false,
      "title": "Prevent Simultaneous",
      "type": "boolean"
    },
    "status": {
      "anyOf": [
        {
          "$ref": "#/$defs/Status"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "required": [
    "id",
    "charging",
    "discharging"
  ],
  "title": "Storage",
  "type": "object"
}

Fields:

id pydantic-field

id: str

Storage identifier.

charging pydantic-field

charging: Flow

Charging flow.

discharging pydantic-field

discharging: Flow

Discharging flow.

capacity pydantic-field

capacity: float | Sizing | Investment | None = None

Maximum stored energy [MWh], Sizing for investment optimization, or None.

eta_charge pydantic-field

eta_charge: Variate = 1.0

Charging efficiency.

eta_discharge pydantic-field

eta_discharge: Variate = 1.0

Discharging efficiency.

relative_loss_per_hour pydantic-field

relative_loss_per_hour: Variate = 0.0

Self-discharge rate [1/h].

prior_level pydantic-field

prior_level: float | None = None

Initial energy level [MWh]; None = unconstrained.

cyclic pydantic-field

cyclic: bool = True

If True, end level must equal start level.

relative_level_min pydantic-field

relative_level_min: Variate = 0.0

Min SOC as fraction of capacity.

relative_level_max pydantic-field

relative_level_max: Variate = 1.0

Max SOC as fraction of capacity.

final_level_min pydantic-field

final_level_min: float | None = None

Lower bound on the level at the last timestep [MWh], per period; None = unconstrained.

final_level_max pydantic-field

final_level_max: float | None = None

Upper bound on the level at the last timestep [MWh], per period; None = unconstrained.

prevent_simultaneous pydantic-field

prevent_simultaneous: bool = False

If True, a binary per timestep excludes charging and discharging at once. Requires sized flows (fixed or Sizing/Investment) for the big-M bound.

status pydantic-field

status: Status | None = 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).

model_post_init

model_post_init(__context: Any) -> None

Validate physical ranges, carrier match, and status/size requirements.

Source code in src/fluxopt/elements.py
@override
def model_post_init(self, __context: Any) -> None:
    """Validate physical ranges, carrier match, and status/size requirements."""
    if isinstance(self.capacity, (int, float)) and self.capacity < 0:
        msg = f'Storage {self.id!r}: capacity is negative ({self.capacity})'
        raise ValueError(msg)
    for name, low, high, low_open in (
        ('eta_charge', 0.0, 1.0, True),
        ('eta_discharge', 0.0, 1.0, True),
        ('relative_loss_per_hour', 0.0, 1.0, False),
    ):
        bad = variate_out_of_range(getattr(self, name), low=low, high=high, low_open=low_open)
        if bad is not None:
            span = f'({low}, {high}]' if low_open else f'[{low}, {high}]'
            msg = f'Storage {self.id!r}: {name} must be in {span}, got {bad}'
            raise ValueError(msg)
    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.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)
    if self.prevent_simultaneous:
        for f in (self.charging, self.discharging):
            if f.size is None:
                msg = (
                    f'Storage {self.id!r}: flow {f.short_id!r} must have a size when '
                    f'prevent_simultaneous is set — the exclusion binary needs an '
                    f'upper bound on the rate (big-M)'
                )
                raise ValueError(msg)

FlowSystem pydantic-model

Bases: BaseModel

A declarative flow-system description (see module docstring).

Show JSON schema:
{
  "$defs": {
    "Carrier": {
      "description": "Physical energy medium (electricity, heat, gas, \u2026).",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "nodes": {
          "items": {
            "type": "string"
          },
          "title": "Nodes",
          "type": "array"
        },
        "unit": {
          "default": "MWh",
          "title": "Unit",
          "type": "string"
        },
        "color": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Color"
        },
        "description": {
          "default": "",
          "title": "Description",
          "type": "string"
        }
      },
      "required": [
        "id"
      ],
      "title": "Carrier",
      "type": "object"
    },
    "Converter": {
      "description": "Conversion between input and output flows.\n\nTwo mutually exclusive modes:\n\n- **Linear** \u2014 ``conversion_factors=[{flow_short_id: a_f}, ...]``,\n  one dict per equation; constraint ``sum_f(a_f * P_{f,t}) = 0``.\n- **Piecewise** \u2014 ``conversion=PiecewiseConversion(...)``; the solver\n  interpolates between breakpoints, optionally with on/off via\n  ``PiecewiseConversion.status``.",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "inputs": {
          "items": {
            "$ref": "#/$defs/Flow"
          },
          "title": "Inputs",
          "type": "array"
        },
        "outputs": {
          "items": {
            "$ref": "#/$defs/Flow"
          },
          "title": "Outputs",
          "type": "array"
        },
        "conversion_factors": {
          "items": {
            "additionalProperties": {
              "$ref": "#/$defs/Variate"
            },
            "type": "object"
          },
          "title": "Conversion Factors",
          "type": "array"
        },
        "conversion": {
          "anyOf": [
            {
              "$ref": "#/$defs/PiecewiseConversion"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        }
      },
      "required": [
        "id",
        "inputs",
        "outputs"
      ],
      "title": "Converter",
      "type": "object"
    },
    "Effect": {
      "description": "A tracked quantity across the optimization horizon (cost, CO\u2082, \u2026).\n\nOne effect is designated as the objective to minimize via the\n``objective`` argument of ``optimize()``. Others can be bounded\nto enforce budgets (e.g. emission caps).\n\nEffects accumulate contributions from two domains:\n\n- **Temporal** \u2014 per-timestep flow costs, running costs, startup costs.\n- **Lump** \u2014 one-time sizing costs and fixed costs.\n\nCross-effect chains (e.g. CO\u2082 \u2192 cost) are supported via\n``contribution_from``.\n\nSee: docs/math/effects.md",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "unit": {
          "default": "",
          "title": "Unit",
          "type": "string"
        },
        "total_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Total Max"
        },
        "total_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Total Min"
        },
        "periodic_max": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "periodic_min": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "contribution_from": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Contribution From",
          "type": "object"
        },
        "period_weights": {
          "anyOf": [
            {
              "items": {
                "type": "number"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Period Weights"
        }
      },
      "required": [
        "id"
      ],
      "title": "Effect",
      "type": "object"
    },
    "Flow": {
      "description": "A single energy flow on a carrier.\n\n``short_id`` defaults to ``carrier`` (or ``carrier:node`` when ``node``\nis set).  Set explicitly to disambiguate multiple flows on the same\ncarrier::\n\n    Flow(carrier='elec')  # short_id='elec'\n    Flow(carrier='heat', node='A')  # short_id='heat:A'\n    Flow(carrier='elec', short_id='base')  # short_id='base'\n\n``short_id`` must be unique within a component.  The system-global\nqualified id ``component(short_id)`` is derived at build time and\nappears as the ``flow`` coordinate in ``ModelData`` and results; the\nflow itself is never modified. Storage resolves colliding short_ids\nto ``charge`` / ``discharge`` at that point.\n\nSee: docs/math/flows.md",
      "properties": {
        "carrier": {
          "title": "Carrier",
          "type": "string"
        },
        "short_id": {
          "default": "",
          "title": "Short Id",
          "type": "string"
        },
        "node": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Node"
        },
        "size": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "$ref": "#/$defs/Sizing"
            },
            {
              "$ref": "#/$defs/Investment"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "relative_rate_min": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "relative_rate_max": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "fixed_relative_profile": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "effects_per_flow_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Flow Hour",
          "type": "object"
        },
        "flow_hours_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Min"
        },
        "flow_hours_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Flow Hours Max"
        },
        "load_factor_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Min"
        },
        "load_factor_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Load Factor Max"
        },
        "ramp_up_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "ramp_down_per_hour": {
          "anyOf": [
            {
              "$ref": "#/$defs/Variate"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "prior_rates": {
          "anyOf": [
            {
              "items": {
                "type": "number"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Prior Rates"
        }
      },
      "required": [
        "carrier"
      ],
      "title": "Flow",
      "type": "object"
    },
    "Investment": {
      "description": "Singular discrete build-timing optimization.\n\nThe solver decides WHEN to build (which period) and at what size.\nOnce built, capacity is available for ``lifetime`` periods.\nSize is decided once \u2014 no growth or partial retirement.",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "lifetime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lifetime"
        },
        "prior_size": {
          "default": 0.0,
          "minimum": 0,
          "title": "Prior Size",
          "type": "number"
        },
        "effects_per_size_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size At Build",
          "type": "object"
        },
        "effects_fixed_at_build": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed At Build",
          "type": "object"
        },
        "effects_per_size_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size Recurring",
          "type": "object"
        },
        "effects_fixed_recurring": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed Recurring",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Investment",
      "type": "object"
    },
    "PiecewiseConversion": {
      "description": "Piecewise-linear conversion linking N flows.\n\nWraps :func:`linopy.piecewise.add_piecewise_formulation`. All flows\nshare interpolation weights \u2014 every operating point lies on the same\npiece of the curve.\n\nTwo input forms:\n\n- **Dict** \u2014 equality-only, terse for the common case::\n\n      PiecewiseConversion(points={'fuel': [0, 50, 100], 'Heat': [0, 45, 70]})\n\n- **List of tuples** \u2014 supports per-flow inequality bounds::\n\n      PiecewiseConversion(\n          points=[\n              ('fuel', [0, 50, 100]),\n              ('Heat', [0, 45, 70], '>='),\n          ]\n      )\n\nSee: docs/math/converters.md",
      "properties": {
        "points": {
          "anyOf": [
            {
              "additionalProperties": {
                "items": {
                  "$ref": "#/$defs/Variate"
                },
                "type": "array"
              },
              "type": "object"
            },
            {
              "items": {
                "anyOf": [
                  {
                    "maxItems": 2,
                    "minItems": 2,
                    "prefixItems": [
                      {
                        "type": "string"
                      },
                      {
                        "items": {
                          "$ref": "#/$defs/Variate"
                        },
                        "type": "array"
                      }
                    ],
                    "type": "array"
                  },
                  {
                    "maxItems": 3,
                    "minItems": 3,
                    "prefixItems": [
                      {
                        "type": "string"
                      },
                      {
                        "items": {
                          "$ref": "#/$defs/Variate"
                        },
                        "type": "array"
                      },
                      {
                        "enum": [
                          "==",
                          "<=",
                          ">="
                        ],
                        "type": "string"
                      }
                    ],
                    "type": "array"
                  }
                ]
              },
              "type": "array"
            }
          ],
          "title": "Points"
        },
        "method": {
          "$ref": "#/$defs/PiecewiseMethod",
          "default": "auto"
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "availability": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        }
      },
      "required": [
        "points"
      ],
      "title": "PiecewiseConversion",
      "type": "object"
    },
    "PiecewiseMethod": {
      "enum": [
        "auto",
        "sos2",
        "incremental",
        "lp"
      ],
      "type": "string"
    },
    "Port": {
      "description": "System boundary importing and exporting carriers.\n\n``imports`` bring flow into the system (sources); ``exports`` send it out\nof the system (sinks).",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "imports": {
          "items": {
            "$ref": "#/$defs/Flow"
          },
          "title": "Imports",
          "type": "array"
        },
        "exports": {
          "items": {
            "$ref": "#/$defs/Flow"
          },
          "title": "Exports",
          "type": "array"
        }
      },
      "required": [
        "id"
      ],
      "title": "Port",
      "type": "object"
    },
    "ProfileRef": {
      "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
      "properties": {
        "dataset": {
          "title": "Dataset",
          "type": "string"
        },
        "variable": {
          "title": "Variable",
          "type": "string"
        }
      },
      "required": [
        "dataset",
        "variable"
      ],
      "title": "ProfileRef",
      "type": "object"
    },
    "Sizing": {
      "description": "Capacity optimization parameters.\n\nThe solver decides the optimal size within [size_min, size_max].\n\n- ``mandatory=True``: continuous, size in [min, max], no binary.\n- ``mandatory=False``: binary indicator gates size: 0 or [min, max].\n- ``size_min == size_max`` with ``mandatory=False``: binary invest\n  at exact size (yes/no).\n\nSee: docs/math/sizing.md",
      "properties": {
        "size_min": {
          "minimum": 0,
          "title": "Size Min",
          "type": "number"
        },
        "size_max": {
          "minimum": 0,
          "title": "Size Max",
          "type": "number"
        },
        "mandatory": {
          "default": true,
          "title": "Mandatory",
          "type": "boolean"
        },
        "effects_per_size": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Size",
          "type": "object"
        },
        "effects_fixed": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Fixed",
          "type": "object"
        }
      },
      "required": [
        "size_min",
        "size_max"
      ],
      "title": "Sizing",
      "type": "object"
    },
    "Status": {
      "description": "Binary on/off behavior parameters.\n\nTogether with relative bounds, gives semi-continuous behavior:\n``{0} U [min, max] * size``.\n\nSee: docs/math/status.md",
      "properties": {
        "uptime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Min"
        },
        "uptime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Uptime Max"
        },
        "downtime_min": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Min"
        },
        "downtime_max": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Downtime Max"
        },
        "effects_per_running_hour": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Running Hour",
          "type": "object"
        },
        "effects_per_startup": {
          "additionalProperties": {
            "$ref": "#/$defs/Variate"
          },
          "title": "Effects Per Startup",
          "type": "object"
        }
      },
      "title": "Status",
      "type": "object"
    },
    "Storage": {
      "description": "Energy storage with level dynamics.\n\nFlow ids are qualified as ``storage(flow)`` at build time. When both\nflows share a short_id, the qualified ids use ``charge`` /\n``discharge`` instead (the flows themselves are never modified)::\n\n    Storage(\n        id='bat', charging=Flow(carrier='elec'), discharging=Flow(carrier='elec')\n    )  # bat(charge), bat(discharge)\n    Storage(\n        id='bat',\n        charging=Flow(carrier='elec', short_id='in'),\n        discharging=Flow(carrier='elec', short_id='out'),\n    )  # bat(in), bat(out)\n\nLevel balance::\n\n    E_{s,t+1} = E_{s,t} (1 - \u03b4)^\u0394t + P^c \u03b7^c \u0394t - P^d / \u03b7^d \u0394t\n\nSee: docs/math/storage.md",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "charging": {
          "$ref": "#/$defs/Flow"
        },
        "discharging": {
          "$ref": "#/$defs/Flow"
        },
        "capacity": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "$ref": "#/$defs/Sizing"
            },
            {
              "$ref": "#/$defs/Investment"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Capacity"
        },
        "eta_charge": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "eta_discharge": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "relative_loss_per_hour": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "prior_level": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Prior Level"
        },
        "cyclic": {
          "default": true,
          "title": "Cyclic",
          "type": "boolean"
        },
        "relative_level_min": {
          "$ref": "#/$defs/Variate",
          "default": 0.0
        },
        "relative_level_max": {
          "$ref": "#/$defs/Variate",
          "default": 1.0
        },
        "final_level_min": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Final Level Min"
        },
        "final_level_max": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Final Level Max"
        },
        "prevent_simultaneous": {
          "default": false,
          "title": "Prevent Simultaneous",
          "type": "boolean"
        },
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        }
      },
      "required": [
        "id",
        "charging",
        "discharging"
      ],
      "title": "Storage",
      "type": "object"
    },
    "Timesteps": {
      "anyOf": [
        {
          "items": {
            "format": "date-time",
            "type": "string"
          },
          "type": "array"
        },
        {
          "items": {
            "type": "integer"
          },
          "type": "array"
        }
      ]
    },
    "Variate": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "integer"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "$ref": "#/$defs/ProfileRef"
        }
      ]
    }
  },
  "description": "A declarative flow-system description (see module docstring).",
  "properties": {
    "timesteps": {
      "$ref": "#/$defs/Timesteps"
    },
    "carriers": {
      "items": {
        "$ref": "#/$defs/Carrier"
      },
      "title": "Carriers",
      "type": "array"
    },
    "effects": {
      "items": {
        "$ref": "#/$defs/Effect"
      },
      "title": "Effects",
      "type": "array"
    },
    "ports": {
      "items": {
        "$ref": "#/$defs/Port"
      },
      "title": "Ports",
      "type": "array"
    },
    "objective": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "additionalProperties": {
            "type": "number"
          },
          "type": "object"
        }
      ],
      "title": "Objective"
    },
    "converters": {
      "items": {
        "$ref": "#/$defs/Converter"
      },
      "title": "Converters",
      "type": "array"
    },
    "storages": {
      "items": {
        "$ref": "#/$defs/Storage"
      },
      "title": "Storages",
      "type": "array"
    },
    "dt": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dt"
    },
    "periods": {
      "anyOf": [
        {
          "items": {
            "type": "integer"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Periods"
    },
    "period_weights": {
      "anyOf": [
        {
          "items": {
            "type": "number"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Period Weights"
    }
  },
  "required": [
    "timesteps",
    "carriers",
    "effects",
    "ports",
    "objective"
  ],
  "title": "FlowSystem",
  "type": "object"
}

Config:

  • default: _PYDANTIC_CFG

Fields:

Validators:

  • _validate_references

timesteps pydantic-field

timesteps: Timesteps

Time index for the optimization horizon.

carriers pydantic-field

carriers: list[Carrier]

Carrier declarations.

effects pydantic-field

effects: list[Effect]

Effects to track (costs, emissions, …).

ports pydantic-field

ports: list[Port]

System boundary ports with imports/exports.

objective pydantic-field

objective: str | dict[str, float]

Effect(s) to minimize — a name or {effect: weight}. Must name at least one non-penalty effect.

converters pydantic-field

converters: list[Converter]

Linear/piecewise converters between carriers.

storages pydantic-field

storages: list[Storage]

Energy storages.

dt pydantic-field

dt: float | list[float] | None = None

Timestep duration in hours. Auto-derived if None.

periods pydantic-field

periods: list[int] | None = None

Integer period labels for multi-period optimization.

period_weights pydantic-field

period_weights: list[float] | None = None

Explicit weights per period. Inferred from gaps if None.

from_dict classmethod

from_dict(data: Mapping[str, Any]) -> FlowSystem

Build a system from a mapping (e.g. parsed YAML/JSON).

Source code in src/fluxopt/flow_system.py
@classmethod
def from_dict(cls, data: Mapping[str, Any]) -> FlowSystem:
    """Build a system from a mapping (e.g. parsed YAML/JSON)."""
    return from_dict(cls, data)

from_yaml classmethod

from_yaml(path: str | Path) -> FlowSystem

Load a system from a YAML file.

Parameters:

Name Type Description Default

path

str | Path

Path to a YAML document describing the system.

required
Source code in src/fluxopt/flow_system.py
@classmethod
def from_yaml(cls, path: str | Path) -> FlowSystem:
    """Load a system from a YAML file.

    Args:
        path: Path to a YAML document describing the system.
    """
    import yaml

    with open(path) as fh:
        return cls.from_dict(yaml.safe_load(fh))

to_dict

to_dict() -> dict[str, Any]

Serialize the system to a JSON-safe dict (profiles as ProfileRef).

Source code in src/fluxopt/flow_system.py
def to_dict(self) -> dict[str, Any]:
    """Serialize the system to a JSON-safe dict (profiles as ``ProfileRef``)."""
    return to_dict(self)

to_yaml

to_yaml(path: str | Path) -> None

Write the system to a YAML file.

Parameters:

Name Type Description Default

path

str | Path

Destination path.

required
Source code in src/fluxopt/flow_system.py
def to_yaml(self, path: str | Path) -> None:
    """Write the system to a YAML file.

    Args:
        path: Destination path.
    """
    import yaml

    with open(path, 'w') as fh:
        yaml.safe_dump(self.to_dict(), fh, sort_keys=False)

required_profiles

required_profiles() -> dict[str, set[str]]

Enumerate the external data this system needs, as {dataset: variables}.

The contract a profiles supply must cover before :meth:build_model / :meth:optimize can run. Empty when every value is inline.

Source code in src/fluxopt/flow_system.py
def required_profiles(self) -> dict[str, set[str]]:
    """Enumerate the external data this system needs, as ``{dataset: variables}``.

    The contract a ``profiles`` supply must cover before
    :meth:`build_model` / :meth:`optimize` can run. Empty when every value
    is inline.
    """
    refs: list[tuple[str, ProfileRef]] = []
    for group in (self.carriers, self.effects, self.ports, self.converters, self.storages):
        _collect_profile_refs(group, '', refs)
    out: dict[str, set[str]] = {}
    for _, ref in refs:
        out.setdefault(ref.dataset, set()).add(ref.variable)
    return out

math

math() -> Any

The equations this system will be solved as, as data.

An :class:lpspec.Model — the whole program, before any of this system's numbers are bound to it. Read it (to_yaml()), typeset it (mathspec.to_latex(...)), or edit it and hand it back to :meth:optimize as math=: adding a named quantity to expressions: or a row to constraints: is how a caller extends the math now, in the same language and with the same load-time checks the shipped program gets.

Source code in src/fluxopt/flow_system.py
def math(self) -> Any:
    """The equations this system will be solved as, as data.

    An :class:`lpspec.Model` — the whole program, before any of this
    system's numbers are bound to it. Read it (``to_yaml()``), typeset it
    (``mathspec.to_latex(...)``), or edit it and hand it back to
    :meth:`optimize` as ``math=``: adding a named quantity to
    ``expressions:`` or a row to ``constraints:`` is how a caller extends
    the math now, in the same language and with the same load-time checks
    the shipped program gets.
    """
    from fluxopt.math import program

    return program()

parameters

parameters(profiles: Mapping[str, Any] | None = None) -> Parameters

The numbers this system binds to its math, as data.

The other half of :meth:math. That returns the equations before any of this system's numbers reach them; this returns the numbers, keyed as the program declares them — so (math, parameters/) is the whole problem, hashable and diffable, and a caller adding a constraint through math= can see what they are adding to.

Derived, never authored: the fold and the pre-scaling have already turned what the user declared into what the solver adds up, so the set is worth keeping and not worth editing. Change the elements or change the program.

Parameters:

Name Type Description Default

profiles

Mapping[str, Any] | None

Mapping from ProfileRef.dataset to a dataset (or mapping) holding the referenced variables, as :meth:build_data takes.

None
Source code in src/fluxopt/flow_system.py
def parameters(self, profiles: Mapping[str, Any] | None = None) -> Parameters:
    """The numbers this system binds to its math, as data.

    The other half of :meth:`math`. That returns the equations before any
    of this system's numbers reach them; this returns the numbers, keyed
    as the program declares them — so `(math, parameters/)` is the
    whole problem, hashable and diffable, and a caller adding a constraint
    through ``math=`` can see what they are adding to.

    Derived, never authored: the fold and the pre-scaling have already
    turned what the user declared into what the solver adds up, so the
    set is worth keeping and not worth editing. Change the elements or
    change the program.

    Args:
        profiles: Mapping from ``ProfileRef.dataset`` to a dataset (or
            mapping) holding the referenced variables, as
            :meth:`build_data` takes.
    """
    return Parameters.of(self.build_data(profiles), self.objective)

build_data

build_data(profiles: Mapping[str, Any] | None = None) -> ModelData

Materialize this declaration's data, resolving profile references.

Resolves ProfileRef references (on a copy — the system stays reusable across different profiles) and builds the ModelData both lanes read.

Parameters:

Name Type Description Default

profiles

Mapping[str, Any] | None

Mapping from ProfileRef.dataset to a dataset (or mapping) holding the referenced variables. Required if the system uses any ProfileRef — see :meth:required_profiles.

None

Raises:

Type Description
KeyError

If any ProfileRef cannot be resolved from profiles; lists every unresolvable ref with its element/field path.

Source code in src/fluxopt/flow_system.py
def build_data(self, profiles: Mapping[str, Any] | None = None) -> ModelData:
    """Materialize this declaration's data, resolving profile references.

    Resolves ``ProfileRef`` references (on a copy — the system stays
    reusable across different ``profiles``) and builds the ``ModelData``
    both lanes read.

    Args:
        profiles: Mapping from ``ProfileRef.dataset`` to a dataset (or mapping)
            holding the referenced variables. Required if the system uses
            any ``ProfileRef`` — see :meth:`required_profiles`.

    Raises:
        KeyError: If any ``ProfileRef`` cannot be resolved from *profiles*;
            lists every unresolvable ref with its element/field path.
    """
    refs: list[tuple[str, ProfileRef]] = []
    for group in (self.carriers, self.effects, self.ports, self.converters, self.storages):
        _collect_profile_refs(group, '', refs)
    _check_profiles_cover(refs, profiles or {})

    carriers, effects, ports, converters, storages = (
        # No refs → nothing to substitute, so no copy or walk needed.
        (self.carriers, self.effects, self.ports, self.converters, self.storages)
        if not refs
        else copy.deepcopy((self.carriers, self.effects, self.ports, self.converters, self.storages))
    )
    if refs:
        for group in (carriers, effects, ports, converters, storages):
            _resolve_refs(group, profiles or {})

    data = ModelData.build(
        self.timesteps,
        carriers,
        effects,
        ports,
        converters,
        storages,
        self.dt,
        periods=self.periods,
        period_weights=self.period_weights,
    )
    return data

optimize

optimize(
    profiles: Mapping[str, Any] | None = None,
    *,
    solver: str = 'highs',
    math: Any = None,
    dimensions: Mapping[str, Any] | None = None,
    lookups: Mapping[str, Any] | None = None,
    parameters: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> Result

Resolve profile references, build the data, and solve.

Parameters:

Name Type Description Default

profiles

Mapping[str, Any] | None

Mapping from ProfileRef.dataset to a dataset (or mapping) holding the referenced variables. Required if the system uses any ProfileRef.

None

solver

str

Solver name — highs, or gurobi with lpspec's extra.

'highs'

math

Any

An edited :meth:math to solve instead of the shipped program. Whatever it declares is checked and lowered exactly as the shipped one is.

None

dimensions

Mapping[str, Any] | None

Labels for dimensions math adds.

None

lookups

Mapping[str, Any] | None

Data for lookups math adds, as {name: frame}.

None

parameters

Mapping[str, Any] | None

Data for parameters math adds.

The three are the language's own declaration blocks, and :meth:parameters returns them under the same three names. All are ignored by the shipped program, which declares nothing a caller has to fill.

None

**kwargs

Any

Passed to the solver verbatim, in its own vocabulary.

{}
Source code in src/fluxopt/flow_system.py
def optimize(
    self,
    profiles: Mapping[str, Any] | None = None,
    *,
    solver: str = 'highs',
    math: Any = None,
    dimensions: Mapping[str, Any] | None = None,
    lookups: Mapping[str, Any] | None = None,
    parameters: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> Result:
    """Resolve profile references, build the data, and solve.

    Args:
        profiles: Mapping from ``ProfileRef.dataset`` to a dataset (or mapping)
            holding the referenced variables. Required if the system uses
            any ``ProfileRef``.
        solver: Solver name — ``highs``, or ``gurobi`` with lpspec's extra.
        math: An edited :meth:`math` to solve instead of the shipped
            program. Whatever it declares is checked and lowered exactly
            as the shipped one is.
        dimensions: Labels for dimensions *math* adds.
        lookups: Data for lookups *math* adds, as ``{name: frame}``.
        parameters: Data for parameters *math* adds.

            The three are the language's own declaration blocks, and
            :meth:`parameters` returns them under the same three names.
            All are ignored by the shipped program, which declares
            nothing a caller has to fill.
        **kwargs: Passed to the solver verbatim, in its own vocabulary.
    """
    from fluxopt.math import solve

    return solve(
        self.build_data(profiles),
        self.objective,
        solver_name=solver,
        solver_options=kwargs or None,
        math=math,
        dimensions=dimensions,
        lookups=lookups,
        parameters=parameters,
    )

Dims dataclass

Dims(timesteps: DataFrame, periods: DataFrame)

The model's temporal axes: what a timestep is, and what a period is.

Two tables, indexed by position like every other container. The labels the user gave ride along as a column, because they are what a result is reported against and nothing else keys on them.

Absence is a missing table: a system with no investment periods has an empty periods, which is what has_periods reads.

Methods:

Name Description
coords

Return shared coordinates for variable/DataArray creation.

build

Build Dims from a time index and optional periods.

Attributes:

Name Type Description
has_periods bool

Whether the system declares an investment period axis.

n_time int

How many timesteps the horizon has.

n_periods int

How many periods the program indexes — one, when none are declared.

period_labels list[Any]

The period labels, or [0] when the system declares no periods.

time DataArray

Timestep labels as a coordinate.

dt DataArray

Timestep durations [h] as a time-indexed array.

weights DataArray

Timestep weights as a time-indexed array.

period DataArray | None

Period labels as a coordinate, or None when there is no period axis.

period_weights DataArray | None

Period duration weights, or None when there is no period axis.

has_periods property

has_periods: bool

Whether the system declares an investment period axis.

n_time property

n_time: int

How many timesteps the horizon has.

n_periods property

n_periods: int

How many periods the program indexes — one, when none are declared.

period_labels property

period_labels: list[Any]

The period labels, or [0] when the system declares no periods.

time property

time: DataArray

Timestep labels as a coordinate.

dt property

Timestep durations [h] as a time-indexed array.

weights property

weights: DataArray

Timestep weights as a time-indexed array.

period property

period: DataArray | None

Period labels as a coordinate, or None when there is no period axis.

period_weights property

period_weights: DataArray | None

Period duration weights, or None when there is no period axis.

coords

coords(*, time: bool = False, period: bool = False) -> dict[str, DataArray]

Return shared coordinates for variable/DataArray creation.

Also the single point of truth for the model's variate dims used by :func:fluxopt.types.as_dataarray: pick the reach a field supports (e.g. coords(time=True, period=True) for operational profiles, coords(period=True) for investment-time fields). When a new variate dim (e.g. scenario) is added, extend this method once and every call site picks it up.

Parameters:

Name Type Description Default

time

bool

Include the time coordinate.

False

period

bool

Include the period coordinate (no-op in single-period mode).

False
Source code in src/fluxopt/model_data.py
def coords(self, *, time: bool = False, period: bool = False) -> dict[str, xr.DataArray]:
    """Return shared coordinates for variable/DataArray creation.

    Also the single point of truth for the model's variate dims used by
    :func:`fluxopt.types.as_dataarray`: pick the reach a field supports
    (e.g. ``coords(time=True, period=True)`` for operational profiles,
    ``coords(period=True)`` for investment-time fields). When a new
    variate dim (e.g. ``scenario``) is added, extend this method once
    and every call site picks it up.

    Args:
        time: Include the time coordinate.
        period: Include the period coordinate (no-op in single-period mode).
    """
    result: dict[str, xr.DataArray] = {}
    if time:
        result['time'] = self.time
    if period and (periods := self.period) is not None:
        result['period'] = periods
    return result

build classmethod

build(
    time: TimeIndex,
    dt: DataArray,
    periods: list[int] | Index | None = None,
    period_weights: list[float] | None = None,
) -> Self

Build Dims from a time index and optional periods.

Parameters:

Name Type Description Default

time

TimeIndex

Normalized time index.

required

dt

DataArray

Timestep durations.

required

periods

list[int] | Index | None

Integer period labels for multi-period optimization.

None

period_weights

list[float] | None

Explicit weights per period. Inferred from gaps if None.

None
Source code in src/fluxopt/model_data.py
@classmethod
def build(
    cls,
    time: TimeIndex,
    dt: xr.DataArray,
    periods: list[int] | pd.Index | None = None,
    period_weights: list[float] | None = None,
) -> Self:
    """Build Dims from a time index and optional periods.

    Args:
        time: Normalized time index.
        dt: Timestep durations.
        periods: Integer period labels for multi-period optimization.
        period_weights: Explicit weights per period. Inferred from gaps if None.
    """
    labels = np.asarray(time)
    timesteps = pl.DataFrame(
        {
            'time': np.arange(len(labels)),
            'label': labels,
            'dt': np.asarray(dt.values, dtype=float),
            'weight': np.ones(len(labels)),
        }
    )
    if periods is None:
        period_table = pl.DataFrame(
            {'period': [], 'label': [], 'weight': []},
            schema={'period': pl.Int64, 'label': pl.Int64, 'weight': pl.Float64},
        )
    else:
        period_idx, weights = _compute_period_weights(periods, period_weights)
        period_table = pl.DataFrame(
            {
                'period': np.arange(len(period_idx)),
                'label': period_idx.values,
                'weight': np.asarray(weights.values, dtype=float),
            },
            schema={'period': pl.Int64, 'label': pl.Int64, 'weight': pl.Float64},
        )
    return cls(timesteps=timesteps, periods=period_table)

ModelData dataclass

ModelData(
    flows: FlowsData,
    carriers: CarriersData,
    converters: ConvertersData | None,
    effects: EffectsData,
    storages: StoragesData | None,
    dims: Dims,
    piecewise: PiecewiseData | None = None,
    status: StatusData | None = None,
)

Methods:

Name Description
__post_init__

Check ids referenced between tables resolve.

save

Write the model data as a directory of tables.

load

Read model data written by :meth:save.

build

Build ModelData from element objects.

__post_init__

__post_init__() -> None

Check ids referenced between tables resolve.

Layer 3, and the clearest case for it: no single table can answer this, and a reloaded file never passed the system layer. See docs/design/validation-layers.md.

Each table validates itself in its own __post_init__; this checks that ids referenced between tables resolve, so a tampered or hand-edited file fails here instead of as a KeyError deep in model building.

Source code in src/fluxopt/model_data.py
def __post_init__(self) -> None:
    """Check ids referenced *between* tables resolve.

    Layer 3, and the clearest case for it: no single table can answer
    this, and a reloaded file never passed the system layer. See
    docs/design/validation-layers.md.

    Each table validates itself in its own ``__post_init__``; this checks
    that ids referenced *between* tables resolve, so a tampered or
    hand-edited file fails here instead of as a ``KeyError`` deep in
    model building.
    """
    flow_ids = set(self.flows.ids)

    def check_flows(ids: list[str], what: str) -> None:
        if unknown := sorted(set(ids) - flow_ids):
            raise ValueError(f'{what} references unknown flow id(s) {unknown}')

    def coord_ids(da: xr.DataArray) -> list[str]:
        return [str(v) for v in da.coords[da.dims[0]].values]

    check_flows(self.carriers.membership['flow'].to_list(), 'carriers.membership')
    if self.flows.sizing is not None:
        check_flows(self.flows.sizing.ids, 'flows.sizing')
    if self.flows.invest is not None:
        check_flows(self.flows.invest.ids, 'flows.invest')
    entities: set[str] = set(self.status.ids) if self.status is not None else set()
    if unknown := sorted(set(self.flows.governed_by['component'].to_list()) - entities):
        raise ValueError(f'flows.governed_by names components without a Status: {unknown}')
    self._check_status_not_degenerate(entities & flow_ids)
    if self.converters is not None:
        check_flows(self.converters.coefficients['flow'].to_list(), 'converters.coefficients')
    if self.piecewise is not None:
        check_flows(self.piecewise.links['flow'].to_list(), 'piecewise.links')
    if self.storages is not None:
        check_flows(self.storages.storages['charge_flow'].to_list(), 'storages.charge_flow')
        check_flows(self.storages.storages['discharge_flow'].to_list(), 'storages.discharge_flow')
        storage_ids = set(self.storages.ids)
        for container, what in (
            (self.storages.sizing, 'storages.sizing'),
            (self.storages.invest, 'storages.invest'),
        ):
            if container is None:
                continue
            ids = container.ids
            if unknown := sorted(set(ids) - storage_ids):
                raise ValueError(f'{what} references unknown storage id(s) {unknown}')

    effect_ids = set(self.effects.ids)
    coeff_effects = set(self.flows.effect_pairs['effect'].unique().to_list())
    if not coeff_effects <= effect_ids:
        raise ValueError(
            f'flows.effect_pairs names effects {sorted(coeff_effects - effect_ids)} that are not in '
            f'the effects table {sorted(effect_ids)}'
        )

save

save(path: str | Path) -> None

Write the model data as a directory of tables.

One parquet file per frame, in a directory per container. Parquet because these are tables: it carries the schema, so a column with no rows still knows it holds strings, and an empty frame reloads as the same empty frame rather than as something that lost its dtypes.

Parameters:

Name Type Description Default

path

str | Path

Directory to write into. Created if absent.

required
Source code in src/fluxopt/model_data.py
def save(self, path: str | Path) -> None:
    """Write the model data as a directory of tables.

    One parquet file per frame, in a directory per container. Parquet
    because these *are* tables: it carries the schema, so a column with no
    rows still knows it holds strings, and an empty frame reloads as the
    same empty frame rather than as something that lost its dtypes.

    Args:
        path: Directory to write into. Created if absent.
    """
    root = Path(path)
    root.mkdir(parents=True, exist_ok=True)
    for name, obj in self._containers().items():
        if obj is None:
            continue
        group = root / name
        group.mkdir(exist_ok=True)
        for frame_name, frame in _frames_of(obj).items():
            frame.write_parquet(group / f'{frame_name}.parquet')
        for cname, sub in _table_containers(obj).items():
            sub_group = group / cname
            sub_group.mkdir(exist_ok=True)
            for frame_name, frame in _frames_of(sub).items():
                frame.write_parquet(sub_group / f'{frame_name}.parquet')

load classmethod

load(path: str | Path) -> ModelData

Read model data written by :meth:save.

Parameters:

Name Type Description Default

path

str | Path

The directory :meth:save wrote.

required

Raises:

Type Description
OSError

If the directory holds no fluxopt model data.

Source code in src/fluxopt/model_data.py
@classmethod
def load(cls, path: str | Path) -> ModelData:
    """Read model data written by :meth:`save`.

    Args:
        path: The directory :meth:`save` wrote.

    Raises:
        OSError: If the directory holds no fluxopt model data.
    """
    root = Path(path)
    if not (root / 'dims').is_dir():
        raise OSError(f'No fluxopt model data found in {root} (missing dims/)')

    def read(name: str, klass: Any, subs: dict[str, Any] | None = None) -> Any:
        group = root / name
        if not group.is_dir():
            return None
        frames = {f.stem: pl.read_parquet(f) for f in sorted(group.glob('*.parquet'))}
        return klass(**frames, **(subs or {})) if frames else None

    def read_subs(name: str, klass: Any) -> dict[str, Any]:
        out: dict[str, Any] = {}
        for field, sub_class in _SUB_CONTAINERS.get(name, {}).items():
            sub = read(f'{name}/{field}', sub_class)
            if sub is not None:
                out[field] = sub
        return out

    return cls(
        flows=read('flows', FlowsData, read_subs('flows', FlowsData)),
        carriers=read('carriers', CarriersData),
        converters=read('converters', ConvertersData),
        effects=read('effects', EffectsData),
        storages=read('storages', StoragesData, read_subs('storages', StoragesData)),
        dims=read('dims', Dims),
        piecewise=read('piecewise', PiecewiseData),
        status=read('status', StatusData),
    )

build classmethod

build(
    timesteps: Timesteps,
    carriers: list[Carrier],
    effects: list[Effect],
    ports: list[Port],
    converters: list[Converter] | None = None,
    storages: list[Storage] | None = None,
    dt: float | list[float] | None = None,
    periods: list[int] | Index | None = None,
    period_weights: list[float] | None = None,
) -> Self

Build ModelData from element objects.

Parameters:

Name Type Description Default

timesteps

Timesteps

Time index for the optimization horizon.

required

carriers

list[Carrier]

Carrier declarations.

required

effects

list[Effect]

Effects to track.

required

ports

list[Port]

System boundary ports.

required

converters

list[Converter] | None

Linear converters.

None

storages

list[Storage] | None

Energy storages.

None

dt

float | list[float] | None

Timestep duration in hours. Auto-derived if None.

None

periods

list[int] | Index | None

Integer period labels for multi-period optimization.

None

period_weights

list[float] | None

Explicit weights per period. Inferred from gaps if None.

None
Source code in src/fluxopt/model_data.py
@classmethod
def build(
    cls,
    timesteps: Timesteps,
    carriers: list[Carrier],
    effects: list[Effect],
    ports: list[Port],
    converters: list[Converter] | None = None,
    storages: list[Storage] | None = None,
    dt: float | list[float] | None = None,
    periods: list[int] | pd.Index | None = None,
    period_weights: list[float] | None = None,
) -> Self:
    """Build ModelData from element objects.

    Args:
        timesteps: Time index for the optimization horizon.
        carriers: Carrier declarations.
        effects: Effects to track.
        ports: System boundary ports.
        converters: Linear converters.
        storages: Energy storages.
        dt: Timestep duration in hours. Auto-derived if None.
        periods: Integer period labels for multi-period optimization.
        period_weights: Explicit weights per period. Inferred from gaps if None.
    """
    from fluxopt.elements import PENALTY_EFFECT_ID, Effect
    from fluxopt.types import compute_dt as _compute_dt

    converters = converters or []
    stor_list = storages or []
    time = normalize_timesteps(timesteps)
    dt_da = _compute_dt(time, dt)

    if not any(e.id == PENALTY_EFFECT_ID for e in effects):
        effects = [*effects, Effect(id=PENALTY_EFFECT_ID)]

    flows, carrier_coeff = _collect_flows(ports, converters, stor_list)
    validate_system(carriers=carriers, effects=effects, ports=ports, converters=converters, storages=stor_list)

    dims = Dims.build(time, dt_da, periods=periods, period_weights=period_weights)

    # Scalar dt for prior duration computation. Pre-horizon steps have no
    # dt of their own, so the first timestep's duration stands in; on a
    # non-uniform grid that is an assumption the user should know about.
    durations = dims.timesteps['dt'].to_numpy()
    dt_scalar = float(durations[0])
    if any(bf.flow.prior_rates is not None for bf in flows) and not np.allclose(durations, dt_scalar):
        warnings.warn(
            f'prior_rates with non-uniform dt: pre-horizon status durations assume the first '
            f'timestep duration ({dt_scalar} h) for every prior step. If your prior steps had '
            f'different durations, adjust prior_rates to compensate.',
            UserWarning,
            stacklevel=2,
        )
    period_idx = pd.Index(dims.periods['label'].to_list()) if dims.has_periods else None

    comp_status_items: list[tuple[str, Status, list[str]]] = [
        (s.id, s.status, [s._charging_id, s._discharging_id]) for s in stor_list if s.status is not None
    ]
    comp_status_items.extend(
        (c.id, c.conversion.status, [bf.id for bf in c._qualified_flows()])
        for c in converters
        if c.conversion is not None and c.conversion.status is not None
    )

    flows_data = FlowsData.build(flows, time, period=period_idx, component_status_items=comp_status_items)
    # One table for one dimension: a flow's own Status and a component's
    # differ only in which rows read the binary, and the program says so
    # with a single `status_entity`. Flows first, so the axis reads in
    # declaration order.
    status_data = StatusData.build(
        [(bf.id, bf.flow.status) for bf in flows if bf.flow.status is not None]
        + [(cid, z) for cid, z, _ in comp_status_items],
        time,
        prior_rates_map={bf.id: bf.flow.prior_rates for bf in flows if bf.flow.prior_rates is not None},
        dt=dt_scalar,
        period=period_idx,
    )
    carriers_data = CarriersData.build(carriers, flows, carrier_coeff)
    converters_data = ConvertersData.build(converters, time)
    effects_data = EffectsData.build(effects, time, period=period_idx)
    storages_data = StoragesData.build(stor_list, time, period=period_idx)
    piecewise_data = PiecewiseData.build(converters, time)

    return cls(
        flows=flows_data,
        carriers=carriers_data,
        converters=converters_data,
        effects=effects_data,
        storages=storages_data,
        dims=dims,
        piecewise=piecewise_data,
        status=status_data,
    )

Result dataclass

Optimization result with solution variables and model data.

Provides access to flow rates, storage levels, effect totals, and investment decisions. Key properties::

result.objective  # scalar objective value
result.flow_rates  # (flow, time) DataArray
result.flow_rate('id')  # single flow time series
result.storage_levels  # (storage, time) DataArray
result.effect_totals  # (effect,) DataArray
result.effects_temporal  # (effect, time) DataArray
result.effects_lump  # (effect,) DataArray
result.sizes  # (flow,) DataArray — invested sizes
result.storage_capacities  # (storage,) DataArray

Per-contributor effect breakdown is available via result.stats.

Parameters:

Name Type Description Default

solution

Dataset

Solved variable values as xr.Dataset.

required

data

ModelData

ModelData used to build the optimization.

required

duals

Dataset

Dual values (shadow prices) from the solver.

Dataset()

expressions

Dataset

Every quantity the model names, evaluated at this solution. A named expression is evaluated against a solve, so it can only be had here — carrying them is what makes a Result answer for the model rather than only for its variables. Read one with :meth:expression; the effect breakdown is a view over them (result.stats.effect_contributions).

Dataset()

Methods:

Name Description
flow_rate

Get flow rate time series for a single flow.

storage_level

Get charge state time series for a single storage.

expression

One quantity the model names, at this solution.

save

Write the result as a directory.

load

Read a result written by :meth:save.

Attributes:

Name Type Description
objective float

Objective function value.

objective_weights dict[str, float]

Effect weights the objective was minimized with (provenance).

flow_rates DataArray

All flow rates as (flow, time) DataArray.

storage_levels DataArray

All storage levels as (storage, time) DataArray.

sizes DataArray

Optimized flow sizes as (flow,) DataArray.

storage_capacities DataArray

Optimized storage capacities as (storage,) DataArray.

effect_totals DataArray

Total effect values as (effect,) DataArray.

effects_temporal DataArray

Per-timestep effect values as (effect, time) DataArray.

effects_lump DataArray

Non-temporal effect values as (effect,) DataArray.

topology dict[Literal['carriers', 'converters'], dict[str, dict[str, list[str]]]]

Carrier and converter connectivity derived from model data.

stats StatsAccessor

Post-processing statistics accessor.

plot PlotAccessor

Plotting accessor (requires fluxopt-plot).

objective property

objective: float

Objective function value.

objective_weights property

objective_weights: dict[str, float]

Effect weights the objective was minimized with (provenance).

Includes the built-in penalty effect (auto-added at 1.0 unless named in objective). Empty for results saved before this field existed.

flow_rates property

flow_rates: DataArray

All flow rates as (flow, time) DataArray.

storage_levels property

storage_levels: DataArray

All storage levels as (storage, time) DataArray.

sizes property

sizes: DataArray

Optimized flow sizes as (flow,) DataArray.

storage_capacities property

storage_capacities: DataArray

Optimized storage capacities as (storage,) DataArray.

effect_totals property

effect_totals: DataArray

Total effect values as (effect,) DataArray.

effects_temporal property

effects_temporal: DataArray

Per-timestep effect values as (effect, time) DataArray.

Reconstructed from flow rates and coefficients — the model carries no per-timestep effect variables (temporal closure).

effects_lump property

effects_lump: DataArray

Non-temporal effect values as (effect,) DataArray.

topology cached property

topology: dict[
    Literal['carriers', 'converters'], dict[str, dict[str, list[str]]]
]

Carrier and converter connectivity derived from model data.

Returns a dict with carriers and converters keys, each mapping element ids to their inputs (flows that produce into the element) and outputs (flows that consume from it).

stats cached property

Post-processing statistics accessor.

plot cached property

plot: PlotAccessor

Plotting accessor (requires fluxopt-plot).

flow_rate

flow_rate(flow_id: str) -> DataArray

Get flow rate time series for a single flow.

Parameters:

Name Type Description Default

flow_id

str

Qualified flow id.

required
Source code in src/fluxopt/results.py
def flow_rate(self, flow_id: str) -> xr.DataArray:
    """Get flow rate time series for a single flow.

    Args:
        flow_id: Qualified flow id.
    """
    return self.flow_rates.sel(flow=flow_id)

storage_level

storage_level(storage_id: str) -> DataArray

Get charge state time series for a single storage.

Parameters:

Name Type Description Default

storage_id

str

Storage id.

required
Source code in src/fluxopt/results.py
def storage_level(self, storage_id: str) -> xr.DataArray:
    """Get charge state time series for a single storage.

    Args:
        storage_id: Storage id.
    """
    return self.storage_levels.sel(storage=storage_id)

expression

expression(name: str) -> DataArray

One quantity the model names, at this solution.

Whatever the program declares under expressions: — including anything a caller added through optimize(math=...).

Parameters:

Name Type Description Default

name

str

The expression's name in the program.

required

Raises:

Type Description
KeyError

Naming what this Result does carry, since a model that declared it and a Result that kept it are different things.

Source code in src/fluxopt/results.py
def expression(self, name: str) -> xr.DataArray:
    """One quantity the model names, at this solution.

    Whatever the program declares under ``expressions:`` — including
    anything a caller added through ``optimize(math=...)``.

    Args:
        name: The expression's name in the program.

    Raises:
        KeyError: Naming what this Result does carry, since a model that
            declared it and a Result that kept it are different things.
    """
    if name not in self.expressions:
        available = ', '.join(sorted(str(n) for n in self.expressions.data_vars)) or 'nothing'
        msg = f'{name!r} is not among the expressions this Result carries ({available})'
        raise KeyError(msg)
    return self.expressions[name]

save

save(path: str | Path) -> None

Write the result as a directory.

The solution and the named quantities are labelled N-dimensional arrays, so they stay netCDF; the model data is tables and goes to parquet under model/.

Parameters:

Name Type Description Default

path

str | Path

Directory to write into. Created if absent.

required
Source code in src/fluxopt/results.py
def save(self, path: str | Path) -> None:
    """Write the result as a directory.

    The solution and the named quantities are labelled N-dimensional
    arrays, so they stay netCDF; the model data is tables and goes to
    parquet under ``model/``.

    Args:
        path: Directory to write into. Created if absent.
    """
    root = Path(path)
    root.mkdir(parents=True, exist_ok=True)
    self.solution.to_netcdf(root / 'solution.nc', mode='w', engine='netcdf4')
    if self.expressions.data_vars:
        self.expressions.to_netcdf(root / 'expressions.nc', mode='w', engine='netcdf4')
    self.data.save(root / 'model')

load classmethod

load(path: str | Path) -> Result

Read a result written by :meth:save.

Parameters:

Name Type Description Default

path

str | Path

The directory :meth:save wrote.

required

Raises:

Type Description
OSError

If the directory holds no solution.

Source code in src/fluxopt/results.py
@classmethod
def load(cls, path: str | Path) -> Result:
    """Read a result written by :meth:`save`.

    Args:
        path: The directory :meth:`save` wrote.

    Raises:
        OSError: If the directory holds no solution.
    """
    import warnings

    from fluxopt.model_data import ModelData

    root = Path(path)
    if not (root / 'solution.nc').is_file():
        raise OSError(f'No fluxopt result found in {root} (missing solution.nc)')
    solution = _read_netcdf(root / 'solution.nc')
    expressions = _read_netcdf(root / 'expressions.nc') if (root / 'expressions.nc').is_file() else xr.Dataset()
    if not expressions.data_vars:
        warnings.warn(
            f'{root} carries none of the quantities the model names — including the '
            'per-contributor effect breakdown. They are evaluated against a solve and '
            'cannot be recovered from the solution alone; re-solve, or re-save a Result '
            'that has them.',
            stacklevel=2,
        )
    return cls(solution=solution, data=ModelData.load(root / 'model'), expressions=expressions)

ProfileRef pydantic-model

Bases: BaseModel

Reference to a time-series stored outside the model definition.

A serializable stand-in for an inline Variate array: the profile lives in a data file / dataset and is named here, so structural definitions round-trip to YAML/JSON without inlining 8760-point series. Resolve it to a :class:xr.DataArray with :meth:resolve before building the model.

Show JSON schema:
{
  "description": "Reference to a time-series stored outside the model definition.\n\nA serializable stand-in for an inline ``Variate`` array: the profile lives\nin a data file / dataset and is named here, so structural definitions\nround-trip to YAML/JSON without inlining 8760-point series. Resolve it to a\n:class:`xr.DataArray` with :meth:`resolve` before building the model.",
  "properties": {
    "dataset": {
      "title": "Dataset",
      "type": "string"
    },
    "variable": {
      "title": "Variable",
      "type": "string"
    }
  },
  "required": [
    "dataset",
    "variable"
  ],
  "title": "ProfileRef",
  "type": "object"
}

Config:

  • frozen: True

Fields:

dataset pydantic-field

dataset: str

Id of the dataset holding the profile (a key into profiles).

variable pydantic-field

variable: str

Variable / column name within dataset.

resolve

Look up the referenced series in profiles.

Parameters:

Name Type Description Default

profiles

Mapping[str, Dataset | Mapping[str, DataArray]]

Mapping from dataset id to a dataset (or mapping) that contains variable.

required

Raises:

Type Description
KeyError

If dataset or variable is absent from profiles.

Source code in src/fluxopt/types.py
def resolve(self, profiles: Mapping[str, xr.Dataset | Mapping[str, xr.DataArray]]) -> xr.DataArray:
    """Look up the referenced series in *profiles*.

    Args:
        profiles: Mapping from dataset id to a dataset (or mapping) that
            contains ``variable``.

    Raises:
        KeyError: If *dataset* or *variable* is absent from *profiles*.
    """
    if self.dataset not in profiles:
        raise KeyError(f'ProfileRef dataset {self.dataset!r} not in profiles {sorted(profiles)}')
    ds = profiles[self.dataset]
    try:
        return xr.DataArray(ds[self.variable])
    except KeyError as exc:
        raise KeyError(f'ProfileRef variable {self.variable!r} not in dataset {self.dataset!r}') from exc

all_element_schemas

all_element_schemas() -> Mapping[str, dict[str, Any]]

Return JSON Schemas for every element type, keyed by class name.

Source code in src/fluxopt/schema.py
def all_element_schemas() -> Mapping[str, dict[str, Any]]:
    """Return JSON Schemas for every element type, keyed by class name."""
    return {t.__name__: element_schema(t) for t in ELEMENT_TYPES}

element_schema

element_schema(element_type: type) -> dict[str, Any]

Return the JSON Schema for one element type.

Parameters:

Name Type Description Default

element_type

type

An element dataclass (e.g. Flow, Effect).

required
Source code in src/fluxopt/schema.py
def element_schema(element_type: type) -> dict[str, Any]:
    """Return the JSON Schema for one element type.

    Args:
        element_type: An element dataclass (e.g. ``Flow``, ``Effect``).
    """
    return TypeAdapter(element_type).json_schema()

from_dict

from_dict(element_type: type[T], data: Mapping[str, Any]) -> T

Reconstruct an element from a dict produced by :func:to_dict.

Parameters:

Name Type Description Default

element_type

type[T]

The element class to build (e.g. Flow).

required

data

Mapping[str, Any]

A mapping of field values.

required
Source code in src/fluxopt/schema.py
def from_dict[T](element_type: type[T], data: Mapping[str, Any]) -> T:
    """Reconstruct an element from a dict produced by :func:`to_dict`.

    Args:
        element_type: The element class to build (e.g. ``Flow``).
        data: A mapping of field values.
    """
    return TypeAdapter(element_type).validate_python(data)

to_dict

to_dict(element: object) -> dict[str, Any]

Serialize an element to a JSON-safe dict.

Nested elements and ProfileRef references are included; inline array-valued Variate fields are not serializable — reference them with a ProfileRef instead.

Parameters:

Name Type Description Default

element

object

Any element instance (e.g. Flow, Converter).

required

Raises:

Type Description
ValueError

If the element holds inline array values, listing each offending field.

Source code in src/fluxopt/schema.py
def to_dict(element: object) -> dict[str, Any]:
    """Serialize an element to a JSON-safe dict.

    Nested elements and ``ProfileRef`` references are
    included; inline array-valued ``Variate`` fields are not serializable —
    reference them with a ``ProfileRef`` instead.

    Args:
        element: Any element instance (e.g. ``Flow``, ``Converter``).

    Raises:
        ValueError: If the element holds inline array values, listing each
            offending field.
    """
    try:
        return TypeAdapter(type(element)).dump_python(element, mode='json')
    except Exception as exc:
        arrays: list[str] = []
        _inline_array_paths(element, type(element).__name__, arrays)
        if arrays:
            msg = (
                'cannot serialize inline array values at: '
                + '; '.join(arrays)
                + ' — reference them with ProfileRef(dataset=..., variable=...) and supply the data via profiles={...} at solve time'
            )
            raise ValueError(msg) from exc
        raise

as_dataarray

as_dataarray(
    value: Variate,
    coords: Mapping[str, Any],
    *,
    name: str = 'value',
    broadcast: bool = True,
) -> DataArray

Convert a Variate to a DataArray aligned to given coordinates.

Pipeline: convert → validate dims → validate coord values → broadcast.

See :data:Variate for accepted inputs. Pandas inputs (Series, DataFrame) follow the same convention as linopy.as_dataarray: the axis name attribute selects the corresponding target dim. For ndarray/list, the dim is selected by length (must be unambiguous). For DataArray, dims must be a subset of coords and coord values must match exactly — alignment errors are surfaced loudly, not silently masked.

Parameters:

Name Type Description Default

value

Variate

Scalar, list, ndarray, Series, DataFrame, or DataArray.

required

coords

Mapping[str, Any]

Target coordinates, e.g. {"time": idx, "period": pidx}. Used both as the reach declaration and as alignment targets.

required

name

str

Name for the resulting DataArray.

'value'

broadcast

bool

Expand result to span all dimensions in coords.

True
Source code in src/fluxopt/types.py
def as_dataarray(
    value: Variate,
    coords: Mapping[str, Any],
    *,
    name: str = 'value',
    broadcast: bool = True,
) -> xr.DataArray:
    """Convert a Variate to a DataArray aligned to given coordinates.

    Pipeline: ``convert → validate dims → validate coord values → broadcast``.

    See :data:`Variate` for accepted inputs. Pandas inputs (``Series``,
    ``DataFrame``) follow the same convention as ``linopy.as_dataarray``: the
    axis ``name`` attribute selects the corresponding target dim. For
    ``ndarray``/``list``, the dim is selected by length (must be unambiguous).
    For ``DataArray``, dims must be a subset of *coords* and coord values must
    match exactly — alignment errors are surfaced loudly, not silently masked.

    Args:
        value: Scalar, list, ndarray, Series, DataFrame, or DataArray.
        coords: Target coordinates, e.g. ``{"time": idx, "period": pidx}``.
            Used both as the reach declaration and as alignment targets.
        name: Name for the resulting DataArray.
        broadcast: Expand result to span all dimensions in *coords*.
    """
    if isinstance(value, ProfileRef):
        raise ValueError(
            f'Unresolved ProfileRef {value!r}: resolve it to an array via ProfileRef.resolve(profiles) '
            f'before building the model.'
        )

    coord_idx = {k: v if isinstance(v, pd.Index) else pd.Index(v) for k, v in coords.items()}

    # --- scalar: 0-dim unless broadcast ---
    if isinstance(value, (int, float)):
        if not broadcast:
            return xr.DataArray(float(value), name=name)
        shape = tuple(len(v) for v in coord_idx.values())
        return xr.DataArray(
            np.full(shape, float(value)),
            dims=list(coord_idx),
            coords=coord_idx,
            name=name,
        )

    # --- 1) Convert to DataArray ---
    da: xr.DataArray
    if isinstance(value, xr.DataArray):
        da = value
    elif isinstance(value, (pd.Series, pd.DataFrame)):
        # Mirror linopy: pandas axes already carry coords; use axis.name as dim.
        # Fall back to length-matching only when no axis is named.
        named = [a.name for a in value.axes if a.name is not None]
        if len(named) == value.ndim:
            da = xr.DataArray(value)
        elif value.ndim == 1 and not named:
            return _from_unnamed_1d(np.asarray(value.values, dtype=float), coord_idx, name, broadcast)
        else:
            raise ValueError(
                f'{type(value).__name__} requires axis.name set on every axis '
                f'(got {[a.name for a in value.axes]!r}). '
                f"Set e.g. df.index.name='time', df.columns.name='period'."
            )
    elif isinstance(value, np.ndarray):
        if value.ndim != 1:
            raise ValueError(
                f'np.ndarray must be 1-D (got ndim={value.ndim}); pass an xr.DataArray '
                f'or pd.DataFrame with named axes for higher-dim inputs.'
            )
        return _from_unnamed_1d(value, coord_idx, name, broadcast)
    elif isinstance(value, list):
        return _from_unnamed_1d(np.asarray(value, dtype=float), coord_idx, name, broadcast)
    else:
        raise TypeError(f'Unsupported Variate type: {type(value)}')

    # --- 2) Validate dims are a subset of the target ---
    foreign = [str(d) for d in da.dims if d not in coord_idx]
    if foreign:
        raise ValueError(
            f'{type(value).__name__} has dims {foreign} not in target coords {list(coord_idx)}. '
            f'Rename before calling as_dataarray().'
        )

    # --- 3) Validate coord values match exactly (close the alignment gap) ---
    for d in da.dims:
        dim_name = str(d)
        if d in da.coords and not pd.Index(da.coords[d].values).equals(coord_idx[dim_name]):
            raise ValueError(
                f'Coord mismatch on dim {dim_name!r}: input coord does not equal target. '
                f"Use the same index as the model's {dim_name}."
            )

    da = da.rename(name)
    if broadcast:
        for dim, idx in coord_idx.items():
            if dim not in da.dims:
                da = da.expand_dims({dim: idx})
        da = da.transpose(*coord_idx)
    return da

optimize

optimize(
    timesteps: Timesteps,
    carriers: list[Carrier],
    effects: list[Effect],
    ports: list[Port],
    objective: str | dict[str, float],
    converters: list[Converter] | None = None,
    storages: list[Storage] | None = None,
    dt: float | list[float] | None = None,
    periods: list[int] | None = None,
    period_weights: list[float] | None = None,
    profiles: Mapping[str, Any] | None = None,
    solver: str = 'highs',
    math: Any = None,
    parameters: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> Result

Build data, build model, optimize, return results.

Parameters:

Name Type Description Default

timesteps

Timesteps

Time index for the optimization horizon.

required

carriers

list[Carrier]

Carrier declarations.

required

effects

list[Effect]

Effects to track (costs, emissions, etc.).

required

ports

list[Port]

System boundary ports with imports/exports.

required

objective

str | dict[str, float]

Effect(s) to minimize. A single name, or a dict mapping effect names to objective weights ({'cost': 1, 'co2': 50}) — tracked effect totals are unaffected by the weighting. The built-in 'penalty' effect is added at weight 1.0 unless the dict names it ({'cost': 1, 'penalty': 0} opts out).

required

converters

list[Converter] | None

Linear converters between carriers.

None

storages

list[Storage] | None

Energy storages.

None

dt

float | list[float] | None

Timestep duration in hours. Auto-derived if None.

None

periods

list[int] | None

Integer period labels for multi-period optimization.

None

period_weights

list[float] | None

Explicit weights per period. Inferred from gaps if None.

None

profiles

Mapping[str, Any] | None

Mapping from ProfileRef.dataset to a dataset (or mapping) holding referenced time series. Required if any element uses a ProfileRef.

None

solver

str

Solver name — highs, or gurobi with lpspec's extra.

'highs'

math

Any

An edited program to solve instead of the shipped one; see :meth:~fluxopt.flow_system.FlowSystem.math.

None

parameters

Mapping[str, Any] | None

Data for the parameters math adds.

None

**kwargs

Any

Passed to the solver verbatim, in its own vocabulary.

{}
Source code in src/fluxopt/__init__.py
def optimize(
    timesteps: Timesteps,
    carriers: list[Carrier],
    effects: list[Effect],
    ports: list[Port],
    objective: str | dict[str, float],
    converters: list[Converter] | None = None,
    storages: list[Storage] | None = None,
    dt: float | list[float] | None = None,
    periods: list[int] | None = None,
    period_weights: list[float] | None = None,
    profiles: Mapping[str, Any] | None = None,
    solver: str = 'highs',
    math: Any = None,
    parameters: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> Result:
    """Build data, build model, optimize, return results.

    Args:
        timesteps: Time index for the optimization horizon.
        carriers: Carrier declarations.
        effects: Effects to track (costs, emissions, etc.).
        ports: System boundary ports with imports/exports.
        objective: Effect(s) to minimize. A single name, or a dict
            mapping effect names to objective weights
            (``{'cost': 1, 'co2': 50}``) — tracked effect totals are
            unaffected by the weighting. The built-in ``'penalty'`` effect
            is added at weight 1.0 unless the dict names it
            (``{'cost': 1, 'penalty': 0}`` opts out).
        converters: Linear converters between carriers.
        storages: Energy storages.
        dt: Timestep duration in hours. Auto-derived if None.
        periods: Integer period labels for multi-period optimization.
        period_weights: Explicit weights per period. Inferred from gaps if None.
        profiles: Mapping from ``ProfileRef.dataset`` to a dataset (or mapping)
            holding referenced time series. Required if any element uses a
            ``ProfileRef``.
        solver: Solver name — ``highs``, or ``gurobi`` with lpspec's extra.
        math: An edited program to solve instead of the shipped one; see
            :meth:`~fluxopt.flow_system.FlowSystem.math`.
        parameters: Data for the parameters *math* adds.
        **kwargs: Passed to the solver verbatim, in its own vocabulary.
    """
    system = FlowSystem(
        timesteps=timesteps,
        carriers=carriers,
        effects=effects,
        ports=ports,
        objective=objective,
        converters=converters or [],
        storages=storages or [],
        dt=dt,
        periods=periods,
        period_weights=period_weights,
    )
    return system.optimize(profiles, solver=solver, math=math, parameters=parameters, **kwargs)