From d394a30f1f358209ae4696f70f4b1cb62db1beb9 Mon Sep 17 00:00:00 2001 From: Francesco Witte Date: Fri, 26 Jun 2026 07:51:35 +0200 Subject: [PATCH 01/11] Add plotting dependencies --- pyproject.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 357e110..1f4bb1d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,8 @@ requires-python = ">=3.11" dependencies = [ "pandas", "numpy", + "plotly>=6.8", + "matplotlib>=3.10", ] license = {text = "MIT"} From 19a24819fb30d518f71ab5704526d24703ad608b Mon Sep 17 00:00:00 2001 From: Francesco Witte Date: Fri, 26 Jun 2026 07:52:17 +0200 Subject: [PATCH 02/11] Boilerplate delegation to separate module for visualizations --- src/exerpy/analyses.py | 209 +++++++++++++++++------------------------ 1 file changed, 84 insertions(+), 125 deletions(-) diff --git a/src/exerpy/analyses.py b/src/exerpy/analyses.py index 31b233f..58e7ce5 100755 --- a/src/exerpy/analyses.py +++ b/src/exerpy/analyses.py @@ -1,7 +1,6 @@ import json import os -import matplotlib.pyplot as plt import numpy as np import pandas as pd from tabulate import tabulate @@ -577,158 +576,118 @@ def convert(x, factor): return df_component_results, df_material_connection_results, df_non_material_connection_results - def plot_exergy_waterfall(self, title=None, figsize=(12, 10), exclude_components=None, show_plot=True): + def plot_sankey( + self, + mode: int = 1, + collapse_passthroughs: bool | list[str] = True, + groups: dict | None = None, + node_colors: dict | None = None, + title: str | None = None, + output_path: str | None = None, + ): """ - Create an exergy destruction waterfall diagram. + Create an interactive Sankey diagram of the exergy flows. - This method visualizes the exergy flow through the system as a waterfall chart, - showing how exergy is destroyed in each component from the exergetic fuel (100%) - down to the exergetic product and losses. + Parameters + ---------- + mode : {1, 2, 3} + 1 – total exergy E per link (default) + 2 – split material links into E_PH + E_CH (requires chemical_exergy_enabled) + 3 – split material links into E_T + E_M + E_CH (requires split_physical_exergy) + collapse_passthroughs : bool or list[str] + True → collapse CycleCloser nodes into their neighbours (default) + False → show every component node + list → collapse only the listed component type names + groups : dict[str, list[str]], optional + Visual grouping: ``{"Group": ["comp1", "comp2"]}`` hides internal connections + and aggregates boundary-crossing links. + node_colors : dict[str, str], optional + Per-component colour overrides keyed by component name, e.g. + ``{"CC": "#C62828", "GT": "#388E3C"}``. Components not listed use + the default node colour. + title : str, optional + Diagram title. + output_path : str, optional + If given, export the diagram to this HTML file path. + + Returns + ------- + plotly.graph_objects.Figure + """ + from .visualization.sankey import SankeyBuilder + + builder = SankeyBuilder( + self, + mode=mode, + collapse_passthroughs=collapse_passthroughs, + groups=groups, + node_colors=node_colors, + ) + fig = builder.to_plotly(title=title) + if output_path is not None: + fig.write_html(output_path) + return fig + + def plot_exergy_waterfall(self, title=None, figsize=(12, 10), exclude_components=None, colors=None, show_plot=True): + """ + Create an exergy destruction waterfall diagram. Parameters ---------- title : str, optional - Title for the plot. If None, no title is displayed. + Title for the plot. figsize : tuple, optional Figure size as (width, height) in inches. Default is (12, 10). exclude_components : list, optional - List of component names to exclude from the diagram. - By default, all components with NaN E_F (Exergetic Fuel) are excluded, - as well as CycleCloser and PowerBus components. + Component names to exclude. Components with NaN E_F are always excluded. + colors : dict, optional + Override bar colors. Keys: "fuel", "destruction", "loss", "product". show_plot : bool, optional Whether to display the plot immediately. Default is True. Returns ------- fig : matplotlib.figure.Figure - The figure object containing the waterfall diagram. ax : matplotlib.axes.Axes - The axes object of the waterfall diagram. - - Raises - ------ - RuntimeError - If the exergy analysis has not been performed yet (analyse() not called). - - Notes - ----- - - The waterfall diagram displays exergy values as percentages of the total fuel exergy. - - Components are sorted by their exergy destruction rate (y [%]) in descending order. - - Each bar represents the remaining exergy after destruction in that component. - - Red bars indicate exergy destruction in components. - - Blue bar represents the initial exergetic fuel (100%). - - Green bar represents the final exergetic product. Examples -------- >>> analysis = ExergyAnalysis.from_tespy(network, Tamb=288.15, pamb=101325) # doctest: +SKIP >>> analysis.analyse(E_F={'inputs': ['fuel']}, E_P={'outputs': ['power']}) # doctest: +SKIP >>> fig, ax = analysis.plot_exergy_waterfall(title='Power Plant Exergy Waterfall') # doctest: +SKIP - >>> fig.savefig('exergy_waterfall.pdf') # doctest: +SKIP - - See Also - -------- - exergy_results : Display tabular exergy analysis results. - print_exergy_summary : Print a text summary of exergy analysis. """ - # Check if analysis has been performed - if not hasattr(self, "epsilon") or self.epsilon is None: - raise RuntimeError("Exergy analysis has not been performed yet. Please call analyse() first.") + from .visualization.waterfall import plot_exergy_waterfall - # Get component results without printing - df_component_results, _, _ = self.exergy_results(print_results=False) + return plot_exergy_waterfall(self, title=title, figsize=figsize, exclude_components=exclude_components, colors=colors, show_plot=show_plot) - # Default exclusions: empty list, but filter for valid E_F - if exclude_components is None: - exclude_components = [] + def plot_exergy_waterfall_plotly(self, title=None, exclude_components=None, colors=None, show_plot=True): + """ + Create an interactive exergy destruction waterfall diagram using Plotly. - # Get total values from df_component_results - total_row = df_component_results[df_component_results["Component"] == "TOT"].iloc[0] - epsilon_total = total_row["epsilon [%]"] - E_L_total = total_row["E_L [kW]"] - E_F_total = total_row["E_F [kW]"] - exergetic_loss_percent = (E_L_total / E_F_total) * 100 if E_F_total != 0 else 0 + Parameters + ---------- + title : str, optional + Title for the plot. + exclude_components : list, optional + Component names to exclude. Components with NaN E_F are always excluded. + colors : dict, optional + Override bar colors. Keys: "fuel", "destruction", "loss", "product". + show_plot : bool, optional + Whether to display the plot immediately. Default is True. - # Filter components (exclude TOT, components with NaN E_F, and specified components) - component_data = df_component_results[ - (df_component_results["Component"] != "TOT") - & (df_component_results["E_F [kW]"].notna()) - & (~df_component_results["Component"].isin(exclude_components)) - & (df_component_results["y [%]"].notna()) - ].copy() - - # Sort by y [%] in descending order - component_data = component_data.sort_values("y [%]", ascending=False) - - # Create bar values: Start at 100%, decrease by each component's y [%] - bar_values = [100.0] - current_value = 100.0 - for y in component_data["y [%]"]: - current_value -= y - bar_values.append(current_value) - bar_values.append(epsilon_total) # Final bar is the exergetic product - - # Create labels for spaces between bars - space_labels = ["Exergetic fuel"] + list(component_data["Component"]) + ["Exergetic loss", "Exergetic product"] - - # Create the figure - fig, ax = plt.subplots(figsize=figsize) - - # Number of bars and spaces - n_bars = len(bar_values) - - # Create horizontal bars at positions 0, 1, 2, ..., n_bars-1 - bar_positions = np.arange(n_bars) - bar_colors = ["#1565C0"] + ["#D32F2F"] * (n_bars - 2) + ["#2E7D32"] - # Blue for fuel, red for destruction, green for product - - for i, (pos, value, color) in enumerate(zip(bar_positions, bar_values, bar_colors, strict=False)): - ax.barh(pos, value, color=color, alpha=0.8, height=0.6) - # Add value label inside the bar on the right - ax.text( - value - 2, pos, f"{value:.2f}%", va="center", ha="right", fontsize=9, fontweight="bold", color="white" - ) + Returns + ------- + fig : plotly.graph_objects.Figure - # Add space labels between bars - # Space positions: between bars, so at 0.5, 1.5, 2.5, ..., and above/below - space_positions = [-0.5] + [i + 0.5 for i in range(n_bars - 1)] + [n_bars - 0.5] - - for i, (space_pos, label) in enumerate(zip(space_positions, space_labels, strict=False)): - if i == 0: # Exergetic fuel - above first bar - ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") - elif i == len(space_labels) - 2: # Exergetic loss - loss_label = f"{label} (-{exergetic_loss_percent:.2f}%)" - ax.text( - 2, space_pos, loss_label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic" - ) - elif i == len(space_labels) - 1: # Exergetic product - below last bar - ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") - else: # Component labels - component_idx = i - 1 - destruction_rate = component_data.iloc[component_idx]["y [%]"] - label_with_rate = f"{label} (-{destruction_rate:.2f}%)" - ax.text(2, space_pos, label_with_rate, va="center", ha="left", fontsize=10, fontweight="bold") - - # Customize plot - ax.set_yticks(bar_positions) - ax.set_yticklabels([""] * n_bars) # Empty labels since we have custom labels - ax.set_xlabel("Exergy [%]", fontsize=12, fontweight="bold") - - if title is not None: - ax.set_title(title, fontsize=14, fontweight="bold", pad=20) - - ax.set_xlim(0, 100) - ax.set_ylim(-1, n_bars) - ax.grid(axis="x", alpha=0.3, linestyle="--") - ax.axvline(x=0, color="black", linewidth=0.8) - ax.invert_yaxis() # Invert so highest bar is at top - - plt.tight_layout() - - if show_plot: - plt.show() - - return fig, ax + Examples + -------- + >>> analysis = ExergyAnalysis.from_tespy(network, Tamb=288.15, pamb=101325) # doctest: +SKIP + >>> analysis.analyse(E_F={'inputs': ['fuel']}, E_P={'outputs': ['power']}) # doctest: +SKIP + >>> fig = analysis.plot_exergy_waterfall_plotly(title='Power Plant Exergy Waterfall') # doctest: +SKIP + """ + from .visualization.waterfall import plot_exergy_waterfall_plotly + + return plot_exergy_waterfall_plotly(self, title=title, exclude_components=exclude_components, colors=colors, show_plot=show_plot) def print_exergy_summary(self): """ From 9e744aeb4f795e170bf0286a0d777d794e058e49 Mon Sep 17 00:00:00 2001 From: Francesco Witte Date: Fri, 26 Jun 2026 07:53:56 +0200 Subject: [PATCH 03/11] Transfer waterfall to own module and implement plotly variant --- src/exerpy/visualization/__init__.py | 3 + src/exerpy/visualization/waterfall.py | 274 ++++++++++++++++++++++++++ 2 files changed, 277 insertions(+) create mode 100644 src/exerpy/visualization/__init__.py create mode 100644 src/exerpy/visualization/waterfall.py diff --git a/src/exerpy/visualization/__init__.py b/src/exerpy/visualization/__init__.py new file mode 100644 index 0000000..57a34c8 --- /dev/null +++ b/src/exerpy/visualization/__init__.py @@ -0,0 +1,3 @@ +from .sankey import SankeyBuilder +from .waterfall import plot_exergy_waterfall +from .waterfall import plot_exergy_waterfall_plotly diff --git a/src/exerpy/visualization/waterfall.py b/src/exerpy/visualization/waterfall.py new file mode 100644 index 0000000..465a9aa --- /dev/null +++ b/src/exerpy/visualization/waterfall.py @@ -0,0 +1,274 @@ +"""Exergy destruction waterfall diagram.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .colors import WATERFALL_COLORS + +if TYPE_CHECKING: + from exerpy.analyses import ExergyAnalysis + + +def _prepare_data(analysis: ExergyAnalysis, exclude_components: list[str] | None): + """Extract and sort component data for waterfall plotting.""" + df, _, _ = analysis.exergy_results(print_results=False) + + if exclude_components is None: + exclude_components = [] + + total_row = df[df["Component"] == "TOT"].iloc[0] + epsilon_total = total_row["epsilon [%]"] + E_F_total = total_row["E_F [kW]"] + loss_percent = (total_row["E_L [kW]"] / E_F_total) * 100 if E_F_total != 0 else 0 + + comp_data = df[ + (df["Component"] != "TOT") + & (df["E_F [kW]"].notna()) + & (~df["Component"].isin(exclude_components)) + & (df["y [%]"].notna()) + ].copy().sort_values("y [%]", ascending=False) + + return comp_data, epsilon_total, loss_percent + + +def _resolve_colors(user_colors: dict[str, str] | None) -> dict[str, str]: + palette = dict(WATERFALL_COLORS) + if user_colors: + palette.update(user_colors) + return palette + + +def plot_exergy_waterfall( + analysis: ExergyAnalysis, + title: str | None = None, + figsize: tuple[float, float] = (12, 10), + exclude_components: list[str] | None = None, + colors: dict[str, str] | None = None, + show_plot: bool = True, +): + """ + Create an exergy destruction waterfall diagram using Matplotlib. + + Visualizes exergy flow through the system as a waterfall chart, showing how + exergy is destroyed in each component from the exergetic fuel (100%) down to + the exergetic product and losses. + + Parameters + ---------- + analysis : ExergyAnalysis + Completed analysis instance (analyse() must be called first). + title : str, optional + Title for the plot. If None, no title is displayed. + figsize : tuple, optional + Figure size as (width, height) in inches. Default is (12, 10). + exclude_components : list, optional + List of component names to exclude from the diagram. + By default, all components with NaN E_F (Exergetic Fuel) are excluded, + as well as CycleCloser and PowerBus components. + colors : dict, optional + Override bar colors. Keys: "fuel", "destruction", "loss", "product". + Defaults are taken from ``exerpy.visualization.colors.WATERFALL_COLORS``. + show_plot : bool, optional + Whether to display the plot immediately. Default is True. + + Returns + ------- + fig : matplotlib.figure.Figure + The figure object containing the waterfall diagram. + ax : matplotlib.axes.Axes + The axes object of the waterfall diagram. + + Raises + ------ + RuntimeError + If the exergy analysis has not been performed yet (analyse() not called). + + Notes + ----- + - The waterfall diagram displays exergy values as percentages of the total fuel exergy. + - Components are sorted by their exergy destruction rate (y [%]) in descending order. + - Each bar represents the remaining exergy after destruction in that component. + - Red bars indicate exergy destruction in components. + - Blue bar represents the initial exergetic fuel (100%). + - Green bar represents the final exergetic product. + + Examples + -------- + >>> analysis = ExergyAnalysis.from_tespy(network, Tamb=288.15, pamb=101325) # doctest: +SKIP + >>> analysis.analyse(E_F={'inputs': ['fuel']}, E_P={'outputs': ['power']}) # doctest: +SKIP + >>> fig, ax = analysis.plot_exergy_waterfall(title='Power Plant Exergy Waterfall') # doctest: +SKIP + >>> fig.savefig('exergy_waterfall.pdf') # doctest: +SKIP + + See Also + -------- + exergy_results : Display tabular exergy analysis results. + print_exergy_summary : Print a text summary of exergy analysis. + plot_exergy_waterfall_plotly : Plotly version of this diagram. + """ + import matplotlib.pyplot as plt + import numpy as np + + if not hasattr(analysis, "epsilon") or analysis.epsilon is None: + raise RuntimeError("Exergy analysis has not been performed yet. Please call analyse() first.") + + palette = _resolve_colors(colors) + comp_data, epsilon_total, loss_percent = _prepare_data(analysis, exclude_components) + + bar_values = [100.0] + current = 100.0 + for y in comp_data["y [%]"]: + current -= y + bar_values.append(current) + bar_values.append(epsilon_total) + + space_labels = ( + ["Exergetic fuel"] + + list(comp_data["Component"]) + + ["Exergetic loss", "Exergetic product"] + ) + + fig, ax = plt.subplots(figsize=figsize) + n_bars = len(bar_values) + bar_positions = np.arange(n_bars) + bar_colors = ( + [palette["fuel"]] + + [palette["destruction"]] * (n_bars - 2) + + [palette["product"]] + ) + + for pos, value, color in zip(bar_positions, bar_values, bar_colors, strict=False): + ax.barh(pos, value, color=color, alpha=0.8, height=0.6) + ax.text(value - 2, pos, f"{value:.2f}%", va="center", ha="right", fontsize=9, fontweight="bold", color="white") + + space_positions = [-0.5] + [i + 0.5 for i in range(n_bars - 1)] + [n_bars - 0.5] + for i, (space_pos, label) in enumerate(zip(space_positions, space_labels, strict=False)): + if i == 0: + ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") + elif i == len(space_labels) - 2: + ax.text(2, space_pos, f"{label} (-{loss_percent:.2f}%)", va="center", ha="left", fontsize=10, fontweight="bold", style="italic") + elif i == len(space_labels) - 1: + ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") + else: + y_rate = comp_data.iloc[i - 1]["y [%]"] + ax.text(2, space_pos, f"{label} (-{y_rate:.2f}%)", va="center", ha="left", fontsize=10, fontweight="bold") + + ax.set_yticks(bar_positions) + ax.set_yticklabels([""] * n_bars) + ax.set_xlabel("Exergy [%]", fontsize=12, fontweight="bold") + if title is not None: + ax.set_title(title, fontsize=14, fontweight="bold", pad=20) + ax.set_xlim(0, 100) + ax.set_ylim(-1, n_bars) + ax.grid(axis="x", alpha=0.3, linestyle="--") + ax.axvline(x=0, color="black", linewidth=0.8) + ax.invert_yaxis() + plt.tight_layout() + + if show_plot: + plt.show() + + return fig, ax + + +def plot_exergy_waterfall_plotly( + analysis: ExergyAnalysis, + title: str | None = None, + exclude_components: list[str] | None = None, + colors: dict[str, str] | None = None, + show_plot: bool = True, +): + """ + Create an exergy destruction waterfall diagram using Plotly. + + Interactive version of the waterfall chart. Components are sorted by their + exergy destruction rate (y [%]) in descending order. + + Parameters + ---------- + analysis : ExergyAnalysis + Completed analysis instance (analyse() must be called first). + title : str, optional + Title for the plot. If None, a default title is used. + exclude_components : list, optional + List of component names to exclude from the diagram. + By default, all components with NaN E_F (Exergetic Fuel) are excluded, + as well as CycleCloser and PowerBus components. + colors : dict, optional + Override bar colors. Keys: "fuel", "destruction", "loss", "product". + Defaults are taken from ``exerpy.visualization.colors.WATERFALL_COLORS``. + show_plot : bool, optional + Whether to display the plot immediately. Default is True. + + Returns + ------- + fig : plotly.graph_objects.Figure + + Raises + ------ + RuntimeError + If the exergy analysis has not been performed yet (analyse() not called). + + Examples + -------- + >>> analysis = ExergyAnalysis.from_tespy(network, Tamb=288.15, pamb=101325) # doctest: +SKIP + >>> analysis.analyse(E_F={'inputs': ['fuel']}, E_P={'outputs': ['power']}) # doctest: +SKIP + >>> fig = analysis.plot_exergy_waterfall_plotly(title='Power Plant Exergy Waterfall') # doctest: +SKIP + >>> fig.write_html('waterfall.html') # doctest: +SKIP + + See Also + -------- + plot_exergy_waterfall : Matplotlib version of this diagram. + """ + import plotly.graph_objects as go + + if not hasattr(analysis, "epsilon") or analysis.epsilon is None: + raise RuntimeError("Exergy analysis has not been performed yet. Please call analyse() first.") + + palette = _resolve_colors(colors) + comp_data, epsilon_total, loss_percent = _prepare_data(analysis, exclude_components) + + names = ( + ["E_F (100%)"] + + [f"{name} (-{y:.2f}%)" for name, y in zip(comp_data["Component"], comp_data["y [%]"], strict=False)] + + [f"E_L (-{loss_percent:.2f}%)", "E_P"] + ) + measures = ["absolute"] + ["relative"] * len(comp_data) + ["relative", "total"] + x_values = [100.0] + [-y for y in comp_data["y [%]"]] + [-loss_percent, 0] + text = ( + ["100.00%"] + + [f"-{y:.2f}%" for y in comp_data["y [%]"]] + + [f"-{loss_percent:.2f}%", f"{epsilon_total:.2f}%"] + ) + bar_colors = ( + [palette["fuel"]] + + [palette["destruction"]] * len(comp_data) + + [palette["loss"], palette["product"]] + ) + + fig = go.Figure(go.Waterfall( + orientation="h", + measure=measures, + y=names, + x=x_values, + text=text, + textposition="inside", + connector={"line": {"color": "#9E9E9E", "dash": "dot", "width": 1}}, + decreasing={"marker": {"color": palette["destruction"]}}, + increasing={"marker": {"color": palette["destruction"]}}, + totals={"marker": {"color": palette["product"]}}, + marker={"color": bar_colors}, + )) + + fig.update_layout( + title_text=title or "Exergy Waterfall Diagram", + xaxis_title="Exergy [%]", + xaxis={"range": [0, 105]}, + font_size=12, + showlegend=False, + ) + + if show_plot: + fig.show() + + return fig From 0e920761233490bab6180c6e6757e666349911be Mon Sep 17 00:00:00 2001 From: Francesco Witte Date: Fri, 26 Jun 2026 07:54:07 +0200 Subject: [PATCH 04/11] Implement sankey draft --- src/exerpy/visualization/colors.py | 83 ++++++ src/exerpy/visualization/sankey.py | 442 +++++++++++++++++++++++++++++ 2 files changed, 525 insertions(+) create mode 100644 src/exerpy/visualization/colors.py create mode 100644 src/exerpy/visualization/sankey.py diff --git a/src/exerpy/visualization/colors.py b/src/exerpy/visualization/colors.py new file mode 100644 index 0000000..ffcd3be --- /dev/null +++ b/src/exerpy/visualization/colors.py @@ -0,0 +1,83 @@ +"""Color utilities for Sankey visualization.""" + +from __future__ import annotations + +# Base colors by fluid identity +_FLUID_BASE: dict[str, str] = { + "water": "#2196F3", + "R245FA": "#4CAF50", + "R134a": "#8BC34A", + "CH4": "#FF9800", + "H2": "#E91E63", + "CO2": "#607D8B", + "air": "#9E9E9E", + "flue_gas": "#795548", + "mixture": "#546E7A", + "default": "#00BCD4", +} + +# Multi-component air identification: subsets of these keys +_AIR_KEYS = {"N2", "O2", "H2O", "CO2", "AR", "Ar", "CH4"} + +POWER_COLOR = "#FFC107" +HEAT_COLOR = "#F44336" +ED_COLOR = "#424242" + +TERMINAL_COLORS: dict[str, str] = { + "E_F": "#8D6E63", + "E_P": "#43A047", + "E_D": "#E53935", + "E_L": "#FB8C00", +} + +WATERFALL_COLORS: dict[str, str] = { + "fuel": "#1565C0", + "destruction": "#D32F2F", + "loss": "#E65100", + "product": "#2E7D32", +} + +DEFAULT_NODE_COLOR = "#78909C" + + +def _hex_to_rgb(hex_color: str) -> tuple[int, int, int]: + h = hex_color.lstrip("#") + return int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16) + + +def hex_to_rgba(hex_color: str, alpha: float = 0.6) -> str: + r, g, b = _hex_to_rgb(hex_color) + return f"rgba({r},{g},{b},{alpha})" + + +def shade(hex_color: str, factor: float) -> str: + """Scale RGB channels by factor (< 1 darkens, > 1 lightens, clamped to 0-255).""" + r, g, b = _hex_to_rgb(hex_color) + r = min(255, max(0, int(r * factor))) + g = min(255, max(0, int(g * factor))) + b = min(255, max(0, int(b * factor))) + return f"#{r:02x}{g:02x}{b:02x}" + + +def fluid_base_color(mass_composition: dict) -> str: + if not mass_composition: + return _FLUID_BASE["default"] + keys = list(mass_composition.keys()) + if len(keys) == 1: + return _FLUID_BASE.get(keys[0], _FLUID_BASE["default"]) + # Multi-component: classify as air or flue gas if it matches the air-component set + if set(keys).issubset(_AIR_KEYS): + co2 = mass_composition.get("CO2", 0) or 0 + return _FLUID_BASE["flue_gas"] if co2 > 0.01 else _FLUID_BASE["air"] + return _FLUID_BASE["mixture"] + + +def connection_base_color(conn_data: dict) -> str: + kind = conn_data.get("kind", "material") + if kind == "power": + return POWER_COLOR + if kind == "heat": + return HEAT_COLOR + return fluid_base_color(conn_data.get("mass_composition") or {}) + + diff --git a/src/exerpy/visualization/sankey.py b/src/exerpy/visualization/sankey.py new file mode 100644 index 0000000..9267ff1 --- /dev/null +++ b/src/exerpy/visualization/sankey.py @@ -0,0 +1,442 @@ +"""Sankey diagram builder for exergy analysis results.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .colors import DEFAULT_NODE_COLOR +from .colors import ED_COLOR +from .colors import TERMINAL_COLORS +from .colors import connection_base_color +from .colors import hex_to_rgba +from .colors import shade + +if TYPE_CHECKING: + from exerpy.analyses import ExergyAnalysis + +# Component types that are true 1:1 pass-throughs (no topological change, no transformation). +# Splitter (1→N) and Mixer (N→1) are NOT pass-throughs: they change topology. +_DEFAULT_PASSTHROUGH_TYPES: frozenset[str] = frozenset({"CycleCloser"}) + +# Skip destruction links for these types (E_D is None or structurally zero) +_NO_DESTRUCTION_TYPES: frozenset[str] = frozenset({"CycleCloser", "PowerBus", "Splitter", "Mixer"}) + +# Special terminal node IDs +_EF = "__E_F__" +_EP = "__E_P__" +_ED = "__E_D__" +_EL = "__E_L__" +_EP_NET = "__E_P_net__" +_EL_NET = "__E_L_net__" + + +class SankeyBuilder: + """ + Builds a Plotly Sankey diagram from a completed ExergyAnalysis result. + + Parameters + ---------- + analysis : ExergyAnalysis + Must have had ``.analyse()`` called. + mode : {1, 2, 3} + 1 – total exergy E per link + 2 – split material links into E_PH + E_CH (requires chemical_exergy_enabled) + 3 – split material links into E_T + E_M + E_CH (requires split_physical_exergy) + collapse_passthroughs : bool or list[str] + True → collapse CycleCloser nodes (default) + False → show every node + list → collapse only the listed component type names + groups : dict[str, list[str]], optional + Visual grouping: ``{"Group name": ["comp1", "comp2", ...]}``. + Internal connections are hidden; only boundary-crossing links are shown. + """ + + def __init__( + self, + analysis: ExergyAnalysis, + mode: int = 1, + collapse_passthroughs: bool | list[str] = True, + groups: dict[str, list[str]] | None = None, + node_colors: dict[str, str] | None = None, + ) -> None: + if not hasattr(analysis, "E_F"): + raise RuntimeError("Call ExergyAnalysis.analyse() before building Sankey.") + if mode == 2 and not analysis.chemical_exergy_enabled: + raise ValueError("Mode 2 requires chemical_exergy_enabled=True.") + if mode == 3 and not analysis.split_physical_exergy: + raise ValueError("Mode 3 requires split_physical_exergy=True.") + + self.analysis = analysis + self.mode = mode + self.groups: dict[str, list[str]] = groups or {} + self.node_colors: dict[str, str] = node_colors or {} + + if collapse_passthroughs is True: + self._passthrough_types: frozenset[str] = _DEFAULT_PASSTHROUGH_TYPES + elif collapse_passthroughs is False: + self._passthrough_types = frozenset() + else: + self._passthrough_types = frozenset(collapse_passthroughs) + + self._nodes: list[dict] = [] + self._links: list[dict] = [] + self._node_idx: dict[str, int] = {} + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def build(self) -> tuple[list[dict], list[dict]]: + """Build and return ``(nodes, links)`` lists.""" + self._nodes = [] + self._links = [] + self._node_idx = {} + + collapsed = self._find_collapsed() + routes = self._build_routes(collapsed) + terminal_ids = self._terminal_conn_ids() + + self._add_component_nodes(collapsed) + self._add_terminal_nodes() + self._add_connection_links(routes, terminal_ids) + self._add_destruction_links(collapsed) + self._add_terminal_links(routes) + + if self.groups: + self._apply_grouping() + + return self._nodes, self._links + + def to_plotly(self, title: str | None = None): + """Return a Plotly ``Figure`` containing the Sankey trace.""" + try: + import plotly.graph_objects as go + except ImportError as exc: + raise ImportError("plotly is required for Sankey diagrams: pip install plotly") from exc + + nodes, links = self.build() + fig = go.Figure(go.Sankey( + arrangement="snap", + node=dict( + pad=15, + thickness=20, + label=[n["label"] for n in nodes], + color=[n["color"] for n in nodes], + ), + link=dict( + source=[lk["source"] for lk in links], + target=[lk["target"] for lk in links], + value=[max(lk["value"], 0.0) for lk in links], + color=[lk["color"] for lk in links], + label=[lk.get("label", "") for lk in links], + ), + )) + fig.update_layout( + title_text=title or "Exergy Sankey Diagram", + font_size=12, + ) + return fig + + def to_html(self, path: str, title: str | None = None) -> None: + """Export the Sankey diagram to an HTML file.""" + self.to_plotly(title=title).write_html(path) + + # ------------------------------------------------------------------ + # Node building + # ------------------------------------------------------------------ + + def _add_node(self, node_id: str, label: str, color: str) -> int: + idx = len(self._nodes) + self._nodes.append({"id": node_id, "label": label, "color": color}) + self._node_idx[node_id] = idx + return idx + + def _find_collapsed(self) -> set[str]: + return { + name + for name, comp in self.analysis.components.items() + if comp.__class__.__name__ in self._passthrough_types + } + + def _add_component_nodes(self, collapsed: set[str]) -> None: + for comp_name in self.analysis.components: + if comp_name in collapsed: + continue + color = self.node_colors.get(comp_name, DEFAULT_NODE_COLOR) + self._add_node(comp_name, comp_name, color) + + def _add_terminal_nodes(self) -> None: + for node_id, label, key in ( + (_EF, "E_F (Fuel)", "E_F"), + (_EP, "E_P (Product)", "E_P"), + (_ED, "E_D (Destruction)", "E_D"), + (_EL, "E_L (Loss)", "E_L"), + ): + self._add_node(node_id, label, TERMINAL_COLORS[key]) + + # Intermediate "net" nodes only when a terminal has both inputs and outputs + if self.analysis.E_P_dict.get("inputs") and self.analysis.E_P_dict.get("outputs"): + self._add_node(_EP_NET, "E_P net", TERMINAL_COLORS["E_P"]) + if self.analysis.E_L_dict.get("inputs") and self.analysis.E_L_dict.get("outputs"): + self._add_node(_EL_NET, "E_L net", TERMINAL_COLORS["E_L"]) + + # ------------------------------------------------------------------ + # Routing (for 1:1 collapsed pass-through nodes) + # ------------------------------------------------------------------ + + def _build_routes( + self, collapsed: set[str] + ) -> dict[str, tuple[str | None, str | None] | None]: + """ + Return ``{conn_id: (eff_source, eff_target)}`` after collapsing pass-through nodes. + + Connections whose *target* is a collapsed node are marked ``None`` (skipped). + Connections whose *source* is a collapsed node get their source remapped to + the real upstream component via the pass-through chain. + """ + conns = self.analysis.connections + + def upstream(comp: str | None, depth: int = 0) -> str | None: + if comp is None or comp not in collapsed or depth > 50: + return comp + for cd in conns.values(): + if cd.get("target_component") == comp: + return upstream(cd.get("source_component"), depth + 1) + return comp + + routes: dict[str, tuple[str | None, str | None] | None] = {} + for conn_id, cd in conns.items(): + src = cd.get("source_component") + tgt = cd.get("target_component") + if tgt in collapsed: + # Input to a collapsed node: skip (the downstream connection takes over) + routes[conn_id] = None + elif src in collapsed: + # Output from collapsed node: remap source to real upstream + routes[conn_id] = (upstream(src), tgt) + else: + routes[conn_id] = (src, tgt) + + return routes + + def _terminal_conn_ids(self) -> set[str]: + result: set[str] = set() + for d in (self.analysis.E_F_dict, self.analysis.E_P_dict, self.analysis.E_L_dict): + result.update(d.get("inputs", [])) + result.update(d.get("outputs", [])) + return result + + # ------------------------------------------------------------------ + # Link building + # ------------------------------------------------------------------ + + def _add_link( + self, source_id: str, target_id: str, value: float, color: str, label: str = "" + ) -> None: + if not (value > 0): + return + src = self._node_idx.get(source_id) + tgt = self._node_idx.get(target_id) + if src is None or tgt is None: + return + self._links.append({"source": src, "target": tgt, "value": value, "color": color, "label": label}) + + def _sub_links(self, conn_data: dict) -> list[tuple[float, str, str]]: + """Return ``[(value_W, rgba_color, sub_label), ...]`` based on self.mode.""" + kind = conn_data.get("kind") + E = conn_data.get("E") or 0 + base = connection_base_color(conn_data) + + # Non-material connections and mode 1 always use total E + if kind in ("power", "heat") or self.mode == 1: + return [(E, hex_to_rgba(base, 0.5), "E")] + + if self.mode == 2: + E_PH = conn_data.get("E_PH") or 0 + E_CH = conn_data.get("E_CH") or 0 + result = [] + if E_PH > 0: + result.append((E_PH, hex_to_rgba(base, 0.6), "E_PH")) + if E_CH > 0: + result.append((E_CH, hex_to_rgba(shade(base, 0.75), 0.5), "E_CH")) + return result or [(E, hex_to_rgba(base, 0.5), "E")] + + # mode 3: E_T = e_T * m, E_M = e_M * m, E_CH from stored value + m = conn_data.get("m") or 0 + e_T = conn_data.get("e_T") or 0 + e_M = conn_data.get("e_M") or 0 + E_T = e_T * m + E_M = e_M * m + E_CH = conn_data.get("E_CH") or 0 + result = [] + if E_T > 0: + result.append((E_T, hex_to_rgba(base, 0.70), "E_T")) + if E_M > 0: + result.append((E_M, hex_to_rgba(shade(base, 0.80), 0.60), "E_M")) + if E_CH > 0: + result.append((E_CH, hex_to_rgba(shade(base, 0.60), 0.50), "E_CH")) + return result or [(E, hex_to_rgba(base, 0.5), "E")] + + def _add_connection_links( + self, routes: dict, terminal_ids: set[str] + ) -> None: + for conn_id, conn_data in self.analysis.connections.items(): + if conn_id in terminal_ids: + continue + route = routes.get(conn_id) + if route is None: + continue + src_id, tgt_id = route + if src_id not in self._node_idx or tgt_id not in self._node_idx: + continue + if src_id == tgt_id: + continue + for value, color, sub_label in self._sub_links(conn_data): + self._add_link(src_id, tgt_id, value, color, f"{conn_id} [{sub_label}]: {value * 1e-3:.1f} kW") + + def _add_destruction_links(self, collapsed: set[str]) -> None: + ed_color = hex_to_rgba(ED_COLOR, 0.4) + for comp_name, comp in self.analysis.components.items(): + if comp_name in collapsed: + continue + if comp.__class__.__name__ in _NO_DESTRUCTION_TYPES: + continue + E_D = getattr(comp, "E_D", None) + if not (E_D and E_D > 0): + continue + self._add_link(comp_name, _ED, E_D, ed_color, f"{comp_name} E_D: {E_D * 1e-3:.1f} kW") + + def _add_terminal_links(self, routes: dict) -> None: + conns = self.analysis.connections + use_ep_net = _EP_NET in self._node_idx + use_el_net = _EL_NET in self._node_idx + + def _color(conn_data: dict) -> str: + return hex_to_rgba(connection_base_color(conn_data), 0.5) + + def _E(conn_id: str) -> float: + cd = conns.get(conn_id) + return (cd.get("E") or 0) if cd else 0 + + def _inside(conn_id: str, side: str) -> str | None: + """Return the component node ID on 'source' or 'target' side that is inside the system.""" + cd = conns.get(conn_id) + if cd is None: + return None + # Use routing if available (handles collapsed nodes) + route = routes.get(conn_id) + if route is not None: + src, tgt = route + else: + src, tgt = cd.get("source_component"), cd.get("target_component") + node_id = src if side == "source" else tgt + return node_id if node_id in self._node_idx else None + + def _lbl(conn_id: str, tag: str) -> str: + return f"{conn_id} [{tag}]: {_E(conn_id) * 1e-3:.1f} kW" + + # --- E_F --- + # "inputs": fuel flows from outside INTO the system → E_F node → component + for c in self.analysis.E_F_dict.get("inputs", []): + tgt = _inside(c, "target") + if tgt: + self._add_link(_EF, tgt, _E(c), _color(conns[c]), _lbl(c, "E_F in")) + + # "outputs": something returns from system back to fuel boundary → component → E_F node + for c in self.analysis.E_F_dict.get("outputs", []): + src = _inside(c, "source") + if src: + self._add_link(src, _EF, _E(c), _color(conns[c]), _lbl(c, "E_F return")) + + # --- E_P --- + # "inputs": product leaves system → component → E_P (or intermediate) + ep_sink = _EP_NET if use_ep_net else _EP + for c in self.analysis.E_P_dict.get("inputs", []): + src = _inside(c, "source") + if src: + self._add_link(src, ep_sink, _E(c), _color(conns[c]), _lbl(c, "E_P")) + + # "outputs": something enters system as part of product definition → intermediate → component + for c in self.analysis.E_P_dict.get("outputs", []): + tgt = _inside(c, "target") + if tgt: + self._add_link(ep_sink, tgt, _E(c), _color(conns[c]), _lbl(c, "E_P return")) + + # Net link from intermediate to final E_P node + if use_ep_net: + net = self.analysis.E_P + self._add_link(_EP_NET, _EP, net, hex_to_rgba(TERMINAL_COLORS["E_P"], 0.7), f"E_P net: {net * 1e-3:.1f} kW") + + # --- E_L --- + el_sink = _EL_NET if use_el_net else _EL + for c in self.analysis.E_L_dict.get("inputs", []): + src = _inside(c, "source") + if src: + self._add_link(src, el_sink, _E(c), _color(conns[c]), _lbl(c, "E_L")) + + for c in self.analysis.E_L_dict.get("outputs", []): + tgt = _inside(c, "target") + if tgt: + self._add_link(el_sink, tgt, _E(c), _color(conns[c]), _lbl(c, "E_L return")) + + if use_el_net: + net = self.analysis.E_L + self._add_link(_EL_NET, _EL, net, hex_to_rgba(TERMINAL_COLORS["E_L"], 0.7), f"E_L net: {net * 1e-3:.1f} kW") + + # ------------------------------------------------------------------ + # Grouping + # ------------------------------------------------------------------ + + def _apply_grouping(self) -> None: + """ + Replace groups of components with single aggregate nodes. + Internal connections (both endpoints in the same group) are dropped. + Cross-boundary connections are remapped to the group node. + """ + comp_to_group: dict[str, str] = { + member: gname + for gname, members in self.groups.items() + for member in members + } + + new_nodes: list[dict] = [] + new_idx: dict[str, int] = {} + group_added: set[str] = set() + + for node in self._nodes: + nid = node["id"] + if nid in comp_to_group: + gname = comp_to_group[nid] + if gname not in group_added: + gidx = len(new_nodes) + new_idx[gname] = gidx + new_nodes.append({"id": gname, "label": f"[{gname}]", "color": "#546E7A"}) + group_added.add(gname) + new_idx[nid] = new_idx[gname] + else: + new_idx[nid] = len(new_nodes) + new_nodes.append(node) + + new_links: list[dict] = [] + for lk in self._links: + old_src_id = self._nodes[lk["source"]]["id"] + old_tgt_id = self._nodes[lk["target"]]["id"] + grp_src = comp_to_group.get(old_src_id) + grp_tgt = comp_to_group.get(old_tgt_id) + + # Drop connections internal to a group + if grp_src and grp_tgt and grp_src == grp_tgt: + continue + + new_src = grp_src or old_src_id + new_tgt = grp_tgt or old_tgt_id + new_src_idx = new_idx[new_src] + new_tgt_idx = new_idx[new_tgt] + if new_src_idx == new_tgt_idx: + continue + + new_links.append({**lk, "source": new_src_idx, "target": new_tgt_idx}) + + self._nodes = new_nodes + self._links = new_links + self._node_idx = {n["id"]: i for i, n in enumerate(new_nodes)} From 364bb4870c062f2994c179d2686e60197bdb9034 Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:09:16 +0200 Subject: [PATCH 05/11] Fix: Mixer exergy destruction in Sankey, plotly waterfall crash and black formatting --- src/exerpy/analyses.py | 13 +++- src/exerpy/visualization/colors.py | 2 - src/exerpy/visualization/sankey.py | 57 +++++++--------- src/exerpy/visualization/waterfall.py | 95 +++++++++++++++------------ 4 files changed, 89 insertions(+), 78 deletions(-) diff --git a/src/exerpy/analyses.py b/src/exerpy/analyses.py index 58e7ce5..67fefd9 100755 --- a/src/exerpy/analyses.py +++ b/src/exerpy/analyses.py @@ -658,7 +658,14 @@ def plot_exergy_waterfall(self, title=None, figsize=(12, 10), exclude_components """ from .visualization.waterfall import plot_exergy_waterfall - return plot_exergy_waterfall(self, title=title, figsize=figsize, exclude_components=exclude_components, colors=colors, show_plot=show_plot) + return plot_exergy_waterfall( + self, + title=title, + figsize=figsize, + exclude_components=exclude_components, + colors=colors, + show_plot=show_plot, + ) def plot_exergy_waterfall_plotly(self, title=None, exclude_components=None, colors=None, show_plot=True): """ @@ -687,7 +694,9 @@ def plot_exergy_waterfall_plotly(self, title=None, exclude_components=None, colo """ from .visualization.waterfall import plot_exergy_waterfall_plotly - return plot_exergy_waterfall_plotly(self, title=title, exclude_components=exclude_components, colors=colors, show_plot=show_plot) + return plot_exergy_waterfall_plotly( + self, title=title, exclude_components=exclude_components, colors=colors, show_plot=show_plot + ) def print_exergy_summary(self): """ diff --git a/src/exerpy/visualization/colors.py b/src/exerpy/visualization/colors.py index ffcd3be..dd541de 100644 --- a/src/exerpy/visualization/colors.py +++ b/src/exerpy/visualization/colors.py @@ -79,5 +79,3 @@ def connection_base_color(conn_data: dict) -> str: if kind == "heat": return HEAT_COLOR return fluid_base_color(conn_data.get("mass_composition") or {}) - - diff --git a/src/exerpy/visualization/sankey.py b/src/exerpy/visualization/sankey.py index 9267ff1..e8766c0 100644 --- a/src/exerpy/visualization/sankey.py +++ b/src/exerpy/visualization/sankey.py @@ -18,8 +18,9 @@ # Splitter (1→N) and Mixer (N→1) are NOT pass-throughs: they change topology. _DEFAULT_PASSTHROUGH_TYPES: frozenset[str] = frozenset({"CycleCloser"}) -# Skip destruction links for these types (E_D is None or structurally zero) -_NO_DESTRUCTION_TYPES: frozenset[str] = frozenset({"CycleCloser", "PowerBus", "Splitter", "Mixer"}) +# Skip destruction links for these types (E_D is None or structurally zero). +# Mixer is NOT in this set: mixing streams of different temperature destroys exergy. +_NO_DESTRUCTION_TYPES: frozenset[str] = frozenset({"CycleCloser", "PowerBus", "Splitter"}) # Special terminal node IDs _EF = "__E_F__" @@ -115,22 +116,24 @@ def to_plotly(self, title: str | None = None): raise ImportError("plotly is required for Sankey diagrams: pip install plotly") from exc nodes, links = self.build() - fig = go.Figure(go.Sankey( - arrangement="snap", - node=dict( - pad=15, - thickness=20, - label=[n["label"] for n in nodes], - color=[n["color"] for n in nodes], - ), - link=dict( - source=[lk["source"] for lk in links], - target=[lk["target"] for lk in links], - value=[max(lk["value"], 0.0) for lk in links], - color=[lk["color"] for lk in links], - label=[lk.get("label", "") for lk in links], - ), - )) + fig = go.Figure( + go.Sankey( + arrangement="snap", + node=dict( + pad=15, + thickness=20, + label=[n["label"] for n in nodes], + color=[n["color"] for n in nodes], + ), + link=dict( + source=[lk["source"] for lk in links], + target=[lk["target"] for lk in links], + value=[max(lk["value"], 0.0) for lk in links], + color=[lk["color"] for lk in links], + label=[lk.get("label", "") for lk in links], + ), + ) + ) fig.update_layout( title_text=title or "Exergy Sankey Diagram", font_size=12, @@ -184,9 +187,7 @@ def _add_terminal_nodes(self) -> None: # Routing (for 1:1 collapsed pass-through nodes) # ------------------------------------------------------------------ - def _build_routes( - self, collapsed: set[str] - ) -> dict[str, tuple[str | None, str | None] | None]: + def _build_routes(self, collapsed: set[str]) -> dict[str, tuple[str | None, str | None] | None]: """ Return ``{conn_id: (eff_source, eff_target)}`` after collapsing pass-through nodes. @@ -230,9 +231,7 @@ def _terminal_conn_ids(self) -> set[str]: # Link building # ------------------------------------------------------------------ - def _add_link( - self, source_id: str, target_id: str, value: float, color: str, label: str = "" - ) -> None: + def _add_link(self, source_id: str, target_id: str, value: float, color: str, label: str = "") -> None: if not (value > 0): return src = self._node_idx.get(source_id) @@ -277,9 +276,7 @@ def _sub_links(self, conn_data: dict) -> list[tuple[float, str, str]]: result.append((E_CH, hex_to_rgba(shade(base, 0.60), 0.50), "E_CH")) return result or [(E, hex_to_rgba(base, 0.5), "E")] - def _add_connection_links( - self, routes: dict, terminal_ids: set[str] - ) -> None: + def _add_connection_links(self, routes: dict, terminal_ids: set[str]) -> None: for conn_id, conn_data in self.analysis.connections.items(): if conn_id in terminal_ids: continue @@ -393,11 +390,7 @@ def _apply_grouping(self) -> None: Internal connections (both endpoints in the same group) are dropped. Cross-boundary connections are remapped to the group node. """ - comp_to_group: dict[str, str] = { - member: gname - for gname, members in self.groups.items() - for member in members - } + comp_to_group: dict[str, str] = {member: gname for gname, members in self.groups.items() for member in members} new_nodes: list[dict] = [] new_idx: dict[str, int] = {} diff --git a/src/exerpy/visualization/waterfall.py b/src/exerpy/visualization/waterfall.py index 465a9aa..2610358 100644 --- a/src/exerpy/visualization/waterfall.py +++ b/src/exerpy/visualization/waterfall.py @@ -22,12 +22,16 @@ def _prepare_data(analysis: ExergyAnalysis, exclude_components: list[str] | None E_F_total = total_row["E_F [kW]"] loss_percent = (total_row["E_L [kW]"] / E_F_total) * 100 if E_F_total != 0 else 0 - comp_data = df[ - (df["Component"] != "TOT") - & (df["E_F [kW]"].notna()) - & (~df["Component"].isin(exclude_components)) - & (df["y [%]"].notna()) - ].copy().sort_values("y [%]", ascending=False) + comp_data = ( + df[ + (df["Component"] != "TOT") + & (df["E_F [kW]"].notna()) + & (~df["Component"].isin(exclude_components)) + & (df["y [%]"].notna()) + ] + .copy() + .sort_values("y [%]", ascending=False) + ) return comp_data, epsilon_total, loss_percent @@ -122,20 +126,12 @@ def plot_exergy_waterfall( bar_values.append(current) bar_values.append(epsilon_total) - space_labels = ( - ["Exergetic fuel"] - + list(comp_data["Component"]) - + ["Exergetic loss", "Exergetic product"] - ) + space_labels = ["Exergetic fuel"] + list(comp_data["Component"]) + ["Exergetic loss", "Exergetic product"] fig, ax = plt.subplots(figsize=figsize) n_bars = len(bar_values) bar_positions = np.arange(n_bars) - bar_colors = ( - [palette["fuel"]] - + [palette["destruction"]] * (n_bars - 2) - + [palette["product"]] - ) + bar_colors = [palette["fuel"]] + [palette["destruction"]] * (n_bars - 2) + [palette["product"]] for pos, value, color in zip(bar_positions, bar_values, bar_colors, strict=False): ax.barh(pos, value, color=color, alpha=0.8, height=0.6) @@ -146,7 +142,16 @@ def plot_exergy_waterfall( if i == 0: ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") elif i == len(space_labels) - 2: - ax.text(2, space_pos, f"{label} (-{loss_percent:.2f}%)", va="center", ha="left", fontsize=10, fontweight="bold", style="italic") + ax.text( + 2, + space_pos, + f"{label} (-{loss_percent:.2f}%)", + va="center", + ha="left", + fontsize=10, + fontweight="bold", + style="italic", + ) elif i == len(space_labels) - 1: ax.text(2, space_pos, label, va="center", ha="left", fontsize=10, fontweight="bold", style="italic") else: @@ -233,39 +238,45 @@ def plot_exergy_waterfall_plotly( + [f"{name} (-{y:.2f}%)" for name, y in zip(comp_data["Component"], comp_data["y [%]"], strict=False)] + [f"E_L (-{loss_percent:.2f}%)", "E_P"] ) - measures = ["absolute"] + ["relative"] * len(comp_data) + ["relative", "total"] - x_values = [100.0] + [-y for y in comp_data["y [%]"]] + [-loss_percent, 0] - text = ( - ["100.00%"] - + [f"-{y:.2f}%" for y in comp_data["y [%]"]] - + [f"-{loss_percent:.2f}%", f"{epsilon_total:.2f}%"] - ) - bar_colors = ( - [palette["fuel"]] - + [palette["destruction"]] * len(comp_data) - + [palette["loss"], palette["product"]] + text = ["100.00%"] + [f"-{y:.2f}%" for y in comp_data["y [%]"]] + [f"-{loss_percent:.2f}%", f"{epsilon_total:.2f}%"] + bar_colors = [palette["fuel"]] + [palette["destruction"]] * len(comp_data) + [palette["loss"], palette["product"]] + + # go.Waterfall only supports three colors (increasing/decreasing/totals), + # so build the floating segments manually with go.Bar to color every bar + # according to the fuel/destruction/loss/product palette. + starts = [0.0] + widths = [100.0] + remaining = 100.0 + for y in comp_data["y [%]"]: + remaining -= y + starts.append(remaining) + widths.append(y) + starts.append(remaining - loss_percent) + widths.append(loss_percent) + starts.append(0.0) + widths.append(epsilon_total) + + fig = go.Figure( + go.Bar( + orientation="h", + y=names, + x=widths, + base=starts, + text=text, + textposition="inside", + marker={"color": bar_colors}, + hovertemplate="%{y}", + ) ) - fig = go.Figure(go.Waterfall( - orientation="h", - measure=measures, - y=names, - x=x_values, - text=text, - textposition="inside", - connector={"line": {"color": "#9E9E9E", "dash": "dot", "width": 1}}, - decreasing={"marker": {"color": palette["destruction"]}}, - increasing={"marker": {"color": palette["destruction"]}}, - totals={"marker": {"color": palette["product"]}}, - marker={"color": bar_colors}, - )) - fig.update_layout( title_text=title or "Exergy Waterfall Diagram", xaxis_title="Exergy [%]", xaxis={"range": [0, 105]}, + yaxis={"autorange": "reversed"}, font_size=12, showlegend=False, + bargap=0.3, ) if show_plot: From ff561388c1efedd5681fe029278a973bdb634f54 Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:09:37 +0200 Subject: [PATCH 06/11] Feat: Add visualization usage example for the ccpp case --- .gitignore | 5 +- examples/visualization/ccpp_diagrams.py | 86 +++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 examples/visualization/ccpp_diagrams.py diff --git a/.gitignore b/.gitignore index ac67ba7..92a9781 100644 --- a/.gitignore +++ b/.gitignore @@ -156,4 +156,7 @@ examples/exergoeconomic_analysis/heatpump/ examples/opt_examples/ # macOS -.DS_Store \ No newline at end of file +.DS_Store +# Generated visualization example outputs +examples/visualization/*.html +examples/visualization/*.png diff --git a/examples/visualization/ccpp_diagrams.py b/examples/visualization/ccpp_diagrams.py new file mode 100644 index 0000000..ba25478 --- /dev/null +++ b/examples/visualization/ccpp_diagrams.py @@ -0,0 +1,86 @@ +""" +Demonstrate the exerpy visualization module (Sankey and waterfall diagrams). + +The script loads the exported exergy analysis results of the combined cycle +power plant (ccpp) example, so no simulator is required to run it: + + python examples/visualization/ccpp_diagrams.py + +The interactive diagrams are written as HTML files next to this script and +opened in the default browser (pass --no-show to only write the files). +""" + +import sys +import webbrowser +from pathlib import Path + +from exerpy import ExergyAnalysis + +HERE = Path(__file__).parent + +# ---------------------------------------------------------------------------------------------------------------------- +# 1. Load the ccpp example results and run the exergy analysis +# ---------------------------------------------------------------------------------------------------------------------- +ean = ExergyAnalysis.from_json(str(HERE / ".." / "exergy_analysis" / "ccpp" / "ccpp_tespy.json")) + +fuel = {"inputs": ["1", "3"], "outputs": []} +product = {"inputs": ["e15", "h1"], "outputs": []} +loss = {"inputs": ["8", "15"], "outputs": ["14"]} + +ean.analyse(E_F=fuel, E_P=product, E_L=loss) + +# ---------------------------------------------------------------------------------------------------------------------- +# 2. Sankey diagrams +# ---------------------------------------------------------------------------------------------------------------------- +# Mode 1: one link per connection carrying the total exergy flow E. +ean.plot_sankey( + mode=1, + title="CCPP – Sankey diagram (total exergy)", + output_path=str(HERE / "ccpp_sankey_total.html"), +) + +# Mode 2: material links are split into physical (E_PH) and chemical (E_CH) +# exergy. Requires an analysis with chemical exergy enabled. +ean.plot_sankey( + mode=2, + title="CCPP – Sankey diagram (physical + chemical exergy)", + output_path=str(HERE / "ccpp_sankey_split.html"), +) + +# Components can be aggregated into groups and nodes can be colored +# individually. Links inside a group are hidden. +ean.plot_sankey( + groups={"Steam cycle": ["DEA", "COND", "FP", "CP", "DP"]}, + node_colors={"CC": "#C62828", "GT": "#388E3C"}, + title="CCPP – Sankey diagram (grouped steam cycle)", + output_path=str(HERE / "ccpp_sankey_grouped.html"), +) + +# ---------------------------------------------------------------------------------------------------------------------- +# 3. Waterfall diagrams +# ---------------------------------------------------------------------------------------------------------------------- +# Interactive plotly variant. Bar colors can be customized via the `colors` +# dict with the keys "fuel", "destruction", "loss" and "product". +fig = ean.plot_exergy_waterfall_plotly(title="CCPP – Exergy waterfall", show_plot=False) +fig.write_html(HERE / "ccpp_waterfall.html") + +# Static matplotlib variant of the same diagram, exported as PNG. +fig, ax = ean.plot_exergy_waterfall(title="CCPP – Exergy waterfall", show_plot=False) +fig.savefig(HERE / "ccpp_waterfall.png", dpi=150, bbox_inches="tight") + +# ---------------------------------------------------------------------------------------------------------------------- +# 4. Show the results +# ---------------------------------------------------------------------------------------------------------------------- +html_files = [ + HERE / "ccpp_sankey_total.html", + HERE / "ccpp_sankey_split.html", + HERE / "ccpp_sankey_grouped.html", + HERE / "ccpp_waterfall.html", +] + +for path in html_files + [HERE / "ccpp_waterfall.png"]: + print(f"written: {path}") + +if "--no-show" not in sys.argv: + for path in html_files: + webbrowser.open(path.resolve().as_uri()) From f051e39a57e018e0ad8fe59edad1591d43e63f14 Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:37:09 +0200 Subject: [PATCH 07/11] Refactor: remove plotly import fallback, plotly is a required dependency --- src/exerpy/visualization/sankey.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/exerpy/visualization/sankey.py b/src/exerpy/visualization/sankey.py index e8766c0..0dcb3bd 100644 --- a/src/exerpy/visualization/sankey.py +++ b/src/exerpy/visualization/sankey.py @@ -110,10 +110,7 @@ def build(self) -> tuple[list[dict], list[dict]]: def to_plotly(self, title: str | None = None): """Return a Plotly ``Figure`` containing the Sankey trace.""" - try: - import plotly.graph_objects as go - except ImportError as exc: - raise ImportError("plotly is required for Sankey diagrams: pip install plotly") from exc + import plotly.graph_objects as go nodes, links = self.build() fig = go.Figure( From ef016fcd0f257f682582cd2cca9d5a644453807c Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:37:09 +0200 Subject: [PATCH 08/11] Test: add tests for the Sankey and waterfall visualization module --- tests/test_visualization.py | 205 ++++++++++++++++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 tests/test_visualization.py diff --git a/tests/test_visualization.py b/tests/test_visualization.py new file mode 100644 index 0000000..603fc88 --- /dev/null +++ b/tests/test_visualization.py @@ -0,0 +1,205 @@ +import os + +import matplotlib + +matplotlib.use("Agg") + +import pytest + +from exerpy import ExergyAnalysis +from exerpy.visualization import SankeyBuilder + +_CCPP_JSON = os.path.join(os.path.dirname(__file__), "../examples/exergy_analysis/ccpp/ccpp_tespy.json") + +_TERMINAL_LABELS = {"E_F (Fuel)", "E_P (Product)", "E_D (Destruction)", "E_L (Loss)", "E_L net"} + + +@pytest.fixture(scope="module") +def ccpp(): + ean = ExergyAnalysis.from_json(_CCPP_JSON) + ean.analyse( + E_F={"inputs": ["1", "3"], "outputs": []}, + E_P={"inputs": ["e15", "h1"], "outputs": []}, + E_L={"inputs": ["8", "15"], "outputs": ["14"]}, + ) + return ean + + +def _flows(nodes, links): + """Return per-node (inflow, outflow) sums keyed by node id.""" + inflow = {n["id"]: 0.0 for n in nodes} + outflow = {n["id"]: 0.0 for n in nodes} + for lk in links: + outflow[nodes[lk["source"]]["id"]] += lk["value"] + inflow[nodes[lk["target"]]["id"]] += lk["value"] + return inflow, outflow + + +class TestSankeyCcpp: + def test_node_and_link_count(self, ccpp): + # 26 components - 1 collapsed CycleCloser + 4 terminals + 1 E_L net helper + nodes, links = SankeyBuilder(ccpp, mode=1).build() + assert len(nodes) == 30 + assert len(links) == 64 + + def test_mode_2_splits_material_links(self, ccpp): + nodes, links = SankeyBuilder(ccpp, mode=2).build() + assert len(nodes) == 30 + assert len(links) > 64 + sub_labels = {lk["label"].split("[")[1].split("]")[0] for lk in links if "[" in lk["label"]} + assert "E_PH" in sub_labels + assert "E_CH" in sub_labels + + def test_mode_3_requires_split_physical_exergy(self, ccpp): + with pytest.raises(ValueError): + SankeyBuilder(ccpp, mode=3) + + def test_all_link_values_positive(self, ccpp): + _, links = SankeyBuilder(ccpp, mode=1).build() + assert all(lk["value"] > 0 for lk in links) + + def test_component_nodes_balance(self, ccpp): + # Everything entering a component node must leave it (to other + # components, terminals or E_D). Only terminal nodes are unbalanced. + nodes, links = SankeyBuilder(ccpp, mode=1).build() + inflow, outflow = _flows(nodes, links) + for node in nodes: + if node["label"] in _TERMINAL_LABELS: + continue + assert inflow[node["id"]] == pytest.approx(outflow[node["id"]], rel=1e-3) + + def test_destruction_total_matches_analysis(self, ccpp): + # Regression: Mixer destruction (deaerator) used to be dropped. + nodes, links = SankeyBuilder(ccpp, mode=1).build() + inflow, _ = _flows(nodes, links) + assert inflow["__E_D__"] == pytest.approx(ccpp.E_D, rel=1e-6) + dea_ed = [lk for lk in links if nodes[lk["source"]]["id"] == "DEA" and nodes[lk["target"]]["id"] == "__E_D__"] + assert len(dea_ed) == 1 + + def test_loss_helper_node(self, ccpp): + # E_L has both inputs and outputs in the ccpp definition, so an + # intermediate "E_L net" node must be present; E_P has only inputs. + nodes, links = SankeyBuilder(ccpp, mode=1).build() + ids = {n["id"] for n in nodes} + assert "__E_L_net__" in ids + assert "__E_P_net__" not in ids + inflow, outflow = _flows(nodes, links) + # gross loss flows enter the helper node, returns and the net loss leave it + assert inflow["__E_L_net__"] == pytest.approx(outflow["__E_L_net__"], rel=1e-3) + # the terminal E_L node receives exactly the net loss of the analysis + assert inflow["__E_L__"] == pytest.approx(ccpp.E_L, rel=1e-6) + + def test_collapse_passthroughs(self, ccpp): + nodes_collapsed, _ = SankeyBuilder(ccpp, collapse_passthroughs=True).build() + nodes_full, _ = SankeyBuilder(ccpp, collapse_passthroughs=False).build() + ids_collapsed = {n["id"] for n in nodes_collapsed} + ids_full = {n["id"] for n in nodes_full} + assert "cycle closer" not in ids_collapsed + assert "cycle closer" in ids_full + assert len(nodes_full) == len(nodes_collapsed) + 1 + + def test_grouping(self, ccpp): + members = ["ECO", "EVA", "SH", "drum", "drum pump"] + nodes, links = SankeyBuilder(ccpp, groups={"HRSG": members}).build() + ids = {n["id"] for n in nodes} + assert "HRSG" in ids + assert not set(members) & ids + # 5 members merge into 1 node; internal links disappear + assert len(nodes) == 26 + assert len(links) < 64 + + def test_node_colors_override(self, ccpp): + nodes, _ = SankeyBuilder(ccpp, node_colors={"CC": "#C62828"}).build() + by_id = {n["id"]: n for n in nodes} + assert by_id["CC"]["color"] == "#C62828" + assert by_id["GT"]["color"] != "#C62828" + + def test_to_plotly_figure(self, ccpp): + fig = ccpp.plot_sankey(mode=1, title="test") + trace = fig.data[0] + assert trace.type == "sankey" + assert len(trace.node.label) == 30 + assert len(trace.link.value) == 64 + + +class TestSankeyEdgeCases: + @staticmethod + def _fake_analysis(E_ab): + """Minimal duck-typed analysis with a single A -> B connection.""" + + class _Analysis: + pass + + analysis = _Analysis() + analysis.E_F = 1000.0 + analysis.E_P = 800.0 + analysis.E_L = 0.0 + analysis.chemical_exergy_enabled = False + analysis.split_physical_exergy = False + analysis.components = { + "A": type("Turbine", (), {"E_D": 50.0})(), + "B": type("Generator", (), {"E_D": None})(), + } + analysis.connections = { + "fuel": {"kind": "power", "source_component": None, "target_component": "A", "E": 1000.0}, + "ab": {"kind": "power", "source_component": "A", "target_component": "B", "E": E_ab}, + "prod": {"kind": "power", "source_component": "B", "target_component": None, "E": 800.0}, + } + analysis.E_F_dict = {"inputs": ["fuel"], "outputs": []} + analysis.E_P_dict = {"inputs": ["prod"], "outputs": []} + analysis.E_L_dict = {"inputs": [], "outputs": []} + return analysis + + def test_negative_exergy_flow_is_dropped(self): + # Negative flows cannot be rendered by a Sankey; they must be + # dropped instead of producing a corrupt diagram. + nodes, links = SankeyBuilder(self._fake_analysis(E_ab=-100.0)).build() + ab_links = [lk for lk in links if nodes[lk["source"]]["id"] == "A" and nodes[lk["target"]]["id"] == "B"] + assert ab_links == [] + assert all(lk["value"] > 0 for lk in links) + + def test_positive_exergy_flow_is_kept(self): + nodes, links = SankeyBuilder(self._fake_analysis(E_ab=900.0)).build() + ab_links = [lk for lk in links if nodes[lk["source"]]["id"] == "A" and nodes[lk["target"]]["id"] == "B"] + assert len(ab_links) == 1 + assert ab_links[0]["value"] == pytest.approx(900.0) + + def test_requires_analysed_instance(self): + class _Empty: + pass + + with pytest.raises(RuntimeError): + SankeyBuilder(_Empty()) + + +class TestWaterfall: + def test_matplotlib_waterfall(self, ccpp): + fig, ax = ccpp.plot_exergy_waterfall(title="test", show_plot=False) + # fuel bar + one bar per component with valid destruction + product bar + n_bars = len(ax.patches) + assert n_bars > 2 + assert ax.patches[0].get_width() == pytest.approx(100.0) + + def test_plotly_waterfall(self, ccpp): + fig = ccpp.plot_exergy_waterfall_plotly(title="test", show_plot=False) + bar = fig.data[0] + assert bar.type == "bar" + # first bar: fuel from 0 to 100 % + assert bar.base[0] == pytest.approx(0.0) + assert bar.x[0] == pytest.approx(100.0) + # last bar: product from 0 to epsilon_total + df, _, _ = ccpp.exergy_results(print_results=False) + epsilon_total = df[df["Component"] == "TOT"].iloc[0]["epsilon [%]"] + assert bar.base[-1] == pytest.approx(0.0) + assert bar.x[-1] == pytest.approx(epsilon_total) + + def test_plotly_waterfall_custom_colors(self, ccpp): + fig = ccpp.plot_exergy_waterfall_plotly(show_plot=False, colors={"fuel": "#111111", "product": "#222222"}) + bar_colors = fig.data[0].marker.color + assert bar_colors[0] == "#111111" + assert bar_colors[-1] == "#222222" + + def test_exclude_components(self, ccpp): + fig_all = ccpp.plot_exergy_waterfall_plotly(show_plot=False) + fig_less = ccpp.plot_exergy_waterfall_plotly(show_plot=False, exclude_components=["CC"]) + assert len(fig_less.data[0].x) == len(fig_all.data[0].x) - 1 From bf15fe6ca74e52c916c6b9838e741d0055ed1a80 Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:37:09 +0200 Subject: [PATCH 09/11] Feat: add visualization examples for ccpp, cgam and heat pump with docs page --- docs/examples.rst | 18 ++++ docs/examples/visualization.rst | 122 ++++++++++++++++++++++++ examples/visualization/ccpp_diagrams.py | 29 ++---- examples/visualization/cgam_diagrams.py | 73 ++++++++++++++ examples/visualization/hp_diagrams.py | 64 +++++++++++++ 5 files changed, 286 insertions(+), 20 deletions(-) create mode 100644 docs/examples/visualization.rst create mode 100644 examples/visualization/cgam_diagrams.py create mode 100644 examples/visualization/hp_diagrams.py diff --git a/docs/examples.rst b/docs/examples.rst index 9563ee6..3ead787 100644 --- a/docs/examples.rst +++ b/docs/examples.rst @@ -125,3 +125,21 @@ In the following example, we demonstrate how to extend the exergy analysis with Exergoeconomic analysis with manually defined component investment costs and input stream specific costs. + +In the following example, we demonstrate how to visualize the results of an exergy analysis with Sankey and waterfall diagrams. + +.. toctree:: + :maxdepth: 1 + :hidden: + + examples/visualization.rst + +.. card:: + :link: examples_visualization_label + :link-type: ref + + **Visualization** + ^^^ + + Interactive Sankey diagrams of the exergy flows and waterfall diagrams of + the exergy destruction per component for the three example systems. diff --git a/docs/examples/visualization.rst b/docs/examples/visualization.rst new file mode 100644 index 0000000..30fca72 --- /dev/null +++ b/docs/examples/visualization.rst @@ -0,0 +1,122 @@ +.. _examples_visualization_label: + +************* +Visualization +************* + +ExerPy can visualize the results of a completed exergy analysis as an +interactive Sankey diagram of all exergy flows and as a waterfall diagram of +the exergy destruction per component. The examples below build the diagrams +from the exported analysis results of the three example systems, so no +simulator is required to run them, e.g.: + +.. code-block:: bash + + python examples/visualization/ccpp_diagrams.py + +Sankey diagram +============== + +:code:`plot_sankey` renders the exergy flows between all components of the +system together with four terminal nodes for the exergetic fuel (E_F), product +(E_P), destruction (E_D) and loss (E_L). If a fuel, product or loss definition +contains both inputs and outputs, an intermediate "net" node displays the +gross flows next to the net value. The most important options are: + +- :code:`mode` – level of detail of the material links: :code:`1` shows the + total exergy flow E per connection, :code:`2` splits each material link into + physical and chemical exergy (requires chemical exergy), and :code:`3` + splits into thermal, mechanical and chemical exergy (requires + :code:`split_physical_exergy=True`). +- :code:`collapse_passthroughs` – hides pure pass-through components + (by default the :code:`CycleCloser`). +- :code:`groups` – aggregates several components into a single node and hides + the connections between them, e.g. to represent a subsystem. +- :code:`node_colors` – overrides the color of individual component nodes. + Link colors are assigned automatically based on the type of flow (power, + heat, or material by fluid). +- :code:`output_path` – exports the interactive diagram to an HTML file. + +Waterfall diagram +================= + +:code:`plot_exergy_waterfall` (Matplotlib) and +:code:`plot_exergy_waterfall_plotly` (interactive Plotly variant) display the +exergy destruction per component as percentages of the total fuel exergy, from +the exergetic fuel (100 %) down to the exergetic product. The bar colors can +be customized with the :code:`colors` dict using the keys :code:`"fuel"`, +:code:`"destruction"`, :code:`"loss"` and :code:`"product"`. + +Examples +======== + +.. tab-set:: + + .. tab-item:: CCPP + + 1. **Load the results and run the analysis** + + .. literalinclude:: /../examples/visualization/ccpp_diagrams.py + :language: python + :start-after: [analysis_section] + :end-before: [sankey_section] + + 2. **Create the Sankey diagrams** + + .. literalinclude:: /../examples/visualization/ccpp_diagrams.py + :language: python + :start-after: [sankey_section] + :end-before: [waterfall_section] + + 3. **Create the waterfall diagrams** + + .. literalinclude:: /../examples/visualization/ccpp_diagrams.py + :language: python + :start-after: [waterfall_section] + :end-before: [show_section] + + .. tab-item:: Heat Pump + + 1. **Load the results and run the analysis** + + .. literalinclude:: /../examples/visualization/hp_diagrams.py + :language: python + :start-after: [analysis_section] + :end-before: [sankey_section] + + 2. **Create the Sankey diagrams** + + .. literalinclude:: /../examples/visualization/hp_diagrams.py + :language: python + :start-after: [sankey_section] + :end-before: [waterfall_section] + + 3. **Create the waterfall diagrams** + + .. literalinclude:: /../examples/visualization/hp_diagrams.py + :language: python + :start-after: [waterfall_section] + :end-before: [show_section] + + .. tab-item:: CGAM + + 1. **Load the results and run the analysis** + + .. literalinclude:: /../examples/visualization/cgam_diagrams.py + :language: python + :start-after: [analysis_section] + :end-before: [sankey_section] + + 2. **Create the Sankey diagrams** + + .. literalinclude:: /../examples/visualization/cgam_diagrams.py + :language: python + :start-after: [sankey_section] + :end-before: [waterfall_section] + + 3. **Create the waterfall diagrams** + + .. literalinclude:: /../examples/visualization/cgam_diagrams.py + :language: python + :start-after: [waterfall_section] + :end-before: [show_section] diff --git a/examples/visualization/ccpp_diagrams.py b/examples/visualization/ccpp_diagrams.py index ba25478..7d66c5a 100644 --- a/examples/visualization/ccpp_diagrams.py +++ b/examples/visualization/ccpp_diagrams.py @@ -1,8 +1,8 @@ """ -Demonstrate the exerpy visualization module (Sankey and waterfall diagrams). +Sankey and waterfall diagrams for the combined cycle power plant (ccpp) example. -The script loads the exported exergy analysis results of the combined cycle -power plant (ccpp) example, so no simulator is required to run it: +The script loads the exported exergy analysis results, so no simulator is +required to run it: python examples/visualization/ccpp_diagrams.py @@ -18,9 +18,7 @@ HERE = Path(__file__).parent -# ---------------------------------------------------------------------------------------------------------------------- -# 1. Load the ccpp example results and run the exergy analysis -# ---------------------------------------------------------------------------------------------------------------------- +# [analysis_section] ean = ExergyAnalysis.from_json(str(HERE / ".." / "exergy_analysis" / "ccpp" / "ccpp_tespy.json")) fuel = {"inputs": ["1", "3"], "outputs": []} @@ -28,10 +26,7 @@ loss = {"inputs": ["8", "15"], "outputs": ["14"]} ean.analyse(E_F=fuel, E_P=product, E_L=loss) - -# ---------------------------------------------------------------------------------------------------------------------- -# 2. Sankey diagrams -# ---------------------------------------------------------------------------------------------------------------------- +# [sankey_section] # Mode 1: one link per connection carrying the total exergy flow E. ean.plot_sankey( mode=1, @@ -50,15 +45,12 @@ # Components can be aggregated into groups and nodes can be colored # individually. Links inside a group are hidden. ean.plot_sankey( - groups={"Steam cycle": ["DEA", "COND", "FP", "CP", "DP"]}, + groups={"HRSG": ["ECO", "EVA", "SH", "drum", "drum pump"]}, node_colors={"CC": "#C62828", "GT": "#388E3C"}, - title="CCPP – Sankey diagram (grouped steam cycle)", + title="CCPP – Sankey diagram (grouped heat recovery steam generator)", output_path=str(HERE / "ccpp_sankey_grouped.html"), ) - -# ---------------------------------------------------------------------------------------------------------------------- -# 3. Waterfall diagrams -# ---------------------------------------------------------------------------------------------------------------------- +# [waterfall_section] # Interactive plotly variant. Bar colors can be customized via the `colors` # dict with the keys "fuel", "destruction", "loss" and "product". fig = ean.plot_exergy_waterfall_plotly(title="CCPP – Exergy waterfall", show_plot=False) @@ -67,10 +59,7 @@ # Static matplotlib variant of the same diagram, exported as PNG. fig, ax = ean.plot_exergy_waterfall(title="CCPP – Exergy waterfall", show_plot=False) fig.savefig(HERE / "ccpp_waterfall.png", dpi=150, bbox_inches="tight") - -# ---------------------------------------------------------------------------------------------------------------------- -# 4. Show the results -# ---------------------------------------------------------------------------------------------------------------------- +# [show_section] html_files = [ HERE / "ccpp_sankey_total.html", HERE / "ccpp_sankey_split.html", diff --git a/examples/visualization/cgam_diagrams.py b/examples/visualization/cgam_diagrams.py new file mode 100644 index 0000000..b0242e9 --- /dev/null +++ b/examples/visualization/cgam_diagrams.py @@ -0,0 +1,73 @@ +""" +Sankey and waterfall diagrams for the CGAM example. + +The script loads the exported exergy analysis results, so no simulator is +required to run it: + + python examples/visualization/cgam_diagrams.py + +The interactive diagrams are written as HTML files next to this script and +opened in the default browser (pass --no-show to only write the files). +""" + +import sys +import webbrowser +from pathlib import Path + +from exerpy import ExergyAnalysis + +HERE = Path(__file__).parent + +# [analysis_section] +ean = ExergyAnalysis.from_json(str(HERE / ".." / "exergy_analysis" / "cgam" / "cgam_tespy.json")) + +fuel = {"inputs": ["1", "10"], "outputs": []} +product = {"inputs": ["e3", "9"], "outputs": ["8"]} +loss = {"inputs": ["7"], "outputs": []} + +ean.analyse(E_F=fuel, E_P=product, E_L=loss) +# [sankey_section] +# Mode 1: one link per connection carrying the total exergy flow E. Since the +# product definition has inputs and outputs, an intermediate "E_P net" node +# shows the gross product flows next to the net product. +ean.plot_sankey( + mode=1, + title="CGAM – Sankey diagram (total exergy)", + output_path=str(HERE / "cgam_sankey_total.html"), +) + +# Mode 2: material links are split into physical (E_PH) and chemical (E_CH) +# exergy. Requires an analysis with chemical exergy enabled. +ean.plot_sankey( + mode=2, + title="CGAM – Sankey diagram (physical + chemical exergy)", + output_path=str(HERE / "cgam_sankey_split.html"), +) + +# Group the heat recovery steam generator into a single node. +ean.plot_sankey( + groups={"HRSG": ["EV", "PH", "DRUM"]}, + node_colors={"CC": "#C62828", "EXP": "#388E3C"}, + title="CGAM – Sankey diagram (grouped heat recovery steam generator)", + output_path=str(HERE / "cgam_sankey_grouped.html"), +) +# [waterfall_section] +fig = ean.plot_exergy_waterfall_plotly(title="CGAM – Exergy waterfall", show_plot=False) +fig.write_html(HERE / "cgam_waterfall.html") + +fig, ax = ean.plot_exergy_waterfall(title="CGAM – Exergy waterfall", show_plot=False) +fig.savefig(HERE / "cgam_waterfall.png", dpi=150, bbox_inches="tight") +# [show_section] +html_files = [ + HERE / "cgam_sankey_total.html", + HERE / "cgam_sankey_split.html", + HERE / "cgam_sankey_grouped.html", + HERE / "cgam_waterfall.html", +] + +for path in html_files + [HERE / "cgam_waterfall.png"]: + print(f"written: {path}") + +if "--no-show" not in sys.argv: + for path in html_files: + webbrowser.open(path.resolve().as_uri()) diff --git a/examples/visualization/hp_diagrams.py b/examples/visualization/hp_diagrams.py new file mode 100644 index 0000000..da21f23 --- /dev/null +++ b/examples/visualization/hp_diagrams.py @@ -0,0 +1,64 @@ +""" +Sankey and waterfall diagrams for the air source heat pump example. + +The script loads the exported exergy analysis results, so no simulator is +required to run it: + + python examples/visualization/hp_diagrams.py + +The interactive diagrams are written as HTML files next to this script and +opened in the default browser (pass --no-show to only write the files). +""" + +import sys +import webbrowser +from pathlib import Path + +from exerpy import ExergyAnalysis + +HERE = Path(__file__).parent + +# [analysis_section] +ean = ExergyAnalysis.from_json(str(HERE / ".." / "exergy_analysis" / "heatpump" / "hp_tespy.json")) + +fuel = {"inputs": ["e1"], "outputs": []} +product = {"inputs": ["23"], "outputs": ["21"]} +loss = {"inputs": ["13"], "outputs": ["11"]} + +ean.analyse(E_F=fuel, E_P=product, E_L=loss) +# [sankey_section] +# The heat pump analysis runs without chemical exergy, so only mode 1 is +# available. Product and loss definitions both have inputs and outputs, so the +# diagram shows intermediate "E_P net" and "E_L net" nodes. +ean.plot_sankey( + mode=1, + title="Heat pump – Sankey diagram (total exergy)", + output_path=str(HERE / "hp_sankey_total.html"), +) + +# Group the electric drives into a single node. +ean.plot_sankey( + groups={"Drives": ["electricity distribution", "MOT1", "MOT2", "MOT3"]}, + node_colors={"COMP": "#C62828", "COND": "#388E3C"}, + title="Heat pump – Sankey diagram (grouped drives)", + output_path=str(HERE / "hp_sankey_grouped.html"), +) +# [waterfall_section] +fig = ean.plot_exergy_waterfall_plotly(title="Heat pump – Exergy waterfall", show_plot=False) +fig.write_html(HERE / "hp_waterfall.html") + +fig, ax = ean.plot_exergy_waterfall(title="Heat pump – Exergy waterfall", show_plot=False) +fig.savefig(HERE / "hp_waterfall.png", dpi=150, bbox_inches="tight") +# [show_section] +html_files = [ + HERE / "hp_sankey_total.html", + HERE / "hp_sankey_grouped.html", + HERE / "hp_waterfall.html", +] + +for path in html_files + [HERE / "hp_waterfall.png"]: + print(f"written: {path}") + +if "--no-show" not in sys.argv: + for path in html_files: + webbrowser.open(path.resolve().as_uri()) From ee87cd8c3bb51e968f2b1cc5795c1d6f9fefaf2b Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:52:31 +0200 Subject: [PATCH 10/11] Build: move plotly and matplotlib to an optional [viz] extra --- docs/examples/visualization.rst | 12 +++++++++--- pyproject.toml | 11 +++++++++-- src/exerpy/visualization/sankey.py | 5 ++++- src/exerpy/visualization/waterfall.py | 10 ++++++++-- 4 files changed, 30 insertions(+), 8 deletions(-) diff --git a/docs/examples/visualization.rst b/docs/examples/visualization.rst index 30fca72..7665818 100644 --- a/docs/examples/visualization.rst +++ b/docs/examples/visualization.rst @@ -6,9 +6,15 @@ Visualization ExerPy can visualize the results of a completed exergy analysis as an interactive Sankey diagram of all exergy flows and as a waterfall diagram of -the exergy destruction per component. The examples below build the diagrams -from the exported analysis results of the three example systems, so no -simulator is required to run them, e.g.: +the exergy destruction per component. The plotting libraries are an optional +dependency, install them with: + +.. code-block:: bash + + pip install exerpy[viz] + +The examples below build the diagrams from the exported analysis results of +the three example systems, so no simulator is required to run them, e.g.: .. code-block:: bash diff --git a/pyproject.toml b/pyproject.toml index 1f4bb1d..be73d43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,9 @@ exclude = [ "examples/exergoeconomic_analysis/cgam/**", "examples/exergoeconomic_analysis/ccpp/**", "examples/exergoeconomic_analysis/heatpump/**", + "examples/visualization/*.html", + "examples/visualization/*.png", + "**/.DS_Store", ] [project] @@ -52,8 +55,6 @@ requires-python = ">=3.11" dependencies = [ "pandas", "numpy", - "plotly>=6.8", - "matplotlib>=3.10", ] license = {text = "MIT"} @@ -64,11 +65,17 @@ Changelog = "https://exerpy.readthedocs.io/en/latest/whats_new.html" "Issue Tracker" = "https://github.com/oemof/exerpy/issues" [project.optional-dependencies] +viz = [ + "plotly>=6.8", + "matplotlib>=3.10", +] dev = [ "build", "flit", "furo", "isort", + "matplotlib>=3.10", + "plotly>=6.8", "pytest", "sphinx>=7.2.2", "sphinx-copybutton", diff --git a/src/exerpy/visualization/sankey.py b/src/exerpy/visualization/sankey.py index 0dcb3bd..6a62321 100644 --- a/src/exerpy/visualization/sankey.py +++ b/src/exerpy/visualization/sankey.py @@ -110,7 +110,10 @@ def build(self) -> tuple[list[dict], list[dict]]: def to_plotly(self, title: str | None = None): """Return a Plotly ``Figure`` containing the Sankey trace.""" - import plotly.graph_objects as go + try: + import plotly.graph_objects as go + except ImportError as exc: + raise ImportError("plotly is required for Sankey diagrams: pip install exerpy[viz]") from exc nodes, links = self.build() fig = go.Figure( diff --git a/src/exerpy/visualization/waterfall.py b/src/exerpy/visualization/waterfall.py index 2610358..7a5a4b4 100644 --- a/src/exerpy/visualization/waterfall.py +++ b/src/exerpy/visualization/waterfall.py @@ -110,7 +110,10 @@ def plot_exergy_waterfall( print_exergy_summary : Print a text summary of exergy analysis. plot_exergy_waterfall_plotly : Plotly version of this diagram. """ - import matplotlib.pyplot as plt + try: + import matplotlib.pyplot as plt + except ImportError as exc: + raise ImportError("matplotlib is required for waterfall diagrams: pip install exerpy[viz]") from exc import numpy as np if not hasattr(analysis, "epsilon") or analysis.epsilon is None: @@ -225,7 +228,10 @@ def plot_exergy_waterfall_plotly( -------- plot_exergy_waterfall : Matplotlib version of this diagram. """ - import plotly.graph_objects as go + try: + import plotly.graph_objects as go + except ImportError as exc: + raise ImportError("plotly is required for waterfall diagrams: pip install exerpy[viz]") from exc if not hasattr(analysis, "epsilon") or analysis.epsilon is None: raise RuntimeError("Exergy analysis has not been performed yet. Please call analyse() first.") From 8a505b8b6a82ed2f765319ecc274b4055f7ba97c Mon Sep 17 00:00:00 2001 From: sertomas Date: Fri, 17 Jul 2026 15:52:31 +0200 Subject: [PATCH 11/11] Test: add synthetic cycle tests for the Sankey builder --- tests/test_visualization.py | 245 ++++++++++++++++++++++++++++++++++++ 1 file changed, 245 insertions(+) diff --git a/tests/test_visualization.py b/tests/test_visualization.py index 603fc88..c3f9ae2 100644 --- a/tests/test_visualization.py +++ b/tests/test_visualization.py @@ -172,6 +172,251 @@ class _Empty: SankeyBuilder(_Empty()) +def _synthetic_analysis(components, connections, E_F, E_P, E_L, chemical=False, split=False): + """Duck-typed stand-in for an analysed ExergyAnalysis instance. + + components: ``{name: (component_class_name, E_D)}`` + connections: ``{name: {...}}`` with the same keys as parsed connection data + """ + + class _Analysis: + pass + + def _net(d): + ins = sum(connections[c]["E"] for c in d.get("inputs", [])) + outs = sum(connections[c]["E"] for c in d.get("outputs", [])) + return ins - outs + + analysis = _Analysis() + analysis.components = {name: type(cls, (), {"E_D": e_d})() for name, (cls, e_d) in components.items()} + analysis.connections = connections + analysis.E_F_dict, analysis.E_P_dict, analysis.E_L_dict = E_F, E_P, E_L + analysis.E_F, analysis.E_P, analysis.E_L = _net(E_F), _net(E_P), _net(E_L) + analysis.chemical_exergy_enabled = chemical + analysis.split_physical_exergy = split + return analysis + + +def _material(src, tgt, E, composition=None, **extra): + conn = { + "kind": "material", + "source_component": src, + "target_component": tgt, + "E": E, + "mass_composition": composition or {"water": 1}, + } + conn.update(extra) + return conn + + +def _rankine(): + """Closed pure-fluid cycle: single fuel input (heat), single product output (power).""" + components = { + "PUMP": ("Pump", 5.0), + "BOIL": ("SimpleHeatExchanger", 100.0), + "TURB": ("Turbine", 45.0), + "COND": ("HeatExchanger", 50.0), + "cc": ("CycleCloser", None), + } + connections = { + "1": _material("PUMP", "BOIL", 295.0), + "2": _material("BOIL", "TURB", 1195.0), + "3": _material("TURB", "COND", 350.0), + "4": _material("COND", "cc", 300.0), + "5": _material("cc", "PUMP", 300.0), + "q": {"kind": "heat", "source_component": None, "target_component": "BOIL", "E": 1000.0}, + "p": {"kind": "power", "source_component": "TURB", "target_component": None, "E": 800.0}, + } + return _synthetic_analysis( + components, + connections, + E_F={"inputs": ["q"], "outputs": []}, + E_P={"inputs": ["p"], "outputs": []}, + E_L={"inputs": [], "outputs": []}, + ) + + +class TestSyntheticCycles: + def test_simple_cycle_structure(self): + nodes, links = SankeyBuilder(_rankine()).build() + ids = {n["id"] for n in nodes} + # CycleCloser is collapsed, terminals are always added + assert "cc" not in ids + assert len(nodes) == 4 + 4 + # 4 material links (loop closed through the collapsed cc), 4 E_D links, + # 1 fuel link, 1 product link + assert len(links) == 10 + + def test_simple_cycle_balances(self): + analysis = _rankine() + nodes, links = SankeyBuilder(analysis).build() + inflow, outflow = _flows(nodes, links) + for comp in ("PUMP", "BOIL", "TURB", "COND"): + assert inflow[comp] == pytest.approx(outflow[comp]) + assert inflow["__E_D__"] == pytest.approx(200.0) + assert inflow["__E_P__"] == pytest.approx(analysis.E_P) + + def test_simple_cycle_loop_closed_through_collapsed_node(self): + nodes, links = SankeyBuilder(_rankine()).build() + cond_pump = [lk for lk in links if nodes[lk["source"]]["id"] == "COND" and nodes[lk["target"]]["id"] == "PUMP"] + assert len(cond_pump) == 1 + assert cond_pump[0]["value"] == pytest.approx(300.0) + + def test_pure_fluid_link_color(self): + # water links must use the water base color (#2196F3) + nodes, links = SankeyBuilder(_rankine()).build() + material = [lk for lk in links if nodes[lk["source"]]["id"] == "PUMP"] + assert material[0]["color"].startswith("rgba(33,150,243") + + def test_mode_2_splits_physical_and_chemical(self): + components = {"A": ("Turbine", 0.0), "B": ("Generator", 0.0)} + connections = { + "f": {"kind": "power", "source_component": None, "target_component": "A", "E": 1000.0}, + "ab": _material("A", "B", 1000.0, E_PH=700.0, E_CH=300.0), + "p": {"kind": "power", "source_component": "B", "target_component": None, "E": 1000.0}, + } + analysis = _synthetic_analysis( + components, + connections, + E_F={"inputs": ["f"], "outputs": []}, + E_P={"inputs": ["p"], "outputs": []}, + E_L={"inputs": [], "outputs": []}, + chemical=True, + ) + nodes, links = SankeyBuilder(analysis, mode=2).build() + ab = [lk for lk in links if nodes[lk["source"]]["id"] == "A" and nodes[lk["target"]]["id"] == "B"] + assert len(ab) == 2 + assert sorted(lk["value"] for lk in ab) == [300.0, 700.0] + assert sum(lk["value"] for lk in ab) == pytest.approx(connections["ab"]["E"]) + labels = {lk["label"].split("[")[1].split("]")[0] for lk in ab} + assert labels == {"E_PH", "E_CH"} + + def test_mode_3_splits_thermal_mechanical_chemical(self): + components = {"A": ("Turbine", 0.0), "B": ("Generator", 0.0)} + connections = { + "f": {"kind": "power", "source_component": None, "target_component": "A", "E": 1000.0}, + "ab": _material("A", "B", 1000.0, E_PH=700.0, E_CH=300.0, e_T=40.0, e_M=30.0, m=10.0), + "p": {"kind": "power", "source_component": "B", "target_component": None, "E": 1000.0}, + } + analysis = _synthetic_analysis( + components, + connections, + E_F={"inputs": ["f"], "outputs": []}, + E_P={"inputs": ["p"], "outputs": []}, + E_L={"inputs": [], "outputs": []}, + chemical=True, + split=True, + ) + nodes, links = SankeyBuilder(analysis, mode=3).build() + ab = [lk for lk in links if nodes[lk["source"]]["id"] == "A" and nodes[lk["target"]]["id"] == "B"] + # E_T = e_T * m = 400, E_M = e_M * m = 300, E_CH = 300 + assert sorted(lk["value"] for lk in ab) == [300.0, 300.0, 400.0] + labels = {lk["label"].split("[")[1].split("]")[0] for lk in ab} + assert labels == {"E_T", "E_M", "E_CH"} + + def test_split_modes_leave_power_links_untouched(self): + components = {"A": ("Turbine", 0.0), "B": ("Generator", 0.0)} + connections = { + "f": {"kind": "power", "source_component": None, "target_component": "A", "E": 1000.0}, + "ab": {"kind": "power", "source_component": "A", "target_component": "B", "E": 1000.0}, + "p": {"kind": "power", "source_component": "B", "target_component": None, "E": 1000.0}, + } + analysis = _synthetic_analysis( + components, + connections, + E_F={"inputs": ["f"], "outputs": []}, + E_P={"inputs": ["p"], "outputs": []}, + E_L={"inputs": [], "outputs": []}, + chemical=True, + ) + nodes, links = SankeyBuilder(analysis, mode=2).build() + ab = [lk for lk in links if nodes[lk["source"]]["id"] == "A" and nodes[lk["target"]]["id"] == "B"] + assert len(ab) == 1 + assert ab[0]["value"] == pytest.approx(1000.0) + + @staticmethod + def _fuel_and_product_with_returns(): + components = {"A": ("SimpleHeatExchanger", 150.0), "B": ("Turbine", 100.0)} + connections = { + "f_in": _material(None, "A", 1000.0), + "f_ret": _material("A", None, 50.0), + "ab": _material("A", "B", 800.0), + "p_out": _material("B", None, 700.0), + "p_ret": _material(None, "B", 100.0), + "l_out": _material("B", None, 100.0), + } + return _synthetic_analysis( + components, + connections, + E_F={"inputs": ["f_in"], "outputs": ["f_ret"]}, + E_P={"inputs": ["p_out"], "outputs": ["p_ret"]}, + E_L={"inputs": ["l_out"], "outputs": []}, + ) + + def test_fuel_with_inputs_and_outputs(self): + # E_F gets no helper node: the return flow links straight back to it + analysis = self._fuel_and_product_with_returns() + nodes, links = SankeyBuilder(analysis).build() + ids = {n["id"] for n in nodes} + assert "__E_P_net__" in ids + assert "__E_L_net__" not in ids + inflow, outflow = _flows(nodes, links) + assert outflow["__E_F__"] == pytest.approx(1000.0) + assert inflow["__E_F__"] == pytest.approx(50.0) + + def test_product_net_node_shows_gross_and_net(self): + analysis = self._fuel_and_product_with_returns() + nodes, links = SankeyBuilder(analysis).build() + inflow, outflow = _flows(nodes, links) + # gross product in, return plus net out + assert inflow["__E_P_net__"] == pytest.approx(700.0) + assert outflow["__E_P_net__"] == pytest.approx(100.0 + analysis.E_P) + assert inflow["__E_P__"] == pytest.approx(analysis.E_P) + assert pytest.approx(600.0) == analysis.E_P + + def test_negative_fuel_input_is_dropped(self): + # e.g. ambient air with negative chemical exergy: the link is dropped, + # so the displayed E_F outflow exceeds the analysis E_F value. + components = {"A": ("Turbine", 0.0)} + connections = { + "f1": _material(None, "A", 1000.0), + "f2": _material(None, "A", -40.0), + "p": {"kind": "power", "source_component": "A", "target_component": None, "E": 960.0}, + } + analysis = _synthetic_analysis( + components, + connections, + E_F={"inputs": ["f1", "f2"], "outputs": []}, + E_P={"inputs": ["p"], "outputs": []}, + E_L={"inputs": [], "outputs": []}, + ) + nodes, links = SankeyBuilder(analysis).build() + assert all(lk["value"] > 0 for lk in links) + inflow, outflow = _flows(nodes, links) + assert outflow["__E_F__"] == pytest.approx(1000.0) + assert pytest.approx(960.0) == analysis.E_F + + +class TestLinkColors: + def test_connection_base_colors(self): + from exerpy.visualization.colors import connection_base_color + + assert connection_base_color({"kind": "power"}) == "#FFC107" + assert connection_base_color({"kind": "heat"}) == "#F44336" + assert connection_base_color(_material("A", "B", 1.0, {"water": 1})) == "#2196F3" + assert connection_base_color(_material("A", "B", 1.0, {"CH4": 1})) == "#FF9800" + + def test_air_and_flue_gas_classification(self): + from exerpy.visualization.colors import fluid_base_color + + air = {"N2": 0.755, "O2": 0.231, "AR": 0.013, "H2O": 0.006, "CO2": 0.0004} + flue_gas = {"N2": 0.74, "O2": 0.14, "AR": 0.013, "H2O": 0.06, "CO2": 0.05} + assert fluid_base_color(air) == "#9E9E9E" + assert fluid_base_color(flue_gas) == "#795548" + # non-air mixtures fall back to the generic mixture color + assert fluid_base_color({"water": 0.5, "R134a": 0.5}) == "#546E7A" + + class TestWaterfall: def test_matplotlib_waterfall(self, ccpp): fig, ax = ccpp.plot_exergy_waterfall(title="test", show_plot=False)