← Files VeraARCHIVED FILE
modules/variance-analysis/scripts/root_cause_bridge_chart.py
48.3 KB · Oct 2, 2026 · 00:29 UTC
"""Root-cause bridge chart rendering from legacy variable-dimension output."""
from __future__ import annotations
import copy
import importlib
import math
import warnings
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import polars as pl
from ibcs_titles import build_ibcs_title, ibcs_title_html, measure_line_segments
from PIL import Image, ImageDraw, ImageFont
__all__ = [
"RootCauseBridgeChartExport",
"build_root_cause_bridge_chart_rows",
"write_root_cause_bridge_png",
]
TOLERANCE = 0.000001
MAX_DRIVER_ROWS = 8
METRIC_COLUMNS = {
"bridge_level",
"bridge_dimensions",
"variance_type",
"variance_amount",
"amount_baseline",
"amount_comparison",
"units_baseline",
"units_comparison",
"bridge_unique_value_weight",
}
COLORS = {
"actual": "#1F1F1F",
"baseline_period": "#A6A6A6",
"connector": "#C8C8C8",
"grid": "#EDEDED",
"negative": "#FF2F2F",
"positive": "#67C40F",
"text": "#1F2328",
"muted": "#666666",
"white": "#FFFFFF",
}
LEGACY_ROOT_CAUSE_RENDERER = (
"modules.charting.plot_charts.plot_root_cause_variable_waterfall"
)
LEGACY_ROOT_CAUSE_SOURCE_FUNCTIONS = [
"modules.variance.variance_decomposition.process_node_combinations",
LEGACY_ROOT_CAUSE_RENDERER,
"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",
]
@dataclass(frozen=True)
class RootCauseBridgeChartExport:
"""Root-cause bridge chart export paths and audit metadata."""
paths: list[str]
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 _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 _is_plan_label(label: str) -> bool:
"""Return whether ``label`` represents a plan-like scenario."""
normalized = label.strip().upper()
return normalized in {"PL", "PLAN", "BUDGET", "BDG", "FORECAST", "FC"}
def _is_period_comparison(recipe: dict[str, Any]) -> bool:
"""Return whether the recipe compares periods rather than scenarios."""
options = recipe.get("options") or {}
return options.get("comparison_basis") == "period"
def _format_number(value: float, *, signed: bool = True) -> str:
"""Return compact chart text with K/M/B suffixes."""
if not math.isfinite(value):
return ""
sign = "+" if signed and value > 0 else "-" if value < 0 else ""
abs_value = abs(value)
if abs_value >= 1_000_000_000:
text = f"{abs_value / 1_000_000_000:,.1f}B"
elif abs_value >= 1_000_000:
text = f"{abs_value / 1_000_000:,.1f}M"
elif abs_value >= 1_000:
text = f"{abs_value / 1_000:,.1f}K"
else:
text = f"{abs_value:,.0f}"
return f"{sign}{text}"
def _safe_float(value: Any, default: float = 0.0) -> float:
"""Return a finite float for chart calculations."""
try:
number = float(value)
except (TypeError, ValueError):
return default
return number if math.isfinite(number) else default
def _driver_percent_delta(value: float, amount_baseline: float) -> float | None:
"""Return row-level percent change for a root-cause driver."""
if amount_baseline <= TOLERANCE:
return None
percent = (value / amount_baseline) * 100
return percent if math.isfinite(percent) else None
def _format_percent_marker(value: float | None) -> str:
"""Return compact percent text for the legacy delta side panel."""
if value is None or not math.isfinite(value):
return ""
abs_value = abs(value)
if abs_value >= 10:
body = f"{abs_value:.0f}"
else:
body = f"{abs_value:.1f}"
sign = "+" if value > 0 else "-" if value < 0 else ""
return f"{sign}{body}"
def _font(size: int, *, bold: bool = False) -> ImageFont.ImageFont:
"""Return a readable font while staying robust in headless runs."""
candidates = [
(
"/System/Library/Fonts/Supplemental/Arial Bold.ttf"
if bold
else "/System/Library/Fonts/Supplemental/Arial.ttf"
),
"/Library/Fonts/Arial Bold.ttf" if bold else "/Library/Fonts/Arial.ttf",
"/System/Library/Fonts/Helvetica.ttc",
]
for candidate in candidates:
try:
return ImageFont.truetype(candidate, size=size)
except OSError:
continue
return ImageFont.load_default()
def _draw_segmented_text(
draw: ImageDraw.ImageDraw,
xy: tuple[int, int],
segments: tuple[tuple[str, bool], ...],
*,
fill: str,
regular_font: ImageFont.ImageFont,
bold_font: ImageFont.ImageFont,
) -> None:
"""Draw one title line with per-segment emphasis."""
x, y = xy
for text, emphasized in segments:
if not text:
continue
font = bold_font if emphasized else regular_font
draw.text((x, y), text, fill=fill, font=font)
bbox = draw.textbbox((x, y), text, font=font)
x += bbox[2] - bbox[0]
def _active_dimensions(bridge_dimensions: str) -> list[str]:
"""Return active dimension names from a legacy bridge-dimension string."""
if not bridge_dimensions or bridge_dimensions == "total":
return []
return [item.strip() for item in bridge_dimensions.split(",") if item.strip()]
def _dimension_columns(bridge: pl.DataFrame) -> list[str]:
"""Return dimension columns emitted by the legacy bridge output."""
return [column for column in bridge.columns if column not in METRIC_COLUMNS]
def _active_filter(
row: dict[str, Any], dimensions: list[str]
) -> tuple[tuple[str, str], ...]:
"""Return active dimension-value filters for a legacy bridge row."""
return tuple(
(dimension, str(row[dimension]))
for dimension in dimensions
if row.get(dimension) not in (None, "All")
)
def _display_labels(recipe: dict[str, Any]) -> dict[str, str]:
language = (
str(recipe.get("language") or "en")
.lower()
.replace("_", "-")
.split("-", maxsplit=1)[0]
)
return {
"it": {
"Total variance": "Scostamento totale",
"Price & volume & mix": "Prezzo, Volume e Mix",
"Price & units & mix": "Prezzo, Unità e Mix",
"Units & mix": "Unità e Mix",
"Total": "Totale",
"Other": "Altro",
"Price": "Prezzo",
"Units": "Unità",
"Volume": "Volume",
"Mix": "Mix",
},
"es": {
"Total variance": "Variación total",
"Price & volume & mix": "Precio, Volumen y Mix",
"Price & units & mix": "Precio, Unidades y Mix",
"Units & mix": "Unidades y Mix",
"Total": "Total",
"Other": "Otros",
"Price": "Precio",
"Units": "Unidades",
"Volume": "Volumen",
"Mix": "Mix",
},
"fr": {
"Total variance": "Écart total",
"Price & volume & mix": "Prix, Volume et Mix",
"Price & units & mix": "Prix, Unités et Mix",
"Units & mix": "Unités et Mix",
"Total": "Total",
"Other": "Autres",
"Price": "Prix",
"Units": "Unités",
"Volume": "Volume",
"Mix": "Mix",
},
"de": {
"Total variance": "Gesamtabweichung",
"Price & volume & mix": "Preis, Volumen und Mix",
"Price & units & mix": "Preis, Menge und Mix",
"Units & mix": "Menge und Mix",
"Total": "Gesamt",
"Other": "Sonstige",
"Price": "Preis",
"Units": "Menge",
"Volume": "Volumen",
"Mix": "Mix",
},
}.get(language, {})
def _translate_variance_type(value: str, labels: dict[str, str]) -> str:
translated = value
for source in sorted(labels, key=len, reverse=True):
translated = translated.replace(source, labels[source])
return translated
def _filter_label(
filters: tuple[tuple[str, str], ...],
labels: dict[str, str],
) -> str:
"""Return a compact chart label for active filters."""
return " / ".join(value for _dimension, value in filters) or labels.get(
"Total", "Total"
)
def _sequence_label(
row: dict[str, Any],
filters: tuple[tuple[str, str], ...],
*,
include_variance_type: bool,
labels: dict[str, str],
) -> str:
"""Return the chart label for one legacy sequence row."""
label = _filter_label(filters, labels)
variance_type = str(row.get("variance_type") or "").strip()
if (
include_variance_type
and variance_type
and variance_type.lower() not in {"total", "none"}
):
return f"{label} - {_translate_variance_type(variance_type, labels)}"
return label
def _legacy_sequence_driver_rows(
bridge: pl.DataFrame,
total_delta: float,
max_drivers: int = MAX_DRIVER_ROWS,
*,
include_variance_type: bool = True,
labels: dict[str, str] | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
"""Return driver rows directly from ``process_node_combinations`` output."""
dimensions = _dimension_columns(bridge)
display_labels = labels or {}
all_selected_rows: list[dict[str, Any]] = []
for row in bridge.to_dicts():
value = _safe_float(row.get("variance_amount"))
if abs(value) <= TOLERANCE:
continue
filters = _active_filter(row, dimensions)
bridge_dimensions = _active_dimensions(str(row.get("bridge_dimensions") or ""))
amount_baseline = _safe_float(row.get("amount_baseline"))
amount_comparison = _safe_float(row.get("amount_comparison"))
row_number = len(all_selected_rows) + 1
all_selected_rows.append(
{
"label": _sequence_label(
row,
filters,
include_variance_type=include_variance_type,
labels=display_labels,
),
"kind": "driver",
"value": value,
"source_value": value,
"row_number": row_number,
"amount_baseline": amount_baseline,
"amount_comparison": amount_comparison,
"percent_delta": _driver_percent_delta(value, amount_baseline),
"filters": filters,
"bridge_dimensions": bridge_dimensions,
"bridge_level": int(row.get("bridge_level") or len(filters)),
"variance_types": [str(row.get("variance_type") or "")],
}
)
if not all_selected_rows:
raise ValueError("Legacy process_node_combinations returned no chart rows.")
requested_driver_count = len(all_selected_rows)
display_limit = max(0, int(max_drivers))
selected_rows = all_selected_rows[:display_limit]
selected_sum = sum(float(row["value"]) for row in selected_rows)
residual = total_delta - selected_sum
if abs(residual) > TOLERANCE:
selected_rows.append(
{
"label": display_labels.get("Other", "Other"),
"kind": "driver",
"value": residual,
"source_value": residual,
"row_number": None,
"amount_baseline": 0.0,
"amount_comparison": residual,
"percent_delta": None,
"filters": tuple(),
"bridge_dimensions": ["residual"],
"bridge_level": 0,
"variance_types": ["Residual"],
}
)
active_dimension_labels = [
",".join(row["bridge_dimensions"])
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
]
unique_active_dimension_labels = list(dict.fromkeys(active_dimension_labels))
audit = {
"candidate_count": bridge.height,
"selected_driver_count": requested_driver_count,
"displayed_legacy_driver_count": len(
[
row
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
]
),
"selected_driver_filters": [
" / ".join(f"{dimension}={value}" for dimension, value in row["filters"])
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
],
"selected_driver_bridge_dimensions": [
[",".join(row["bridge_dimensions"])]
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
],
"selected_driver_active_dimensions": [
row["bridge_dimensions"]
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
],
"selected_bridge_levels": [
row["bridge_level"]
for row in selected_rows
if row["label"] != display_labels.get("Other", "Other")
],
"selected_sequence_unique_bridge_dimensions": unique_active_dimension_labels,
"selected_sequence_has_mixed_dimensions": len(unique_active_dimension_labels)
> 1,
"other_included": any(
row["label"] == display_labels.get("Other", "Other")
for row in selected_rows
),
"selection_strategy": "legacy_process_node_combinations",
"selection_truncated": requested_driver_count > display_limit,
}
return selected_rows, audit
def build_root_cause_bridge_chart_rows(
bridge: pl.DataFrame,
result: pl.DataFrame,
recipe: dict[str, Any],
*,
max_drivers: int = MAX_DRIVER_ROWS,
include_variance_type: bool = True,
) -> tuple[pl.DataFrame, dict[str, Any]]:
"""Build a readable bridge table from legacy root-cause output.
The legacy runtime owns the variable-dimension decomposition. This function
renders the ordered rows returned by ``process_node_combinations`` and only
adds ``Other`` for the mathematical residual.
"""
baseline_label, comparison_label = _periods(recipe)
baseline_total = _sum_column(result, "amount_baseline")
comparison_total = _sum_column(result, "amount_comparison")
total_delta = comparison_total - baseline_total
driver_rows, selection_audit = _legacy_sequence_driver_rows(
bridge,
total_delta,
max_drivers=max_drivers,
include_variance_type=include_variance_type,
labels=_display_labels(recipe),
)
rows = [
{
"label": baseline_label,
"kind": "baseline",
"value": baseline_total,
"row_number": None,
"amount_baseline": baseline_total,
"amount_comparison": None,
"percent_delta": None,
"percent_label": "",
},
*[
{
"label": str(row["label"]),
"kind": "driver",
"value": float(row["value"]),
"row_number": row.get("row_number"),
"amount_baseline": float(row.get("amount_baseline") or 0.0),
"amount_comparison": float(row.get("amount_comparison") or 0.0),
"percent_delta": row.get("percent_delta"),
"percent_label": _format_percent_marker(row.get("percent_delta")),
}
for row in driver_rows
],
{
"label": comparison_label,
"kind": "comparison",
"value": comparison_total,
"row_number": None,
"amount_baseline": None,
"amount_comparison": comparison_total,
"percent_delta": _driver_percent_delta(total_delta, baseline_total),
"percent_label": _format_percent_marker(
_driver_percent_delta(total_delta, baseline_total)
),
},
]
frame = pl.DataFrame(rows)
audit = {
"displayed_driver_count": len(driver_rows),
"max_driver_count": max_drivers,
"total_delta": total_delta,
"displayed_driver_sum": sum(float(row["value"]) for row in driver_rows),
"selected_bridge_dimensions": "legacy_sequence",
"selected_bridge_summary": "sequential residual driver sequence",
**selection_audit,
}
audit["chart_reconciliation_delta"] = (
float(audit["displayed_driver_sum"]) - total_delta
)
return frame, audit
def _x_position(value: float, low: float, high: float, left: int, width: int) -> int:
"""Map a value to a horizontal pixel position."""
if abs(high - low) < TOLERANCE:
high = low + 1.0
return int(left + ((value - low) / (high - low)) * width)
def _draw_bar(
draw: ImageDraw.ImageDraw,
box: tuple[int, int, int, int],
*,
fill: str,
outline: str | None = None,
width: int = 1,
) -> None:
"""Draw a rounded bar, using outline only when supplied."""
draw.rounded_rectangle(box, radius=4, fill=fill, outline=outline, width=width)
def _bar_box_from_zero(
zero_x: int,
value_x: int,
y0: int,
y1: int,
) -> tuple[int, int, int, int]:
"""Return a horizontal bar box that is valid on either side of zero."""
x0 = min(zero_x, value_x)
x1 = max(zero_x, value_x)
return (x0, y0, max(x1, x0 + 3), y1)
def _fit_text(
draw: ImageDraw.ImageDraw,
text: str,
font: ImageFont.ImageFont,
max_width: int,
) -> str:
"""Return text truncated to fit a fixed pixel width."""
if draw.textbbox((0, 0), text, font=font)[2] <= max_width:
return text
ellipsis = "..."
low = 0
high = len(text)
while low < high:
mid = (low + high + 1) // 2
candidate = f"{text[:mid].rstrip()}{ellipsis}"
if draw.textbbox((0, 0), candidate, font=font)[2] <= max_width:
low = mid
else:
high = mid - 1
return f"{text[:low].rstrip()}{ellipsis}"
def _chart_kind(audit: dict[str, Any]) -> str:
"""Return the IBCS title kind for the selected bridge sequence."""
artifact = str(audit.get("artifact") or "")
variance_mode = str(audit.get("root_cause_variance_mode") or "")
has_mixed_dimensions = bool(audit.get("selected_sequence_has_mixed_dimensions"))
if "drilldown" in artifact:
return "root_cause_drilldown"
if variance_mode == "component_variance":
return "root_cause_component"
if variance_mode == "total_variance":
return (
"variable_root_cause_total" if has_mixed_dimensions else "root_cause_total"
)
if has_mixed_dimensions:
return "variable_root_cause"
return "root_cause"
def _collect_if_lazy(frame: pl.DataFrame | pl.LazyFrame) -> pl.DataFrame:
"""Return an eager Polars frame."""
return frame.collect() if isinstance(frame, pl.LazyFrame) else frame
def _legacy_run_name(artifact_name: str, names: dict[str, str]) -> str:
"""Return the legacy report run name matching the chart artifact."""
if "drilldown" in artifact_name:
return names["drilldownReportRunName"]
return names["mainReportRunName"]
def _legacy_chart_audit(
captured_output: Any,
capture: Any,
*,
renderer: str,
plotly_export_error: str | None,
) -> dict[str, Any]:
"""Return audit data proving the legacy Plotly path was executed."""
frame = _collect_if_lazy(captured_output.frame)
layout_json = captured_output.figure.layout.to_plotly_json()
trace_types = [
str(getattr(trace, "type", "unknown")) for trace in captured_output.figure.data
]
percent_trace_count = sum(
1
for trace in captured_output.figure.data
if str(getattr(trace, "xaxis", "")) == "x2"
or str(getattr(trace, "yaxis", "")) == "y2"
)
measure_column = next(
(column for column in frame.columns if column.lower() == "measure"),
None,
)
measure_values = (
frame.get_column(measure_column).to_list() if measure_column else []
)
audit = {
"renderer": renderer,
"legacy_reference_function": LEGACY_ROOT_CAUSE_RENDERER,
"legacy_reference_function_call_mode": "executed_headless",
"source_functions": LEGACY_ROOT_CAUSE_SOURCE_FUNCTIONS,
"captured_chart_output_count": len(capture.chart_outputs),
"captured_plotly_chart_count": len(capture.plotly_charts),
"legacy_chart_ready_row_count": frame.height,
"legacy_chart_ready_columns": frame.columns,
"legacy_trace_types": trace_types,
"legacy_percent_side_panel": "xaxis2" in layout_json or percent_trace_count > 0,
"legacy_initial_final_values": (
len(measure_values) >= 2
and str(measure_values[0]) == "absolute"
and str(measure_values[-1]) == "absolute"
),
}
if plotly_export_error:
audit["plotly_export_error"] = plotly_export_error
return audit
def _legacy_root_cause_canvas(row_count: int) -> tuple[int, int]:
"""Return a readable export canvas for variable root-cause bridges."""
width = 1280
height = max(650, 160 + max(row_count, 7) * 72)
return width, height
def _screenshot_plotly_html(
html_path: Path,
output_path: Path,
*,
width: int,
height: int,
) -> str | None:
"""Screenshot a standalone legacy Plotly HTML file when a browser is available."""
try:
from playwright.sync_api import Error as PlaywrightError
from playwright.sync_api import sync_playwright
except ImportError as exc:
return str(exc)
try:
with sync_playwright() as playwright:
try:
browser = playwright.chromium.launch(
channel="chrome",
headless=True,
args=[
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
],
)
except PlaywrightError:
browser = playwright.chromium.launch(
headless=True,
args=[
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
],
)
page = browser.new_page(
viewport={"width": width, "height": height},
device_scale_factor=1,
)
page.goto(html_path.resolve().as_uri(), wait_until="networkidle")
page.locator(".plotly-graph-div").first.wait_for(
state="visible",
timeout=10_000,
)
page.wait_for_timeout(500)
page.screenshot(path=str(output_path), full_page=True)
browser.close()
except (OSError, RuntimeError, ValueError, PlaywrightError) as exc:
return str(exc)
return None
def _write_legacy_root_cause_png(
legacy_frame: pl.DataFrame | pl.LazyFrame,
recipe: dict[str, Any],
audit: dict[str, Any],
output_path: Path,
*,
legacy_param: dict[str, Any],
legacy_chart: dict[str, Any],
legacy_index_cols: list[str],
baseline_total: float,
comparison_total: float,
) -> dict[str, Any]:
"""Render the root-cause bridge through the legacy Plotly waterfall path."""
from legacy_adapter import _ensure_legacy_import_path
_ensure_legacy_import_path()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
config = importlib.import_module("modules.utilities.config")
chart_primitives = importlib.import_module("modules.charting.chart_primitives")
plot_charts = importlib.import_module("modules.charting.plot_charts")
ui_notifier = importlib.import_module("modules.utilities.ui_notifier")
names = config.get_naming_params()
param = copy.deepcopy(legacy_param)
chart = copy.deepcopy(legacy_chart)
legacy_source = _collect_if_lazy(legacy_frame)
available_index_cols = [
column for column in legacy_index_cols if column in legacy_source.columns
]
if available_index_cols:
legacy_source = legacy_source.with_columns(
[
pl.when(pl.col(column).cast(pl.Utf8) == "All")
.then(pl.lit(""))
.otherwise(pl.col(column).cast(pl.Utf8))
.alias(column)
for column in available_index_cols
]
)
chart[names["processingChoice"]] = names["runVariableDimensionalAnalysis"]
chart[names["varianceAggregation"]] = names["totalVarianceAggregation"]
chart[names["showInitialAndFinalValues"]] = True
chart[names["varianceInPercent"]] = False
chart[names["shareOfTotalMarket"]] = False
chart[names["plotSmallMultiplesWaterfall"]] = False
chart[names["mainDimension"]] = list(available_index_cols)
param[names["columnHash"]] = param.get(names["columnHash"], {})
param[names["totalAmountPeriodZero"]] = baseline_total
param[names["totalAmountPeriodOne"]] = comparison_total
param[names["totalVarianceValue"]] = comparison_total - baseline_total
param[names["periodZeroSum"]] = baseline_total
param[names["periodOneSum"]] = comparison_total
color_dict = chart_primitives.get_color_dictionary(chart)
capture = ui_notifier.HeadlessChartCapture()
run = _legacy_run_name(str(audit.get("artifact") or ""), names)
with ui_notifier.use_ui_notifier(capture):
plot_charts.plot_root_cause_variable_waterfall(
legacy_source,
list(available_index_cols),
param,
chart,
color_dict,
run,
)
captured_output = capture.last_chart_output()
fig = captured_output.figure
captured_frame = _collect_if_lazy(captured_output.frame)
export_width, export_height = _legacy_root_cause_canvas(captured_frame.height)
ibcs_title = build_ibcs_title(recipe, chart_kind=str(audit["chart_kind"]))
fig.update_layout(
width=export_width,
height=export_height,
autosize=False,
title={
"text": ibcs_title_html(ibcs_title),
"x": 0.01,
"xanchor": "left",
"font": {"size": 18},
},
)
margin = dict(fig.layout.margin.to_plotly_json()) if fig.layout.margin else {}
margin["l"] = max(int(margin.get("l") or 0), 360)
margin["r"] = max(int(margin.get("r") or 0), 135)
margin["t"] = max(int(margin.get("t") or 0), 145)
margin["b"] = max(int(margin.get("b") or 0), 70)
fig.update_layout(margin=margin)
renderer = "legacy_plotly+kaleido"
plotly_export_error: str | None = None
html_path: Path | None = None
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
fig.write_image(str(output_path), format="png")
except (OSError, RuntimeError, ValueError) as exc:
plotly_export_error = str(exc)
html_path = output_path.with_suffix(".html")
fig.write_html(str(html_path), include_plotlyjs=True)
screenshot_error = _screenshot_plotly_html(
html_path,
output_path,
width=export_width,
height=export_height,
)
if screenshot_error is None and output_path.exists():
renderer = "legacy_plotly+browser_screenshot"
else:
renderer = "legacy_plotly+html"
output_path = html_path
legacy_audit = _legacy_chart_audit(
captured_output,
capture,
renderer=renderer,
plotly_export_error=plotly_export_error,
)
legacy_audit.update(
{
"path": str(output_path),
"bytes": output_path.stat().st_size,
"legacy_variance_aggregation": chart[names["varianceAggregation"]],
"legacy_processing_choice": chart[names["processingChoice"]],
"legacy_report_run": run,
"legacy_requested_index_cols": legacy_index_cols,
"legacy_rendered_index_cols": available_index_cols,
"export_width": export_width,
"export_height": export_height,
}
)
if html_path is not None:
legacy_audit["legacy_plotly_html_path"] = str(html_path)
legacy_audit["legacy_plotly_html_bytes"] = html_path.stat().st_size
if renderer == "legacy_plotly+html" and html_path is not None:
legacy_audit["browser_screenshot_error"] = screenshot_error
return legacy_audit
def _draw_ibcs_title(
draw: ImageDraw.ImageDraw,
recipe: dict[str, Any],
audit: dict[str, Any],
) -> None:
"""Draw the standard three-row IBCS title."""
who_font = _font(18)
title_font = _font(18)
title_subject_font = _font(18, bold=True)
subtitle_font = _font(17)
title_lines = audit.get("chart_title_lines")
if not isinstance(title_lines, list) or not title_lines:
title_lines = build_ibcs_title(
recipe,
chart_kind=_chart_kind(audit),
).lines()
if title_lines:
draw.text((58, 26), str(title_lines[0]), fill=COLORS["muted"], font=who_font)
if len(title_lines) > 1:
_draw_segmented_text(
draw,
(58, 51),
measure_line_segments(str(title_lines[1])),
fill=COLORS["text"],
regular_font=title_font,
bold_font=title_subject_font,
)
if len(title_lines) > 2:
draw.text(
(58, 77),
str(title_lines[2]),
fill=COLORS["muted"],
font=subtitle_font,
)
def _write_total_variance_png(
rows: pl.DataFrame,
recipe: dict[str, Any],
audit: dict[str, Any],
output_path: Path,
) -> None:
"""Render total root-cause rows as overlaid initial/final value bars."""
baseline_label, comparison_label = _periods(recipe)
period_mode = _is_period_comparison(recipe)
row_count = rows.height
width = 1280
row_height = 74
top = 142
bottom = 46
height = max(560, top + row_height * row_count + bottom)
left = 430
right = 230
plot_width = width - left - right
bar_height = 16
row_overlay_offset = 7
label_x = 58
delta_panel_left = width - right + 42
delta_panel_width = right - 78
delta_value_x = delta_panel_left - 88
label_font = _font(18)
value_font = _font(16, bold=True)
small_font = _font(14)
subtitle_font = _font(17)
value_candidates: list[float] = []
for row in rows.to_dicts():
if row["kind"] in {"baseline", "comparison"}:
value_candidates.append(float(row["value"]))
else:
value_candidates.extend(
[
float(row.get("amount_baseline") or 0.0),
float(row.get("amount_comparison") or 0.0),
]
)
low = min(0.0, *value_candidates)
high = max(0.0, *value_candidates)
padding = max((high - low) * 0.04, 1.0)
low -= padding
high += padding
image = Image.new("RGB", (width, height), COLORS["white"])
draw = ImageDraw.Draw(image)
_draw_ibcs_title(draw, recipe, audit)
zero_x = _x_position(0.0, low, high, left, plot_width)
draw.line(
(zero_x, top - 18, zero_x, height - bottom + 12),
fill=COLORS["grid"],
width=1,
)
percent_values = [
float(row["percent_delta"])
for row in rows.to_dicts()[1:-1]
if row.get("percent_delta") is not None
and math.isfinite(float(row["percent_delta"]))
]
if percent_values:
pct_low = min(0.0, min(percent_values))
pct_high = max(0.0, max(percent_values))
pct_padding = max((pct_high - pct_low) * 0.18, 1.0)
pct_low -= pct_padding
pct_high += pct_padding
pct_zero_x = _x_position(
0.0,
pct_low,
pct_high,
delta_panel_left,
delta_panel_width,
)
draw.text(
(delta_panel_left, top - 54),
"\u0394%",
fill=COLORS["text"],
font=subtitle_font,
)
draw.line(
(pct_zero_x, top - 18, pct_zero_x, height - bottom + 12),
fill=COLORS["grid"],
width=1,
)
else:
pct_low = pct_high = 0.0
pct_zero_x = delta_panel_left
for index, row in enumerate(rows.to_dicts()):
y = top + index * row_height
kind = str(row["kind"])
text_x = label_x
label = _fit_text(
draw,
str(row["label"]),
label_font,
max_width=left - text_x - 20,
)
draw.text((text_x, y + 10), label, fill=COLORS["text"], font=label_font)
if kind in {"baseline", "comparison"}:
value = float(row["value"])
x1 = _x_position(value, low, high, left, plot_width)
color = COLORS["actual"] if kind == "comparison" else COLORS["white"]
outline = COLORS["actual"] if kind == "baseline" else None
if (
kind == "baseline"
and period_mode
and not _is_plan_label(baseline_label)
):
color = COLORS["baseline_period"]
outline = None
_draw_bar(
draw,
_bar_box_from_zero(zero_x, x1, y + 8, y + 8 + bar_height),
fill=color,
outline=outline,
width=2 if outline else 1,
)
draw.text(
(x1 + 10, y + 7),
_format_number(value, signed=False),
fill=COLORS["text"],
font=value_font,
)
continue
baseline_value = float(row.get("amount_baseline") or 0.0)
comparison_value = float(row.get("amount_comparison") or 0.0)
delta_value = comparison_value - baseline_value
baseline_x = _x_position(baseline_value, low, high, left, plot_width)
comparison_x = _x_position(comparison_value, low, high, left, plot_width)
baseline_top = y + 12
comparison_top = baseline_top + row_overlay_offset
_draw_bar(
draw,
_bar_box_from_zero(
zero_x,
baseline_x,
baseline_top,
baseline_top + bar_height,
),
fill=COLORS["baseline_period"],
)
_draw_bar(
draw,
_bar_box_from_zero(
zero_x,
comparison_x,
comparison_top,
comparison_top + bar_height,
),
fill=COLORS["actual"],
)
baseline_label_y = baseline_top - 12
comparison_label_y = comparison_top + bar_height - 2
draw.text(
(baseline_x + 8, baseline_label_y),
_format_number(baseline_value, signed=False),
fill=COLORS["muted"],
font=small_font,
)
draw.text(
(comparison_x + 8, comparison_label_y),
_format_number(comparison_value, signed=False),
fill=COLORS["text"],
font=small_font,
)
delta_label = _format_number(delta_value)
delta_color = COLORS["positive"] if delta_value >= 0 else COLORS["negative"]
draw.text(
(delta_value_x, comparison_top - 5),
delta_label,
fill=delta_color,
font=value_font,
)
percent_delta = row.get("percent_delta")
percent_label = str(row.get("percent_label") or "")
if (
percent_label
and percent_delta is not None
and math.isfinite(float(percent_delta))
):
center_y = comparison_top + (bar_height // 2)
pct_x = _x_position(
float(percent_delta),
pct_low,
pct_high,
delta_panel_left,
delta_panel_width,
)
line_color = (
COLORS["positive"] if float(percent_delta) >= 0 else COLORS["negative"]
)
draw.line(
(min(pct_zero_x, pct_x), center_y, max(pct_zero_x, pct_x), center_y),
fill=line_color,
width=2,
)
marker_box = (pct_x - 4, center_y - 4, pct_x + 4, center_y + 4)
draw.rectangle(marker_box, fill=COLORS["actual"])
label_box = draw.textbbox((0, 0), percent_label, font=value_font)
label_width = label_box[2] - label_box[0]
percent_label_y = comparison_top - 5
if pct_x >= pct_zero_x:
draw.text(
(pct_x + 8, percent_label_y),
percent_label,
fill=COLORS["text"],
font=value_font,
)
else:
draw.text(
(pct_x - label_width - 8, percent_label_y),
percent_label,
fill=COLORS["text"],
font=value_font,
)
image.save(output_path)
def _write_png(
rows: pl.DataFrame,
recipe: dict[str, Any],
audit: dict[str, Any],
output_path: Path,
) -> None:
"""Render a legacy-style root-cause bridge PNG with Pillow."""
baseline_label, comparison_label = _periods(recipe)
period_mode = _is_period_comparison(recipe)
row_count = rows.height
width = 1280
row_height = 62
top = 142
bottom = 46
height = max(520, top + row_height * row_count + bottom)
left = 430
right = 230
plot_width = width - left - right
bar_height = 24
label_x = 58
delta_panel_left = width - right + 42
delta_panel_width = right - 78
values = rows["value"].to_list()
baseline_total = float(values[0])
comparison_total = float(values[-1])
cumulative = baseline_total
low = min(0.0, baseline_total, comparison_total)
high = max(baseline_total, comparison_total)
for row in rows.to_dicts()[1:-1]:
next_value = cumulative + float(row["value"])
low = min(low, cumulative, next_value)
high = max(high, cumulative, next_value)
cumulative = next_value
padding = max((high - low) * 0.04, 1.0)
low -= padding
high += padding
image = Image.new("RGB", (width, height), COLORS["white"])
draw = ImageDraw.Draw(image)
subtitle_font = _font(17)
label_font = _font(18)
value_font = _font(16, bold=True)
small_font = _font(14)
_draw_ibcs_title(draw, recipe, audit)
zero_x = _x_position(0.0, low, high, left, plot_width)
draw.line(
(zero_x, top - 18, zero_x, height - bottom + 12),
fill=COLORS["grid"],
width=1,
)
percent_values = [
float(row["percent_delta"])
for row in rows.to_dicts()[1:-1]
if row.get("percent_delta") is not None
and math.isfinite(float(row["percent_delta"]))
]
if percent_values:
pct_low = min(0.0, min(percent_values))
pct_high = max(0.0, max(percent_values))
pct_padding = max((pct_high - pct_low) * 0.18, 1.0)
pct_low -= pct_padding
pct_high += pct_padding
pct_zero_x = _x_position(
0.0,
pct_low,
pct_high,
delta_panel_left,
delta_panel_width,
)
draw.text(
(delta_panel_left, top - 54),
"\u0394%",
fill=COLORS["text"],
font=subtitle_font,
)
draw.line(
(pct_zero_x, top - 18, pct_zero_x, height - bottom + 12),
fill=COLORS["grid"],
width=1,
)
else:
pct_low = pct_high = 0.0
pct_zero_x = delta_panel_left
cumulative = baseline_total
previous_x: int | None = None
previous_y: int | None = None
for index, row in enumerate(rows.to_dicts()):
y = top + index * row_height
center_y = y + bar_height // 2
kind = str(row["kind"])
text_x = label_x
label = _fit_text(
draw,
str(row["label"]),
label_font,
max_width=left - text_x - 20,
)
value = float(row["value"])
draw.text((text_x, y + 1), label, fill=COLORS["text"], font=label_font)
if kind in {"baseline", "comparison"}:
x0 = _x_position(0.0, low, high, left, plot_width)
x1 = _x_position(value, low, high, left, plot_width)
if kind == "baseline":
if period_mode and not _is_plan_label(baseline_label):
_draw_bar(
draw,
(x0, y, x1, y + bar_height),
fill=COLORS["baseline_period"],
)
else:
_draw_bar(
draw,
(x0, y, x1, y + bar_height),
fill=COLORS["white"],
outline=COLORS["actual"],
width=2,
)
else:
_draw_bar(draw, (x0, y, x1, y + bar_height), fill=COLORS["actual"])
draw.text(
(x1 + 10, y + 1),
_format_number(value, signed=False),
fill=COLORS["text"],
font=value_font,
)
previous_x = x1
previous_y = center_y
if kind == "baseline":
cumulative = value
continue
start = cumulative
end = cumulative + value
x0 = _x_position(min(start, end), low, high, left, plot_width)
x1 = _x_position(max(start, end), low, high, left, plot_width)
connector_target = x0 if value >= 0 else x1
if previous_x is not None and previous_y is not None:
draw.line(
(previous_x, previous_y, previous_x, center_y),
fill=COLORS["connector"],
width=1,
)
draw.line(
(previous_x, center_y, connector_target, center_y),
fill=COLORS["connector"],
width=1,
)
color = COLORS["positive"] if value >= 0 else COLORS["negative"]
_draw_bar(draw, (x0, y, max(x1, x0 + 3), y + bar_height), fill=color)
text = _format_number(value)
if value >= 0:
draw.text((x1 + 10, y + 1), text, fill=COLORS["text"], font=value_font)
previous_x = x1
else:
text_box = draw.textbbox((0, 0), text, font=value_font)
draw.text(
(x0 - (text_box[2] - text_box[0]) - 10, y + 1),
text,
fill=COLORS["text"],
font=value_font,
)
previous_x = x0
previous_y = center_y
percent_delta = row.get("percent_delta")
percent_label = str(row.get("percent_label") or "")
if (
percent_label
and percent_delta is not None
and math.isfinite(float(percent_delta))
):
pct_x = _x_position(
float(percent_delta),
pct_low,
pct_high,
delta_panel_left,
delta_panel_width,
)
line_color = (
COLORS["positive"] if float(percent_delta) >= 0 else COLORS["negative"]
)
draw.line(
(min(pct_zero_x, pct_x), center_y, max(pct_zero_x, pct_x), center_y),
fill=line_color,
width=2,
)
marker_box = (pct_x - 4, center_y - 4, pct_x + 4, center_y + 4)
draw.rectangle(marker_box, fill=COLORS["actual"])
label_box = draw.textbbox((0, 0), percent_label, font=value_font)
label_width = label_box[2] - label_box[0]
if pct_x >= pct_zero_x:
draw.text(
(pct_x + 8, y + 1),
percent_label,
fill=COLORS["text"],
font=value_font,
)
else:
draw.text(
(pct_x - label_width - 8, y + 1),
percent_label,
fill=COLORS["text"],
font=value_font,
)
cumulative = end
image.save(output_path)
def write_root_cause_bridge_png(
bridge: pl.DataFrame,
result: pl.DataFrame,
recipe: dict[str, Any],
output_dir: Path,
*,
artifact_name: str = "root_cause_bridge.png",
variance_mode: str = "component_variance",
legacy_frame: pl.DataFrame | pl.LazyFrame | None = None,
legacy_param: dict[str, Any] | None = None,
legacy_chart: dict[str, Any] | None = None,
legacy_index_cols: list[str] | None = None,
legacy_plotly_renderer: bool = False,
) -> RootCauseBridgeChartExport:
"""Write an IBCS-style PNG from legacy root-cause bridge output."""
uses_legacy_sequence = legacy_frame is not None
normalized_variance_mode = (
"total_variance" if variance_mode == "total_variance" else "component_variance"
)
audit: dict[str, Any] = {
"enabled": True,
"artifact": artifact_name,
"format": "png",
"root_cause_variance_mode": normalized_variance_mode,
"renderer": (
"pillow_from_legacy_root_cause_sequence"
if uses_legacy_sequence
else "pillow_root_cause_sequence"
),
}
if uses_legacy_sequence:
renderer_function = (
"root_cause_bridge_chart._write_total_variance_png"
if normalized_variance_mode == "total_variance"
else "root_cause_bridge_chart._write_png"
)
audit.update(
{
"legacy_reference_function": (
"modules.variance.variance_decomposition."
"process_node_combinations"
),
"legacy_reference_function_call_mode": "sequence_already_materialized",
"source_functions": [
(
"modules.variance.variance_decomposition."
"process_node_combinations"
),
f"plugins.variance-analysis.scripts.{renderer_function}",
],
}
)
rows, row_audit = build_root_cause_bridge_chart_rows(
bridge,
result,
recipe,
include_variance_type=normalized_variance_mode == "component_variance",
)
audit.update(row_audit)
chart_kind = _chart_kind(audit)
ibcs_title = build_ibcs_title(recipe, chart_kind=chart_kind)
audit["chart_title"] = " / ".join(ibcs_title.lines())
audit["chart_title_lines"] = ibcs_title.lines()
audit["chart_kind"] = chart_kind
chart_path = output_dir / artifact_name
if (
legacy_frame is not None
and legacy_param is not None
and legacy_chart is not None
and legacy_index_cols is not None
and legacy_plotly_renderer
):
audit.update(
_write_legacy_root_cause_png(
legacy_frame,
recipe,
audit,
chart_path,
legacy_param=legacy_param,
legacy_chart=legacy_chart,
legacy_index_cols=legacy_index_cols,
baseline_total=_sum_column(result, "amount_baseline"),
comparison_total=_sum_column(result, "amount_comparison"),
)
)
path = Path(str(audit["path"]))
audit.update(
{
"status": "written",
"format": path.suffix.lstrip(".") or "png",
"bytes": path.stat().st_size,
}
)
return RootCauseBridgeChartExport(paths=[str(path)], audit=audit)
if normalized_variance_mode == "total_variance":
_write_total_variance_png(rows, recipe, audit, chart_path)
renderer_version = "root_cause_total_initial_final_overlay_v2"
else:
_write_png(rows, recipe, audit, chart_path)
renderer_version = "root_cause_component_contribution_v1"
audit.update(
{
"status": "written",
"path": str(chart_path),
"bytes": chart_path.stat().st_size,
"pillow_renderer_version": renderer_version,
"row_number_markers": False,
"delta_percent_side_panel": True,
"delta_percent_basis": "driver_variance_amount_over_driver_baseline",
"initial_final_value_bars": normalized_variance_mode == "total_variance",
}
)
return RootCauseBridgeChartExport(paths=[str(chart_path)], audit=audit)
SHA-256: bc9c536a0d95a6fd000d3e00a5d9710c60deb5544a87a1cfae4c082a6250c518