alert_email

frequenz.lib.notebooks.alerts.alert_email

This module provides functionality for generating email alert notifications.

It includes functions for formatting and structuring alert-related emails, such as: - Generating a summary of alerts per microgrid (optionally grouped by component ID). - Creating an HTML table representation of alert details. - Constructing a complete alert email with formatted content. - Sorting alerts by severity (optional) and applying color-coded styling. - Generating structured JSON output for alerts. - Filtering groups with no errors or warnings (optional, enabled by default).

Example Usage:

import pandas as pd
from frequenz.lib.notebooks.alerts.alert_email import AlertEmailConfig, generate_alert_email

def example():
    # Example alert records dataframe
    alert_records = pd.DataFrame(
        [
            {
                "microgrid_id": 1,
                "component_id": 1,
                "state_type": "error",
                "state_value": "UNDERVOLTAGE",
                "start_time": "2025-03-14 15:06:30",
                "end_time": "2025-03-14 17:00:00",
            },
            {
                "microgrid_id": 2,
                "component_id": 1,
                "state_type": "state",
                "state_value": "DISCHARGING",
                "start_time": "2025-03-14 15:06:30",
                "end_time": None,
            },
        ]
    )

    # Configuration for email generation
    alert_email_config = AlertEmailConfig(
        notebook_url="http://alerts.example.com",
        displayed_rows=10,
        sort_by_severity=True,
        group_by_component=False,
        filter_no_alerts=True,
    )

    # Generate the HTML body of the alert email
    html_email = generate_alert_email(alert_records=alert_records, config=alert_email_config)

    # Output the HTML or send it via email
    print(html_email)

Classes

frequenz.lib.notebooks.alerts.alert_email.AlertEmailConfig dataclass

Configuration for generating alert emails.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

@dataclass(kw_only=True)
class AlertEmailConfig:
    """Configuration for generating alert emails."""

    notebook_url: str = field(
        default="",
        metadata={
            "description": "URL to manage alert preferences.",
        },
    )

    displayed_rows: int = field(
        default=20,
        metadata={
            "description": "Number of alert rows to display in the HTML table.",
            "validate": lambda x: x > 0,
        },
    )

    sort_by_severity: bool = field(
        default=False,
        metadata={
            "description": "Whether to sort alerts by severity level in the HTML table.",
        },
    )

    group_by_component: bool = field(
        default=False,
        metadata={
            "description": (
                "Whether to group summary by component_id in addition to "
                "microgrid_id in the HTML table."
            ),
        },
    )

    filter_no_alerts: bool = field(
        default=True,
        metadata={
            "description": (
                "Whether to exclude groups with no errors or warnings "
                "in the alert email."
            ),
        },
    )

frequenz.lib.notebooks.alerts.alert_email.AlertPlotType

Bases: str, Enum

Possible plot types for alert visualisations.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

class AlertPlotType(str, Enum):
    """Possible plot types for alert visualisations."""

    SUMMARY = "summary"
    """Plot of alert counts per microgrid/component."""

    STATE_TRANSITIONS = "state_transitions"
    """Plot of state transitions over time."""

    ALL = "all"
    """Generate all available plots."""

Attributes

ALL class-attribute instance-attribute

ALL = 'all'

Generate all available plots.

STATE_TRANSITIONS class-attribute instance-attribute

STATE_TRANSITIONS = 'state_transitions'

Plot of state transitions over time.

SUMMARY class-attribute instance-attribute

SUMMARY = 'summary'

Plot of alert counts per microgrid/component.

frequenz.lib.notebooks.alerts.alert_email.ExportOptions dataclass

Configuration for exporting and/or displaying plots.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

@dataclass
class ExportOptions:
    """Configuration for exporting and/or displaying plots."""

    format: str | ImageExportFormat | list[str | ImageExportFormat] | None = field(
        default=None,
        metadata={
            "description": (
                "Export format(s) for the plots. Examples: 'png', ['html', 'svg']. "
                "All options: 'png', 'html', 'svg', 'pdf', 'jpeg', 'json'. "
                "If None, plots will not be saved."
            ),
        },
    )

    output_dir: str | Path | None = field(
        default=None,
        metadata={
            "description": (
                "Directory to save the exported plots. "
                "If None, uses the current directory."
            ),
        },
    )

    show: bool = field(
        default=True,
        metadata={"description": "Whether to display the plots interactively."},
    )

frequenz.lib.notebooks.alerts.alert_email.ImageExportFormat

Bases: str, Enum

Export formats for images.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

class ImageExportFormat(str, Enum):
    """Export formats for images."""

    PNG = "png"
    HTML = "html"
    SVG = "svg"
    PDF = "pdf"
    JPEG = "jpeg"
    JSON = "json"

Functions

frequenz.lib.notebooks.alerts.alert_email.compute_time_since

compute_time_since(row: SeriesType, ts_column: str) -> str

Calculate the time elapsed since a given timestamp (start or end time).

PARAMETER	DESCRIPTION
row	DataFrame row containing timestamps. TYPE: SeriesType
ts_column	Column name ("start_time" or "end_time") to compute from. TYPE: str

RETURNS	DESCRIPTION
str	Time elapsed as a formatted string (e.g., "3h 47m", "2d 5h").

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def compute_time_since(row: SeriesType, ts_column: str) -> str:
    """Calculate the time elapsed since a given timestamp (start or end time).

    Args:
        row: DataFrame row containing timestamps.
        ts_column: Column name ("start_time" or "end_time") to compute from.

    Returns:
        Time elapsed as a formatted string (e.g., "3h 47m", "2d 5h").
    """
    timestamp = _parse_and_localize_timestamp(row[ts_column])
    now = pd.Timestamp.utcnow()

    if pd.isna(timestamp):
        return "N/A"

    if ts_column == "start_time":
        end_time = _parse_and_localize_timestamp(row["end_time"])
        reference_time = end_time if pd.notna(end_time) else now
    else:
        reference_time = now

    return _format_timedelta(reference_time - timestamp)

frequenz.lib.notebooks.alerts.alert_email.generate_alert_email

generate_alert_email(
    alert_records: DataFrame,
    config: AlertEmailConfig,
    checks: list[str] | None = None,
) -> str

Generate a full HTML email for alerts.

PARAMETER	DESCRIPTION
alert_records	DataFrame containing alert records. TYPE: DataFrame
config	Configuration object for email generation. TYPE: AlertEmailConfig
checks	A list of conditions checked by the alert system. TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
str	Full HTML email body.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_alert_email(
    alert_records: pd.DataFrame,
    config: AlertEmailConfig,
    checks: list[str] | None = None,
) -> str:
    """Generate a full HTML email for alerts.

    Args:
        alert_records: DataFrame containing alert records.
        config: Configuration object for email generation.
        checks: A list of conditions checked by the alert system.

    Returns:
        Full HTML email body.
    """
    return f"""
    <html>
        <head>{EMAIL_CSS}</head>
        <body>
            <h1>Microgrid Alert</h1>
            <h2>Summary:</h2>
            {generate_alert_summary(alert_records, config.group_by_component,
                                    config.filter_no_alerts)}
            <h2>Alert Details:</h2>
            {generate_alert_table(alert_records, config.displayed_rows, config.sort_by_severity)}
            {generate_check_status(checks)}
            <p style='color: #665;'><em>This is an automated notification.</em></p>
            {_generate_email_footer(config.notebook_url)}
        </body>
    </html>
    """

frequenz.lib.notebooks.alerts.alert_email.generate_alert_json

generate_alert_json(
    alert_records: DataFrame,
    group_by_component: bool = False,
) -> dict[str, Any]

Generate a JSON representation of the alert data.

The data can be optionally grouped by component ID

PARAMETER	DESCRIPTION
alert_records	DataFrame containing alert records. TYPE: DataFrame
group_by_component	Whether to group alerts by component ID. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
dict[str, Any]	Dictionary representing the alert data in JSON format.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_alert_json(
    alert_records: pd.DataFrame, group_by_component: bool = False
) -> dict[str, Any]:
    """Generate a JSON representation of the alert data.

    The data can be optionally grouped by component ID

    Args:
        alert_records: DataFrame containing alert records.
        group_by_component: Whether to group alerts by component ID.

    Returns:
        Dictionary representing the alert data in JSON format.
    """
    if alert_records.empty:
        return {"summary": "<p>No alerts recorded.</p>"}

    group_columns = ["microgrid_id"]
    if group_by_component:
        group_columns.append("component_id")

    return {
        "summary": {
            idx: group.to_dict(orient="records")
            for idx, group in alert_records.groupby(group_columns)
        }
    }

frequenz.lib.notebooks.alerts.alert_email.generate_alert_summary

generate_alert_summary(
    alert_records: DataFrame,
    group_by_component: bool = False,
    filter_no_alerts: bool = True,
) -> str

Generate a summary of alerts per microgrid, optionally grouped by component ID.

PARAMETER	DESCRIPTION
alert_records	DataFrame containing alert records. TYPE: DataFrame
group_by_component	Whether to group alerts by component ID. TYPE: bool DEFAULT: False
filter_no_alerts	Whether to exclude groups with zero errors and warnings. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	HTML summary string.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_alert_summary(
    alert_records: pd.DataFrame,
    group_by_component: bool = False,
    filter_no_alerts: bool = True,
) -> str:
    """Generate a summary of alerts per microgrid, optionally grouped by component ID.

    Args:
        alert_records: DataFrame containing alert records.
        group_by_component: Whether to group alerts by component ID.
        filter_no_alerts: Whether to exclude groups with zero errors and warnings.

    Returns:
        HTML summary string.
    """
    if alert_records.empty:
        return "<p>No alerts recorded.</p>"

    group_columns = ["microgrid_id"]
    if group_by_component:
        group_columns.append("component_id")

    summary_data = (
        alert_records.groupby(group_columns)
        .agg(
            total_errors=(
                "state_type",
                lambda x: (x.fillna("").str.lower() == "error").sum(),
            ),
            total_warnings=(
                "state_type",
                lambda x: (x.fillna("").str.lower() == "warning").sum(),
            ),
            unique_states=(
                "state_value",
                lambda x: [html.escape(str(s)) for s in x.unique()],
            ),
            unique_components=("component_id", lambda x: list(x.unique())),
        )
        .reset_index()
    )

    if filter_no_alerts:
        summary_data = summary_data[
            (summary_data["total_errors"] > 0) | (summary_data["total_warnings"] > 0)
        ]

    summary_html = "".join(
        [
            f"""
            <p><strong>Microgrid {row['microgrid_id']}{
                ", Component " + str(row['component_id']) if group_by_component else ""
                }:</strong></p>
            <ul>
                <li><strong>Total errors:</strong> {row['total_errors']}</li>
                <li><strong>Total warnings:</strong> {row['total_warnings']}</li>
                <li><strong>States:</strong>
                    <ul>
                        <li>Unique states found: {len(row['unique_states'])}</li>
                        <li>Unique States: {row['unique_states']}</li>
                    </ul>
                </li>
            </ul>
            """
            + (
                f"""
                <ul>
                    <li><strong>Components:</strong>
                        <ul>
                            <li>Alerts found for {len(row['unique_components'])} components</li>
                            <li>Components: {row['unique_components']}</li>
                        </ul>
                    </li>
                </ul>
                """
                if not group_by_component
                else ""
            )
            + "</p>"
            for _, row in summary_data.iterrows()
        ]
    )

    return summary_html

frequenz.lib.notebooks.alerts.alert_email.generate_alert_table

generate_alert_table(
    alert_records: DataFrame,
    displayed_rows: int = 20,
    sort_by_severity: bool = False,
) -> str

Generate a formatted HTML table for alert details with color-coded severity levels.

PARAMETER	DESCRIPTION
alert_records	DataFrame containing alert records. TYPE: DataFrame
displayed_rows	Number of rows to display. TYPE: int DEFAULT: 20
sort_by_severity	Whether to sort alerts by severity. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str	HTML string of the table with color-coded rows.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_alert_table(
    alert_records: pd.DataFrame,
    displayed_rows: int = 20,
    sort_by_severity: bool = False,
) -> str:
    """Generate a formatted HTML table for alert details with color-coded severity levels.

    Args:
        alert_records: DataFrame containing alert records.
        displayed_rows: Number of rows to display.
        sort_by_severity: Whether to sort alerts by severity.

    Returns:
        HTML string of the table with color-coded rows.
    """
    if alert_records.empty:
        return "<p>No alerts recorded.</p>"

    if sort_by_severity:
        alert_records = alert_records.copy()
        alert_records["state_type"] = alert_records["state_type"].str.lower()
        alert_records["state_type"] = pd.Categorical(
            alert_records["state_type"], categories=SEVERITY_ORDER, ordered=True
        )
        alert_records = alert_records.sort_values("state_type")

    if len(alert_records) > displayed_rows:
        note = f"""
        <p><strong>Note:</strong> Table limited to {displayed_rows} rows.
        Download the attached file to view all {len(alert_records)} rows.</p>
        """
    else:
        note = ""

    severity_colors = {
        "error": "background-color: #D32F2F; color: white;",
        "warning": "background-color: #F57C00; color: black;",
    }

    # general table styling
    # We use cast(Any, ...) here to bypass mypy's strict type checking on .set_table_styles().
    # Pandas internally expects CSSDict, which we want to avoid importing to prevent
    # unnecessary dependencies (like with Jinja2 library).
    # Since the structure is guaranteed to be correct for Pandas at runtime, using Any
    # ensures type flexibility while avoiding compatibility issues with future Pandas versions.
    table_styles = cast(
        Any,
        [
            {
                "selector": "th",
                "props": [("background-color", "#f4f4f4"), ("font-weight", "bold")],
            },
            {
                "selector": "td, th",
                "props": [("border", "1px solid #ddd"), ("padding", "8px")],
            },
        ],
    )

    # apply severity color to entire rows
    styled_table = (
        alert_records.head(displayed_rows)
        .style.apply(
            lambda row: [severity_colors.get(row["state_type"], "")] * len(row), axis=1
        )
        .set_table_styles(table_styles, overwrite=False)
        .hide(axis="index")
        .to_html()
    )
    return f"{note}{styled_table}"

frequenz.lib.notebooks.alerts.alert_email.generate_check_status

generate_check_status(checks: list[str] | None) -> str

Generate a clean HTML bullet list summarising what was checked.

PARAMETER	DESCRIPTION
checks	A list of plain text items (e.g. conditions, rules). TYPE: list[str] | None

RETURNS	DESCRIPTION
str	An HTML unordered list.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_check_status(checks: list[str] | None) -> str:
    """Generate a clean HTML bullet list summarising what was checked.

    Args:
        checks: A list of plain text items (e.g. conditions, rules).

    Returns:
        An HTML unordered list.
    """
    if not checks:
        return ""
    items = "".join(f"<li>{html.escape(check)}</li>" for check in checks)
    return (
        "<h3>✅ Conditions Checked:</h3>"
        "<ul style='line-height: 1.6;'>"
        f"{items}"
        "</ul>"
    )

frequenz.lib.notebooks.alerts.alert_email.generate_no_alerts_email

generate_no_alerts_email(
    checks: list[str] | None = None, notebook_url: str = ""
) -> str

Generate an HTML email when no alerts are found.

PARAMETER	DESCRIPTION
checks	A list of conditions checked. TYPE: list[str] | None DEFAULT: None
notebook_url	Optional link to manage preferences. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
str	HTML email body.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def generate_no_alerts_email(
    checks: list[str] | None = None,
    notebook_url: str = "",
) -> str:
    """Generate an HTML email when no alerts are found.

    Args:
        checks: A list of conditions checked.
        notebook_url: Optional link to manage preferences.

    Returns:
        HTML email body.
    """
    return f"""
    <html>
    <body style='font-family: Arial, sans-serif; line-height: 1.6; color: #333;'>
        <h2 style='color: #28a745;'>✅ No Alerts Detected</h2>
        <p>
        This is a confirmation that the alerting system ran successfully and no
        alerts were found at this time.
        </p>
        {generate_check_status(checks)}
        <p style='color: #665;'><em>This is an automated notification.</em></p>
        {_generate_email_footer(notebook_url)}
    </body>
    </html>
    """

frequenz.lib.notebooks.alerts.alert_email.plot_alerts

plot_alerts(
    records: DataFrame,
    *,
    plot_type: str | AlertPlotType = SUMMARY,
    export_options: ExportOptions | None = None,
    **kwargs: Any
) -> list[str] | None

Generate alert visualisations and optionally save them as image files.

Behaviour based on export_options format and show fields: - format=None, show=False: Do nothing. - format=None, show=True: Display plots only (default). - format=[...], show=False: Save plots to multiple formats only. - format=[...], show=True: Save plots to multiple formats and display them.

PARAMETER	DESCRIPTION
records	DataFrame containing alert records with expected columns "microgrid_id", "component_id", "state_value", and "start_time". TYPE: DataFrame
plot_type	Which plot to create. Options: - 'summary': Plot of alert counts per microgrid/component. - 'state_transitions': Plot of state transitions over time. - 'all': Generate both types. TYPE: str | AlertPlotType DEFAULT: SUMMARY
export_options	Configuration for exporting and/or displaying the plots. TYPE: ExportOptions | None DEFAULT: None
**kwargs	Additional arguments for the plot functions. - 'stacked': Whether to use a stacked bar representation per microgrid (only for 'summary' plot type). TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[str] | None	List of file paths if plots are exported, otherwise None.

Source code in frequenz/lib/notebooks/alerts/alert_email.py

def plot_alerts(
    records: pd.DataFrame,
    *,
    plot_type: str | AlertPlotType = AlertPlotType.SUMMARY,
    export_options: ExportOptions | None = None,
    **kwargs: Any,
) -> list[str] | None:
    """Generate alert visualisations and optionally save them as image files.

    Behaviour based on `export_options` `format` and `show` fields:
        - format=None, show=False: Do nothing.
        - format=None, show=True: Display plots only (default).
        - format=[...], show=False: Save plots to multiple formats only.
        - format=[...], show=True: Save plots to multiple formats and display them.

    Args:
        records: DataFrame containing alert records with expected columns
            "microgrid_id", "component_id", "state_value", and "start_time".
        plot_type: Which plot to create. Options:
            - 'summary': Plot of alert counts per microgrid/component.
            - 'state_transitions': Plot of state transitions over time.
            - 'all': Generate both types.
        export_options: Configuration for exporting and/or displaying the plots.
        **kwargs: Additional arguments for the plot functions.
            - 'stacked': Whether to use a stacked bar representation per microgrid
                (only for 'summary' plot type).

    Returns:
        List of file paths if plots are exported, otherwise None.
    """
    if records.empty:
        _log.info("Records are empty, no plots generated.")
        return None

    if (
        export_options is not None
        and export_options.format is None
        and not export_options.show
    ):
        _log.info("No export format and show=False: no plots generated.")
        return None

    plot_type = _coerce_enum(plot_type, AlertPlotType, "plot_type")
    figs = {}
    if plot_type in (AlertPlotType.SUMMARY, AlertPlotType.ALL):
        figs["alert_summary.html"] = _plot_alert_summary(records, **kwargs)
    if plot_type in (AlertPlotType.STATE_TRANSITIONS, AlertPlotType.ALL):
        figs["state_transitions.html"] = _plot_state_transitions(records)

    filepaths = None
    if export_options is not None:
        if export_options.format is not None:
            export_formats = _coerce_formats(export_options.format)
            filepaths = []
            for fmt in export_formats:
                filepaths += _save_figures(figs, fmt, export_options.output_dir)

        if export_options.show:
            for fig in figs.values():
                fig.show()
    return filepaths