fluxopt.results

Classes:

Name	Description
Result	Optimization result with solution variables and model data.

Result dataclass

Result(
    solution: Dataset,
    data: ModelData,
    duals: Dataset = Dataset(),
    contributions: Dataset | None = None,
)

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()
contributions	Dataset | None	Cached direct per-contributor effect breakdown (no cross-effect propagation). Surfaced via result.stats.effect_contributions_direct; result.stats.effect_contributions applies Leontief on top.	None

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.
to_netcdf	Write solution and model data to NetCDF.
from_netcdf	Read a Result from a NetCDF file.
from_model	Extract solution from a solved linopy model.

Attributes:

Name	Type	Description
objective	float	Objective function value.
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.

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.

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

stats: StatsAccessor

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)

to_netcdf

to_netcdf(path: str | Path) -> None

Write solution and model data to NetCDF.

Parameters:

Name	Type	Description	Default
path	str | Path	Output file path.	required

Source code in src/fluxopt/results.py

def to_netcdf(self, path: str | Path) -> None:
    """Write solution and model data to NetCDF.

    Args:
        path: Output file path.
    """
    p = Path(path)
    self.solution.to_netcdf(p, mode='w', engine='netcdf4')
    self.data.to_netcdf(p)
    if self.contributions is not None:
        self.contributions.to_netcdf(p, mode='a', group='contributions', engine='netcdf4')

from_netcdf classmethod

from_netcdf(path: str | Path) -> Result

Read a Result from a NetCDF file.

Parameters:

Name	Type	Description	Default
path	str | Path	Input file path.	required

Source code in src/fluxopt/results.py

@classmethod
def from_netcdf(cls, path: str | Path) -> Result:
    """Read a Result from a NetCDF file.

    Args:
        path: Input file path.
    """
    from fluxopt.model_data import ModelData

    p = Path(path)
    solution = xr.load_dataset(p, engine='netcdf4')
    data = ModelData.from_netcdf(p)

    try:
        contributions = xr.load_dataset(p, group='contributions', engine='netcdf4')
    except OSError:
        contributions = None
        import warnings

        warnings.warn(
            f"NetCDF file {p} has no 'contributions' group; per-contributor effect "
            'breakdown will be re-derived from solution + ModelData on first access. '
            'Results may differ from the original solve if the contribution-decomposition '
            'logic has changed since the file was written. Re-save the Result to refresh '
            'the cached breakdown.',
            stacklevel=2,
        )

    return cls(solution=solution, data=data, contributions=contributions)

from_model classmethod

from_model(model: FlowSystem) -> Result

Extract solution from a solved linopy model.

Parameters:

Name	Type	Description	Default
model	FlowSystem	Solved FlowSystem instance.	required

Source code in src/fluxopt/results.py

@classmethod
def from_model(cls, model: FlowSystem) -> Result:
    """Extract solution from a solved linopy model.

    Args:
        model: Solved FlowSystem instance.
    """
    sol_vars: dict[str, xr.DataArray] = {
        'flow--rate': model.flow_rate.solution,
        'effect--total': model.effect_total.solution,
        'effect--temporal': model.effect_temporal.solution,
        'effect--lump': model.effect_lump.solution,
    }

    if model.storage_level is not None:
        sol_vars['storage--level'] = model.storage_level.solution
    if model.flow_size is not None:
        sol_vars['flow--size'] = model.flow_size.solution
    if model.flow_size_indicator is not None:
        sol_vars['flow--size_indicator'] = model.flow_size_indicator.solution
    if model.storage_capacity is not None:
        sol_vars['storage--capacity'] = model.storage_capacity.solution
    if model.storage_capacity_indicator is not None:
        sol_vars['storage--size_indicator'] = model.storage_capacity_indicator.solution
    if model.invest_size is not None:
        sol_vars['invest--size'] = model.invest_size.solution
    if model.invest_build is not None:
        sol_vars['invest--build'] = model.invest_build.solution
    if model.invest_active is not None:
        sol_vars['invest--active'] = model.invest_active.solution
    if model.invest_size_at_build is not None:
        sol_vars['invest--size_at_build'] = model.invest_size_at_build.solution
    if model.flow_on is not None:
        sol_vars['flow--on'] = model.flow_on.solution
    if model.flow_startup is not None:
        sol_vars['flow--startup'] = model.flow_startup.solution
    if model.flow_shutdown is not None:
        sol_vars['flow--shutdown'] = model.flow_shutdown.solution
    if model.component_on is not None:
        sol_vars['component--on'] = model.component_on.solution
    if model.component_startup is not None:
        sol_vars['component--startup'] = model.component_startup.solution
    if model.component_shutdown is not None:
        sol_vars['component--shutdown'] = model.component_shutdown.solution

    # Piecewise auxiliary variables (from linopy.add_piecewise_formulation).
    # Stored under their linopy-generated names so they survive IO roundtrip.
    for formulation in model._piecewise.values():
        for var_name in formulation.variable_names:
            if var_name not in sol_vars:
                sol_vars[var_name] = model.m.variables[var_name].solution

    # Include custom variables added after build()
    for var_name in model.m.variables:
        if var_name not in model._builtin_var_names and var_name not in sol_vars:
            sol_vars[var_name] = model.m.variables[var_name].solution

    raw = model.m.objective.value
    obj_val = float(raw) if raw is not None else 0.0

    solution = xr.Dataset(sol_vars, attrs={'objective': obj_val})
    duals = model.m.dual

    from fluxopt.contributions import _with_cross_effects, compute_effect_contributions

    try:
        # Cache the direct (no cross-effect) view — it's the primitive both
        # accessors build on. effect_contributions applies Leontief on top
        # via _with_cross_effects, which is cheap relative to _compute_direct.
        contributions = compute_effect_contributions(solution, model.data, cross_effects=False)
        # Sanity-check at solve time: applying cross-effects must reproduce
        # the solver's effect--total. Result discarded; caches stay direct.
        _with_cross_effects(contributions, model.data, solution)
    except Exception as exc:
        import warnings

        warnings.warn(
            f'Failed to compute effect contributions during solve ({exc!r}); '
            'result.contributions will be None (re-derive via result.stats.effect_contributions)',
            stacklevel=2,
        )
        contributions = None

    return cls(solution=solution, data=model.data, duals=duals, contributions=contributions)