# Copyright 2022 - 2026 The PyMC Labs Developers
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
GenerateReport pipeline step.
Collects effect summaries, plots, and sensitivity check results from the
pipeline context and renders a structured HTML report.
"""
from __future__ import annotations
import base64
import io
import logging
from pathlib import Path
from typing import Any
from jinja2 import Environment, FileSystemLoader
from causalpy.pipeline import PipelineContext
logger = logging.getLogger(__name__)
_TEMPLATE_DIR = Path(__file__).resolve().parent.parent / "templates"
[docs]
class GenerateReport:
"""Pipeline step that generates an HTML report from pipeline results.
Parameters
----------
include_plots : bool, default True
Whether to include diagnostic plots in the report. Covers both the
experiment plot and any figures the sensitivity checks attached to
their results, so turning it off leaves the report imageless.
include_effect_summary : bool, default True
Whether to include the effect summary section.
include_sensitivity : bool, default True
Whether to include sensitivity analysis results.
output_file : str or Path, optional
If provided, write the HTML report to this file.
Examples
--------
>>> import causalpy as cp # doctest: +SKIP
>>> step = cp.GenerateReport(
... include_plots=True, output_file="report.html"
... ) # doctest: +SKIP
"""
[docs]
def __init__(
self,
include_plots: bool = True,
include_effect_summary: bool = True,
include_sensitivity: bool = True,
output_file: str | Path | None = None,
) -> None:
self.include_plots = include_plots
self.include_effect_summary = include_effect_summary
self.include_sensitivity = include_sensitivity
self.output_file = Path(output_file) if output_file else None
[docs]
def validate(self, context: PipelineContext) -> None:
"""GenerateReport has no strict prerequisites; it gracefully handles
missing data.
Parameters
----------
context : PipelineContext
Pipeline context (unused; required by the step interface).
"""
@staticmethod
def _encode_figure(fig: Any) -> str:
"""Encode a matplotlib figure as a base64 PNG string."""
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=100, bbox_inches="tight")
buf.seek(0)
return base64.b64encode(buf.read()).decode("utf-8")
def _render_plot(self, experiment: Any) -> list[str]:
"""Render experiment plots as base64-encoded PNG strings."""
plots: list[str] = []
try:
import matplotlib.pyplot as plt
fig, _ = experiment.plot()
plots.append(self._encode_figure(fig))
plt.close(fig)
except Exception as exc:
logger.debug("Could not render plot: %s", exc)
return plots
def _render_check_figures(self, check_result: Any) -> list[str]:
"""Render the figures a check attached to its result.
The figures are not closed: they belong to the ``CheckResult`` the
caller still holds, and closing them would blank it.
"""
figures: list[str] = []
for fig in check_result.figures:
try:
figures.append(self._encode_figure(fig))
except Exception as exc:
logger.debug(
"Could not render figure from check %s: %s",
check_result.check_name,
exc,
)
return figures
[docs]
def run(self, context: PipelineContext) -> PipelineContext:
"""Generate the HTML report and store it in the context.
Parameters
----------
context : PipelineContext
Pipeline context providing ``experiment``, ``effect_summary``,
and ``sensitivity_results`` (any of which may be ``None``).
Returns
-------
PipelineContext
The same context with ``report`` populated.
"""
env = Environment(
loader=FileSystemLoader(str(_TEMPLATE_DIR)),
autoescape=True,
)
template = env.get_template("report.html")
effect_summary = None
effect_summary_table_html = None
if self.include_effect_summary and context.effect_summary is not None:
effect_summary = context.effect_summary
if effect_summary.table is not None:
effect_summary_table_html = effect_summary.table.to_html(
classes="", index=False, border=0
)
plots: list[str] = []
if self.include_plots and context.experiment is not None:
plots = self._render_plot(context.experiment)
sensitivity_results: list[dict[str, Any]] = []
if self.include_sensitivity and context.sensitivity_results:
for cr in context.sensitivity_results:
entry: dict[str, Any] = {
"check_name": cr.check_name,
"passed": cr.passed,
"text": cr.text,
"table_html": None,
"figures": (
self._render_check_figures(cr) if self.include_plots else []
),
}
if cr.table is not None:
entry["table_html"] = cr.table.to_html(
classes="", index=False, border=0
)
sensitivity_results.append(entry)
html = template.render(
effect_summary=effect_summary,
effect_summary_table_html=effect_summary_table_html,
plots=plots,
sensitivity_results=sensitivity_results,
)
context.report = html
if self.output_file is not None:
self.output_file.write_text(html, encoding="utf-8")
logger.info("Report written to %s", self.output_file)
return context
def __repr__(self) -> str:
"""Return a string representation of the step."""
return (
f"GenerateReport(include_plots={self.include_plots}, "
f"include_effect_summary={self.include_effect_summary}, "
f"include_sensitivity={self.include_sensitivity})"
)
