diff --git a/docs/stylesheets/extra.css b/docs/stylesheets/extra.css
index 78946b9ad..0b600fb08 100644
--- a/docs/stylesheets/extra.css
+++ b/docs/stylesheets/extra.css
@@ -763,6 +763,26 @@ button:focus-visible {
scrollbar-color: var(--md-default-fg-color--lighter) var(--md-default-bg-color);
}
+/* ============================================================================
+ Color Swatches for Carrier Documentation
+ ========================================================================= */
+
+/* Inline color swatch - a small colored square */
+.color-swatch {
+ display: inline-block;
+ width: 1em;
+ height: 1em;
+ border-radius: 3px;
+ vertical-align: middle;
+ margin-right: 0.3em;
+ border: 1px solid rgba(0, 0, 0, 0.15);
+ box-shadow: 0 1px 2px rgba(0, 0, 0, 0.1);
+}
+
+[data-md-color-scheme="slate"] .color-swatch {
+ border-color: rgba(255, 255, 255, 0.2);
+}
+
/* ============================================================================
Footer Alignment Fix
========================================================================= */
diff --git a/docs/user-guide/core-concepts.md b/docs/user-guide/core-concepts.md
index 401b34705..78e38ade7 100644
--- a/docs/user-guide/core-concepts.md
+++ b/docs/user-guide/core-concepts.md
@@ -31,6 +31,17 @@ $$\sum inputs = \sum outputs$$
This balance constraint is what makes your model physically meaningful — energy can't appear or disappear.
+### Carriers
+
+Buses can be assigned a **carrier** — a type of energy or material (electricity, heat, gas, etc.). Carriers enable automatic coloring in plots and help organize your system semantically:
+
+```python
+heat_bus = fx.Bus('HeatNetwork', carrier='heat') # Uses default heat color
+elec_bus = fx.Bus('Grid', carrier='electricity')
+```
+
+See [Color Management](results-plotting.md#color-management) for details.
+
## Flows: What Moves Between Elements
A [`Flow`][flixopt.elements.Flow] represents the movement of energy or material. Every flow connects a component to a bus, with a defined direction.
diff --git a/docs/user-guide/mathematical-notation/elements/Bus.md b/docs/user-guide/mathematical-notation/elements/Bus.md
index 464381fe8..ca089bfec 100644
--- a/docs/user-guide/mathematical-notation/elements/Bus.md
+++ b/docs/user-guide/mathematical-notation/elements/Bus.md
@@ -2,6 +2,29 @@
A Bus is where flows meet and must balance — inputs equal outputs at every timestep.
+## Carriers
+
+Buses can optionally be assigned a **carrier** — a type of energy or material (e.g., electricity, heat, gas). Carriers enable:
+
+- **Automatic coloring** in plots based on energy type
+- **Unit tracking** for better result visualization
+- **Semantic grouping** of buses by type
+
+```python
+# Assign a carrier by name (uses CONFIG.Carriers defaults)
+heat_bus = fx.Bus('HeatNetwork', carrier='heat')
+elec_bus = fx.Bus('Grid', carrier='electricity')
+
+# Or register custom carriers on the FlowSystem
+biogas = fx.Carrier('biogas', color='#228B22', unit='kW', description='Biogas fuel')
+flow_system.add_carrier(biogas)
+gas_bus = fx.Bus('BiogasNetwork', carrier='biogas')
+```
+
+See [Color Management](../../../user-guide/results-plotting.md#color-management) for more on how carriers affect visualization.
+
+---
+
## Basic: Balance Equation
$$
diff --git a/docs/user-guide/results-plotting.md b/docs/user-guide/results-plotting.md
index 4f1932e53..bfb6c8777 100644
--- a/docs/user-guide/results-plotting.md
+++ b/docs/user-guide/results-plotting.md
@@ -318,6 +318,113 @@ flow_system.statistics.plot.balance('Bus', colors={
})
```
+## Color Management
+
+flixOpt provides centralized color management through the `flow_system.colors` accessor and carriers. This ensures consistent colors across all visualizations.
+
+### Carriers
+
+[`Carriers`][flixopt.carrier.Carrier] define energy or material types with associated colors. Built-in carriers are available in `CONFIG.Carriers`:
+
+| Carrier | Color | Description |
+|---------|-------|-------------|
+| `electricity` | `#FECB52` | Yellow - lightning/energy |
+| `heat` | `#D62728` | Red - warmth/fire |
+| `gas` | `#1F77B4` | Blue - natural gas |
+| `hydrogen` | `#9467BD` | Purple - clean/future |
+| `fuel` | `#8C564B` | Brown - fossil/oil |
+| `biomass` | `#2CA02C` | Green - organic/renewable |
+
+Colors are from the D3/Plotly palettes for professional consistency.
+
+Assign carriers to buses for automatic coloring:
+
+```python
+# Buses use carrier colors automatically
+heat_bus = fx.Bus('HeatNetwork', carrier='heat')
+elec_bus = fx.Bus('Grid', carrier='electricity')
+
+# Plots automatically use carrier colors for bus-related elements
+flow_system.statistics.plot.sankey() # Buses colored by carrier
+```
+
+### Custom Carriers
+
+Register custom carriers on your FlowSystem:
+
+```python
+# Create a custom carrier
+biogas = fx.Carrier('biogas', color='#228B22', unit='kW', description='Biogas fuel')
+hydrogen = fx.Carrier('hydrogen', color='#00CED1', unit='kg/h')
+
+# Register with FlowSystem (overrides CONFIG.Carriers defaults)
+flow_system.add_carrier(biogas)
+flow_system.add_carrier(hydrogen)
+
+# Access registered carriers
+flow_system.carriers # CarrierContainer with locally registered carriers
+flow_system.get_carrier('biogas') # Returns Carrier object
+```
+
+### Color Accessor
+
+The `flow_system.colors` accessor provides centralized color configuration:
+
+```python
+# Configure colors for components
+flow_system.colors.setup({
+ 'Boiler': '#D35400',
+ 'CHP': '#8E44AD',
+ 'HeatPump': '#27AE60',
+})
+
+# Or set individual colors
+flow_system.colors.set_component_color('Boiler', '#D35400')
+flow_system.colors.set_carrier_color('biogas', '#228B22')
+
+# Load from file
+flow_system.colors.setup('colors.json') # or .yaml
+```
+
+### Context-Aware Coloring
+
+Plot colors are automatically resolved based on context:
+
+- **Bus balance plots**: Colors based on the connected component
+- **Component balance plots**: Colors based on the connected bus/carrier
+- **Sankey diagrams**: Buses use carrier colors, components use configured colors
+
+```python
+# Plotting a bus balance → flows colored by their parent component
+flow_system.statistics.plot.balance('ElectricityBus')
+
+# Plotting a component balance → flows colored by their connected bus/carrier
+flow_system.statistics.plot.balance('CHP')
+```
+
+### Color Resolution Priority
+
+Colors are resolved in this order:
+
+1. **Explicit colors** passed to plot methods (always override)
+2. **Component/bus colors** set via `flow_system.colors.setup()`
+3. **Element `meta_data['color']`** if present
+4. **Carrier colors** from FlowSystem or CONFIG.Carriers
+5. **Default colorscale** (controlled by `CONFIG.Plotting.default_qualitative_colorscale`)
+
+### Persistence
+
+Color configurations are automatically saved with the FlowSystem:
+
+```python
+# Colors are persisted
+flow_system.to_netcdf('my_system.nc')
+
+# And restored
+loaded = fx.FlowSystem.from_netcdf('my_system.nc')
+loaded.colors # Configuration restored
+```
+
### Display Control
Control whether plots are shown automatically:
diff --git a/examples/01_Simple/simple_example.py b/examples/01_Simple/simple_example.py
index 13781c973..e8185e248 100644
--- a/examples/01_Simple/simple_example.py
+++ b/examples/01_Simple/simple_example.py
@@ -21,7 +21,12 @@
# --- Define Energy Buses ---
# These represent nodes, where the used medias are balanced (electricity, heat, and gas)
- flow_system.add_elements(fx.Bus(label='Strom'), fx.Bus(label='Fernwärme'), fx.Bus(label='Gas'))
+ # Carriers provide automatic color assignment in plots (yellow for electricity, red for heat, etc.)
+ flow_system.add_elements(
+ fx.Bus(label='Strom', carrier='electricity'),
+ fx.Bus(label='Fernwärme', carrier='heat'),
+ fx.Bus(label='Gas', carrier='gas'),
+ )
# --- Define Effects (Objective and CO2 Emissions) ---
# Cost effect: used as the optimization objective --> minimizing costs
@@ -109,13 +114,14 @@
# Plotting through statistics accessor - returns PlotResult with .data and .figure
flow_system.statistics.plot.balance('Fernwärme')
flow_system.statistics.plot.balance('Storage')
- flow_system.statistics.plot.heatmap('CHP(Q_th)|flow_rate')
- flow_system.statistics.plot.heatmap('Storage|charge_state')
+ flow_system.statistics.plot.heatmap('CHP(Q_th)')
+ flow_system.statistics.plot.heatmap('Storage')
+ flow_system.statistics.plot.heatmap('Storage')
# Access data as xarray Datasets
print(flow_system.statistics.flow_rates)
print(flow_system.statistics.charge_states)
# Duration curve and effects analysis
- flow_system.statistics.plot.duration_curve('Boiler(Q_th)|flow_rate')
+ flow_system.statistics.plot.duration_curve('Boiler(Q_th)')
print(flow_system.statistics.temporal_effects)
diff --git a/examples/02_Complex/complex_example.py b/examples/02_Complex/complex_example.py
index f1b524a2b..3f38ff954 100644
--- a/examples/02_Complex/complex_example.py
+++ b/examples/02_Complex/complex_example.py
@@ -32,10 +32,11 @@
# --- Define Energy Buses ---
# Represent node balances (inputs=outputs) for the different energy carriers (electricity, heat, gas) in the system
+ # Carriers provide automatic color assignment in plots (yellow for electricity, red for heat, blue for gas)
flow_system.add_elements(
- fx.Bus('Strom', imbalance_penalty_per_flow_hour=imbalance_penalty),
- fx.Bus('Fernwärme', imbalance_penalty_per_flow_hour=imbalance_penalty),
- fx.Bus('Gas', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Strom', carrier='electricity', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Fernwärme', carrier='heat', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Gas', carrier='gas', imbalance_penalty_per_flow_hour=imbalance_penalty),
)
# --- Define Effects ---
@@ -200,7 +201,7 @@
flow_system.to_netcdf('results/complex_example.nc')
# Plot results using the statistics accessor
- flow_system.statistics.plot.heatmap('BHKW2(Q_th)|flow_rate')
+ flow_system.statistics.plot.heatmap('BHKW2(Q_th)') # Flow label - auto-resolves to flow_rate
flow_system.statistics.plot.balance('BHKW2')
- flow_system.statistics.plot.heatmap('Speicher|charge_state')
+ flow_system.statistics.plot.heatmap('Speicher') # Storage label - auto-resolves to charge_state
flow_system.statistics.plot.balance('Fernwärme')
diff --git a/examples/03_Optimization_modes/example_optimization_modes.py b/examples/03_Optimization_modes/example_optimization_modes.py
index 3dcd8bd1c..1f9968357 100644
--- a/examples/03_Optimization_modes/example_optimization_modes.py
+++ b/examples/03_Optimization_modes/example_optimization_modes.py
@@ -69,10 +69,10 @@ def get_solutions(optimizations: list, variable: str) -> xr.Dataset:
flow_system = fx.FlowSystem(timesteps)
flow_system.add_elements(
- fx.Bus('Strom', imbalance_penalty_per_flow_hour=imbalance_penalty),
- fx.Bus('Fernwärme', imbalance_penalty_per_flow_hour=imbalance_penalty),
- fx.Bus('Gas', imbalance_penalty_per_flow_hour=imbalance_penalty),
- fx.Bus('Kohle', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Strom', carrier='electricity', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Fernwärme', carrier='heat', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Gas', carrier='gas', imbalance_penalty_per_flow_hour=imbalance_penalty),
+ fx.Bus('Kohle', carrier='fuel', imbalance_penalty_per_flow_hour=imbalance_penalty),
)
# Effects
diff --git a/examples/04_Scenarios/scenario_example.py b/examples/04_Scenarios/scenario_example.py
index e3c6f5fd3..820336e93 100644
--- a/examples/04_Scenarios/scenario_example.py
+++ b/examples/04_Scenarios/scenario_example.py
@@ -89,7 +89,12 @@
# --- Define Energy Buses ---
# These represent nodes, where the used medias are balanced (electricity, heat, and gas)
- flow_system.add_elements(fx.Bus(label='Strom'), fx.Bus(label='Fernwärme'), fx.Bus(label='Gas'))
+ # Carriers provide automatic color assignment in plots (yellow for electricity, red for heat, blue for gas)
+ flow_system.add_elements(
+ fx.Bus(label='Strom', carrier='electricity'),
+ fx.Bus(label='Fernwärme', carrier='heat'),
+ fx.Bus(label='Gas', carrier='gas'),
+ )
# --- Define Effects (Objective and CO2 Emissions) ---
# Cost effect: used as the optimization objective --> minimizing costs
@@ -199,10 +204,10 @@
# --- Analyze Results ---
# Plotting through statistics accessor - returns PlotResult with .data and .figure
- flow_system.statistics.plot.heatmap('CHP(Q_th)|flow_rate')
+ flow_system.statistics.plot.heatmap('CHP(Q_th)') # Flow label - auto-resolves to flow_rate
flow_system.statistics.plot.balance('Fernwärme')
flow_system.statistics.plot.balance('Storage')
- flow_system.statistics.plot.heatmap('Storage|charge_state')
+ flow_system.statistics.plot.heatmap('Storage') # Storage label - auto-resolves to charge_state
# Access data as xarray Datasets
print(flow_system.statistics.flow_rates)
diff --git a/examples/05_Two-stage-optimization/two_stage_optimization.py b/examples/05_Two-stage-optimization/two_stage_optimization.py
index 8dea1713b..bf7f13a39 100644
--- a/examples/05_Two-stage-optimization/two_stage_optimization.py
+++ b/examples/05_Two-stage-optimization/two_stage_optimization.py
@@ -37,11 +37,12 @@
gas_price = filtered_data['Gaspr.€/MWh'].to_numpy()
flow_system = fx.FlowSystem(timesteps)
+ # Carriers provide automatic color assignment in plots
flow_system.add_elements(
- fx.Bus('Strom'),
- fx.Bus('Fernwärme'),
- fx.Bus('Gas'),
- fx.Bus('Kohle'),
+ fx.Bus('Strom', carrier='electricity'),
+ fx.Bus('Fernwärme', carrier='heat'),
+ fx.Bus('Gas', carrier='gas'),
+ fx.Bus('Kohle', carrier='fuel'),
fx.Effect('costs', '€', 'Kosten', is_standard=True, is_objective=True),
fx.Effect('CO2', 'kg', 'CO2_e-Emissionen'),
fx.Effect('PE', 'kWh_PE', 'Primärenergie'),
diff --git a/flixopt/__init__.py b/flixopt/__init__.py
index 8874811b3..3c4edf7e8 100644
--- a/flixopt/__init__.py
+++ b/flixopt/__init__.py
@@ -14,6 +14,7 @@
# Import commonly used classes and functions
from . import linear_converters, plotting, results, solvers
+from .carrier import Carrier, CarrierContainer
from .clustering import ClusteringParameters
from .components import (
LinearConverter,
@@ -34,6 +35,8 @@
__all__ = [
'TimeSeriesData',
'CONFIG',
+ 'Carrier',
+ 'CarrierContainer',
'Flow',
'Bus',
'Effect',
diff --git a/flixopt/carrier.py b/flixopt/carrier.py
new file mode 100644
index 000000000..8a663eca9
--- /dev/null
+++ b/flixopt/carrier.py
@@ -0,0 +1,159 @@
+"""Carrier class for energy/material type definitions.
+
+Carriers represent types of energy or materials that flow through buses,
+such as electricity, heat, gas, or water. They provide consistent styling
+and metadata across visualizations.
+"""
+
+from __future__ import annotations
+
+from .structure import ContainerMixin, Interface, register_class_for_io
+
+
+@register_class_for_io
+class Carrier(Interface):
+ """Definition of an energy or material carrier type.
+
+ Carriers represent the type of energy or material flowing through a Bus.
+ They provide consistent color, unit, and description across all visualizations
+ and can be shared between multiple buses of the same type.
+
+ Inherits from Interface to provide serialization capabilities.
+
+ Args:
+ name: Identifier for the carrier (e.g., 'electricity', 'heat', 'gas').
+ color: Hex color string for visualizations (e.g., '#FFD700').
+ unit: Unit string for display (e.g., 'kW', 'kW_th', 'm³/h').
+ description: Optional human-readable description.
+
+ Examples:
+ Creating custom carriers:
+
+ ```python
+ import flixopt as fx
+
+ # Define custom carriers
+ electricity = fx.Carrier('electricity', '#FFD700', 'kW', 'Electrical power')
+ district_heat = fx.Carrier('district_heat', '#FF6B6B', 'kW_th', 'District heating')
+ hydrogen = fx.Carrier('hydrogen', '#00CED1', 'kg/h', 'Hydrogen fuel')
+
+ # Register with FlowSystem
+ flow_system.add_carrier(electricity)
+ flow_system.add_carrier(district_heat)
+
+ # Use with buses (just reference by name)
+ elec_bus = fx.Bus('MainGrid', carrier='electricity')
+ heat_bus = fx.Bus('HeatingNetwork', carrier='district_heat')
+ ```
+
+ Using predefined carriers from CONFIG:
+
+ ```python
+ # Access built-in carriers
+ elec = fx.CONFIG.Carriers.electricity
+ heat = fx.CONFIG.Carriers.heat
+
+ # Use directly
+ bus = fx.Bus('Grid', carrier='electricity')
+ ```
+
+ Adding custom carriers to CONFIG:
+
+ ```python
+ # Add a new carrier globally
+ fx.CONFIG.Carriers.add(fx.Carrier('biogas', '#228B22', 'kW', 'Biogas'))
+
+ # Now available as
+ fx.CONFIG.Carriers.biogas
+ ```
+
+ Note:
+ Carriers are compared by name for equality, allowing flexible usage
+ patterns where the same carrier type can be referenced by name string
+ or Carrier object interchangeably.
+ """
+
+ def __init__(
+ self,
+ name: str,
+ color: str = '',
+ unit: str = '',
+ description: str = '',
+ ) -> None:
+ """Initialize a Carrier.
+
+ Args:
+ name: Identifier for the carrier (normalized to lowercase).
+ color: Hex color string for visualizations.
+ unit: Unit string for display.
+ description: Optional human-readable description.
+ """
+ self.name = name.lower()
+ self.color = color
+ self.unit = unit
+ self.description = description
+
+ def transform_data(self, name_prefix: str = '') -> None:
+ """Transform data to match FlowSystem dimensions.
+
+ Carriers don't have time-series data, so this is a no-op.
+
+ Args:
+ name_prefix: Ignored for Carrier.
+ """
+ pass # Carriers have no data to transform
+
+ @property
+ def label(self) -> str:
+ """Label for container keying (alias for name)."""
+ return self.name
+
+ def __hash__(self):
+ return hash(self.name)
+
+ def __eq__(self, other):
+ if isinstance(other, Carrier):
+ return self.name == other.name
+ if isinstance(other, str):
+ return self.name == other.lower()
+ return False
+
+ def __repr__(self):
+ return f"Carrier('{self.name}', color='{self.color}', unit='{self.unit}')"
+
+ def __str__(self):
+ return self.name
+
+
+class CarrierContainer(ContainerMixin['Carrier']):
+ """Container for Carrier objects.
+
+ Uses carrier.name for keying. Provides dict-like access to carriers
+ registered with a FlowSystem.
+
+ Examples:
+ ```python
+ # Access via FlowSystem
+ carriers = flow_system.carriers
+
+ # Dict-like access
+ elec = carriers['electricity']
+ 'heat' in carriers # True/False
+
+ # Iteration
+ for name in carriers:
+ print(name)
+ ```
+ """
+
+ def __init__(self, carriers: list[Carrier] | dict[str, Carrier] | None = None):
+ """Initialize a CarrierContainer.
+
+ Args:
+ carriers: Initial carriers to add.
+ """
+ super().__init__(elements=carriers, element_type_name='carriers')
+
+ def _get_label(self, carrier: Carrier) -> str:
+ """Extract name from Carrier for keying."""
+ return carrier.name
diff --git a/flixopt/config.py b/flixopt/config.py
index 043142cbe..4d4571dcd 100644
--- a/flixopt/config.py
+++ b/flixopt/config.py
@@ -575,6 +575,36 @@ class Plotting:
default_sequential_colorscale: str = _DEFAULTS['plotting']['default_sequential_colorscale']
default_qualitative_colorscale: str = _DEFAULTS['plotting']['default_qualitative_colorscale']
+ class Carriers:
+ """Default carrier definitions for common energy types.
+
+ Provides convenient defaults for carriers. Colors are from D3/Plotly palettes.
+
+ Predefined: electricity, heat, gas, hydrogen, fuel, biomass
+
+ Examples:
+ ```python
+ import flixopt as fx
+
+ # Access predefined carriers
+ fx.CONFIG.Carriers.electricity # Carrier with color '#FECB52'
+ fx.CONFIG.Carriers.heat.color # '#D62728'
+
+ # Use with buses
+ bus = fx.Bus('Grid', carrier='electricity')
+ ```
+ """
+
+ from .carrier import Carrier
+
+ # Default carriers - colors from D3/Plotly palettes
+ electricity: Carrier = Carrier('electricity', '#FECB52') # Yellow
+ heat: Carrier = Carrier('heat', '#D62728') # Red
+ gas: Carrier = Carrier('gas', '#1F77B4') # Blue
+ hydrogen: Carrier = Carrier('hydrogen', '#9467BD') # Purple
+ fuel: Carrier = Carrier('fuel', '#8C564B') # Brown
+ biomass: Carrier = Carrier('biomass', '#2CA02C') # Green
+
config_name: str = _DEFAULTS['config_name']
@classmethod
@@ -601,6 +631,16 @@ def reset(cls) -> None:
for key, value in _DEFAULTS['plotting'].items():
setattr(cls.Plotting, key, value)
+ # Reset Carriers to defaults
+ from .carrier import Carrier
+
+ cls.Carriers.electricity = Carrier('electricity', '#FECB52')
+ cls.Carriers.heat = Carrier('heat', '#D62728')
+ cls.Carriers.gas = Carrier('gas', '#1F77B4')
+ cls.Carriers.hydrogen = Carrier('hydrogen', '#9467BD')
+ cls.Carriers.fuel = Carrier('fuel', '#8C564B')
+ cls.Carriers.biomass = Carrier('biomass', '#2CA02C')
+
cls.config_name = _DEFAULTS['config_name']
# Reset logging to default (silent)
diff --git a/flixopt/elements.py b/flixopt/elements.py
index 74ed7bde4..d2ebf7ac0 100644
--- a/flixopt/elements.py
+++ b/flixopt/elements.py
@@ -93,8 +93,9 @@ def __init__(
status_parameters: StatusParameters | None = None,
prevent_simultaneous_flows: list[Flow] | None = None,
meta_data: dict | None = None,
+ color: str | None = None,
):
- super().__init__(label, meta_data=meta_data)
+ super().__init__(label, meta_data=meta_data, color=color)
self.inputs: list[Flow] = inputs or []
self.outputs: list[Flow] = outputs or []
self.status_parameters = status_parameters
@@ -194,6 +195,9 @@ class Bus(Element):
Args:
label: The label of the Element. Used to identify it in the FlowSystem.
+ carrier: Name of the energy/material carrier type (e.g., 'electricity', 'heat', 'gas').
+ Carriers are registered via ``flow_system.add_carrier()`` or available as
+ predefined defaults in CONFIG.Carriers. Used for automatic color assignment in plots.
imbalance_penalty_per_flow_hour: Penalty costs for bus balance violations.
When None (default), no imbalance is allowed (hard constraint). When set to a
value > 0, allows bus imbalances at penalty cost.
@@ -201,30 +205,30 @@ class Bus(Element):
in results. Only use Python native types.
Examples:
- Electrical bus with strict balance:
+ Using predefined carrier names:
```python
- electricity_bus = Bus(
- label='main_electrical_bus',
- imbalance_penalty_per_flow_hour=None, # No imbalance allowed
- )
+ electricity_bus = Bus(label='main_grid', carrier='electricity')
+ heat_bus = Bus(label='district_heating', carrier='heat')
```
- Heat network with penalty for imbalances:
+ Registering custom carriers on FlowSystem:
```python
- heat_network = Bus(
- label='district_heating_network',
- imbalance_penalty_per_flow_hour=1000, # €1000/MWh penalty for imbalance
- )
+ import flixopt as fx
+
+ fs = fx.FlowSystem(timesteps)
+ fs.add_carrier(fx.Carrier('biogas', '#228B22', 'kW'))
+ biogas_bus = fx.Bus(label='biogas_network', carrier='biogas')
```
- Material flow with time-varying penalties:
+ Heat network with penalty for imbalances:
```python
- material_hub = Bus(
- label='material_processing_hub',
- imbalance_penalty_per_flow_hour=waste_disposal_costs, # Time series
+ heat_bus = Bus(
+ label='district_heating',
+ carrier='heat',
+ imbalance_penalty_per_flow_hour=1000,
)
```
@@ -245,6 +249,7 @@ class Bus(Element):
def __init__(
self,
label: str,
+ carrier: str | None = None,
imbalance_penalty_per_flow_hour: Numeric_TPS | None = None,
meta_data: dict | None = None,
**kwargs,
@@ -254,6 +259,7 @@ def __init__(
kwargs, 'excess_penalty_per_flow_hour', 'imbalance_penalty_per_flow_hour', imbalance_penalty_per_flow_hour
)
self._validate_kwargs(kwargs)
+ self.carrier = carrier.lower() if carrier else None # Store as lowercase string
self.imbalance_penalty_per_flow_hour = imbalance_penalty_per_flow_hour
self.inputs: list[Flow] = []
self.outputs: list[Flow] = []
diff --git a/flixopt/flow_system.py b/flixopt/flow_system.py
index 5fda024f7..ee1a10261 100644
--- a/flixopt/flow_system.py
+++ b/flixopt/flow_system.py
@@ -4,6 +4,7 @@
from __future__ import annotations
+import json
import logging
import pathlib
import warnings
@@ -39,6 +40,8 @@
from .solvers import _Solver
from .types import Effect_TPS, Numeric_S, Numeric_TPS, NumericOrBool
+from .carrier import Carrier, CarrierContainer
+
logger = logging.getLogger('flixopt')
@@ -217,6 +220,9 @@ def __init__(
# Statistics accessor cache - lazily initialized, invalidated on new solution
self._statistics: StatisticsAccessor | None = None
+ # Carrier container - local carriers override CONFIG.Carriers
+ self._carriers: CarrierContainer = CarrierContainer()
+
# Use properties to validate and store scenario dimension settings
self.scenario_independent_sizes = scenario_independent_sizes
self.scenario_independent_flow_rates = scenario_independent_flow_rates
@@ -578,6 +584,14 @@ def to_dataset(self) -> xr.Dataset:
else:
ds.attrs['has_solution'] = False
+ # Include carriers if any are registered
+ if self._carriers:
+ carriers_structure = {}
+ for name, carrier in self._carriers.items():
+ carrier_ref, _ = carrier._create_reference_structure()
+ carriers_structure[name] = carrier_ref
+ ds.attrs['carriers'] = json.dumps(carriers_structure)
+
return ds
@classmethod
@@ -661,6 +675,13 @@ def from_dataset(cls, ds: xr.Dataset) -> FlowSystem:
solution_ds = solution_ds.rename({'solution_time': 'time'})
flow_system.solution = solution_ds
+ # Restore carriers if present
+ if 'carriers' in reference_structure:
+ carriers_structure = json.loads(reference_structure['carriers'])
+ for carrier_data in carriers_structure.values():
+ carrier = cls._resolve_reference_structure(carrier_data, {})
+ flow_system._carriers.add(carrier)
+
return flow_system
def to_netcdf(self, path: str | pathlib.Path, compression: int = 0, overwrite: bool = True):
@@ -813,6 +834,8 @@ def connect_and_transform(self):
return
self._connect_network()
+ self._register_missing_carriers()
+ self._assign_element_colors()
for element in chain(self.components.values(), self.effects.values(), self.buses.values()):
element.transform_data()
@@ -821,6 +844,40 @@ def connect_and_transform(self):
self._connected_and_transformed = True
+ def _register_missing_carriers(self) -> None:
+ """Auto-register carriers from CONFIG for buses that reference unregistered carriers."""
+ for bus in self.buses.values():
+ if bus.carrier and bus.carrier not in self._carriers:
+ # Try to get from CONFIG defaults
+ default_carrier = getattr(CONFIG.Carriers, bus.carrier, None)
+ if default_carrier is not None:
+ self._carriers[bus.carrier] = default_carrier
+ logger.debug(f"Auto-registered carrier '{bus.carrier}' from CONFIG")
+
+ def _assign_element_colors(self) -> None:
+ """Auto-assign colors to elements that don't have explicit colors set.
+
+ Components and buses without explicit colors are assigned colors from the
+ default qualitative colorscale. This ensures zero-config color support
+ while still allowing users to override with explicit colors.
+ """
+ from .color_processing import process_colors
+
+ # Collect elements without colors (components only - buses use carrier colors)
+ elements_without_colors = [comp.label for comp in self.components.values() if comp.color is None]
+
+ if not elements_without_colors:
+ return
+
+ # Generate colors from the default colorscale
+ colorscale = CONFIG.Plotting.default_qualitative_colorscale
+ color_mapping = process_colors(colorscale, elements_without_colors)
+
+ # Assign colors to elements
+ for label, color in color_mapping.items():
+ self.components[label].color = color
+ logger.debug(f"Auto-assigned color '{color}' to component '{label}'")
+
def add_elements(self, *elements: Element) -> None:
"""
Add Components(Storages, Boilers, Heatpumps, ...), Buses or Effects to the FlowSystem
@@ -859,6 +916,84 @@ def add_elements(self, *elements: Element) -> None:
element_type = type(new_element).__name__
logger.info(f'Registered new {element_type}: {new_element.label_full}')
+ def add_carriers(self, *carriers: Carrier) -> None:
+ """Register a custom carrier for this FlowSystem.
+
+ Custom carriers registered on the FlowSystem take precedence over
+ CONFIG.Carriers defaults when resolving colors and units for buses.
+
+ Args:
+ carrier: A Carrier object defining the carrier properties.
+
+ Examples:
+ ```python
+ import flixopt as fx
+
+ fs = fx.FlowSystem(timesteps)
+
+ # Define and register custom carriers
+ biogas = fx.Carrier('biogas', '#228B22', 'kW', 'Biogas fuel')
+ fs.add_carrier(biogas)
+
+ # Now buses can reference this carrier by name
+ bus = fx.Bus('BioGasNetwork', carrier='biogas')
+ fs.add_elements(bus)
+
+ # The carrier color will be used in plots automatically
+ ```
+ """
+ if self.connected_and_transformed:
+ warnings.warn(
+ 'You are adding a carrier to an already connected FlowSystem. This is not recommended (But it works).',
+ stacklevel=2,
+ )
+ self._connected_and_transformed = False
+
+ for carrier in list(carriers):
+ if not isinstance(carrier, Carrier):
+ raise TypeError(f'Expected Carrier object, got {type(carrier)}')
+ self._carriers.add(carrier)
+ logger.debug(f'Adding carrier {carrier} to FlowSystem')
+
+ def get_carrier(self, label: str) -> Carrier | None:
+ """Get the carrier for a bus or flow.
+
+ Args:
+ label: Bus label (e.g., 'Fernwärme') or flow label (e.g., 'Boiler(Q_th)').
+
+ Returns:
+ Carrier or None if not found.
+
+ Note:
+ To access a carrier directly by name, use ``flow_system.carriers['electricity']``.
+
+ Raises:
+ RuntimeError: If FlowSystem is not connected_and_transformed.
+ """
+ if not self.connected_and_transformed:
+ raise RuntimeError(
+ 'FlowSystem is not connected_and_transformed. Call FlowSystem.connect_and_transform() first.'
+ )
+
+ # Try as bus label
+ bus = self.buses.get(label)
+ if bus and bus.carrier:
+ return self._carriers.get(bus.carrier.lower())
+
+ # Try as flow label
+ flow = self.flows.get(label)
+ if flow and flow.bus:
+ bus = self.buses.get(flow.bus)
+ if bus and bus.carrier:
+ return self._carriers.get(bus.carrier.lower())
+
+ return None
+
+ @property
+ def carriers(self) -> CarrierContainer:
+ """Carriers registered on this FlowSystem."""
+ return self._carriers
+
def create_model(self, normalize_weights: bool = True) -> FlowSystemModel:
"""
Create a linopy model from the FlowSystem.
diff --git a/flixopt/statistics_accessor.py b/flixopt/statistics_accessor.py
index 9f6bb01be..2927c42a6 100644
--- a/flixopt/statistics_accessor.py
+++ b/flixopt/statistics_accessor.py
@@ -738,6 +738,77 @@ def __init__(self, statistics: StatisticsAccessor) -> None:
self._stats = statistics
self._fs = statistics._fs
+ def _get_color_map_for_balance(self, node: str, flow_labels: list[str]) -> dict[str, str]:
+ """Build color map for balance plot.
+
+ - Bus balance: colors from component.color
+ - Component balance: colors from flow's carrier
+
+ Raises:
+ RuntimeError: If FlowSystem is not connected_and_transformed.
+ """
+ if not self._fs.connected_and_transformed:
+ raise RuntimeError(
+ 'FlowSystem is not connected_and_transformed. Call FlowSystem.connect_and_transform() first.'
+ )
+
+ is_bus = node in self._fs.buses
+ color_map = {}
+ uncolored = []
+
+ for label in flow_labels:
+ if is_bus:
+ color = self._fs.components[self._fs.flows[label].component].color
+ else:
+ carrier = self._fs.get_carrier(label) # get_carrier accepts flow labels
+ color = carrier.color if carrier else None
+
+ if color:
+ color_map[label] = color
+ else:
+ uncolored.append(label)
+
+ if uncolored:
+ color_map.update(process_colors(CONFIG.Plotting.default_qualitative_colorscale, uncolored))
+
+ return color_map
+
+ def _resolve_variable_names(self, variables: list[str], solution: xr.Dataset) -> list[str]:
+ """Resolve flow labels to variable names with fallback.
+
+ For each variable:
+ 1. First check if it exists in the dataset as-is
+ 2. If not found and doesn't contain '|', try adding '|flow_rate' suffix
+ 3. If still not found, try '|charge_state' suffix (for storages)
+
+ Args:
+ variables: List of flow labels or variable names.
+ solution: The solution dataset to check variable existence.
+
+ Returns:
+ List of resolved variable names.
+ """
+ resolved = []
+ for var in variables:
+ if var in solution:
+ # Variable exists as-is, use it directly
+ resolved.append(var)
+ elif '|' not in var:
+ # Not found and no '|', try common suffixes
+ flow_rate_var = f'{var}|flow_rate'
+ charge_state_var = f'{var}|charge_state'
+ if flow_rate_var in solution:
+ resolved.append(flow_rate_var)
+ elif charge_state_var in solution:
+ resolved.append(charge_state_var)
+ else:
+ # Let it fail with the original name for clear error message
+ resolved.append(var)
+ else:
+ # Contains '|' but not in solution - let it fail with original name
+ resolved.append(var)
+ return resolved
+
def balance(
self,
node: str,
@@ -801,6 +872,10 @@ def balance(
ds = _apply_selection(ds, select)
actual_facet_col, actual_facet_row = _resolve_facets(ds, facet_col, facet_row)
+ # Build color map from Element.color attributes if no colors specified
+ if colors is None:
+ colors = self._get_color_map_for_balance(node, list(ds.data_vars))
+
fig = _create_stacked_bar(
ds,
colors=colors,
@@ -836,7 +911,9 @@ def heatmap(
reshaping is skipped and variables are shown on the y-axis with time on x-axis.
Args:
- variables: Variable name(s) from solution.
+ variables: Flow label(s) or variable name(s). Flow labels like 'Boiler(Q_th)'
+ are automatically resolved to 'Boiler(Q_th)|flow_rate'. Full variable
+ names like 'Storage|charge_state' are used as-is.
select: xarray-style selection, e.g. {'scenario': 'Base Case'}.
reshape: Time reshape frequencies as (outer, inner), e.g. ('D', 'h') for
days × hours. Set to None to disable reshaping.
@@ -856,7 +933,10 @@ def heatmap(
if isinstance(variables, str):
variables = [variables]
- ds = solution[variables]
+ # Resolve flow labels to variable names
+ resolved_variables = self._resolve_variable_names(variables, solution)
+
+ ds = solution[resolved_variables]
ds = _apply_selection(ds, select)
# Stack variables into single DataArray
@@ -972,8 +1052,8 @@ def flows(
for flow in self._fs.flows.values():
# Get bus label (could be string or Bus object)
- bus_label = flow.bus if isinstance(flow.bus, str) else flow.bus.label
- comp_label = flow.component.label if hasattr(flow.component, 'label') else str(flow.component)
+ bus_label = flow.bus
+ comp_label = flow.component.label_full
# start/end filtering based on flow direction
if flow.is_input_in_component:
@@ -1075,8 +1155,8 @@ def sankey(
# Determine source/target based on flow direction
# is_input_in_component: True means bus -> component, False means component -> bus
- bus_label = flow.bus if isinstance(flow.bus, str) else flow.bus.label
- comp_label = flow.component.label if hasattr(flow.component, 'label') else str(flow.component)
+ bus_label = flow.bus
+ comp_label = flow.component.label_full
if flow.is_input_in_component:
source = bus_label
@@ -1203,8 +1283,10 @@ def duration_curve(
"""Plot load duration curves (sorted time series).
Args:
- variables: Flow label(s) to plot (e.g., 'Boiler(Q_th)').
- Uses flow_rates from statistics.
+ variables: Flow label(s) or variable name(s). Flow labels like 'Boiler(Q_th)'
+ are looked up in flow_rates. Full variable names like 'Boiler(Q_th)|flow_rate'
+ are stripped to their flow label. Other variables (e.g., 'Storage|charge_state')
+ are looked up in the solution directly.
select: xarray-style selection.
normalize: If True, normalize x-axis to 0-100%.
colors: Color specification (colorscale name, color list, or label-to-color dict).
@@ -1215,13 +1297,36 @@ def duration_curve(
Returns:
PlotResult with sorted duration curve data.
"""
- self._stats._require_solution()
+ solution = self._stats._require_solution()
if isinstance(variables, str):
variables = [variables]
- # Use flow_rates from statistics (already has clean labels without |flow_rate suffix)
- ds = self._stats.flow_rates[variables]
+ # Normalize variable names: strip |flow_rate suffix for flow_rates lookup
+ flow_rates = self._stats.flow_rates
+ normalized_vars = []
+ for var in variables:
+ # Strip |flow_rate suffix if present
+ if var.endswith('|flow_rate'):
+ var = var[: -len('|flow_rate')]
+ normalized_vars.append(var)
+
+ # Try to get from flow_rates first, fall back to solution for non-flow variables
+ ds_parts = []
+ for var in normalized_vars:
+ if var in flow_rates:
+ ds_parts.append(flow_rates[[var]])
+ elif var in solution:
+ ds_parts.append(solution[[var]])
+ else:
+ # Try with |flow_rate suffix as last resort
+ flow_rate_var = f'{var}|flow_rate'
+ if flow_rate_var in solution:
+ ds_parts.append(solution[[flow_rate_var]].rename({flow_rate_var: var}))
+ else:
+ raise KeyError(f"Variable '{var}' not found in flow_rates or solution")
+
+ ds = xr.merge(ds_parts)
ds = _apply_selection(ds, select)
if 'time' not in ds.dims:
diff --git a/flixopt/structure.py b/flixopt/structure.py
index 8bec197bc..d00066683 100644
--- a/flixopt/structure.py
+++ b/flixopt/structure.py
@@ -1015,6 +1015,7 @@ def __init__(
self,
label: str,
meta_data: dict | None = None,
+ color: str | None = None,
_variable_names: list[str] | None = None,
_constraint_names: list[str] | None = None,
):
@@ -1022,11 +1023,13 @@ def __init__(
Args:
label: The label of the element
meta_data: used to store more information about the Element. Is not used internally, but saved in the results. Only use python native types.
+ color: Optional color for visualizations (e.g., '#FF6B6B'). If not provided, a color will be automatically assigned during FlowSystem.connect_and_transform().
_variable_names: Internal. Variable names for this element (populated after modeling).
_constraint_names: Internal. Constraint names for this element (populated after modeling).
"""
self.label = Element._valid_label(label)
self.meta_data = meta_data if meta_data is not None else {}
+ self.color = color
self.submodel = None
self._flow_system: FlowSystem | None = None
# Variable/constraint names - populated after modeling, serialized for results
@@ -1127,16 +1130,20 @@ def __init__(
elements: list[T] | dict[str, T] | None = None,
element_type_name: str = 'elements',
truncate_repr: int | None = None,
+ item_name: str | None = None,
):
"""
Args:
elements: Initial elements to add (list or dict)
element_type_name: Name for display (e.g., 'components', 'buses')
truncate_repr: Maximum number of items to show in repr. If None, show all items. Default: None
+ item_name: Singular name for error messages (e.g., 'Component', 'Carrier').
+ If None, inferred from first added item's class name.
"""
super().__init__()
self._element_type_name = element_type_name
self._truncate_repr = truncate_repr
+ self._item_name = item_name
if elements is not None:
if isinstance(elements, dict):
@@ -1158,13 +1165,28 @@ def _get_label(self, element: T) -> str:
"""
raise NotImplementedError('Subclasses must implement _get_label()')
+ def _get_item_name(self) -> str:
+ """Get the singular item name for error messages.
+
+ Returns the explicitly set item_name, or infers from the first item's class name.
+ Falls back to 'Item' if container is empty and no name was set.
+ """
+ if self._item_name is not None:
+ return self._item_name
+ # Infer from first item's class name
+ if self:
+ first_item = next(iter(self.values()))
+ return first_item.__class__.__name__
+ return 'Item'
+
def add(self, element: T) -> None:
"""Add an element to the container."""
label = self._get_label(element)
if label in self:
+ item_name = element.__class__.__name__
raise ValueError(
- f'Element with label "{label}" already exists in {self._element_type_name}. '
- f'Each element must have a unique label.'
+ f'{item_name} with label "{label}" already exists in {self._element_type_name}. '
+ f'Each {item_name.lower()} must have a unique label.'
)
self[label] = element
@@ -1195,8 +1217,9 @@ def __getitem__(self, label: str) -> T:
return super().__getitem__(label)
except KeyError:
# Provide helpful error with close matches suggestions
+ item_name = self._get_item_name()
suggestions = get_close_matches(label, self.keys(), n=3, cutoff=0.6)
- error_msg = f'Element "{label}" not found in {self._element_type_name}.'
+ error_msg = f'{item_name} "{label}" not found in {self._element_type_name}.'
if suggestions:
error_msg += f' Did you mean: {", ".join(suggestions)}?'
else: