flixopt.transform_accessor ¶

Transform accessor for FlowSystem.

This module provides the TransformAccessor class that enables transformations on FlowSystem like clustering, selection, and resampling.

Classes¶

TransformAccessor ¶

TransformAccessor(flow_system: FlowSystem)

Accessor for transformation methods on FlowSystem.

This class provides transformations that create new FlowSystem instances with modified structure or data, accessible via flow_system.transform.

Examples:

Clustered optimization:

>>> clustered_fs = flow_system.transform.cluster(params)
>>> clustered_fs.optimize(solver)
>>> print(clustered_fs.solution)

Future MGA:

>>> mga_fs = flow_system.transform.mga(alternatives=5)
>>> mga_fs.optimize(solver)

Initialize the accessor with a reference to the FlowSystem.

Parameters:

Name	Type	Description	Default
`flow_system`	`FlowSystem`	The FlowSystem to transform.	required

Functions¶

cluster ¶

cluster(parameters: ClusteringParameters, components_to_clusterize: list | None = None) -> FlowSystem

Create a clustered FlowSystem for time series aggregation.

This method creates a new FlowSystem that can be optimized with clustered time series data. The clustering reduces computational complexity by identifying representative time periods.

The returned FlowSystem: - Has the same timesteps as the original (clustering works via constraints, not reduction) - Has aggregated time series data (if aggregate_data_and_fix_non_binary_vars=True) - Will have clustering constraints added during build_model()

Parameters:

Name	Type	Description	Default
`parameters`	`ClusteringParameters`	Clustering parameters specifying period duration, number of periods, and aggregation settings.	required
`components_to_clusterize`	`list \| None`	List of components to apply clustering to. If None, all components are clustered.	`None`

Returns:

Type	Description
`FlowSystem`	A new FlowSystem configured for clustered optimization.

Raises:

Type	Description
`ValueError`	If timestep sizes are inconsistent.
`ValueError`	If hours_per_period is not a multiple of timestep size.

Examples:

Basic clustered optimization:

>>> from flixopt import ClusteringParameters
>>> params = ClusteringParameters(
...     hours_per_period=24,
...     nr_of_periods=8,
...     fix_storage_flows=True,
...     aggregate_data_and_fix_non_binary_vars=True,
... )
>>> clustered_fs = flow_system.transform.cluster(params)
>>> clustered_fs.optimize(solver)
>>> print(clustered_fs.solution)

With model modifications:

>>> clustered_fs = flow_system.transform.cluster(params)
>>> clustered_fs.build_model()
>>> clustered_fs.model.add_constraints(...)
>>> clustered_fs.solve(solver)

sel ¶

sel(time: str | slice | list[str] | Timestamp | DatetimeIndex | None = None, period: int | slice | list[int] | Index | None = None, scenario: str | slice | list[str] | Index | None = None) -> FlowSystem

Select a subset of the FlowSystem by label.

Creates a new FlowSystem with data selected along the specified dimensions. The returned FlowSystem has no solution (it must be re-optimized).

Parameters:

Name	Type	Description	Default
`time`	`str \| slice \| list[str] \| Timestamp \| DatetimeIndex \| None`	Time selection (e.g., slice('2023-01-01', '2023-12-31'), '2023-06-15')	`None`
`period`	`int \| slice \| list[int] \| Index \| None`	Period selection (e.g., slice(2023, 2024), or list of periods)	`None`
`scenario`	`str \| slice \| list[str] \| Index \| None`	Scenario selection (e.g., 'scenario1', or list of scenarios)	`None`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New FlowSystem with selected data (no solution).

Examples:

>>> # Select specific time range
>>> fs_jan = flow_system.transform.sel(time=slice('2023-01-01', '2023-01-31'))
>>> fs_jan.optimize(solver)

>>> # Select single scenario
>>> fs_base = flow_system.transform.sel(scenario='Base Case')

isel ¶

isel(time: int | slice | list[int] | None = None, period: int | slice | list[int] | None = None, scenario: int | slice | list[int] | None = None) -> FlowSystem

Select a subset of the FlowSystem by integer indices.

Creates a new FlowSystem with data selected along the specified dimensions. The returned FlowSystem has no solution (it must be re-optimized).

Parameters:

Name	Type	Description	Default
`time`	`int \| slice \| list[int] \| None`	Time selection by integer index (e.g., slice(0, 100), 50, or [0, 5, 10])	`None`
`period`	`int \| slice \| list[int] \| None`	Period selection by integer index	`None`
`scenario`	`int \| slice \| list[int] \| None`	Scenario selection by integer index	`None`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New FlowSystem with selected data (no solution).

Examples:

>>> # Select first 24 timesteps
>>> fs_day1 = flow_system.transform.isel(time=slice(0, 24))
>>> fs_day1.optimize(solver)

>>> # Select first scenario
>>> fs_first = flow_system.transform.isel(scenario=0)

resample ¶

resample(time: str, method: Literal['mean', 'sum', 'max', 'min', 'first', 'last', 'std', 'var', 'median', 'count'] = 'mean', hours_of_last_timestep: int | float | None = None, hours_of_previous_timesteps: int | float | ndarray | None = None, fill_gaps: Literal['ffill', 'bfill', 'interpolate'] | None = None, **kwargs: Any) -> FlowSystem

Create a resampled FlowSystem by resampling data along the time dimension.

Creates a new FlowSystem with resampled time series data. The returned FlowSystem has no solution (it must be re-optimized).

Parameters:

Name	Type	Description	Default
`time`	`str`	Resampling frequency (e.g., '3h', '2D', '1M')	required
`method`	`Literal['mean', 'sum', 'max', 'min', 'first', 'last', 'std', 'var', 'median', 'count']`	Resampling method. Recommended: 'mean', 'first', 'last', 'max', 'min'	`'mean'`
`hours_of_last_timestep`	`int \| float \| None`	Duration of the last timestep after resampling. If None, computed from the last time interval.	`None`
`hours_of_previous_timesteps`	`int \| float \| ndarray \| None`	Duration of previous timesteps after resampling. If None, computed from the first time interval. Can be a scalar or array.	`None`
`fill_gaps`	`Literal['ffill', 'bfill', 'interpolate'] \| None`	Strategy for filling gaps (NaN values) that arise when resampling irregular timesteps to regular intervals. Options: 'ffill' (forward fill), 'bfill' (backward fill), 'interpolate' (linear interpolation). If None (default), raises an error when gaps are detected.	`None`
`**kwargs`	`Any`	Additional arguments passed to xarray.resample()	`{}`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New resampled FlowSystem (no solution).

Raises:

Type	Description
`ValueError`	If resampling creates gaps and fill_gaps is not specified.

Examples:

>>> # Resample to 4-hour intervals
>>> fs_4h = flow_system.transform.resample(time='4h', method='mean')
>>> fs_4h.optimize(solver)

>>> # Resample to daily with max values
>>> fs_daily = flow_system.transform.resample(time='1D', method='max')

fix_sizes ¶

fix_sizes(sizes: Dataset | dict[str, float] | None = None, decimal_rounding: int | None = 5) -> FlowSystem

Create a new FlowSystem with investment sizes fixed to specified values.

This is useful for two-stage optimization workflows: 1. Solve a sizing problem (possibly resampled for speed) 2. Fix sizes and solve dispatch at full resolution

The returned FlowSystem has InvestParameters with fixed_size set, making those sizes mandatory rather than decision variables.

Parameters:

Name	Type	Description	Default
`sizes`	`Dataset \| dict[str, float] \| None`	The sizes to fix. Can be: - None: Uses sizes from this FlowSystem's solution (must be solved) - xr.Dataset: Dataset with size variables (e.g., from statistics.sizes) - dict: Mapping of component names to sizes (e.g., {'Boiler(Q_fu)': 100})	`None`
`decimal_rounding`	`int \| None`	Number of decimal places to round sizes to. Rounding helps avoid numerical infeasibility. Set to None to disable.	`5`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New FlowSystem with fixed sizes (no solution).

Raises:

Type	Description
`ValueError`	If no sizes provided and FlowSystem has no solution.
`KeyError`	If a specified size doesn't match any InvestParameters.

Examples:

Two-stage optimization:

>>> # Stage 1: Size with resampled data
>>> fs_sizing = flow_system.transform.resample('2h')
>>> fs_sizing.optimize(solver)
>>>
>>> # Stage 2: Fix sizes and optimize at full resolution
>>> fs_dispatch = flow_system.transform.fix_sizes(fs_sizing.statistics.sizes)
>>> fs_dispatch.optimize(solver)

Using a dict:

>>> fs_fixed = flow_system.transform.fix_sizes(
...     {
...         'Boiler(Q_fu)': 100,
...         'Storage': 500,
...     }
... )
>>> fs_fixed.optimize(solver)