flixopt.flow_system ¶

This module contains the FlowSystem class, which is used to collect instances of many other classes by the end User.

Attributes¶

Classes¶

FlowSystem ¶

Python

FlowSystem(timesteps: DatetimeIndex, periods: Index | None = None, scenarios: Index | None = None, hours_of_last_timestep: int | float | None = None, hours_of_previous_timesteps: int | float | ndarray | None = None, weights: PeriodicDataUser | None = None, scenario_independent_sizes: bool | list[str] = True, scenario_independent_flow_rates: bool | list[str] = False)

Bases: Interface, CompositeContainerMixin[Element]

A FlowSystem organizes the high level Elements (Components, Buses, Effects & Flows).

This is the main container class that users work with to build and manage their energy or material flow system. FlowSystem provides both direct container access (via .components, .buses, .effects, .flows) and a unified dict-like interface for accessing any element by label across all container types.

Parameters:

Name	Type	Description	Default
`timesteps`	`DatetimeIndex`	The timesteps of the model.	required
`periods`	`Index \| None`	The periods of the model.	`None`
`scenarios`	`Index \| None`	The scenarios of the model.	`None`
`hours_of_last_timestep`	`int \| float \| None`	Duration of the last timestep. If None, computed from the last time interval.	`None`
`hours_of_previous_timesteps`	`int \| float \| ndarray \| None`	Duration of previous timesteps. If None, computed from the first time interval. Can be a scalar (all previous timesteps have same duration) or array (different durations). Used to calculate previous values (e.g., consecutive_on_hours).	`None`
`weights`	`PeriodicDataUser \| None`	The weights of each period and scenario. If None, all scenarios have the same weight (normalized to 1). Its recommended to normalize the weights to sum up to 1.	`None`
`scenario_independent_sizes`	`bool \| list[str]`	Controls whether investment sizes are equalized across scenarios. - True: All sizes are shared/equalized across scenarios - False: All sizes are optimized separately per scenario - list[str]: Only specified components (by label_full) are equalized across scenarios	`True`
`scenario_independent_flow_rates`	`bool \| list[str]`	Controls whether flow rates are equalized across scenarios. - True: All flow rates are shared/equalized across scenarios - False: All flow rates are optimized separately per scenario - list[str]: Only specified flows (by label_full) are equalized across scenarios	`False`

Examples:

Creating a FlowSystem and accessing elements:

Python Console Session

>>> import flixopt as fx
>>> import pandas as pd
>>> timesteps = pd.date_range('2023-01-01', periods=24, freq='h')
>>> flow_system = fx.FlowSystem(timesteps)
>>>
>>> # Add elements to the system
>>> boiler = fx.Component('Boiler', inputs=[heat_flow], on_off_parameters=...)
>>> heat_bus = fx.Bus('Heat', excess_penalty_per_flow_hour=1e4)
>>> costs = fx.Effect('costs', is_objective=True, is_standard=True)
>>> flow_system.add_elements(boiler, heat_bus, costs)

Unified dict-like access (recommended for most cases):

Python Console Session

>>> # Access any element by label, regardless of type
>>> boiler = flow_system['Boiler']  # Returns Component
>>> heat_bus = flow_system['Heat']  # Returns Bus
>>> costs = flow_system['costs']  # Returns Effect
>>>
>>> # Check if element exists
>>> if 'Boiler' in flow_system:
...     print('Boiler found in system')
>>>
>>> # Iterate over all elements
>>> for label in flow_system.keys():
...     element = flow_system[label]
...     print(f'{label}: {type(element).__name__}')
>>>
>>> # Get all element labels and objects
>>> all_labels = list(flow_system.keys())
>>> all_elements = list(flow_system.values())
>>> for label, element in flow_system.items():
...     print(f'{label}: {element}')

Direct container access for type-specific operations:

Python Console Session

>>> # Access specific container when you need type filtering
>>> for component in flow_system.components.values():
...     print(f'{component.label}: {len(component.inputs)} inputs')
>>>
>>> # Access buses directly
>>> for bus in flow_system.buses.values():
...     print(f'{bus.label}')
>>>
>>> # Flows are automatically collected from all components

Power user pattern - Efficient chaining without conversion overhead:

Python Console Session

>>> # Instead of chaining (causes multiple conversions):
>>> result = flow_system.sel(time='2020-01').resample('2h')  # Slow
>>>
>>> # Use dataset methods directly (single conversion):
>>> ds = flow_system.to_dataset()
>>> ds = FlowSystem._dataset_sel(ds, time='2020-01')
>>> ds = flow_system._dataset_resample(ds, freq='2h', method='mean')
>>> result = FlowSystem.from_dataset(ds)  # Fast!
>>>
>>> # Available dataset methods:
>>> # - FlowSystem._dataset_sel(dataset, time=..., period=..., scenario=...)
>>> # - FlowSystem._dataset_isel(dataset, time=..., period=..., scenario=...)
>>> # - flow_system._dataset_resample(dataset, freq=..., method=..., **kwargs)
>>> for flow in flow_system.flows.values():
...     print(f'{flow.label_full}: {flow.size}')
>>>
>>> # Access effects
>>> for effect in flow_system.effects.values():
...     print(f'{effect.label}')

Notes

The dict-like interface (flow_system['element']) searches across all containers (components, buses, effects, flows) to find the element with the matching label.
Element labels must be unique across all container types. Attempting to add elements with duplicate labels will raise an error, ensuring each label maps to exactly one element.
The .all_elements property is deprecated. Use the dict-like interface instead: flow_system['element'], 'element' in flow_system, flow_system.keys(), flow_system.values(), or flow_system.items().
Direct container access (.components, .buses, .effects, .flows) is useful when you need type-specific filtering or operations.
The .flows container is automatically populated from all component inputs and outputs.
Creates an empty registry for components and buses, an empty EffectCollection, and a placeholder for a SystemModel.
The instance starts disconnected (self._connected_and_transformed == False) and will be connected_and_transformed automatically when trying to solve a calculation.

Attributes¶

all_elements `property` ¶

Python

all_elements: dict[str, Element]

Get all elements as a dictionary.

.. deprecated:: 3.2.0 Use dict-like interface instead: flow_system['element'], 'element' in flow_system, flow_system.keys(), flow_system.values(), or flow_system.items(). This property will be removed in v4.0.0.

Returns:

Type	Description
`dict[str, Element]`	Dictionary mapping element labels to element objects.

scenario_independent_sizes `property` `writable` ¶

Python

scenario_independent_sizes: bool | list[str]

Controls whether investment sizes are equalized across scenarios.

Returns:

Type	Description
`bool \| list[str]`	bool or list[str]: Configuration for scenario-independent sizing

scenario_independent_flow_rates `property` `writable` ¶

Python

scenario_independent_flow_rates: bool | list[str]

Controls whether flow rates are equalized across scenarios.

Returns:

Type	Description
`bool \| list[str]`	bool or list[str]: Configuration for scenario-independent flow rates

Functions¶

calculate_hours_per_timestep `staticmethod` ¶

Python

calculate_hours_per_timestep(timesteps_extra: DatetimeIndex) -> xr.DataArray

Calculate duration of each timestep as a 1D DataArray.

to_dataset ¶

Python

to_dataset() -> xr.Dataset

Convert the FlowSystem to an xarray Dataset. Ensures FlowSystem is connected before serialization.

Returns:

Type	Description
`Dataset`	xr.Dataset: Dataset containing all DataArrays with structure in attributes

from_dataset `classmethod` ¶

Python

from_dataset(ds: Dataset) -> FlowSystem

Create a FlowSystem from an xarray Dataset. Handles FlowSystem-specific reconstruction logic.

Parameters:

Name	Type	Description	Default
`ds`	`Dataset`	Dataset containing the FlowSystem data	required

Returns:

Type	Description
`FlowSystem`	FlowSystem instance

to_netcdf ¶

Python

to_netcdf(path: str | Path, compression: int = 0)

Save the FlowSystem to a NetCDF file. Ensures FlowSystem is connected before saving.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	The path to the netCDF file.	required
`compression`	`int`	The compression level to use when saving the file.	`0`

get_structure ¶

Python

get_structure(clean: bool = False, stats: bool = False) -> dict

Get FlowSystem structure. Ensures FlowSystem is connected before getting structure.

Parameters:

Name	Type	Description	Default
`clean`	`bool`	If True, remove None and empty dicts and lists.	`False`
`stats`	`bool`	If True, replace DataArray references with statistics	`False`

to_json ¶

Python

to_json(path: str | Path)

Save the flow system to a JSON file. Ensures FlowSystem is connected before saving.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	The path to the JSON file.	required

fit_to_model_coords ¶

Python

fit_to_model_coords(name: str, data: TemporalDataUser | PeriodicDataUser | None, dims: Collection[FlowSystemDimensions] | None = None) -> TemporalData | PeriodicData | None

Fit data to model coordinate system (currently time, but extensible).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the data	required
`data`	`TemporalDataUser \| PeriodicDataUser \| None`	Data to fit to model coordinates	required
`dims`	`Collection[FlowSystemDimensions] \| None`	Collection of dimension names to use for fitting. If None, all dimensions are used.	`None`

Returns:

Type	Description
`TemporalData \| PeriodicData \| None`	xr.DataArray aligned to model coordinate system. If data is None, returns None.

fit_effects_to_model_coords ¶

Python

fit_effects_to_model_coords(label_prefix: str | None, effect_values: TemporalEffectsUser | PeriodicEffectsUser | None, label_suffix: str | None = None, dims: Collection[FlowSystemDimensions] | None = None, delimiter: str = '|') -> TemporalEffects | PeriodicEffects | None

Transform EffectValues from the user to Internal Datatypes aligned with model coordinates.

connect_and_transform ¶

Python

connect_and_transform()

Transform data for all elements using the new simplified approach.

add_elements ¶

Python

add_elements(*elements: Element) -> None

Add Components(Storages, Boilers, Heatpumps, ...), Buses or Effects to the FlowSystem

Parameters:

Name	Type	Description	Default
`*elements`	`Element`	childs of Element like Boiler, HeatPump, Bus,... modeling Elements	`()`

create_model ¶

Python

create_model(normalize_weights: bool = True) -> FlowSystemModel

Create a linopy model from the FlowSystem.

Parameters:

Name	Type	Description	Default
`normalize_weights`	`bool`	Whether to automatically normalize the weights (periods and scenarios) to sum up to 1 when solving.	`True`

plot_network ¶

Python

plot_network(path: bool | str | Path = 'flow_system.html', controls: bool | list[Literal['nodes', 'edges', 'layout', 'interaction', 'manipulation', 'physics', 'selection', 'renderer']] = True, show: bool | None = None) -> pyvis.network.Network | None

Visualizes the network structure of a FlowSystem using PyVis, saving it as an interactive HTML file.

Parameters:

Name	Type	Description	Default
`path`	`bool \| str \| Path`	Path to save the HTML visualization. - `False`: Visualization is created but not saved. - `str` or `Path`: Specifies file path (default: 'flow_system.html').	`'flow_system.html'`
`controls`	`bool \| list[Literal['nodes', 'edges', 'layout', 'interaction', 'manipulation', 'physics', 'selection', 'renderer']]`	UI controls to add to the visualization. - `True`: Enables all available controls. - `List`: Specify controls, e.g., ['nodes', 'layout']. - Options: 'nodes', 'edges', 'layout', 'interaction', 'manipulation', 'physics', 'selection', 'renderer'.	`True`
`show`	`bool \| None`	Whether to open the visualization in the web browser.	`None`

Returns: - 'pyvis.network.Network' | None: The Network instance representing the visualization, or None if pyvis is not installed.

Examples:

Python Console Session

>>> flow_system.plot_network()
>>> flow_system.plot_network(show=False)
>>> flow_system.plot_network(path='output/custom_network.html', controls=['nodes', 'layout'])

Notes: - This function requires pyvis. If not installed, the function prints a warning and returns None. - Nodes are styled based on type (e.g., circles for buses, boxes for components) and annotated with node information.

start_network_app ¶

Python

start_network_app()

Visualizes the network structure of a FlowSystem using Dash, Cytoscape, and networkx. Requires optional dependencies: dash, dash-cytoscape, dash-daq, networkx, flask, werkzeug.

stop_network_app ¶

Python

stop_network_app()

Stop the network visualization server.

sel ¶

Python

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.

For power users: Use FlowSystem._dataset_sel() to chain operations on datasets without conversion overhead. See _dataset_sel() documentation.

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

isel ¶

Python

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.

For power users: Use FlowSystem._dataset_isel() to chain operations on datasets without conversion overhead. See _dataset_sel() documentation.

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 (e.g., slice(0, 100), 50, or [0, 5, 10])	`None`
`scenario`	`int \| slice \| list[int] \| None`	Scenario selection by integer index (e.g., slice(0, 3), 50, or [0, 5, 10])	`None`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New FlowSystem with selected data

resample ¶

Python

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, **kwargs: Any) -> FlowSystem

Create a resampled FlowSystem by resampling data along the time dimension (like xr.Dataset.resample()). Only resamples data variables that have a time dimension.

For power users: Use FlowSystem._dataset_resample() to chain operations on datasets without conversion overhead. See _dataset_sel() documentation.

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`
`**kwargs`	`Any`	Additional arguments passed to xarray.resample()	`{}`

Returns:

Name	Type	Description
`FlowSystem`	`FlowSystem`	New resampled FlowSystem

keys ¶

Python

keys() -> list[str]

Return all element labels across all containers.

values ¶

Python

values() -> list[T_element]

Return all element objects across all containers.

items ¶

Python

items() -> list[tuple[str, T_element]]

Return (label, element) pairs for all elements.

transform_data ¶

Python

transform_data(flow_system: FlowSystem, name_prefix: str = '') -> None

Transform the data of the interface to match the FlowSystem's dimensions.

Parameters:

Name	Type	Description	Default
`flow_system`	`FlowSystem`	The FlowSystem containing timing and dimensional information	required
`name_prefix`	`str`	The prefix to use for the names of the variables. Defaults to '', which results in no prefix.	`''`

Raises:

Type	Description
`NotImplementedError`	Must be implemented by subclasses

from_netcdf `classmethod` ¶

Python

from_netcdf(path: str | Path) -> Interface

Load an instance from a NetCDF file.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the NetCDF file	required

Returns:

Type	Description
`Interface`	Interface instance

Raises:

Type	Description
`IOError`	If file cannot be read
`ValueError`	If file format is invalid

copy ¶

Python

copy() -> Interface

Create a copy of the Interface object.

Uses the existing serialization infrastructure to ensure proper copying of all DataArrays and nested objects.

Returns:

Type	Description
`Interface`	A new instance of the same class with copied data.

flixopt.flow_system ¶

Attributes¶

Classes¶

FlowSystem ¶

Attributes¶

all_elements property ¶

scenario_independent_sizes property writable ¶

scenario_independent_flow_rates property writable ¶

Functions¶

calculate_hours_per_timestep staticmethod ¶

to_dataset ¶

from_dataset classmethod ¶

to_netcdf ¶

get_structure ¶

to_json ¶

fit_to_model_coords ¶

fit_effects_to_model_coords ¶

connect_and_transform ¶

add_elements ¶

create_model ¶

plot_network ¶

start_network_app ¶

stop_network_app ¶

sel ¶

isel ¶

resample ¶

keys ¶

values ¶

items ¶

transform_data ¶

from_netcdf classmethod ¶

copy ¶

all_elements `property` ¶

scenario_independent_sizes `property` `writable` ¶

scenario_independent_flow_rates `property` `writable` ¶

calculate_hours_per_timestep `staticmethod` ¶

from_dataset `classmethod` ¶

from_netcdf `classmethod` ¶