Add color parameter to components and bulk color assignment via topology accessor (#585)

* Add color attribute to remaining classes

* 1. Added color parameter to all component classes (flixopt/components.py, flixopt/linear_converters.py):
  - LinearConverter, Storage, Transmission, Source, Sink
  - Boiler, Power2Heat, HeatPump, CoolingTower, CHP, HeatPumpWithSource

  2. Added bulk color assignment methods to TopologyAccessor (flixopt/topology_accessor.py):

  # Single component
  flow_system.topology.set_component_color('Boiler', '#D35400')

  # Multiple components - direct assignment
  flow_system.topology.set_component_colors({
      'Boiler': '#D35400',
      'CHP': 'darkred',
  })

  # Multiple components - colorscale for groups
  flow_system.topology.set_component_colors({
      'Oranges': ['Solar1', 'Solar2'],
      'Blues': ['Wind1', 'Wind2'],
  })

  # All components with colorscale
  flow_system.topology.set_component_colors('turbo')

  # Carrier color (affects bus colors)
  flow_system.topology.set_carrier_color('electricity', '#FECB52')

  3. Cache invalidation - _invalidate_color_caches() resets cached color dicts when colors change.

  4. Updated documentation (docs/user-guide/results-plotting.md) - Fixed the incorrect flow_system.colors references to use flow_system.topology.

  Read-only accessors still available:
  flow_system.topology.component_colors  # dict[str, str]
  flow_system.topology.carrier_colors    # dict[str, str]
  flow_system.topology.bus_colors        # dict[str, str]

* The overwrite=False parameter works correctly.

  Summary:

  # Set colors only for components that don't have one yet
  flow_system.topology.set_component_colors('turbo', overwrite=False)

  # With dict - skips components that already have colors
  flow_system.topology.set_component_colors({
      'Boiler': 'red',    # Skipped if Boiler already has a color
      'CHP': 'blue',      # Skipped if CHP already has a color
  }, overwrite=False)

  Returns only the colors that were actually assigned, so you can see what changed.

* ⏺ Done. Reduced from ~50 lines to ~30 lines while keeping the same functionality:

  def set_component_colors(
      self,
      colors: dict[str, str | list[str]] | str,
      overwrite: bool = True,
  ) -> dict[str, str]:
      components = self._fs.components

      # Normalize to {label: color} mapping
      if isinstance(colors, str):
          color_map = process_colors(colors, list(components.keys()))
      else:
          color_map = {}
          for key, value in colors.items():
              if isinstance(value, list):
                  missing = [c for c in value if c not in components]
                  if missing:
                      raise KeyError(f"Components not found: {missing}")
                  color_map.update(process_colors(key, value))
              else:
                  if key not in components:
                      raise KeyError(f"Component '{key}' not found")
                  color_map[key] = value

      # Apply colors (respecting overwrite flag)
      result = {}
      for label, color in color_map.items():
          if overwrite or components[label].color is None:
              components[label].color = color
              result[label] = color

      self._invalidate_color_caches()
      return result

  Key improvements:
  - Normalize all inputs to a {label: color} dict first
  - Single pass to apply colors
  - Cleaner error messages
  - Shorter docstring with inline examples

* Update CHANGELOG.md

* Add colors to snakey/flows plots

* Add missing color parameter

Author: FBumann

CHANGELOG.md

@@ -243,6 +243,21 @@ comp.diff('baseline')  # vs named case
 - Mirrors all `StatisticsPlotAccessor` methods (`balance`, `carrier_balance`, `flows`, `sizes`, `duration_curve`, `effects`, `charge_states`, `heatmap`, `storage`)
 - Existing plotting infrastructure automatically handles faceting by `'case'`
 
+#### Component Color Parameter (#585)
+
+All component classes now accept a `color` parameter for visualization customization:
+
+```python
+# Set color at instantiation
+boiler = fx.Boiler('Boiler', ..., color='#D35400')
+storage = fx.Storage('Battery', ..., color='green')
+
+# Bulk assignment via topology accessor
+flow_system.topology.set_component_colors({'Boiler': 'red', 'CHP': 'blue'})
+flow_system.topology.set_component_colors({'Oranges': ['Solar1', 'Solar2']})  # Colorscale
+flow_system.topology.set_component_colors('turbo', overwrite=False)  # Only unset colors
+```
+
 ### 💥 Breaking Changes
 
 #### tsam v3 Migration

docs/user-guide/results-plotting.md

@@ -394,24 +394,38 @@ flow_system.carriers  # CarrierContainer with locally registered carriers
 flow_system.get_carrier('biogas')  # Returns Carrier object
 ```
 
-### Color Accessor
+### Setting Colors via Topology Accessor
 
-The `flow_system.colors` accessor provides centralized color configuration:
+The `flow_system.topology` accessor provides methods for bulk color assignment:
 
 ```python
 # Configure colors for components
-flow_system.colors.setup({
+flow_system.topology.set_component_colors({
     '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')
+flow_system.topology.set_component_color('Boiler', '#D35400')
+flow_system.topology.set_carrier_color('electricity', '#FECB52')
+
+# Apply a colorscale to groups of components
+flow_system.topology.set_component_colors({
+    'Oranges': ['Solar1', 'Solar2', 'Solar3'],  # Colorscale for solar
+    'Blues': ['Wind1', 'Wind2'],                 # Colorscale for wind
+    'Battery': 'green',                          # Direct color assignment
+})
+
+# Apply a colorscale to all components
+flow_system.topology.set_component_colors('turbo')
+```
 
-# Load from file
-flow_system.colors.setup('colors.json')  # or .yaml
+You can also set colors at component instantiation:
+
+```python
+boiler = fx.Boiler('Boiler', ..., color='#D35400')
+storage = fx.Storage('Battery', ..., color='green')
 ```
 
 ### Context-Aware Coloring
@@ -435,22 +449,21 @@ flow_system.statistics.plot.balance('CHP')
 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`)
+2. **Component colors** set via `flow_system.topology.set_component_colors()` or at instantiation
+3. **Carrier colors** from FlowSystem or CONFIG.Carriers (for buses)
+4. **Default colorscale** (controlled by `CONFIG.Plotting.default_qualitative_colorscale`)
 
 ### Persistence
 
 Color configurations are automatically saved with the FlowSystem:
 
 ```python
-# Colors are persisted
+# Colors are persisted (component.color attributes)
 flow_system.to_netcdf('my_system.nc')
 
 # And restored
 loaded = fx.FlowSystem.from_netcdf('my_system.nc')
-loaded.colors  # Configuration restored
+loaded.topology.component_colors  # Colors are restored
 ```
 
 ### Display Control

flixopt/components.py

@@ -172,8 +172,9 @@ def __init__(
         conversion_factors: list[dict[str, Numeric_TPS]] | None = None,
         piecewise_conversion: PiecewiseConversion | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
-        super().__init__(label, inputs, outputs, status_parameters, meta_data=meta_data)
+        super().__init__(label, inputs, outputs, status_parameters, meta_data=meta_data, color=color)
         self.conversion_factors = conversion_factors or []
         self.piecewise_conversion = piecewise_conversion
 
@@ -417,6 +418,7 @@ def __init__(
         balanced: bool = False,
         cluster_mode: Literal['independent', 'cyclic', 'intercluster', 'intercluster_cyclic'] = 'intercluster_cyclic',
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # TODO: fixed_relative_chargeState implementieren
         super().__init__(
@@ -425,6 +427,7 @@ def __init__(
             outputs=[discharging],
             prevent_simultaneous_flows=[charging, discharging] if prevent_simultaneous_charge_and_discharge else None,
             meta_data=meta_data,
+            color=color,
         )
 
         self.charging = charging
@@ -744,6 +747,7 @@ def __init__(
         prevent_simultaneous_flows_in_both_directions: bool = True,
         balanced: bool = False,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         super().__init__(
             label,
@@ -754,6 +758,7 @@ def __init__(
             if in2 is None or prevent_simultaneous_flows_in_both_directions is False
             else [in1, in2],
             meta_data=meta_data,
+            color=color,
         )
         self.in1 = in1
         self.out1 = out1
@@ -1702,13 +1707,15 @@ def __init__(
         outputs: list[Flow] | None = None,
         prevent_simultaneous_flow_rates: bool = True,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         super().__init__(
             label,
             inputs=inputs,
             outputs=outputs,
             prevent_simultaneous_flows=(inputs or []) + (outputs or []) if prevent_simultaneous_flow_rates else None,
             meta_data=meta_data,
+            color=color,
         )
         self.prevent_simultaneous_flow_rates = prevent_simultaneous_flow_rates
 
@@ -1795,13 +1802,15 @@ def __init__(
         outputs: list[Flow] | None = None,
         meta_data: dict | None = None,
         prevent_simultaneous_flow_rates: bool = False,
+        color: str | None = None,
     ):
         self.prevent_simultaneous_flow_rates = prevent_simultaneous_flow_rates
         super().__init__(
             label,
             outputs=outputs,
             meta_data=meta_data,
             prevent_simultaneous_flows=outputs if prevent_simultaneous_flow_rates else None,
+            color=color,
         )
 
 
@@ -1888,6 +1897,7 @@ def __init__(
         inputs: list[Flow] | None = None,
         meta_data: dict | None = None,
         prevent_simultaneous_flow_rates: bool = False,
+        color: str | None = None,
     ):
         """Initialize a Sink (consumes flow from the system).
 
@@ -1897,6 +1907,7 @@ def __init__(
             meta_data: Arbitrary metadata attached to the element.
             prevent_simultaneous_flow_rates: If True, prevents simultaneous nonzero flow rates
                 across the element's inputs by wiring that restriction into the base Component setup.
+            color: Optional color for visualizations.
         """
 
         self.prevent_simultaneous_flow_rates = prevent_simultaneous_flow_rates
@@ -1905,4 +1916,5 @@ def __init__(
             inputs=inputs,
             meta_data=meta_data,
             prevent_simultaneous_flows=inputs if prevent_simultaneous_flow_rates else None,
+            color=color,
         )

flixopt/linear_converters.py

@@ -81,6 +81,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if fuel_flow is None:
@@ -96,6 +97,7 @@ def __init__(
             outputs=[thermal_flow],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
         self.fuel_flow = fuel_flow
         self.thermal_flow = thermal_flow
@@ -176,6 +178,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if electrical_flow is None:
@@ -191,6 +194,7 @@ def __init__(
             outputs=[thermal_flow],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
 
         self.electrical_flow = electrical_flow
@@ -271,6 +275,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if electrical_flow is None:
@@ -287,6 +292,7 @@ def __init__(
             conversion_factors=[],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
         self.electrical_flow = electrical_flow
         self.thermal_flow = thermal_flow
@@ -368,6 +374,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if electrical_flow is None:
@@ -381,6 +388,7 @@ def __init__(
             outputs=[],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
 
         self.electrical_flow = electrical_flow
@@ -472,6 +480,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if fuel_flow is None:
@@ -492,6 +501,7 @@ def __init__(
             conversion_factors=[{}, {}],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
 
         self.fuel_flow = fuel_flow
@@ -602,6 +612,7 @@ def __init__(
         thermal_flow: Flow | None = None,
         status_parameters: StatusParameters | None = None,
         meta_data: dict | None = None,
+        color: str | None = None,
     ):
         # Validate required parameters
         if electrical_flow is None:
@@ -619,6 +630,7 @@ def __init__(
             outputs=[thermal_flow],
             status_parameters=status_parameters,
             meta_data=meta_data,
+            color=color,
         )
         self.electrical_flow = electrical_flow
         self.heat_source_flow = heat_source_flow

flixopt/statistics_accessor.py

@@ -1051,20 +1051,22 @@ def _create_figure(
         return fig
 
     def _get_node_colors(self, node_list: list[str], colors: ColorType | None) -> list[str]:
-        """Get colors for nodes: buses use cached bus_colors, components use process_colors."""
-        # Get fallback colors from process_colors
-        fallback_colors = process_colors(colors, node_list)
-
-        # Use cached bus colors for efficiency
+        """Get colors for nodes: buses use bus_colors, components use component_colors."""
+        # Get cached colors
         bus_colors = self._stats.bus_colors
+        component_colors = self._stats.component_colors
+
+        # Get fallback colors for nodes without explicit colors
+        uncolored = [n for n in node_list if n not in bus_colors and n not in component_colors]
+        fallback_colors = process_colors(colors, uncolored) if uncolored else {}
 
         node_colors = []
         for node in node_list:
-            # Check if node is a bus with a cached color
             if node in bus_colors:
                 node_colors.append(bus_colors[node])
+            elif node in component_colors:
+                node_colors.append(component_colors[node])
             else:
-                # Fall back to process_colors
                 node_colors.append(fallback_colors[node])
 
         return node_colors