Feature/add zone level component assignment

epinterface/analysis/zone_assignment_viz.py

@@ -0,0 +1,161 @@
+"""matplotlib plots comparing resolved zone-assignment scalars and sim energy."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+import pandas as pd
+
+if TYPE_CHECKING:
+    from epinterface.sbem.builder import ModelRunResults
+
+DEFAULT_ASSIGNMENT_METRICS: tuple[str, ...] = (
+    "LightingPowerDensity",
+    "EquipmentPowerDensity",
+    "FacadeRValue",
+)
+
+DEFAULT_SIM_ENERGY_METRICS: tuple[str, ...] = (
+    "sim_lighting_kwh_per_m2",
+    "sim_equipment_kwh_per_m2",
+    "sim_heating_kwh_per_m2",
+    "sim_cooling_kwh_per_m2",
+    "sim_total_kwh_per_m2",
+)
+
+
+def _import_plt():
+    try:
+        import matplotlib.pyplot as plt
+        from matplotlib.axes import Axes
+        from matplotlib.figure import Figure
+    except ImportError as exc:
+        msg = "matplotlib is required for plotting (install dev deps: uv sync --group dev)"
+        raise ImportError(msg) from exc
+    return plt, Axes, Figure
+
+
+def _plot_label_column(
+    df: pd.DataFrame, groupby: Literal["zone", "storey_role"]
+) -> pd.Series:
+    if groupby == "zone":
+        return df["ep_zone_name"].astype(str)
+    return "S" + df["storey_index"].astype(str) + " / " + df["role"].astype(str)
+
+
+def _horizontal_metric_bars(
+    df: pd.DataFrame,
+    metrics: tuple[str, ...],
+    *,
+    groupby: Literal["zone", "storey_role"],
+    title_prefix: str = "",
+    ax=None,
+):
+    plt, Axes, Figure = _import_plt()
+    plot_df = df.copy()
+    plot_df["_plot_label"] = _plot_label_column(plot_df, groupby)
+
+    present = [m for m in metrics if m in plot_df.columns]
+    if not present:
+        msg = f"none of the metrics {metrics!r} appear in columns {list(plot_df.columns)!r}"
+        raise ValueError(msg)
+
+    roles = (
+        sorted(plot_df["role"].astype(str).unique())
+        if "role" in plot_df.columns
+        else ["all"]
+    )
+    cmap = plt.colormaps["tab10"]
+    role_color = {r: cmap(i % 10) for i, r in enumerate(roles)}
+
+    if ax is not None:
+        if len(present) > 1:
+            msg = "pass a single metric when ax is provided"
+            raise ValueError(msg)
+        metric = present[0]
+        if not isinstance(ax, Axes):
+            msg = "ax must be a matplotlib Axes when provided"
+            raise TypeError(msg)
+        sub = plot_df.sort_values(["storey_index", "role", "_plot_label"])
+        colors = [role_color.get(str(r), cmap(0)) for r in sub.get("role", sub.index)]
+        ax.barh(sub["_plot_label"], sub[metric], color=colors)
+        ax.set_xlabel(metric)
+        ax.set_title(f"{title_prefix}{metric}".strip())
+        return ax.figure
+
+    fig_h = max(3.0, 2.0 * len(present))
+    fig, axes = plt.subplots(len(present), 1, figsize=(9, fig_h), squeeze=False)
+    for axi, metric in zip(axes.flatten(), present, strict=True):
+        sub = plot_df.sort_values(["storey_index", "role", "_plot_label"])
+        colors = [role_color.get(str(r), cmap(0)) for r in sub.get("role", sub.index)]
+        axi.barh(sub["_plot_label"], sub[metric], color=colors)
+        axi.set_xlabel(metric)
+        axi.set_title(f"{title_prefix}{metric}".strip())
+    fig.tight_layout()
+    if not isinstance(fig, Figure):
+        msg = "expected matplotlib Figure"
+        raise TypeError(msg)
+    return fig
+
+
+def plot_zone_comparison(
+    summary: pd.DataFrame,
+    metrics: tuple[str, ...] = DEFAULT_ASSIGNMENT_METRICS,
+    *,
+    groupby: Literal["zone", "storey_role"] = "zone",
+    ax=None,
+):
+    """Horizontal bars for resolved assignment scalars (inputs)."""
+    return _horizontal_metric_bars(summary, metrics, groupby=groupby, ax=ax)
+
+
+def plot_zone_energy_comparison(
+    summary: pd.DataFrame,
+    metrics: tuple[str, ...] = DEFAULT_SIM_ENERGY_METRICS,
+    *,
+    groupby: Literal["zone", "storey_role"] = "zone",
+    ax=None,
+    results: ModelRunResults | None = None,
+):
+    """Horizontal bars of simulated annual site energy (kWh/m2) per zone."""
+    if results is not None:
+        from epinterface.analysis.zone_energy import (
+            merge_assignment_and_energy,
+            zone_energy_summary,
+        )
+
+        energy = zone_energy_summary(results.sql, results.idf)
+        summary = merge_assignment_and_energy(summary, energy)
+    present = tuple(m for m in metrics if m in summary.columns)
+    return _horizontal_metric_bars(
+        summary,
+        present or DEFAULT_SIM_ENERGY_METRICS,
+        groupby=groupby,
+        title_prefix="sim: ",
+        ax=ax,
+    )
+
+
+def plot_assignment_and_energy(
+    summary: pd.DataFrame,
+    *,
+    results: ModelRunResults,
+    groupby: Literal["zone", "storey_role"] = "zone",
+):
+    """Return (assignment_figure, energy_figure, merged_summary)."""
+    from epinterface.analysis.zone_energy import (
+        merge_assignment_and_energy,
+        zone_energy_summary,
+    )
+
+    energy = zone_energy_summary(results.sql, results.idf)
+    merged = merge_assignment_and_energy(summary, energy)
+    assign_metrics = tuple(m for m in DEFAULT_ASSIGNMENT_METRICS if m in merged.columns)
+    sim_metrics = tuple(m for m in DEFAULT_SIM_ENERGY_METRICS if m in merged.columns)
+    assign_fig = _horizontal_metric_bars(
+        merged, assign_metrics, groupby=groupby, title_prefix="assigned: "
+    )
+    energy_fig = _horizontal_metric_bars(
+        merged, sim_metrics, groupby=groupby, title_prefix="sim: "
+    )
+    return assign_fig, energy_fig, merged

epinterface/analysis/zone_energy.py

@@ -0,0 +1,154 @@
+"""Per-zone annual energy from EnergyPlus SQL (post-simulation)."""
+
+from __future__ import annotations
+
+import pandas as pd
+from archetypal.idfclass import IDF
+from archetypal.idfclass.sql import Sql
+
+from epinterface.geometry import get_zone_floor_area
+
+J_to_kWh = 1.0 / 3_600_000.0
+GJ_to_kWh = 277.778
+
+_ZONE_ENERGY_VARS: tuple[tuple[str, str], ...] = (
+    ("Zone Lights Electricity Energy", "sim_lighting_kwh_per_m2"),
+    ("Zone Electric Equipment Electricity Energy", "sim_equipment_kwh_per_m2"),
+    ("Zone Ideal Loads Zone Total Heating Energy", "sim_heating_kwh_per_m2"),
+    ("Zone Ideal Loads Zone Total Cooling Energy", "sim_cooling_kwh_per_m2"),
+)
+
+
+def _norm_zone_key(name: str) -> str:
+    return name.replace("_", " ").upper()
+
+
+def _zone_name_lookup(zone_names: list[str]) -> dict[str, str]:
+    return {_norm_zone_key(zn): zn for zn in zone_names}
+
+
+def _match_zone_name(key_value: str, zone_lookup: dict[str, str]) -> str | None:
+    """Map SQL KeyValue (zone or ideal-loads system name) to EP zone name."""
+    kv = key_value.strip()
+    if kv in zone_lookup.values():
+        return kv
+    nkv = _norm_zone_key(kv)
+    if nkv in zone_lookup:
+        return zone_lookup[nkv]
+    suffix = " IDEAL LOADS AIR SYSTEM"
+    if nkv.endswith(suffix):
+        candidate = nkv[: -len(suffix)]
+        if candidate in zone_lookup:
+            return zone_lookup[candidate]
+    return None
+
+
+def _annual_kwh_per_m2_from_hourly(
+    sql: Sql, variable_name: str, idf: IDF
+) -> dict[str, float]:
+    """Sum hourly zone energy [J] and normalize by zone floor area."""
+    zone_names = [z.Name for z in idf.idfobjects["ZONE"]]
+    zone_lookup = _zone_name_lookup(zone_names)
+    try:
+        hourly = sql.timeseries_by_name([variable_name], "Hourly")
+    except Exception:
+        return {}
+    if hourly.empty:
+        return {}
+
+    totals: dict[str, float] = {}
+    if isinstance(hourly.columns, pd.MultiIndex):
+        for col in hourly.columns:
+            key_value = str(col[1]) if len(col) > 1 else str(col[0])
+            zn = _match_zone_name(key_value, zone_lookup)
+            if zn is None:
+                continue
+            totals[zn] = totals.get(zn, 0.0) + float(hourly[col].sum())
+    else:
+        annual_j = hourly.sum()
+        for key, joules in annual_j.items():
+            zn = _match_zone_name(str(key), zone_lookup)
+            if zn is None:
+                continue
+            totals[zn] = totals.get(zn, 0.0) + float(joules)
+
+    return {
+        zn: joules * J_to_kWh / float(get_zone_floor_area(idf, zn))
+        for zn, joules in totals.items()
+        if float(get_zone_floor_area(idf, zn)) > 0
+    }
+
+
+def _lighting_kwh_per_m2_from_tabular(sql: Sql, idf: IDF) -> dict[str, float]:
+    """Lighting consumption from LightingSummary when hourly vars are absent."""
+    zone_lookup = _zone_name_lookup([z.Name for z in idf.idfobjects["ZONE"]])
+    try:
+        tbl = sql.tabular_data_by_name(
+            "LightingSummary", "Interior Lighting", "Entire Facility"
+        )
+    except Exception:
+        return {}
+    zone_col = ("Zone Name", "")
+    cons_col = ("Consumption", "GJ")
+    area_col = ("Space Area", "m2")
+    if cons_col not in tbl.columns or zone_col not in tbl.columns:
+        return {}
+    out: dict[str, float] = {}
+    for _, row in tbl.iterrows():
+        zn = row.loc[zone_col]
+        if not isinstance(zn, str) or not zn.strip():
+            continue
+        cons_val = row.loc[cons_col]
+        if pd.isna(cons_val):
+            continue
+        gj = float(cons_val)
+        if area_col in tbl.columns:
+            area_val = row.loc[area_col]
+            area = 0.0 if pd.isna(area_val) else float(area_val)
+        else:
+            area = 0.0
+        if area <= 0:
+            continue
+        zn_idf = zone_lookup.get(_norm_zone_key(zn.strip()))
+        if zn_idf is None:
+            continue
+        out[zn_idf] = gj * GJ_to_kWh / area
+    return out
+
+
+def zone_energy_summary(sql: Sql, idf: IDF) -> pd.DataFrame:
+    """Annual simulated site energy per zone, normalized to kWh/m2."""
+    zone_names = [z.Name for z in idf.idfobjects["ZONE"]]
+    rows: list[dict[str, float | str | None]] = []
+    col_data: dict[str, dict[str, float]] = {}
+
+    lighting = _annual_kwh_per_m2_from_hourly(
+        sql, "Zone Lights Electricity Energy", idf
+    )
+    if not lighting:
+        lighting = _lighting_kwh_per_m2_from_tabular(sql, idf)
+    col_data["sim_lighting_kwh_per_m2"] = lighting
+
+    for var_name, col_name in _ZONE_ENERGY_VARS[1:]:
+        col_data[col_name] = _annual_kwh_per_m2_from_hourly(sql, var_name, idf)
+
+    for zn in zone_names:
+        row: dict[str, float | str | None] = {"ep_zone_name": zn}
+        for col_name in [c for _, c in _ZONE_ENERGY_VARS]:
+            row[col_name] = col_data.get(col_name, {}).get(zn)
+        sim_total = sum(
+            float(v)
+            for k, v in row.items()
+            if k.startswith("sim_") and isinstance(v, int | float)
+        )
+        row["sim_total_kwh_per_m2"] = sim_total if sim_total else None
+        rows.append(row)
+    return pd.DataFrame(rows)
+
+
+def merge_assignment_and_energy(
+    assignment: pd.DataFrame,
+    energy: pd.DataFrame,
+) -> pd.DataFrame:
+    """Join resolved assignment summary with simulated zone energy."""
+    return assignment.merge(energy, on="ep_zone_name", how="left")