diff --git a/CHANGELOG.md b/CHANGELOG.md index 6635b786b..01165c249 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/user-guide/results-plotting.md b/docs/user-guide/results-plotting.md index 91a88cd80..9bab035ec 100644 --- a/docs/user-guide/results-plotting.md +++ b/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 diff --git a/flixopt/components.py b/flixopt/components.py index eb64d9879..20b4b6ae5 100644 --- a/flixopt/components.py +++ b/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,6 +1707,7 @@ 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, @@ -1709,6 +1715,7 @@ def __init__( 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,6 +1802,7 @@ 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__( @@ -1802,6 +1810,7 @@ def __init__( 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, ) diff --git a/flixopt/linear_converters.py b/flixopt/linear_converters.py index 8326fe6c5..c5c9afd4d 100644 --- a/flixopt/linear_converters.py +++ b/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 diff --git a/flixopt/statistics_accessor.py b/flixopt/statistics_accessor.py index 0092d4989..21d0619d1 100644 --- a/flixopt/statistics_accessor.py +++ b/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 diff --git a/flixopt/topology_accessor.py b/flixopt/topology_accessor.py index eb5f05876..cbcb3fd6e 100644 --- a/flixopt/topology_accessor.py +++ b/flixopt/topology_accessor.py @@ -239,6 +239,103 @@ def effect_units(self) -> dict[str, str]: self._effect_units = {effect.label: effect.unit or '' for effect in self._fs.effects.values()} return self._effect_units + def _invalidate_color_caches(self) -> None: + """Reset all color caches so they are rebuilt on next access.""" + self._carrier_colors = None + self._component_colors = None + self._bus_colors = None + + def set_component_color(self, label: str, color: str) -> None: + """Set the color for a single component. + + Args: + label: Component label. + color: Color string (hex like '#FF0000', named like 'red', etc.). + + Raises: + KeyError: If component with given label doesn't exist. + + Examples: + >>> flow_system.topology.set_component_color('Boiler', '#D35400') + >>> flow_system.topology.set_component_color('CHP', 'darkred') + """ + if label not in self._fs.components: + raise KeyError(f"Component '{label}' not found. Available: {list(self._fs.components.keys())}") + self._fs.components[label].color = color + self._invalidate_color_caches() + + def set_component_colors( + self, + colors: dict[str, str | list[str]] | str, + overwrite: bool = True, + ) -> dict[str, str]: + """Set colors for multiple components at once. + + Args: + colors: Color configuration: + - ``str``: Colorscale name for all components (e.g., ``'turbo'``) + - ``dict``: Component-to-color mapping (``{'Boiler': 'red'}``) or + colorscale-to-components (``{'Blues': ['Wind1', 'Wind2']}``) + overwrite: If False, skip components that already have colors. + + Returns: + Mapping of colors that were actually assigned. + + Examples: + >>> flow_system.topology.set_component_colors('turbo') + >>> flow_system.topology.set_component_colors({'Boiler': 'red', 'CHP': '#0000FF'}) + >>> flow_system.topology.set_component_colors({'Blues': ['Wind1', 'Wind2']}) + >>> flow_system.topology.set_component_colors('turbo', overwrite=False) + """ + 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): + # Colorscale -> component 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: + # Direct assignment + 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 + + def set_carrier_color(self, carrier: str, color: str) -> None: + """Set the color for a carrier. + + This affects bus colors derived from this carrier. + + Args: + carrier: Carrier name (case-insensitive). + color: Color string (hex like '#FF0000', named like 'red', etc.). + + Examples: + >>> flow_system.topology.set_carrier_color('electricity', '#FECB52') + >>> flow_system.topology.set_carrier_color('heat', 'firebrick') + """ + carrier_obj = self._fs.get_carrier(carrier) + if carrier_obj is None: + raise KeyError(f"Carrier '{carrier}' not found.") + carrier_obj.color = color + self._invalidate_color_caches() + def infos(self) -> tuple[dict[str, dict[str, str]], dict[str, dict[str, str]]]: """ Get network topology information as dictionaries.