← Files ClaraARCHIVED FILE
modules/variance-analysis/vendor/modules/charting/draw_waterfall.py
69.4 KB · Oct 4, 2026 · 12:28 UTC
"""Headless legacy-style Plotly waterfall rendering for variance outputs.
This module intentionally keeps the plotting surface small: it ports the legacy
vertical waterfall chart style into a non-UI renderer that can be packaged with
the Codex plugin.
"""
from __future__ import annotations
import html
import math
from dataclasses import dataclass
from typing import Any
import plotly.graph_objects as go
import polars as pl
from PIL import Image, ImageDraw, ImageFont
from plotly.subplots import make_subplots
__all__ = [
"LegacyWaterfallFigure",
"draw_pvm_decomposition_ladder",
"draw_vertical_waterfall_chart",
"make_vertical_waterfall_chart_title",
"write_waterfall_fallback_png",
]
LEGACY_SOURCE_FUNCTIONS = (
"modules.charting.draw_waterfall.draw_vertical_waterfall_chart",
"modules.charting.make_titles.make_vertical_waterfall_chart_title",
"modules.charting.update_layouts.update_waterfall_layout_one_dimension",
"modules.charting.update_layouts.update_waterfall_layout_small_multiples",
)
MAX_SMALL_MULTIPLES = 12
TOLERANCE = 0.000001
COLORS = {
"red": "#FF0000",
"green": "#7ACA00",
"grey": "#A6A6A6",
"light_grey": "#D9D9D9",
"black": "#343434",
"white": "#FFFFFF",
}
ORDER_PREFIX_SEPARATOR = "||"
PLAN_TOTAL_LABELS = {"PL", "PLAN", "BUDGET", "BDG", "FORECAST", "FC"}
ACTUAL_TOTAL_LABELS = {"AC", "ACTUAL", "ACT"}
@dataclass(frozen=True)
class LegacyWaterfallFigure:
"""Plotly figure plus audit metadata for a rendered waterfall."""
figure: go.Figure
audit: dict[str, Any]
def _sum_column(df: pl.DataFrame, column: str) -> float:
"""Return the numeric sum for ``column`` or zero when missing."""
if column not in df.schema or df.is_empty():
return 0.0
value = df.select(pl.col(column).sum()).item()
return float(value or 0.0)
def _format_number(value: float) -> str:
"""Return compact chart text with legacy-style signs."""
if not math.isfinite(value):
return ""
abs_value = abs(value)
if abs_value >= 1_000_000_000:
text = f"{value / 1_000_000_000:,.1f}B"
elif abs_value >= 1_000_000:
text = f"{value / 1_000_000:,.1f}M"
elif abs_value >= 1_000:
text = f"{value / 1_000:,.1f}K"
else:
text = f"{value:,.0f}"
if value > 0 and not text.startswith("+"):
return f"+{text}"
return text
def _periods(recipe: dict[str, Any]) -> tuple[str, str]:
"""Return baseline and comparison labels from the recipe."""
mappings = recipe["mappings"]
return str(mappings["baseline_period"]), str(mappings["comparison_period"])
def _normalized_total_label(label: str) -> str:
"""Normalize a total label for semantic IBCS-style total bar coloring."""
text = _strip_order_prefix(_clean_html(str(label))).strip()
for suffix in (" total", " Total", " TOTAL"):
if text.endswith(suffix):
text = text[: -len(suffix)]
break
return text.strip().upper()
def _ordered_axis_label(index: int, label: str) -> str:
"""Return a category label prefixed for stable Plotly waterfall ordering."""
return f"{index:02d}{ORDER_PREFIX_SEPARATOR}{label}"
def _strip_order_prefix(label: str) -> str:
"""Remove an internal category-order prefix from a display label."""
text = str(label)
if ORDER_PREFIX_SEPARATOR in text:
prefix, value = text.split(ORDER_PREFIX_SEPARATOR, 1)
if prefix.isdigit():
return value
return text
def _is_plan_total_label(label: str) -> bool:
"""Return whether a total label is plan, budget, or forecast."""
return _normalized_total_label(label) in PLAN_TOTAL_LABELS
def _is_actual_total_label(label: str) -> bool:
"""Return whether a total label is actual."""
return _normalized_total_label(label) in ACTUAL_TOTAL_LABELS
def _fallback_total_bar_style(label: str, measure: str) -> dict[str, Any]:
"""Return fallback PNG fill/outline colors for IBCS-style total bars."""
if measure == "total" or _is_actual_total_label(label):
return {"fill": COLORS["black"], "outline": COLORS["black"], "width": 1}
if measure == "absolute" and _is_plan_total_label(label):
return {"fill": COLORS["white"], "outline": COLORS["black"], "width": 2}
return {"fill": COLORS["grey"], "outline": COLORS["grey"], "width": 1}
def _component_rows(df: pl.DataFrame) -> list[tuple[str, float]]:
"""Return variance components in the legacy chart order."""
components = [
("Price", _sum_column(df, "price_variance")),
("Volume", _sum_column(df, "volume_variance")),
("Mix", _sum_column(df, "mix_variance")),
]
visible = [(label, value) for label, value in components if abs(value) > TOLERANCE]
total_delta = _sum_column(df, "total_delta")
residual = total_delta - sum(value for _label, value in visible)
if abs(residual) > TOLERANCE:
visible.append(("Other", residual))
if not visible:
visible.append(("Variance", total_delta))
return visible
def _aggregate_component_frame(df: pl.DataFrame) -> pl.DataFrame:
"""Return one result row containing summed standard-variance components."""
numeric_columns = [
"amount_baseline",
"amount_comparison",
"amount_delta",
"total_delta",
"amount_pct_change",
"price_baseline",
"price_comparison",
"price_variance",
"volume_variance",
"mix_variance",
"component_reconciliation_delta",
"units_baseline",
"units_comparison",
"units_delta",
"calculated_price_baseline",
"calculated_price_comparison",
"net_baseline",
"net_comparison",
"net_delta",
"discount_baseline",
"discount_comparison",
"discount_delta",
"cogs_baseline",
"cogs_comparison",
"cogs_delta",
"margin_baseline",
"margin_comparison",
"margin_delta",
"margin_price_variance",
"margin_volume_variance",
"margin_mix_variance",
"margin_component_reconciliation_delta",
]
row = {
column: _sum_column(df, column)
for column in numeric_columns
if column in df.schema
}
return pl.DataFrame([row])
def _component_only_recipe(recipe: dict[str, Any]) -> dict[str, Any]:
"""Return a copy of the recipe with no row dimensions for component bridges."""
component_recipe = dict(recipe)
mappings = dict(recipe.get("mappings") or {})
mappings["dimensions"] = []
component_recipe["mappings"] = mappings
return component_recipe
def _pvm_ladder_aggregations(
names: dict[str, str], recipe: dict[str, Any]
) -> list[str]:
"""Return legacy aggregation names for the decomposition ladder."""
if recipe["mappings"].get("units_column"):
return [
names["totalVarianceAggregation"],
names["priceAndUnitsAggregation"],
names["mixAndUnitsAggregation"],
]
return [
names["totalVarianceAggregation"],
names["priceAndVolumeAggregation"],
names["mixAndVolumeAggregation"],
]
def _pvm_ladder_panel_titles(recipe: dict[str, Any]) -> list[str]:
"""Return business-readable panel titles for the PVM ladder."""
unit_label = "Units" if recipe["mappings"].get("units_column") else "Volume"
return [
"Total variance",
f"Price + {unit_label} & Mix",
f"Price + {unit_label} + Mix",
]
def _delta_percent_label(
baseline_total: float,
comparison_total: float,
names: dict[str, str],
) -> str:
"""Return the legacy-style total movement percent label."""
if baseline_total <= TOLERANCE:
return ""
percent_change = ((comparison_total - baseline_total) / baseline_total) * 100
if not math.isfinite(percent_change):
return ""
return f"{names['deltaName']}{int(round(percent_change, 0))}%"
def _pvm_ladder_components(
aggregation: str,
df: pl.DataFrame,
names: dict[str, str],
recipe: dict[str, Any],
) -> list[tuple[str, float]]:
"""Return the component split for one legacy PVM ladder aggregation."""
total_delta = _sum_column(df, "total_delta")
price = _sum_column(df, "price_variance")
volume = _sum_column(df, "volume_variance")
mix = _sum_column(df, "mix_variance")
residual = total_delta - price - volume - mix
unit_label = "Units" if recipe["mappings"].get("units_column") else "Volume"
if aggregation == names["totalVarianceAggregation"]:
return [(f"Price & {unit_label.lower()} & mix", total_delta)]
if aggregation in {
names["priceAndUnitsAggregation"],
names["priceAndVolumeAggregation"],
}:
return [
("Price", price),
(f"{unit_label} & mix", total_delta - price),
]
components = [
("Price", price),
(unit_label, volume),
("Mix", mix),
]
if abs(residual) > TOLERANCE:
components.append(("Other", residual))
return components
def _pvm_ladder_chart_frame(
df: pl.DataFrame,
recipe: dict[str, Any],
names: dict[str, str],
config: dict[str, Any],
aggregation: str,
variance_array: list[str],
) -> pl.DataFrame:
"""Build a legacy waterfall frame for one PVM ladder panel."""
baseline, comparison = _periods(recipe)
periods = config["periodsArray"]
separator = names["separatorString"]
sales_p0 = names["monetaryLocalCurrencyName"] + separator + periods[0]
sales_p1 = names["monetaryLocalCurrencyName"] + separator + periods[1]
measure = names["measureName"]
work_column = names["workColumn"]
work_column_two = names["workColumnTwo"]
variance_amount = names["varianceAmountName"]
variance_type = names["varianceTypeName"]
label = names["labelName"]
baseline_total = _sum_column(df, "amount_baseline")
comparison_total = _sum_column(df, "amount_comparison")
component_map = {
component_label: value
for component_label, value in _pvm_ladder_components(
aggregation, df, names, recipe
)
}
component_labels = [
str(item)
for item in variance_array
if str(item) not in {baseline, comparison, names["pyName"], names["acName"]}
]
rows: list[dict[str, Any]] = [
{
measure: "absolute",
work_column: baseline,
work_column_two: baseline_total,
variance_amount: baseline_total,
variance_type: "",
label: _format_number(baseline_total),
sales_p0: baseline_total,
sales_p1: comparison_total,
}
]
for component_label in component_labels:
value = float(component_map.get(component_label, 0.0) or 0.0)
rows.append(
{
measure: "relative",
work_column: component_label,
work_column_two: value,
variance_amount: value,
variance_type: component_label,
label: _format_number(value) if abs(value) > TOLERANCE else "",
sales_p0: baseline_total,
sales_p1: comparison_total,
}
)
rows.append(
{
measure: "total",
work_column: comparison,
work_column_two: comparison_total,
variance_amount: comparison_total,
variance_type: "",
label: _format_number(comparison_total),
sales_p0: baseline_total,
sales_p1: comparison_total,
}
)
return pl.DataFrame(rows).with_columns(
pl.col(variance_amount).cast(pl.Float64, strict=False),
pl.col(work_column_two).cast(pl.Float64, strict=False),
pl.col(sales_p0).cast(pl.Float64, strict=False),
pl.col(sales_p1).cast(pl.Float64, strict=False),
)
def draw_pvm_decomposition_ladder(
df: pl.DataFrame,
recipe: dict[str, Any],
*,
legacy_frame: pl.DataFrame | None = None,
) -> LegacyWaterfallFigure:
"""Draw the legacy PVM decomposition ladder used for variance explanation."""
if df.is_empty():
raise ValueError("Cannot render PVM decomposition ladder for an empty result.")
from modules.charting import legacy_draw_waterfall
from modules.charting.adjust_position import move_labels_up
from modules.charting.chart_helpers import make_one_dimensional_variance_subplots
from modules.charting.chart_primitives import get_color_dictionary
from modules.charting.update_layouts import update_waterfall_layout_small_multiples
from modules.variance.variance_orchestrator import build_variance_calculation_array
component_recipe = _component_only_recipe(recipe)
names, config, param, chart = _legacy_context(component_recipe)
aggregations = _pvm_ladder_aggregations(names, component_recipe)
chart[names["varianceAggregationOptionsArray"]] = aggregations
chart[names["varianceAggregation"]] = aggregations[-1]
chart[names["varianceDifferentCalculations"]] = True
chart[names["showInitialAndFinalValues"]] = True
variance_array = build_variance_calculation_array(aggregations, chart)
panel_titles = _pvm_ladder_panel_titles(component_recipe)
fig, count_rows, count_cols, _count, number_of_cols, number_of_rows = (
make_one_dimensional_variance_subplots(panel_titles, numberOfCols=3)
)
baseline_total = _sum_column(df, "amount_baseline")
comparison_total = _sum_column(df, "amount_comparison")
delta_percent_label = _delta_percent_label(
baseline_total,
comparison_total,
names,
)
for key in (
names["totalAmountPeriodZero"],
names["totalAmountPeriodZeroFiltered"],
names["periodZeroSum"],
):
param[key] = baseline_total
for key in (
names["totalAmountPeriodOne"],
names["totalAmountPeriodOneFiltered"],
names["periodOneSum"],
):
param[key] = comparison_total
param[names["totalVarianceValue"]] = comparison_total - baseline_total
color_dict = get_color_dictionary(chart)
run = names["runOneDimensionalAnalysis"]
frames: list[pl.DataFrame] = []
number_format = ""
for aggregation in aggregations:
panel_chart = dict(chart)
panel_chart[names["varianceAggregation"]] = aggregation
panel_df = _pvm_ladder_chart_frame(
df,
component_recipe,
names,
config,
aggregation,
variance_array,
)
frames.append(panel_df)
fig_det, number_format, _panel_chart = (
legacy_draw_waterfall.draw_vertical_waterfall_chart(
panel_df,
color_dict,
dict(param),
panel_chart,
run,
)
)
if fig_det.data:
fig_det.data[0].showlegend = False
fig.add_trace(fig_det.data[0], row=count_rows, col=count_cols)
if count_cols < number_of_cols:
count_cols += 1
else:
count_cols = 1
count_rows += 1
layout_df = pl.concat(frames) if frames else df
fig, _width = update_waterfall_layout_small_multiples(
layout_df,
fig,
chart,
number_of_rows,
number_of_cols,
)
# Legacy figures also contain marker/label and absolute-value bar traces.
# Only waterfall traces carry the measures used by this renderer.
traces = [trace for trace in fig.data if trace.type == "waterfall"]
if traces:
global_range = _combined_trace_extent(traces)
fig.update_xaxes(range=list(global_range), matches="x")
fig = move_labels_up(fig, chart, panel_titles)
fig.update_yaxes(type="category", autorange="reversed", ticks="")
if delta_percent_label:
for index in range(len(aggregations)):
fig.add_annotation(
text=delta_percent_label,
x=comparison_total,
# Numeric-looking period labels are parsed as coordinates by Plotly.
y=len(variance_array) - 1,
showarrow=False,
xshift=42,
yshift=-18,
font={"size": 11, "color": COLORS["grey"]},
row=1,
col=index + 1,
)
fig.update_layout(
title={
"text": make_vertical_waterfall_chart_title(df, component_recipe),
"x": 0.01,
"xanchor": "left",
},
meta={"pvm_delta_percent_label": delta_percent_label},
)
return LegacyWaterfallFigure(
figure=fig,
audit={
"mode": "legacy_pvm_decomposition_ladder",
"title": make_vertical_waterfall_chart_title(df, component_recipe),
"aggregation_count": len(aggregations),
"aggregations": aggregations,
"panel_titles": panel_titles,
"delta_percent_label": delta_percent_label,
"variance_array": variance_array,
"row_count": df.height,
"legacy_source_row_count": (
legacy_frame.height if legacy_frame is not None else None
),
"legacy_reference_function": (
"modules.charting.plot_charts."
"plot_one_dimensional_variance_chart_different_calculations"
),
"source_functions": [
"modules.variance.variance_orchestrator.build_variance_calculation_array",
"modules.charting.chart_helpers.make_one_dimensional_variance_subplots",
"modules.charting.legacy_draw_waterfall.draw_vertical_waterfall_chart",
"modules.charting.update_layouts.update_waterfall_layout_small_multiples",
"modules.charting.adjust_position.move_labels_up",
],
"number_format": number_format,
},
)
def _waterfall_arrays(
df: pl.DataFrame, recipe: dict[str, Any]
) -> tuple[list[str], list[str], list[float], list[str]]:
"""Return labels, Plotly measures, values and text for one waterfall."""
baseline, comparison = _periods(recipe)
labels = [baseline]
measures = ["absolute"]
values = [_sum_column(df, "amount_baseline")]
for label, value in _component_rows(df):
labels.append(label)
measures.append("relative")
values.append(value)
labels.append(comparison)
measures.append("total")
values.append(_sum_column(df, "amount_comparison"))
text = [_format_number(value) for value in values]
return labels, measures, values, text
def _component_columns(recipe: dict[str, Any]) -> list[tuple[str, str]]:
"""Return plugin result columns in legacy display order."""
if not recipe["mappings"].get("units_column"):
return [("Total variance", "total_delta")]
volume_label = "Units"
return [
("Price", "price_variance"),
(volume_label, "volume_variance"),
("Mix", "mix_variance"),
]
def make_vertical_waterfall_chart_title(
df: pl.DataFrame,
recipe: dict[str, Any],
*,
dimension: str | None = None,
dimension_value: str | None = None,
) -> str:
"""Return a legacy-style waterfall title."""
baseline, comparison = _periods(recipe)
dimension_text = f" by {dimension}" if dimension else ""
if dimension and dimension_value:
dimension_text = f" by {dimension}: {dimension_value}"
return f"<b>Sales variance</b>{dimension_text}<br>{comparison} vs {baseline}"
def _add_waterfall_trace(
fig: go.Figure,
df: pl.DataFrame,
recipe: dict[str, Any],
*,
row: int,
col: int,
show_labels: bool = True,
) -> None:
"""Add one horizontal waterfall trace to ``fig``."""
labels, measures, values, text = _waterfall_arrays(df, recipe)
fig.add_trace(
go.Waterfall(
orientation="h",
measure=measures,
y=labels,
x=values,
text=text if show_labels else None,
textinfo="text" if show_labels else "none",
textposition="outside",
cliponaxis=False,
increasing={"marker": {"color": COLORS["green"]}},
decreasing={"marker": {"color": COLORS["red"]}},
totals={"marker": {"color": COLORS["black"]}},
connector={
"mode": "between",
"line": {"width": 1, "color": "rgb(169,169,169)", "dash": "solid"},
},
),
row=row,
col=col,
)
def _update_layout(fig: go.Figure, *, title: str, height: int, width: int) -> None:
"""Apply legacy waterfall layout defaults."""
fig.update_layout(
template="simple_white",
title={"text": title, "x": 0.01, "xanchor": "left"},
width=width,
height=height,
margin={"r": 40, "l": 20, "b": 35, "t": 95, "pad": 20},
showlegend=False,
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(0,0,0,0)",
font={"size": 12, "family": "Arial"},
)
fig.update_xaxes(
automargin=True,
zeroline=True,
zerolinecolor="black",
zerolinewidth=1,
showticklabels=False,
ticks="",
)
fig.update_yaxes(autorange="reversed", ticks="", showticklabels=True)
def _clean_html(text: Any) -> str:
"""Return readable plain text from simple Plotly HTML labels."""
cleaned = (
str(text or "")
.replace("<b>", "")
.replace("</b>", "")
.replace("<br>", " ")
.replace("<BR>", " ")
)
return html.unescape(cleaned)
def _clean_html_lines(text: Any) -> list[str]:
"""Return readable title lines from simple Plotly HTML labels."""
raw = str(text or "").replace("<b>", "").replace("</b>", "").replace("<BR>", "<br>")
lines = []
for line in raw.split("<br>"):
cleaned = _clean_html(line).strip()
if cleaned:
lines.append(cleaned)
return lines
def _font(size: int) -> ImageFont.ImageFont:
"""Return a readable default font."""
try:
return ImageFont.truetype("Arial.ttf", size)
except OSError:
return ImageFont.load_default()
def _trace_extent(trace: Any) -> tuple[float, float]:
"""Return min/max x extent for a waterfall trace."""
current = 0.0
low = 0.0
high = 0.0
for measure, value in zip(trace.measure, trace.x, strict=False):
value = float(value or 0.0)
if measure == "absolute":
start, end = 0.0, value
current = value
elif measure == "total":
start, end = 0.0, value
current = value
else:
start, end = current, current + value
current = end
low = min(low, start, end)
high = max(high, start, end)
if abs(high - low) < TOLERANCE:
high = low + 1.0
padding = (high - low) * 0.12
return low - padding, high + padding
def _combined_trace_extent(traces: list[Any]) -> tuple[float, float]:
"""Return one x extent shared by every waterfall trace."""
extents = [_trace_extent(trace) for trace in traces]
low = min(extent[0] for extent in extents)
high = max(extent[1] for extent in extents)
if abs(high - low) < TOLERANCE:
high = low + 1.0
return low, high
def _draw_fallback_trace(
draw: ImageDraw.ImageDraw,
trace: Any,
*,
top: int,
left: int,
right: int,
title: str,
extent: tuple[float, float] | None = None,
delta_percent_label: str = "",
font_size: int | None = None,
) -> int:
"""Draw one waterfall trace with Pillow and return the next y position."""
title_font = _font(font_size or 18)
body_font = _font(font_size or 13)
label_font = _font(font_size or 12)
row_height = 34
bar_height = 20
labels = [_strip_order_prefix(_clean_html(label)) for label in trace.y]
measures = list(trace.measure)
values = [float(value or 0.0) for value in trace.x]
low, high = extent or _trace_extent(trace)
def x_pos(value: float) -> int:
return int(left + ((value - low) / (high - low)) * (right - left))
if title:
draw.text((left, top), title, fill=COLORS["black"], font=title_font)
axis_y = top + 38 + len(labels) * row_height
zero_x = x_pos(0.0)
draw.line((zero_x, top + 28, zero_x, axis_y), fill=COLORS["light_grey"], width=1)
current = 0.0
for index, (label, measure, value) in enumerate(
zip(labels, measures, values, strict=False)
):
y_mid = top + 42 + index * row_height
if measure == "absolute":
start, end = 0.0, value
current = value
style = _fallback_total_bar_style(label, measure)
elif measure == "total":
start, end = 0.0, value
current = value
style = _fallback_total_bar_style(label, measure)
else:
start, end = current, current + value
current = end
style = {
"fill": COLORS["green"] if value >= 0 else COLORS["red"],
"outline": None,
"width": 1,
}
x0, x1 = sorted((x_pos(start), x_pos(end)))
if x0 == x1:
x1 = x0 + 1
draw.text((left - 140, y_mid - 8), label, fill=COLORS["black"], font=body_font)
draw.rounded_rectangle(
(x0, y_mid - bar_height // 2, x1, y_mid + bar_height // 2),
radius=3,
fill=style["fill"],
outline=style["outline"],
width=style["width"],
)
text = _format_number(value)
# Keep signed amounts out of the reserved row-label column. This also
# makes negative accounting balances readable in compact panels.
text_x = x1 + 6
text_y = y_mid - 7
text_width = draw.textbbox((0, 0), text, font=label_font)[2]
if text_x + text_width > right:
text_x = right - text_width
text_y = y_mid - 26
draw.text((text_x, text_y), text, fill=COLORS["black"], font=label_font)
if measure == "total" and delta_percent_label:
draw.text(
(text_x, y_mid + 9),
delta_percent_label,
fill=COLORS["grey"],
font=label_font,
)
return axis_y + 28
def write_waterfall_fallback_png(fig: go.Figure, path: str) -> None:
"""Write a deterministic PNG fallback when Kaleido/Chrome is unavailable."""
traces = [trace for trace in fig.data if trace.type == "waterfall"]
if not traces:
raise ValueError("No waterfall traces are available for fallback rendering.")
title_lines = _clean_html_lines(getattr(fig.layout.title, "text", "Sales variance"))
panel_heights = [86 + len(getattr(trace, "y", [])) * 34 for trace in traces]
annotations = [
_clean_html(getattr(annotation, "text", ""))
for annotation in getattr(fig.layout, "annotations", [])
]
meta = fig.layout.meta if isinstance(fig.layout.meta, dict) else {}
font_size = meta.get("waterfall_font_size")
delta_percent_label = str(meta.get("pvm_delta_percent_label") or "")
small_multiples_grid = meta.get("waterfall_small_multiples_grid")
delta_percent_labels = [
str(label or "") for label in (meta.get("waterfall_delta_percent_labels") or [])
]
panel_titles = [
str(title or "") for title in (meta.get("waterfall_panel_titles") or [])
]
if isinstance(small_multiples_grid, dict) and len(traces) > 1:
cols = max(1, int(small_multiples_grid.get("cols") or 1))
rows = max(
1, int(small_multiples_grid.get("rows") or math.ceil(len(traces) / cols))
)
panel_width = 480
panel_height = max(250, max(panel_heights))
width = max(int(fig.layout.width or 0), 70 + panel_width * cols)
top = 40 + max(len(title_lines), 1) * 28
height = max(int(fig.layout.height or 0), top + panel_height * rows + 35)
image = Image.new("RGB", (width, height), COLORS["white"])
draw = ImageDraw.Draw(image)
title_font = _font(font_size or 22)
for index, line in enumerate(title_lines):
draw.text(
(36, 24 + index * 28),
line,
fill=COLORS["black"],
font=title_font,
)
shared_extent = _combined_trace_extent(traces)
for index, trace in enumerate(traces):
row = index // cols
col = index % cols
panel_left = 40 + col * panel_width
panel_top = top + row * panel_height
title = panel_titles[index] if index < len(panel_titles) else ""
delta_label = (
delta_percent_labels[index] if index < len(delta_percent_labels) else ""
)
_draw_fallback_trace(
draw,
trace,
top=panel_top,
left=panel_left + 165,
right=panel_left + panel_width - 20,
title=title,
extent=shared_extent,
delta_percent_label=delta_label,
)
image.save(path, format="PNG")
return
if 1 < len(traces) <= 3:
panel_width = 520
width = max(int(fig.layout.width or 0), 80 + panel_width * len(traces))
top = 40 + max(len(title_lines), 1) * 28
height = max(int(fig.layout.height or 0), top + max(panel_heights) + 50)
image = Image.new("RGB", (width, height), COLORS["white"])
draw = ImageDraw.Draw(image)
title_font = _font(font_size or 22)
for index, line in enumerate(title_lines):
draw.text(
(36, 24 + index * 28),
line,
fill=COLORS["black"],
font=title_font,
)
shared_extent = _combined_trace_extent(traces)
for index, trace in enumerate(traces):
panel_left = 40 + index * panel_width
panel_title = annotations[index] if index < len(annotations) else ""
_draw_fallback_trace(
draw,
trace,
top=top,
left=panel_left + 165,
right=panel_left + panel_width - 22,
title=panel_title,
extent=shared_extent,
delta_percent_label=delta_percent_label,
font_size=font_size,
)
image.save(path, format="PNG")
return
width = max(1000, int(fig.layout.width or 1200))
height = max(
int(fig.layout.height or 0),
48 + max(len(title_lines), 1) * 28 + sum(panel_heights),
)
image = Image.new("RGB", (width, height), COLORS["white"])
draw = ImageDraw.Draw(image)
title_font = _font(font_size or 22)
line_height = 28
for index, line in enumerate(title_lines):
draw.text(
(36, 24 + index * line_height),
line,
fill=COLORS["black"],
font=title_font,
)
top = 40 + max(len(title_lines), 1) * line_height
left = 210
right = width - 80
shared_extent = _combined_trace_extent(traces) if len(traces) > 1 else None
for index, trace in enumerate(traces):
panel_title = annotations[index] if index < len(annotations) else ""
top = _draw_fallback_trace(
draw,
trace,
top=top,
left=left,
right=right,
title=panel_title,
extent=shared_extent,
delta_percent_label=delta_percent_label,
font_size=font_size,
)
image.save(path, format="PNG")
def _legacy_context(
recipe: dict[str, Any],
) -> tuple[dict[str, str], dict[str, Any], dict[str, Any], dict[str, Any]]:
"""Build the legacy state dictionaries needed by original chart functions."""
from modules.utilities.config import get_config_params, get_naming_params
names = get_naming_params()
config = get_config_params()
mappings = recipe["mappings"]
not_met = names["notMetConditionValue"]
report_dimensions = list(mappings.get("dimensions") or [])
has_units = bool(mappings.get("units_column"))
variance_choice = (
names["mixAndUnitsAggregation"]
if has_units
else names["totalVarianceAggregation"]
)
param = {
names["numberOfPeriodsFound"]: 2,
names["unitsColFound"]: has_units,
names["volumeColFound"]: False,
names["discountColFound"]: bool(mappings.get("discount_column")),
names["cogsColFound"]: bool(mappings.get("cogs_column")),
names["marginColFound"]: bool(mappings.get("cogs_column")),
names["monetaryLocalCurrencyColFound"]: True,
names["calculateDriverVariance"]: False,
names["driverArray"]: [],
names["isFilteredKey"]: not_met,
names["selectedPeriods"]: [
str(mappings["baseline_period"]),
str(mappings["comparison_period"]),
],
names["allPeriodsList"]: [
str(mappings["baseline_period"]),
str(mappings["comparison_period"]),
],
names["impossibleToProcessFile"]: False,
names["fileUploadDisabled"]: not_met,
names["dropLowCorrelationCols"]: False,
names["toTitleCase"]: False,
names["reverseSortPeriods"]: False,
names["isColumnMultiplied"]: False,
names["renameTitlesDict"]: {},
}
total_baseline = 0.0
total_comparison = 0.0
currency = str(recipe.get("options", {}).get("currency") or "EUR")
chart = {
names["selectedPeriods"]: [
str(mappings["baseline_period"]),
str(mappings["comparison_period"]),
],
names["varianceAggregation"]: variance_choice,
names["processingChoice"]: names["runOneDimensionalAnalysis"],
names["mainDimension"]: report_dimensions,
names["reverseSortPeriods"]: False,
names["showInitialAndFinalValues"]: False,
names["varianceInPercent"]: False,
names["shareOfTotalMarket"]: False,
names["plotSmallMultiplesWaterfall"]: False,
names["filterDates"]: False,
names["compareScenariosOrPeriods"]: names["comparePeriods"],
names["chosenChart"]: names["verticalWaterfallChart"],
names["varianceAnalysisChart"]: True,
names["currencyChoice"]: currency,
names["fullCurrencyName"]: currency,
}
for key in (
names["totalAmountPeriodZero"],
names["totalAmountPeriodZeroFiltered"],
names["periodZeroSum"],
):
param[key] = total_baseline
for key in (
names["totalAmountPeriodOne"],
names["totalAmountPeriodOneFiltered"],
names["periodOneSum"],
):
param[key] = total_comparison
param[names["totalVarianceValue"]] = total_comparison - total_baseline
return names, config, param, chart
def _legacy_chart_frame(
df: pl.DataFrame,
recipe: dict[str, Any],
names: dict[str, str],
config: dict[str, Any],
*,
include_zero_components: bool = False,
) -> pl.DataFrame:
"""Convert plugin result rows into the original waterfall input schema."""
baseline, comparison = _periods(recipe)
periods = config["periodsArray"]
separator = names["separatorString"]
sales_p0 = names["monetaryLocalCurrencyName"] + separator + periods[0]
sales_p1 = names["monetaryLocalCurrencyName"] + separator + periods[1]
measure = names["measureName"]
work_column = names["workColumn"]
work_column_two = names["workColumnTwo"]
variance_amount = names["varianceAmountName"]
variance_type = names["varianceTypeName"]
label = names["labelName"]
report_dimensions = [
dimension
for dimension in recipe["mappings"].get("dimensions", [])
if dimension in df.schema
]
baseline_total = _sum_column(df, "amount_baseline")
comparison_total = _sum_column(df, "amount_comparison")
rows: list[dict[str, Any]] = [
{
measure: "absolute",
work_column: baseline,
work_column_two: baseline_total,
variance_amount: baseline_total,
variance_type: "",
label: _format_number(baseline_total),
sales_p0: baseline_total,
sales_p1: comparison_total,
}
]
for source_row in df.to_dicts():
dimension_values = [
str(source_row.get(dimension))
for dimension in report_dimensions
if source_row.get(dimension) not in (None, "")
]
prefix = " - ".join(dimension_values)
for component_label, column in _component_columns(recipe):
value = float(source_row.get(column) or 0.0)
if abs(value) <= TOLERANCE and not include_zero_components:
continue
display_label = (
f" {prefix} - {component_label}" if prefix else component_label
)
rows.append(
{
measure: "relative",
work_column: display_label,
work_column_two: value,
variance_amount: value,
variance_type: component_label,
label: _format_number(value),
sales_p0: float(source_row.get("amount_baseline") or 0.0),
sales_p1: float(source_row.get("amount_comparison") or 0.0),
}
)
residual = float(source_row.get("component_reconciliation_delta") or 0.0)
if abs(residual) > TOLERANCE:
display_label = f" {prefix} - Other" if prefix else "Other"
rows.append(
{
measure: "relative",
work_column: display_label,
work_column_two: residual,
variance_amount: residual,
variance_type: "Other",
label: _format_number(residual),
sales_p0: float(source_row.get("amount_baseline") or 0.0),
sales_p1: float(source_row.get("amount_comparison") or 0.0),
}
)
rows.append(
{
measure: "total",
work_column: comparison,
work_column_two: comparison_total,
variance_amount: comparison_total,
variance_type: "",
label: _format_number(comparison_total),
sales_p0: baseline_total,
sales_p1: comparison_total,
}
)
return pl.DataFrame(rows).with_columns(
pl.col(variance_amount).cast(pl.Float64, strict=False),
pl.col(work_column_two).cast(pl.Float64, strict=False),
pl.col(sales_p0).cast(pl.Float64, strict=False),
pl.col(sales_p1).cast(pl.Float64, strict=False),
)
def _legacy_small_multiples_aggregation(
names: dict[str, str],
recipe: dict[str, Any],
legacy_frame: pl.DataFrame,
) -> str:
"""Return the legacy compact variance aggregation for dimension panels."""
schema = legacy_frame.schema
if (
recipe["mappings"].get("units_column")
and names["priceVariance"] in schema
and names["volumeVariance"] in schema
):
return names["priceAndUnitsAggregation"]
if names["totalVariance"] in schema:
return names["totalVarianceAggregation"]
return names["mixAndUnitsAggregation"]
def _legacy_small_multiples_variance_types(
aggregation: str,
names: dict[str, str],
) -> list[str]:
"""Return source variance columns shown by a legacy small-multiple panel."""
if aggregation in {
names["priceAndUnitsAggregation"],
names["priceAndVolumeAggregation"],
}:
return [names["priceVariance"], names["volumeVariance"]]
if aggregation == names["totalVarianceAggregation"]:
return [names["totalVariance"]]
if aggregation in {
names["mixAndUnitsAggregation"],
names["mixAndVolumeAggregation"],
}:
return [
names["priceVariance"],
names["pureVolumeVarianceName"],
names["mixVariance"],
]
return []
def _legacy_small_multiples_other_label() -> str:
"""Return the legacy label for the aggregated residual member panel."""
try:
from modules.utilities.config import get_naming_params
return str(get_naming_params()["aggregateOtherWaterfallsName"])
except (ImportError, KeyError):
return "Others aggregated"
def _legacy_frame_with_panel_bucket(
legacy_frame: pl.DataFrame,
dimension: str,
selected_values: list[str],
*,
other_label: str,
include_other: bool,
) -> pl.DataFrame:
"""Return a legacy frame with non-selected members bucketed as Other."""
frame = legacy_frame.with_columns(
pl.col(dimension).cast(pl.Utf8).fill_null("").alias(dimension)
)
if not include_other:
return frame
return frame.with_columns(
pl.when(pl.col(dimension).is_in(selected_values))
.then(pl.col(dimension))
.otherwise(pl.lit(other_label))
.alias(dimension)
)
def _replace_legacy_period_labels(
frame: pl.DataFrame | pl.LazyFrame,
recipe: dict[str, Any],
names: dict[str, str],
) -> pl.DataFrame:
"""Replace legacy Period Zero/One row labels with recipe labels."""
df = frame.collect() if isinstance(frame, pl.LazyFrame) else frame
if df.height < 2:
return df
baseline, comparison = _periods(recipe)
work_column = names["workColumn"]
measure = names["measureName"]
return (
df.with_row_index("__row_nr")
.with_columns(
pl.when(pl.col("__row_nr") == 0)
.then(pl.lit(baseline))
.when(pl.col("__row_nr") == df.height - 1)
.then(pl.lit(comparison))
.otherwise(pl.col(work_column))
.alias(work_column),
pl.when(pl.col("__row_nr") == df.height - 1)
.then(pl.lit("total"))
.otherwise(pl.col(measure))
.alias(measure),
)
.drop("__row_nr")
)
def _legacy_small_multiples_display_types(
aggregation: str,
names: dict[str, str],
recipe: dict[str, Any],
) -> list[str]:
"""Return row labels expected inside one compact variance panel."""
if aggregation in {
names["priceAndUnitsAggregation"],
names["priceAndVolumeAggregation"],
}:
volume_label = (
"Units & mix" if recipe["mappings"].get("units_column") else "Volume & mix"
)
return [names["priceVariance"], volume_label]
if aggregation == names["totalVarianceAggregation"]:
unit_word = "units" if recipe["mappings"].get("units_column") else "volume"
return [f"Price & {unit_word} & mix"]
if aggregation in {
names["mixAndUnitsAggregation"],
names["mixAndVolumeAggregation"],
}:
volume_label = "Units" if recipe["mappings"].get("units_column") else "Volume"
return [names["priceVariance"], volume_label, names["mixVariance"]]
return []
def _order_legacy_small_multiple_rows(
frame: pl.DataFrame,
recipe: dict[str, Any],
names: dict[str, str],
aggregation: str,
) -> pl.DataFrame:
"""Keep all small-multiple panels in the same waterfall row order."""
# Position follows the waterfall role, never the user-facing wording.
# Preserve the calculation's contribution order between the two totals.
measure = pl.col(names["measureName"])
order_expr = (
pl.when(measure == "absolute")
.then(pl.lit(0))
.when(measure == "total")
.then(pl.lit(2))
.otherwise(pl.lit(1))
)
return (
frame.with_row_index("__row_nr")
.with_columns(order_expr.alias("__order"))
.sort(["__order", "__row_nr"])
.drop(["__order", "__row_nr"])
)
def _prefix_legacy_small_multiple_axis_labels(
frame: pl.DataFrame,
recipe: dict[str, Any],
names: dict[str, str],
aggregation: str,
) -> tuple[pl.DataFrame, dict[str, str], list[str], list[str]]:
"""Prefix category labels while returning visible tick labels."""
# The frame is already ordered. Include every contribution in the axis
# order, including labels that differ from the aggregation's usual wording.
visible_order = list(
dict.fromkeys(str(value).strip() for value in frame[names["workColumn"]])
)
label_map = {
label: _ordered_axis_label(index, label)
for index, label in enumerate(visible_order, start=1)
}
expr = pl.col(names["workColumn"])
for visible_label, prefixed_label in label_map.items():
label_key = pl.col(names["workColumn"]).cast(pl.Utf8).str.strip_chars()
expr = (
pl.when(label_key == visible_label)
.then(pl.lit(prefixed_label))
.otherwise(expr)
)
return (
frame.with_columns(expr.alias(names["workColumn"])),
label_map,
[label_map[label] for label in visible_order],
visible_order,
)
def _panel_period_totals(
frame: pl.DataFrame,
dimension: str,
value: str,
names: dict[str, str],
config: dict[str, Any],
) -> tuple[float, float]:
"""Return baseline and comparison totals for one small-multiple panel."""
periods = config["periodsArray"]
separator = names["separatorString"]
sales_p0 = names["monetaryLocalCurrencyName"] + separator + periods[0]
sales_p1 = names["monetaryLocalCurrencyName"] + separator + periods[1]
panel = frame.filter(pl.col(dimension) == value)
return _sum_column(panel, sales_p0), _sum_column(panel, sales_p1)
def _legacy_color_first_bar_shape(
df: pl.DataFrame,
param: dict[str, Any],
chart: dict[str, Any],
color_dict: dict[str, str],
run: str,
count: int,
shapes: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Return legacy shape metadata that colors the first total bar."""
from modules.charting.legacy_draw_waterfall import set_semantic_bar_color
from modules.utilities.config import get_config_params, get_naming_params
names = get_naming_params()
config = get_config_params()
work_column = names["workColumn"]
variance_amount = names["varianceAmountName"]
first_label = str(df[work_column][0]).lower()
is_expected_data = any(
element in first_label for element in config[names["planStemArray"]]
)
can_show_initial_and_final = True
if names["varianceAggregation"] in chart:
if (
chart[names["varianceAggregation"]]
not in [
names["totalVarianceAggregation"],
names["marginVarianceAggregation"],
]
and names["drilldownReportRunName"] in run
):
can_show_initial_and_final = False
if (
names["showInitialAndFinalValues"] in chart
and chart[names["showInitialAndFinalValues"]]
and can_show_initial_and_final
):
first_bar_color, line_width, line_color = set_semantic_bar_color(
is_expected_data,
color_dict,
param,
)
shapes.append(
{
"type": "rect",
"fillcolor": first_bar_color,
"opacity": 1,
"line_width": line_width,
"line_color": line_color,
"xref": f"x{count}",
"yref": f"y{count}",
"y0": -0.4,
"y1": 0.4,
"x0": 0,
"x1": df[variance_amount][0],
}
)
return shapes
def _legacy_line_shape(
df: pl.DataFrame,
param: dict[str, Any],
chart: dict[str, Any],
color_dict: dict[str, str],
run: str,
count: int,
shapes: list[dict[str, Any]],
x0_value: float,
x1_value: float,
number_of_charts: int,
*,
is_arrow: bool,
is_period_zero: bool,
count_rows: int,
) -> list[dict[str, Any]]:
"""Return legacy line/arrow shape metadata for small-multiple totals."""
from modules.charting.adjust_position import get_y1_y0_values
from modules.charting.draw_charts_utils import get_polars_value_at_index
from modules.utilities.config import get_naming_params
names = get_naming_params()
can_show_initial_and_final = True
if names["varianceAggregation"] in chart:
if (
chart[names["varianceAggregation"]]
not in [
names["totalVarianceAggregation"],
names["marginVarianceAggregation"],
]
and names["drilldownReportRunName"] in run
):
can_show_initial_and_final = False
if not (
names["showInitialAndFinalValues"] in chart
and chart[names["showInitialAndFinalValues"]]
and can_show_initial_and_final
):
return shapes
line_width = 1
line_color = color_dict["lightGreyColor"]
panel_lazy = df.lazy()
period_one_value = get_polars_value_at_index(
panel_lazy.filter(
pl.col(names["workColumn"]) == chart[names["selectedPeriods"]][1]
),
names["varianceAmountName"],
0,
)
period_zero_value = get_polars_value_at_index(
panel_lazy,
names["varianceAmountName"],
0,
)
if is_arrow:
line_width = 2
line_color = (
color_dict["greenColor"]
if period_one_value >= period_zero_value
else color_dict["redColor"]
)
y0_value, y1_value, _yshift, line_color = get_y1_y0_values(
number_of_charts,
False,
is_arrow,
count,
is_period_zero,
line_color,
chart,
count_rows,
)
shapes.append(
{
"type": "line",
"opacity": 1,
"line_width": line_width,
"line_color": line_color,
"yref": "paper",
"xref": f"x{count}",
"y0": y0_value,
"y1": y1_value,
"x0": x0_value,
"x1": x1_value,
"layer": "below",
}
)
return shapes
def _legacy_delta_annotation(
df: pl.DataFrame,
param: dict[str, Any],
chart: dict[str, Any],
color_dict: dict[str, str],
run: str,
count: int,
annotations: list[dict[str, Any]],
number_of_charts: int,
*,
is_text: bool,
is_arrow: bool,
count_rows: int,
) -> list[dict[str, Any]]:
"""Return legacy annotation metadata for total change arrows and labels."""
from modules.charting.adjust_position import get_y1_y0_values
from modules.charting.chart_primitives import divide_by_value_prefix
from modules.charting.draw_charts_utils import get_polars_value_at_index
from modules.utilities.config import get_naming_params
names = get_naming_params()
can_show_initial_and_final = True
if names["varianceAggregation"] in chart:
if (
chart[names["varianceAggregation"]]
not in [
names["totalVarianceAggregation"],
names["marginVariance"],
]
and names["drilldownReportRunName"] in run
):
can_show_initial_and_final = False
if not (
names["showInitialAndFinalValues"] in chart
and chart[names["showInitialAndFinalValues"]]
and can_show_initial_and_final
):
return annotations
panel_lazy = df.lazy()
period_one_value = get_polars_value_at_index(
panel_lazy.filter(
pl.col(names["workColumn"]) == chart[names["selectedPeriods"]][1]
),
names["varianceAmountName"],
0,
)
period_zero_value = get_polars_value_at_index(
panel_lazy,
names["varianceAmountName"],
0,
)
arrow_color = (
color_dict["greenColor"]
if period_one_value >= period_zero_value
else color_dict["redColor"]
)
y0_value, _y1_value, yshift, _line_color = get_y1_y0_values(
number_of_charts,
is_text,
is_arrow,
count,
True,
arrow_color,
chart,
count_rows,
)
if period_zero_value:
difference = divide_by_value_prefix(
period_one_value - period_zero_value,
chart,
False,
)
percent_change = (
(period_one_value - period_zero_value) / period_zero_value
) * 100
percent_text = (
f"<i>({int(round(percent_change, 0))}%)</i>"
if period_zero_value > 0 and math.isfinite(percent_change)
else ""
)
change_value = f"{names['deltaName']} {difference} {percent_text}"
else:
change_value = f"{names['deltaName']} nan"
annotations.append(
{
"showarrow": is_arrow,
"arrowcolor": arrow_color,
"text": None if is_arrow else change_value,
"xshift": 0 if is_arrow else 30,
"yshift": yshift,
"arrowhead": 5,
"arrowsize": 1,
"ay": y0_value,
"y": y0_value,
"yref": "paper",
"ax": period_zero_value,
"x": period_zero_value,
"xref": f"x{count}",
"axref": f"x{count}",
}
)
return annotations
def _delete_black_vertical_lines(fig: go.Figure) -> go.Figure:
"""Apply the legacy axis cleanup used by small-multiple waterfalls."""
fig.update_layout(
yaxis={
"showline": False,
"showticklabels": True,
}
)
fig.update_yaxes(showline=False)
fig.update_yaxes(zeroline=True, zerolinecolor="black", zerolinewidth=2)
return fig
def _draw_with_legacy_runtime(
df: pl.DataFrame,
recipe: dict[str, Any],
*,
legacy_frame: pl.DataFrame,
include_zero_components: bool = False,
) -> LegacyWaterfallFigure:
"""Draw the standard waterfall through the original legacy chart functions."""
from modules.charting import legacy_draw_waterfall
from modules.charting.chart_primitives import (
add_title_as_annotation,
get_color_dictionary,
)
from modules.charting.make_titles import make_vertical_waterfall_chart_title
from modules.charting.update_layouts import update_waterfall_layout_one_dimension
names, config, param, chart = _legacy_context(recipe)
chart_df = _legacy_chart_frame(
df,
recipe,
names,
config,
include_zero_components=include_zero_components,
)
baseline_total = _sum_column(df, "amount_baseline")
comparison_total = _sum_column(df, "amount_comparison")
for key in (
names["totalAmountPeriodZero"],
names["totalAmountPeriodZeroFiltered"],
names["periodZeroSum"],
):
param[key] = baseline_total
for key in (
names["totalAmountPeriodOne"],
names["totalAmountPeriodOneFiltered"],
names["periodOneSum"],
):
param[key] = comparison_total
param[names["totalVarianceValue"]] = comparison_total - baseline_total
color_dict = get_color_dictionary(chart)
run = names["runOneDimensionalAnalysis"]
fig, number_format, chart = legacy_draw_waterfall.draw_vertical_waterfall_chart(
chart_df,
color_dict,
param,
chart,
run,
)
layout_chart = dict(chart)
layout_chart.pop(names["varianceDifferentCalculations"], None)
fig = update_waterfall_layout_one_dimension(chart_df, fig, layout_chart)
fig.update_yaxes(autorange="reversed", ticks="")
title_chart = dict(chart)
if names["companySales"] in names.values():
title_chart[names["datasetTypeName"]] = names["companySales"]
title_chart[names["varianceDifferentCalculations"]] = True
title, param, title_chart = make_vertical_waterfall_chart_title(
chart_df,
names["verticalWaterfallChart"],
param,
None,
names["monetaryLocalCurrencyName"],
title_chart,
names["pyName"],
names["acName"],
)
fig = add_title_as_annotation(
fig,
title,
names["verticalWaterfallChart"],
title_chart,
)
baseline_label, comparison_label = _periods(recipe)
title = title.replace(
f"{names['acName']} vs {names['pyName']}",
f"{comparison_label} vs {baseline_label}",
)
fig.layout.annotations[-1].text = title
layout_title = _clean_html(getattr(fig.layout.title, "text", ""))
if layout_title:
fig.update_layout(
title={
"text": str(fig.layout.title.text).replace(
f"{names['acName']} vs {names['pyName']}",
f"{comparison_label} vs {baseline_label}",
)
}
)
return LegacyWaterfallFigure(
figure=fig,
audit={
"mode": "legacy_single",
"title": title,
"number_format": number_format,
"row_count": df.height,
"legacy_chart_row_count": chart_df.height,
"legacy_source_row_count": legacy_frame.height,
"source_functions": [
"modules.charting.draw_waterfall.draw_vertical_waterfall_chart",
"modules.charting.legacy_draw_waterfall.draw_vertical_waterfall_chart",
"modules.charting.update_layouts.update_waterfall_layout_one_dimension",
"modules.charting.make_titles.make_vertical_waterfall_chart_title",
"modules.charting.chart_primitives.add_title_as_annotation",
],
},
)
def _draw_component_with_legacy_runtime(
df: pl.DataFrame,
recipe: dict[str, Any],
*,
legacy_frame: pl.DataFrame,
) -> LegacyWaterfallFigure:
"""Draw the aggregate standard component bridge through legacy code."""
rendered = _draw_with_legacy_runtime(
_aggregate_component_frame(df),
_component_only_recipe(recipe),
legacy_frame=legacy_frame,
include_zero_components=True,
)
audit = dict(rendered.audit)
audit.update(
{
"mode": "legacy_component_single",
"panel_bridge": "standard_variance_components",
"source_row_count": df.height,
}
)
return LegacyWaterfallFigure(figure=rendered.figure, audit=audit)
def _legacy_small_multiples_figure(
df: pl.DataFrame,
recipe: dict[str, Any],
dimension: str,
*,
legacy_frame: pl.DataFrame,
) -> LegacyWaterfallFigure:
"""Build legacy fixed-dimension variance small multiples."""
from modules.charting import legacy_draw_waterfall
from modules.charting.adjust_position import move_labels_up
from modules.charting.chart_helpers import make_one_dimensional_variance_subplots
from modules.charting.chart_primitives import get_color_dictionary
from modules.charting.draw_charts_utils import get_polars_value_at_index
from modules.charting.plot_charts import plot_waterfall_small_multiples
from modules.charting.update_layouts import update_waterfall_layout_small_multiples
from modules.data.waterfall_data_prep import prepare_data_for_waterfall
from modules.utilities.ui_notifier import HeadlessChartCapture, use_ui_notifier
from modules.variance.variance_orchestrator import melt_data_on_variance_cols
from modules.variance.variance_utils import (
group_by_and_sort_data_for_variance_calculation,
)
slices = _small_multiple_values(df, dimension)
if not slices:
return _draw_with_legacy_runtime(df, recipe, legacy_frame=legacy_frame)
names, config, param, chart = _legacy_context(recipe)
aggregation = _legacy_small_multiples_aggregation(names, recipe, legacy_frame)
other_label = _legacy_small_multiples_other_label()
selected_values = [value for value, _slice in slices if value != other_label]
include_other = any(value == other_label for value, _slice in slices)
panel_values = [
other_label if value == other_label else value for value, _ in slices
]
legacy_panel_frame = _legacy_frame_with_panel_bucket(
legacy_frame,
dimension,
selected_values,
other_label=other_label,
include_other=include_other,
)
for key in (
names["totalAmountPeriodZero"],
names["totalAmountPeriodZeroFiltered"],
names["periodZeroSum"],
):
param[key] = _sum_column(
legacy_panel_frame,
names["monetaryLocalCurrencyName"]
+ names["separatorString"]
+ config["periodsArray"][0],
)
for key in (
names["totalAmountPeriodOne"],
names["totalAmountPeriodOneFiltered"],
names["periodOneSum"],
):
param[key] = _sum_column(
legacy_panel_frame,
names["monetaryLocalCurrencyName"]
+ names["separatorString"]
+ config["periodsArray"][1],
)
param[names["totalVarianceValue"]] = (
param[names["totalAmountPeriodOne"]] - param[names["totalAmountPeriodZero"]]
)
chart[names["mainDimension"]] = [dimension]
chart[names["varianceAggregation"]] = aggregation
chart[names["plotSmallMultiplesWaterfall"]] = True
chart[names["aggregateOtherWaterfalls"]] = include_other
chart[names["numberOfSmallMultiplesWaterfall"]] = max(1, len(selected_values))
chart[names["showInitialAndFinalValues"]] = True
chart[names["varianceAggregationOptionsArray"]] = [aggregation]
chart[names["smallMultiplesWaterfall"]] = panel_values
melted, param = melt_data_on_variance_cols(
legacy_panel_frame,
param,
chart,
[dimension],
)
variance_type = names["varianceTypeName"]
variance_amount = names["varianceAmountName"]
source_variance_types = _legacy_small_multiples_variance_types(aggregation, names)
if source_variance_types:
melted = melted.filter(pl.col(variance_type).is_in(source_variance_types))
grouped, _group_by_cols, _sum_cols = (
group_by_and_sort_data_for_variance_calculation(
melted,
[dimension, variance_type],
[variance_amount],
)
)
grouped = grouped.collect() if isinstance(grouped, pl.LazyFrame) else grouped
color_dict = get_color_dictionary(chart)
run = names["runOneDimensionalAnalysis"]
param[names["columnHash"]] = param.get(names["columnHash"], {})
capture = HeadlessChartCapture()
with use_ui_notifier(capture):
param, chart = plot_waterfall_small_multiples(
grouped,
legacy_panel_frame,
[dimension, variance_type],
param,
chart,
color_dict,
run,
)
captured_output = capture.last_chart_output()
fig = captured_output.figure
export_frame = captured_output.frame
panel_values = [
str(value)
for value in chart.get(names["smallMultiplesWaterfall"], panel_values)
]
delta_labels = [
_delta_percent_label(
*_panel_period_totals(legacy_panel_frame, dimension, value, names, config),
names,
)
for value in panel_values
]
title = make_vertical_waterfall_chart_title(df, recipe, dimension=dimension)
annotations = []
for annotation in fig.layout.annotations or ():
if "Sales variance" in _clean_html(getattr(annotation, "text", "")):
continue
annotations.append(annotation)
fig.layout.annotations = tuple(annotations)
traces = [trace for trace in fig.data if trace.type == "waterfall"]
if traces:
global_range = _combined_trace_extent(traces)
fig.update_xaxes(range=list(global_range), matches="x")
number_of_cols = min(3, max(1, len(panel_values)))
number_of_rows = math.ceil(max(1, len(panel_values)) / number_of_cols)
fig.update_yaxes(autorange="reversed", ticks="")
fig.update_layout(
title={"text": title, "x": 0.01, "xanchor": "left"},
meta={
"waterfall_small_multiples_grid": {
"rows": number_of_rows,
"cols": number_of_cols,
},
"waterfall_panel_titles": panel_values,
"waterfall_delta_percent_labels": delta_labels,
},
)
export_row_count = (
export_frame.height if isinstance(export_frame, pl.DataFrame) else None
)
return LegacyWaterfallFigure(
figure=fig,
audit={
"mode": "legacy_variance_small_multiples",
"title": title,
"small_multiples_dimension": dimension,
"small_multiples_panel_bridge": "legacy_price_units_mix",
"small_multiples_aggregation": aggregation,
"small_multiples_count": len(panel_values),
"small_multiples_limit": MAX_SMALL_MULTIPLES,
"row_count": df.height,
"legacy_source_row_count": legacy_frame.height,
"legacy_reference_function": (
"modules.charting.plot_charts.plot_waterfall_small_multiples"
),
"legacy_reference_function_call_mode": "executed_headless",
"source_functions": [
"modules.variance.variance_orchestrator.melt_data_on_variance_cols",
"modules.variance.variance_utils.group_by_and_sort_data_for_variance_calculation",
"modules.charting.plot_charts.plot_waterfall_small_multiples",
"modules.data.waterfall_data_prep.prepare_data_for_waterfall",
"modules.charting.legacy_draw_waterfall.draw_vertical_waterfall_chart",
"modules.charting.chart_helpers.set_up_tab_for_show_or_download_chart",
"modules.utilities.ui_notifier.HeadlessChartCapture",
],
"captured_chart_output_count": len(capture.chart_outputs),
"captured_plotly_chart_count": len(capture.plotly_charts),
"delta_percent_labels": delta_labels,
"number_format": "",
"export_row_count": export_row_count,
},
)
def _single_figure(df: pl.DataFrame, recipe: dict[str, Any]) -> LegacyWaterfallFigure:
"""Build a single legacy-style waterfall figure."""
labels, _measures, _values, _text = _waterfall_arrays(df, recipe)
height = max(520, 90 * len(labels) + 210)
fig = make_subplots(rows=1, cols=1, specs=[[{"type": "waterfall"}]])
_add_waterfall_trace(fig, df, recipe, row=1, col=1)
title = make_vertical_waterfall_chart_title(df, recipe)
_update_layout(fig, title=title, height=height, width=1200)
return LegacyWaterfallFigure(
figure=fig,
audit={
"mode": "single",
"title": title,
"row_count": df.height,
"source_functions": list(LEGACY_SOURCE_FUNCTIONS),
},
)
def _small_multiple_values(
df: pl.DataFrame, dimension: str
) -> list[tuple[str, pl.DataFrame]]:
"""Return top dimension slices for small-multiple waterfalls."""
try:
from modules.utilities.config import get_naming_params
names = get_naming_params()
other_label = names["aggregateOtherWaterfallsName"]
except (ImportError, KeyError):
other_label = "Other"
limit = MAX_SMALL_MULTIPLES
ranked = (
df.group_by(dimension)
.agg(pl.col("total_delta").abs().sum().alias("_abs_delta"))
.sort("_abs_delta", descending=True)
)
all_values = [str(value) for value in ranked[dimension].to_list()]
selected_values = all_values[:limit]
include_other = len(all_values) > limit
if include_other:
selected_values = all_values[: max(1, limit - 1)]
slices = [
(
value,
df.filter(pl.col(dimension).cast(pl.Utf8) == value),
)
for value in selected_values
]
if include_other:
slices.append(
(
other_label,
df.filter(~pl.col(dimension).cast(pl.Utf8).is_in(selected_values)),
)
)
return slices
def _small_multiples_figure(
df: pl.DataFrame, recipe: dict[str, Any], dimension: str
) -> LegacyWaterfallFigure:
"""Build a small-multiple legacy-style waterfall figure."""
slices = _small_multiple_values(df, dimension)
if not slices:
return _single_figure(df, recipe)
cols = min(3, len(slices))
rows = math.ceil(len(slices) / cols)
fig = make_subplots(
rows=rows,
cols=cols,
specs=[[{"type": "waterfall"} for _ in range(cols)] for _ in range(rows)],
subplot_titles=[value for value, _slice in slices],
horizontal_spacing=0.12,
vertical_spacing=0.18,
)
for index, (_value, slice_df) in enumerate(slices):
row = index // cols + 1
col = index % cols + 1
_add_waterfall_trace(fig, slice_df, recipe, row=row, col=col, show_labels=False)
title = make_vertical_waterfall_chart_title(df, recipe, dimension=dimension)
_update_layout(fig, title=title, height=max(520, rows * 360), width=1500)
return LegacyWaterfallFigure(
figure=fig,
audit={
"mode": "small_multiples",
"title": title,
"small_multiples_dimension": dimension,
"small_multiples_count": len(slices),
"small_multiples_limit": MAX_SMALL_MULTIPLES,
"row_count": df.height,
"source_functions": list(LEGACY_SOURCE_FUNCTIONS),
},
)
def draw_vertical_waterfall_chart(
df: pl.DataFrame,
recipe: dict[str, Any],
*,
small_multiples: bool = False,
small_multiples_dimension: str | None = None,
legacy_frame: pl.DataFrame | None = None,
) -> LegacyWaterfallFigure:
"""Build a legacy-style vertical waterfall chart for plugin output."""
if df.is_empty():
raise ValueError("Cannot render waterfall chart for an empty variance result.")
if small_multiples:
dimension = small_multiples_dimension or next(
(
column
for column in recipe["mappings"].get("dimensions", [])
if column in df.schema
),
None,
)
if legacy_frame is not None and dimension and dimension in df.schema:
return _legacy_small_multiples_figure(
df,
recipe,
dimension,
legacy_frame=legacy_frame,
)
if dimension and dimension in df.schema:
return _small_multiples_figure(df, recipe, dimension)
if legacy_frame is not None:
return _draw_component_with_legacy_runtime(
df,
recipe,
legacy_frame=legacy_frame,
)
return _single_figure(df, recipe)
SHA-256: 367d2a8c798abe5b40470acd5cd59702ab3576dc743213455bca8e12d411e370