from __future__ import annotations
import json
import os
import warnings
from .Simulator import simulate_3day as _simulate_3day, simulate_annual as _simulate_annual
from ecoengine.objects.building.ClimateZone import ClimateZone as _ClimateZone
_MAPS_PATH = os.path.join(os.path.dirname(__file__), "../data/preformanceMaps/maps.json")
_WS_LOOKUP = os.path.join(os.path.dirname(__file__), "../data/climate_data/WeatherStation_ClimateZone_Lookup.csv")
[docs]
def get_oat_buckets(
zip_code: str | int | None = None,
zone_id: int | None = None,
weather_station: str | None = None,
) -> dict[float, int]:
"""
Return the distribution of daily average outdoor air temperatures across
5°F buckets for a given location's typical meteorological year.
Exactly one of ``zip_code``, ``zone_id``, or ``weather_station`` must be
provided to identify the climate zone.
Parameters
----------
zip_code : str | int | None
A California 5-digit zip code.
zone_id : int | None
A numeric climate zone ID (1-96).
weather_station : str | None
A weather station name as it appears in the lookup table.
Returns
-------
dict[float, int]
Mapping of bucket temperature [°F] → number of days per year with
a daily average OAT in that 5°F bucket. Only populated buckets
are included.
Examples
--------
>>> get_oat_buckets(zip_code=90210)
{50.0: 46, 55.0: 83, ...}
>>> get_oat_buckets(zone_id=19)
{65.0: 25, ...}
"""
provided = sum(x is not None for x in (zip_code, zone_id, weather_station))
if provided != 1:
raise ValueError(
"Provide exactly one of zip_code, zone_id, or weather_station."
)
if zip_code is not None:
cz = _ClimateZone.from_zip_code(zip_code)
elif zone_id is not None:
cz = _ClimateZone.from_zone_id(zone_id)
else:
cz = _ClimateZone.from_weather_station(weather_station)
return cz.get_oat_buckets()
[docs]
def get_list_of_models(
multi_pass: bool = False,
include_residential: bool = True,
exclude_models: list[str] | None = None,
sgip_models_only: bool = True,
) -> list[list[str]]:
"""
Return available HPWH model codes and display names.
Parameters
----------
multi_pass : bool
``True`` → return only multi-pass (``_MP``) models.
``False`` → return only single-pass (``_SP``) models.
include_residential : bool
``True`` → include residential (``_R_``) models.
``False`` → commercial (``_C_``) models only.
exclude_models : list[str] | None
Model codes to omit from the result. Defaults to no exclusions.
sgip_models_only : bool
``True`` (default) → restrict to models flagged ``SGIP_avail`` in
the performance-map database.
Returns
-------
list[list[str]]
Each element is ``[model_code, display_name]`` where ``model_code``
is the string accepted by ``WaterHeater.from_model_name()`` and
``display_name`` is the human-readable label.
"""
exclude = set(exclude_models or [])
result: list[list[str]] = []
with open(_MAPS_PATH) as f:
data: dict = json.load(f)
for model_code, meta in data.items():
if model_code in exclude:
continue
if sgip_models_only and not meta.get("SGIP_avail", False):
continue
is_commercial = model_code[-4] == "C" # e.g. MODELS_Foo_C_MP → [-4]='C'
if not include_residential and not is_commercial:
continue
is_mp = model_code.endswith("_MP")
if multi_pass and not is_mp:
continue
if not multi_pass and is_mp:
continue
result.append([model_code, meta["name"]])
return result
[docs]
def get_weather_stations(
exclude_zones: list[int] | None = None,
) -> list[list]:
"""
Return all available weather stations and their corresponding climate zone IDs.
Parameters
----------
exclude_zones : list[int], optional
Climate zone IDs to omit from the result. Defaults to ``[96]``.
Returns
-------
list[list]
Each element is ``[weather_station_name, climate_zone_id]`` where
``weather_station_name`` is the string accepted by
``ClimateZone.from_weather_station()`` and ``climate_zone_id`` is the
corresponding integer zone number. Weather data sourced from
https://energyplus.net/weather.
"""
import csv
if exclude_zones is None:
exclude_zones = [96]
exclude = set(exclude_zones)
data = []
with open(_WS_LOOKUP, "r", newline="") as csvfile:
reader = csv.reader(csvfile)
next(reader) # skip header
for row in reader:
if len(row) == 2:
station, zone_id = row
zone_id = int(zone_id)
if zone_id not in exclude:
data.append([station, zone_id])
return data
[docs]
def get_hpwh_output_capacity(
model: str,
oat_f: float,
inlet_water_temp_f: float,
outlet_water_temp_f: float,
num_heaters: int = 1,
return_as_kw: bool = True,
defrost_derate: float = 0.0,
) -> float:
"""
Return the output capacity of an HPWH model at the given operating conditions.
Parameters
----------
model : str
Model code string (see ``get_list_of_models()``).
oat_f : float
Outdoor air temperature [°F].
inlet_water_temp_f : float
Incoming cold-water temperature [°F].
outlet_water_temp_f : float
Hot storage outlet temperature [°F].
num_heaters : int
Number of HPWH units. Output is multiplied by this. Default 1.
return_as_kw : bool
If True (default), return output in kW. If False, return in kBTU/hr.
defrost_derate : float
Fractional capacity reduction due to defrost [0.0–1.0]. Default 0.0.
Returns
-------
float
Output capacity after defrost derate in the requested unit.
Raises
------
ValueError
If ``defrost_derate`` is outside [0, 1] or ``num_heaters`` < 1.
"""
if not 0.0 <= defrost_derate <= 1.0:
raise ValueError("defrost_derate must be between 0.0 and 1.0")
if not isinstance(num_heaters, int) or num_heaters < 1:
raise ValueError("num_heaters must be an integer >= 1")
from ecoengine.objects.components.heating.PerformanceMap import PerformanceMap
perf_map = PerformanceMap.from_model_name(model)
capacity_kbtuh = perf_map.get_capacity_kbtuh(oat_f, outlet_water_temp_f, inlet_water_temp_f)
capacity_kbtuh *= num_heaters * (1.0 - defrost_derate)
if return_as_kw:
return capacity_kbtuh / 3.41214 # kBTU/hr → kW
return capacity_kbtuh
_MONTH_NAMES = ['Jan','Feb','Mar','Apr','May','Jun','Jul','Aug','Sep','Oct','Nov','Dec']
_ANNUAL_DURATION_MIN = 365 * 24 * 60
[docs]
def get_annual_utility_comparison_graph(
sim_run_hp,
sim_run_iwh,
utility_cost_tracker,
return_as_div: bool = False,
return_as_array: bool = False,
):
"""
Build a stacked bar chart comparing monthly utility costs for an HPWH
system vs. an instantaneous electric resistance (UER) baseline.
Each month shows two adjacent bars — one for the HP system, one for UER —
stacked by charge type: base, energy, and demand.
Parameters
----------
sim_run_hp : SimulationRun
Annual simulation result for the heat pump system.
sim_run_iwh : SimulationRun
Annual simulation result for the instantaneous/UER baseline.
utility_cost_tracker : UtilityCostTracker
Rate structure applied to both runs.
return_as_div : bool
If True, return an HTML ``<div>`` string instead of a Figure.
return_as_array : bool
If True, return ``(hp_breakdown, iwh_breakdown)`` dicts instead of a
figure. Each dict has keys ``'energy'`` (12 floats), ``'demand'``
(12 floats), and ``'base'`` (float).
Returns
-------
plotly.graph_objects.Figure, str, or tuple
"""
if sim_run_hp.duration_min != _ANNUAL_DURATION_MIN or sim_run_iwh.duration_min != _ANNUAL_DURATION_MIN:
raise ValueError(
"Both simulation runs must be annual (365-day) to generate a "
"utility comparison graph."
)
hp_bd = sim_run_hp.get_monthly_cost_breakdown_detailed(utility_cost_tracker)
iwh_bd = sim_run_iwh.get_monthly_cost_breakdown_detailed(utility_cost_tracker)
if return_as_array:
return hp_bd, iwh_bd
try:
import plotly.graph_objects as go
except ImportError:
raise ImportError(
"plotly is required for get_annual_utility_comparison_graph(). "
"Install with: pip install plotly"
)
# Interleave HP and IWH bars: for each of 24 x-positions, even index = HP,
# odd index = IWH. Trailing-space trick makes Plotly treat them as
# separate categories while keeping the visual pairing per month.
x_labels = []
for name in _MONTH_NAMES:
x_labels.extend([name, f"{name} "])
base = utility_cost_tracker.monthly_base_charge
def _interleave(values, slot):
"""slot=0 → HP (even positions), slot=1 → IWH (odd positions)."""
out = []
for v in values:
if slot == 0:
out.extend([v, 0.0])
else:
out.extend([0.0, v])
return out
categories = [
("Base Charges", _interleave([base] * 12, 0), _interleave([base] * 12, 1)),
("Peak Energy Charges", _interleave(hp_bd["energy_peak"], 0), _interleave(iwh_bd["energy_peak"], 1)),
("Off-Peak Energy Charges", _interleave(hp_bd["energy_off_peak"], 0), _interleave(iwh_bd["energy_off_peak"], 1)),
("Peak Demand Charges", _interleave(hp_bd["demand_peak"], 0), _interleave(iwh_bd["demand_peak"], 1)),
("Off-Peak Demand Charges", _interleave(hp_bd["demand_off_peak"], 0), _interleave(iwh_bd["demand_off_peak"], 1)),
]
fig = go.Figure()
for label, hp_vals, iwh_vals in categories:
fig.add_trace(go.Bar(
x=x_labels, y=hp_vals,
name=f"{label} for HP",
hovertemplate=f"{label} (HP)<br>%{{x}}<br>$%{{y:.2f}}<extra></extra>",
))
fig.add_trace(go.Bar(
x=x_labels, y=iwh_vals,
name=f"{label} for UER",
hovertemplate=f"{label} (UER)<br>%{{x}}<br>$%{{y:.2f}}<extra></extra>",
))
fig.update_layout(
barmode="stack",
title="Utility Cost Comparison: Heat Pump (HP) vs. Unitary Electric Resistance (UER)",
yaxis_title="Cost ($)",
xaxis=dict(
tickvals=_MONTH_NAMES,
ticktext=_MONTH_NAMES,
title="Month",
),
)
if return_as_div:
return fig.to_html(full_html=False, include_plotlyjs=False)
return fig
# ---------------------------------------------------------------------------
# Standalone sizing-curve plot helper
# ---------------------------------------------------------------------------
[docs]
def get_sizing_curve_plot(
x: list,
y: list,
start_index: int,
load_shifting: bool = False,
er_sized: bool = False,
return_as_div: bool = False,
):
"""
Build a Plotly sizing-curve figure from pre-computed x/y points.
This is a lightweight alternative to ``EcosizerEngine.plot_sizing_curve()``
for callers that already have the curve data (e.g. from
``EcosizerEngine.plot_sizing_curve(return_with_x_y_points=True)``).
Parameters
----------
x : list
X-axis values. Interpretation depends on the mode:
* Normal sizing — storage volume [gal at storage temperature]
* Load-shifting — load-shift coverage percentile [%]
* ER sizing — percent of building covered [%]
y : list
Y-axis values. Interpretation depends on the mode:
* Normal sizing — heating capacity [kBTU/hr]
* Load-shifting — storage volume [gal at storage temperature]
* ER sizing — ER element capacity [kW]
start_index : int
Index into ``x``/``y`` where the recommended-size diamond starts.
load_shifting : bool
If True, use load-shift axis labels. Default False.
er_sized : bool
If True, use ER-sizing axis labels (overrides ``load_shifting``).
Default False.
return_as_div : bool
If True, return an HTML ``<div>`` string instead of a Figure.
Default False.
Returns
-------
plotly.graph_objects.Figure or str
Raises
------
ImportError
If ``plotly`` is not installed.
"""
try:
import plotly.graph_objects as go
except ImportError:
raise ImportError(
"plotly is required for get_sizing_curve_plot(). "
"Install it with: pip install plotly"
)
if er_sized:
x_label = "Percent Coverage (%)"
y_label = "ER Heating Capacity (kW)"
title = "Electric Resistance Sizing Curve"
hover = "Coverage: <b>%{x:.1f}%</b><br>ER Capacity: <b>%{y:.1f} kW</b><extra></extra>"
def _label(i): return f"Coverage: <b>{x[i]:.1f}%</b>, ER Capacity: <b>{y[i]:.1f} kW</b>"
elif load_shifting:
x_label = "Load-Shift Days Covered (%)"
y_label = "Primary Tank Volume (gal at Storage Temperature)"
title = "Load-Shift Sizing Curve"
hover = "Coverage: <b>%{x:.1f}%</b><br>Storage: <b>%{y:.1f} gal</b><extra></extra>"
def _label(i): return f"Coverage: <b>{x[i]:.1f}%</b>, Storage: <b>{y[i]:.1f} gal</b>"
else:
x_label = "Primary Tank Volume (gal at Storage Temperature)"
y_label = "Heating Capacity (kBTU/hr)"
title = "Primary Sizing Curve"
hover = "Storage: <b>%{x:.1f} gal</b><br>Capacity: <b>%{y:.1f} kBTU/hr</b><extra></extra>"
def _label(i): return f"Storage: <b>{x[i]:.1f} gal</b>, Capacity: <b>{y[i]:.1f} kBTU/hr</b>"
fig = go.Figure()
fig.add_trace(go.Scatter(
x=x, y=y,
mode="lines",
line=dict(color="#28a745", width=3),
hovertemplate=hover,
showlegend=False,
))
fig.add_trace(go.Scatter(
x=[x[start_index]], y=[y[start_index]],
mode="markers",
marker=dict(symbol="diamond", color="#2EA3F2", size=12),
hovertemplate=hover,
showlegend=False,
))
fig.update_layout(
title=title,
xaxis_title=x_label,
yaxis_title=y_label,
showlegend=False,
)
if return_as_div:
return fig.to_html(full_html=False, include_plotlyjs=False)
return fig
# ---------------------------------------------------------------------------
# Schematic → DHWSystem class registry
# ---------------------------------------------------------------------------
_RECIRC_SCHEMATICS = {"parallel_loop", "swing_tank", "single_pass_rtp", "multi_pass_rtp"}
[docs]
class EcosizerEngine:
"""
Primary backend interface for the Ecosizer tool.
Accepts building and system parameters, builds and sizes the appropriate
DHWSystem subclass, and runs simulations. All input-range validation is
expected to be handled by the frontend before calling this class.
Construction immediately builds and sizes the system — no separate build()
or size() calls are required::
engine = EcosizerEngine(
building_type = "multi_family",
magnitude = 100,
zip_code_or_climate_zone = {"design_oat_f": 47, "design_inlet_water_temp_f": 47},
supply_temp_f = 125.0,
storage_temp_f = 150.0,
schematic = "primary_no_recirc",
)
result = engine.simulate_3day()
result.to_plotly(filepath="output.html")
"""
[docs]
def __init__(
self,
building_type: str,
magnitude: int | float,
zip_code_or_climate_zone,
supply_temp_f: float,
storage_temp_f: float,
schematic: str,
# Building inputs
gpdpp: float | None = None,
custom_peak_load_shape: list[float] | None = None,
custom_avg_load_shape: list[float] | None = None,
# Primary heater inputs
num_heaters: int = 1,
hpwh_model: str | None = None,
max_daily_run_hr: float = 16.0,
defrost_factor: float = 1.0,
# Aquastat / controls
aquastat_fract: float = 0.5,
off_sensor_fract: float = 0.1,
on_trigger_t_f: float | None = None,
off_trigger_t_f: float | None = None,
# Load shift (optional)
load_shift_schedule: list[int] | None = None,
load_up_hours: int = 0,
shed_aquastat_fract: float | None = None,
load_up_aquastat_fract: float | None = None,
shed_off_sensor_fract: float | None = None,
load_up_off_sensor_fract: float | None = None,
load_up_off_trigger_t_f: float | None = None,
load_shift_percent: float = 0.95,
# Recirculation (required for recirc schematics)
return_temp_f: float | None = None,
return_flow_gpm: float | None = None,
# ParallelLoop TM controls
tm_on_temp_f: float | None = None,
tm_off_temp_f: float | None = None,
tm_off_time_hr: float = 0.5,
tm_safety_factor: float = 1.75,
# Utility cost (optional)
utility_cost_tracker=None,
# Pre-sized system (optional — skip sizing when capacity and volume are known)
storage_volume_storageT_gal: float | None = None,
heating_capacity_kbtuh: float | None = None,
tm_storage_vol: float | None = None,
tm_capacity_kbtuh: float | None = None,
tm_model: str | None = None,
num_tm_heaters: int = 1,
):
"""
Parameters
----------
building_type : str
Building use type (e.g. ``'multi_family'``, ``'office'``).
magnitude : int or float
Occupancy metric appropriate to the building type (people, units, etc.).
zip_code_or_climate_zone : str | int | dict
Climate lookup key. Accepted forms:
* 5-digit CA zip code string or int → ``ClimateZone.from_zip_code()``
* Integer 1–96 → ``ClimateZone.from_zone_id()``
* Non-numeric string → ``ClimateZone.from_weather_station()``
* Dict with keys ``'design_oat_f'`` and/or ``'design_inlet_water_temp_f'``
→ ``ClimateZone.from_design_conditions()``
supply_temp_f : float
DHW delivery temperature [°F].
storage_temp_f : float
Hot water storage setpoint [°F].
schematic : str
Piping configuration. Supported values:
* ``'primary_no_recirc'`` — base DHWSystem (heat pump + stratified tank)
* ``'parallel_loop'`` — separate TM tank in parallel for recirc losses
* ``'swing_tank'`` — swing-tank recirc system
* ``'single_pass_rtp'`` — single-pass return-to-primary
* ``'multi_pass_rtp'`` — multi-pass return-to-primary
* ``'instant_wh'`` — instantaneous (tankless) water heater, no storage
gpdpp : float, optional
Gallons per person per day. If None, building-type defaults are used.
custom_peak_load_shape : list[float], optional
24-element normalized peak-hour DHW load shape. Required when
``building_type`` is None.
custom_avg_load_shape : list[float], optional
24-element normalized average-hour DHW load shape. Used for
annual simulation load profiles.
num_heaters : int
Number of primary HPWH units. Default 1.
hpwh_model : str, optional
HPWH model name for performance map lookup (future use).
max_daily_run_hr : float
Maximum hours the primary heating system may run per day. Default 16. NOT YET USED FOR MPRTP
defrost_factor : float
Fraction of rated capacity available after defrost cycles (0–1). Default 1.0.
aquastat_fract : float
Fractional tank height of the ON aquastat (0–1). Default 0.5.
off_sensor_fract : float
Fractional tank height of the OFF aquastat (0–1). Default 0.1.
on_trigger_t_f : float, optional
Temperature at which the ON aquastat fires. Defaults to supply_temp_f.
off_trigger_t_f : float, optional
Temperature at which the OFF aquastat fires. Defaults to storage_temp_f.
load_shift_schedule : list[int], optional
24-element list (0 = shed hour, 1 = run hour). None → no load shifting.
load_up_hours : int
Hours spent in load-up mode before the first shed window.
shed_aquastat_fract : float, optional
ON aquastat fraction during shed hours (higher → more storage).
load_up_aquastat_fract : float, optional
ON aquastat fraction during load-up hours (lower → fires sooner).
shed_off_sensor_fract : float, optional
OFF aquastat fraction during shed hours. Defaults to off_sensor_fract.
load_up_off_sensor_fract : float, optional
OFF aquastat fraction during load-up hours. Defaults to off_sensor_fract.
load_up_off_trigger_t_f : float, optional
OFF trigger temperature during load-up hours. Defaults to off_trigger_t_f.
load_shift_percent : float
Percentile of days the load-shift sizing must cover [0.25, 1.0].
Default 0.95 — size for 95% of days, accepting that the highest-demand
5% of days may occasionally breach the shed window. Only applied
during load-shift sizing; normal sizing always uses the full daily demand.
return_temp_f : float, optional
Recirc loop return temperature [°F]. Required for recirc schematics.
return_flow_gpm : float, optional
Recirc loop flow rate [GPM]. Required for recirc schematics.
tm_on_temp_f : float, optional
TM element turn-on temperature [°F]. Defaults to supply_temp_f − 5.
tm_off_temp_f : float, optional
TM element turn-off temperature [°F]. Defaults to supply_temp_f.
tm_off_time_hr : float
Max TM heater off-cycle duration [hr]. Default 0.5.
tm_safety_factor : float
TM capacity safety multiplier (> 1.0). Default 1.2.
utility_cost_tracker : UtilityCostTracker, optional
Attached to the building for annual cost estimates.
storage_volume_storageT_gal : float, optional
Pre-sized primary storage tank volume at storage temperature [gallons].
When provided together with ``heating_capacity_kbtuh`` (or
``hpwh_model``), the sizing step is skipped entirely.
heating_capacity_kbtuh : float, optional
Pre-sized total primary heating capacity [kBTU/hr]. Must be paired
with ``storage_volume_storageT_gal`` (except for ``instant_wh``).
For multi-heater systems this is the combined fleet capacity;
each WaterHeater is assigned ``heating_capacity_kbtuh / num_heaters``.
Ignored when ``hpwh_model`` is also provided.
tm_storage_vol : float, optional
Pre-sized TM tank volume [gallons]. Required for ``parallel_loop``
and ``swing_tank`` when using the pre-sized path.
tm_capacity_kbtuh : float, optional
Pre-sized TM heater capacity [kBTU/hr]. Required for ``parallel_loop``
and ``swing_tank`` when ``tm_model`` is not provided.
tm_model : str, optional
HPWH model name for the TM heater's performance map. When provided,
``tm_capacity_kbtuh`` is used only as the ER fallback capacity.
``return_temp_f`` is used as ``design_inlet_temp_f``.
num_tm_heaters : int
Number of identical TM heater units for ``parallel_loop`` schematics.
The single ``tm_water_heater`` object represents one unit; its output
is multiplied by this value before being applied to the TM tank and
recorded. For the sized path, per-unit capacity =
``_minimum_tm_capacity_kbtuh / num_tm_heaters``; for the pre-sized
path with ``tm_capacity_kbtuh``, per-unit =
``tm_capacity_kbtuh / num_tm_heaters``. Default 1.
"""
self.building_type = building_type
self.magnitude = magnitude
self.zip_code_or_climate_zone = zip_code_or_climate_zone
self.supply_temp_f = supply_temp_f
self.storage_temp_f = storage_temp_f
self.schematic = schematic
self.gpdpp = gpdpp
self.custom_peak_load_shape = custom_peak_load_shape
self.custom_avg_load_shape = custom_avg_load_shape
self.num_heaters = num_heaters
self.hpwh_model = hpwh_model
self.max_daily_run_hr = max_daily_run_hr
self.defrost_factor = defrost_factor
self.aquastat_fract = aquastat_fract
self.off_sensor_fract = off_sensor_fract
self.on_trigger_t_f = on_trigger_t_f
self.off_trigger_t_f = off_trigger_t_f
self.load_shift_schedule = load_shift_schedule
self.load_up_hours = load_up_hours
self.shed_aquastat_fract = shed_aquastat_fract
self.load_up_aquastat_fract = load_up_aquastat_fract
self.shed_off_sensor_fract = shed_off_sensor_fract
self.load_up_off_sensor_fract = load_up_off_sensor_fract
self.load_up_off_trigger_t_f = load_up_off_trigger_t_f
self.load_shift_percent = load_shift_percent
self.return_temp_f = return_temp_f
self.return_flow_gpm = return_flow_gpm
self.tm_on_temp_f = tm_on_temp_f if tm_on_temp_f is not None else supply_temp_f
self.tm_off_temp_f = tm_off_temp_f if tm_off_temp_f is not None else supply_temp_f + 8.0
self.tm_off_time_hr = tm_off_time_hr
self.tm_safety_factor = tm_safety_factor
self.utility_cost_tracker = utility_cost_tracker
self.storage_volume_storageT_gal = storage_volume_storageT_gal
self.heating_capacity_kbtuh = heating_capacity_kbtuh
self.tm_storage_vol = tm_storage_vol
self.tm_capacity_kbtuh = tm_capacity_kbtuh
self.tm_model = tm_model
self.num_tm_heaters = num_tm_heaters
self._building = None
self._dhw_system = None
# Build and size immediately
self.build()
# ------------------------------------------------------------------
# Build orchestration
# ------------------------------------------------------------------
[docs]
def build(self) -> None:
"""Build the Building and DHWSystem (including sizing) from stored params."""
self._building = self._build_building()
self._dhw_system = self._build_dhw_system()
def _build_building(self):
"""Construct and return the Building for the configured type and climate."""
from ecoengine.objects.building.Building import Building
from ecoengine.objects.building.ClimateZone import ClimateZone
zone = self._build_climate_zone(ClimateZone)
return Building.from_building_type(
building_type = self.building_type,
magnitude = self.magnitude,
climate_zone = zone,
gpdpp = self.gpdpp,
custom_peak_load_shape = self.custom_peak_load_shape,
custom_avg_load_shape = self.custom_avg_load_shape,
)
def _build_climate_zone(self, ClimateZone):
"""Resolve zip_code_or_climate_zone to a ClimateZone instance."""
czv = self.zip_code_or_climate_zone
if isinstance(czv, dict):
return ClimateZone.from_design_conditions(**czv)
if isinstance(czv, str) and czv.isdigit() and len(czv) == 5:
return ClimateZone.from_zip_code(czv)
if isinstance(czv, int) and 1 <= czv <= 96:
return ClimateZone.from_zone_id(czv)
if isinstance(czv, str):
return ClimateZone.from_weather_station(czv)
raise ValueError(
f"Cannot determine ClimateZone from {czv!r}. "
"Pass a 5-digit CA zip code string, a zone ID int (1–96), "
"a weather station name string, or a dict with 'design_oat_f' "
"and/or 'design_inlet_water_temp_f'."
)
def _build_control_map(self):
"""
Build the control_schedule and control_map from stored aquastat parameters.
Returns (control_schedule, control_map).
"""
from ecoengine.objects.components.heating.Controls import Controls
on_t = self.on_trigger_t_f if self.on_trigger_t_f is not None else self.supply_temp_f
off_t = self.off_trigger_t_f if self.off_trigger_t_f is not None else self.storage_temp_f
normal = Controls(
on_sensor_fract = self.aquastat_fract,
on_trigger_t_f = on_t,
off_sensor_fract = self.off_sensor_fract,
off_trigger_t_f = off_t,
outlet_temp_f = self.storage_temp_f,
)
if not self.load_shift_schedule:
return ["normal"] * 24, {"normal": normal}
# Build schedule from 24-element 0/1 list (0 = shed, 1 = run)
schedule = []
shed_hours = [h for h, v in enumerate(self.load_shift_schedule) if v == 0]
first_shed = shed_hours[0] if shed_hours else None
for hour in range(24):
if self.load_shift_schedule[hour] == 0:
schedule.append("shed")
elif (
first_shed is not None
and self.load_up_hours > 0
and first_shed - self.load_up_hours <= hour < first_shed
):
schedule.append("loadUp")
else:
schedule.append("normal")
cmap = {"normal": normal}
shed_on_fract = self.shed_aquastat_fract if self.shed_aquastat_fract is not None else self.aquastat_fract
shed_off_fract = self.shed_off_sensor_fract if self.shed_off_sensor_fract is not None else self.off_sensor_fract
cmap["shed"] = Controls(
on_sensor_fract = shed_on_fract,
on_trigger_t_f = on_t,
off_sensor_fract = shed_off_fract,
off_trigger_t_f = off_t,
outlet_temp_f = self.storage_temp_f,
)
if self.load_up_hours > 0 and self.load_up_aquastat_fract is not None:
lu_off_fract = self.load_up_off_sensor_fract if self.load_up_off_sensor_fract is not None else self.off_sensor_fract
lu_off_t = self.load_up_off_trigger_t_f if self.load_up_off_trigger_t_f is not None else off_t
cmap["loadUp"] = Controls(
on_sensor_fract = self.load_up_aquastat_fract,
on_trigger_t_f = on_t,
off_sensor_fract = lu_off_fract,
off_trigger_t_f = lu_off_t,
outlet_temp_f = self.storage_temp_f,
)
return schedule, cmap
def _build_dhw_system(self):
"""
Construct and return the appropriate sized DHWSystem subclass based on
self.schematic.
Supported schematics
--------------------
* ``'primary_no_recirc'`` → DHWSystem
* ``'parallel_loop'`` → ParallelLoopSystem
* ``'swing_tank'`` → SwingSystem
* ``'single_pass_rtp'`` → SinglePassRTPSystem
* ``'multi_pass_rtp'`` → MultiPassRTPSystem
* ``'instant_wh'`` → InstantWHSystem
Raises
------
ValueError
If the schematic is not recognised or required recirc params are missing.
NotImplementedError
If the schematic is recognised but not yet fully implemented.
"""
if self.storage_volume_storageT_gal is not None:
return self._build_presized_dhw_system()
from ecoengine.objects.dhwsystems.DHWSystem import DHWSystem, _load_shift_fract_total_vol
control_schedule, control_map = self._build_control_map()
ls_fract = (
_load_shift_fract_total_vol(self.load_shift_percent)
if self.load_shift_schedule
else 1.0
)
if self.schematic in ["primary_no_recirc", "singlepass_norecirc"]:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return DHWSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
max_daily_run_hr = self.max_daily_run_hr,
defrost_factor = self.defrost_factor,
control_schedule = control_schedule,
control_map = control_map,
load_shift_fract_total_vol = ls_fract,
)
if self.schematic in ["parallel_loop", "paralleltank"]:
from ecoengine.objects.dhwsystems.recirc_systems.ParallelLoopSystem import ParallelLoopSystem
self._require_recirc_params()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return ParallelLoopSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
return_temp_f = self.return_temp_f,
return_flow_gpm = self.return_flow_gpm,
tm_on_temp_f = self.tm_on_temp_f,
tm_off_temp_f = self.tm_off_temp_f,
tm_off_time_hr = self.tm_off_time_hr,
tm_safety_factor = self.tm_safety_factor,
max_daily_run_hr = self.max_daily_run_hr,
defrost_factor = self.defrost_factor,
control_schedule = control_schedule,
control_map = control_map,
load_shift_fract_total_vol = ls_fract,
)
if self.schematic in ["swing_tank", "swingtank"]:
from ecoengine.objects.dhwsystems.recirc_systems.SwingSystem import SwingSystem
self._require_recirc_params()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return SwingSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
return_temp_f = self.return_temp_f,
return_flow_gpm = self.return_flow_gpm,
tm_safety_factor = self.tm_safety_factor,
max_daily_run_hr = self.max_daily_run_hr,
defrost_factor = self.defrost_factor,
control_schedule = control_schedule,
control_map = control_map,
load_shift_fract_total_vol = ls_fract,
)
if self.schematic in ["single_pass_rtp", "sprtp"]:
from ecoengine.objects.dhwsystems.rtp_systems.SinglePassRTPSystem import SinglePassRTPSystem
self._require_recirc_params()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return SinglePassRTPSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
return_temp_f = self.return_temp_f,
return_flow_gpm = self.return_flow_gpm,
max_daily_run_hr = self.max_daily_run_hr,
defrost_factor = self.defrost_factor,
tm_safety_factor = self.tm_safety_factor,
control_schedule = control_schedule,
control_map = control_map,
load_shift_fract_total_vol = ls_fract,
)
if self.schematic in ["multi_pass_rtp", "mprtp"]:
from ecoengine.objects.dhwsystems.rtp_systems.MultiPassRTPSystem import MultiPassRTPSystem
self._require_recirc_params()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return MultiPassRTPSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
return_temp_f = self.return_temp_f,
return_flow_gpm = self.return_flow_gpm,
max_daily_run_hr = 14,
defrost_factor = self.defrost_factor,
tm_safety_factor = self.tm_safety_factor,
control_schedule = control_schedule,
control_map = control_map,
)
if self.schematic == "instant_wh":
from ecoengine.objects.dhwsystems.InstantWHSystem import InstantWHSystem
return InstantWHSystem.from_size(
building = self._building,
supply_temp_f = self.supply_temp_f,
storage_temp_f = self.storage_temp_f,
defrost_factor = self.defrost_factor,
)
raise ValueError(
f"Unknown schematic {self.schematic!r}. "
"Supported values: 'primary_no_recirc', 'parallel_loop', 'swing_tank', "
"'single_pass_rtp', 'multi_pass_rtp', 'instant_wh'."
)
def _require_recirc_params(self) -> None:
missing = [
name for name, val in [
("return_temp_f", self.return_temp_f),
("return_flow_gpm", self.return_flow_gpm),
]
if val is None
]
if missing:
raise ValueError(
f"Schematic '{self.schematic}' requires: {', '.join(missing)}."
)
def _require_tm_params(self) -> None:
if self.tm_storage_vol is None:
raise ValueError(
f"Schematic '{self.schematic}' requires tm_storage_vol in pre-sized mode."
)
if self.tm_capacity_kbtuh is None and self.tm_model is None:
raise ValueError(
f"Schematic '{self.schematic}' requires tm_capacity_kbtuh or tm_model "
"in pre-sized mode."
)
def _build_tm_water_heater(self, tm_controls, force_electric_resistance : bool = False):
"""Build the TM WaterHeater (single unit) from tm_model or tm_capacity_kbtuh."""
from ecoengine.objects.components.heating.WaterHeater import WaterHeater
if force_electric_resistance:
self.num_tm_heaters = 1
per_unit_kbtuh = (
self.tm_capacity_kbtuh / self.num_tm_heaters
if self.tm_capacity_kbtuh is not None
else None
)
if self.tm_model is not None and not force_electric_resistance:
return WaterHeater.from_model_name(
model_name=self.tm_model,
control_schedule=["normal"] * 24,
control_map={"normal": tm_controls},
design_inlet_temp_f=self.return_temp_f if self.return_temp_f is not None else 50.0,
nominal_capacity_kbtuh=per_unit_kbtuh, # used by ER fallback when OAT < map minimum
)
return WaterHeater.from_nominal_capacity(
nominal_capacity_kbtuh=per_unit_kbtuh,
control_schedule=["normal"] * 24,
control_map={"normal": tm_controls},
)
def _build_presized_dhw_system(self):
"""
Build the DHWSystem from explicitly provided storage volume and capacity,
skipping the sizing step entirely.
For schematics with a TM sub-system (``parallel_loop``, ``swing_tank``),
the TM is still auto-sized from recirc parameters since TM sizing does
not depend on building load.
"""
from ecoengine.objects.components.heating.WaterHeater import WaterHeater
from ecoengine.objects.components.storage.StratifiedTank import StratifiedTank
control_schedule, control_map = self._build_control_map()
if self.hpwh_model is not None:
design_inlet_temp_f = self._building.get_design_inlet_water_temp_f() or 50.0
# Use num_units instead of N separate objects — all units share the
# same controls and tank sensor so they always act in unison. A single
# WaterHeater with num_units=N avoids N redundant interpolator calls
# per timestep (critical for pkl-backed performance maps at scale).
water_heaters = [
WaterHeater.from_model_name(
model_name=self.hpwh_model,
control_schedule=control_schedule,
control_map=control_map,
design_inlet_temp_f=design_inlet_temp_f,
num_units=self.num_heaters,
)
]
else:
capacity_per_heater = self.heating_capacity_kbtuh / self.num_heaters
water_heaters = [
WaterHeater.from_nominal_capacity(
nominal_capacity_kbtuh=capacity_per_heater,
control_schedule=control_schedule,
control_map=control_map,
)
for _ in range(self.num_heaters)
]
def _primary_tank(strat_slope=None):
if self.storage_volume_storageT_gal is None:
raise ValueError(
f"storage_volume_storageT_gal is required for schematic '{self.schematic}'."
)
kwargs = {"total_volume_gal": self.storage_volume_storageT_gal}
if strat_slope is not None:
kwargs["strat_slope"] = strat_slope
return StratifiedTank(**kwargs)
if self.schematic in ["primary_no_recirc", "singlepass_norecirc"]:
from ecoengine.objects.dhwsystems.DHWSystem import DHWSystem
return DHWSystem(
water_heaters=water_heaters,
storage_tank=_primary_tank(),
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
max_daily_run_hr=self.max_daily_run_hr,
defrost_factor=self.defrost_factor,
)
if self.schematic in ["parallel_loop", "paralleltank"]:
from ecoengine.objects.dhwsystems.recirc_systems.ParallelLoopSystem import ParallelLoopSystem
from ecoengine.objects.components.heating.Controls import Controls
from ecoengine.objects.components.storage.MixedStorageTank import MixedStorageTank
self._require_recirc_params()
self._require_tm_params()
tm_controls = Controls(
on_sensor_fract=0.5,
on_trigger_t_f=self.tm_on_temp_f,
off_sensor_fract=0.5,
off_trigger_t_f=self.tm_off_temp_f,
outlet_temp_f=self.tm_off_temp_f,
)
return ParallelLoopSystem(
water_heaters=water_heaters,
storage_tank=_primary_tank(),
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
return_temp_f=self.return_temp_f,
return_flow_gpm=self.return_flow_gpm,
tm_on_temp_f=self.tm_on_temp_f,
tm_off_temp_f=self.tm_off_temp_f,
tm_off_time_hr=self.tm_off_time_hr,
tm_safety_factor=self.tm_safety_factor,
tm_storage_tank=MixedStorageTank(total_volume_gal=self.tm_storage_vol),
tm_water_heater=self._build_tm_water_heater(tm_controls),
num_tm_heaters=self.num_tm_heaters,
max_daily_run_hr=self.max_daily_run_hr,
defrost_factor=self.defrost_factor,
)
if self.schematic in ["swing_tank", "swingtank"]:
from ecoengine.objects.dhwsystems.recirc_systems.SwingSystem import SwingSystem, _ELEMENT_DEADBAND_F
from ecoengine.objects.components.heating.Controls import Controls
from ecoengine.objects.components.storage.MixedStorageTank import MixedStorageTank
self._require_recirc_params()
self._require_tm_params()
tm_controls = Controls(
on_sensor_fract=0.5,
on_trigger_t_f=self.supply_temp_f,
off_sensor_fract=0.5,
off_trigger_t_f=self.supply_temp_f + _ELEMENT_DEADBAND_F,
outlet_temp_f=self.supply_temp_f + _ELEMENT_DEADBAND_F,
)
system = SwingSystem(
water_heaters=water_heaters,
storage_tank=_primary_tank(),
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
return_temp_f=self.return_temp_f,
return_flow_gpm=self.return_flow_gpm,
tm_safety_factor=self.tm_safety_factor,
max_daily_run_hr=self.max_daily_run_hr,
defrost_factor=self.defrost_factor,
)
system.tm_storage_tank = MixedStorageTank(total_volume_gal=self.tm_storage_vol)
system.tm_water_heater = self._build_tm_water_heater(tm_controls, force_electric_resistance = True)
return system
if self.schematic in ["single_pass_rtp", "sprtp"]:
from ecoengine.objects.dhwsystems.rtp_systems.SinglePassRTPSystem import (
SinglePassRTPSystem, _SPRTP_STRAT_SLOPE,
)
self._require_recirc_params()
return SinglePassRTPSystem(
water_heaters=water_heaters,
storage_tank=_primary_tank(strat_slope=_SPRTP_STRAT_SLOPE),
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
return_temp_f=self.return_temp_f,
return_flow_gpm=self.return_flow_gpm,
max_daily_run_hr=self.max_daily_run_hr,
defrost_factor=self.defrost_factor,
tm_safety_factor=self.tm_safety_factor,
)
if self.schematic in ["multi_pass_rtp", "mprtp"]:
from ecoengine.objects.dhwsystems.rtp_systems.MultiPassRTPSystem import (
MultiPassRTPSystem, _MPRTP_STRAT_SLOPE, _MPRTP_MAX_DAILY_RUN_HR,
)
from ecoengine.objects.components.storage.SlugOverlayTank import SlugOverlayTank
self._require_recirc_params()
cold_temp_f = self._building.get_design_inlet_water_temp_f() or 50.0
return MultiPassRTPSystem(
water_heaters=water_heaters,
storage_tank=SlugOverlayTank(
total_volume_gal=self.storage_volume_storageT_gal,
cold_temp_f=cold_temp_f,
storage_temp_f=self.storage_temp_f,
supply_temp_f=self.supply_temp_f,
strat_slope=_MPRTP_STRAT_SLOPE,
),
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
return_temp_f=self.return_temp_f,
return_flow_gpm=self.return_flow_gpm,
max_daily_run_hr=_MPRTP_MAX_DAILY_RUN_HR,
defrost_factor=self.defrost_factor,
tm_safety_factor=self.tm_safety_factor,
)
if self.schematic == "instant_wh":
from ecoengine.objects.dhwsystems.InstantWHSystem import InstantWHSystem
system = InstantWHSystem(
supply_temp_f=self.supply_temp_f,
storage_temp_f=self.storage_temp_f,
defrost_factor=self.defrost_factor,
)
system.water_heaters = water_heaters
return system
raise ValueError(
f"Unknown schematic {self.schematic!r}. "
"Supported values: 'primary_no_recirc', 'parallel_loop', 'swing_tank', "
"'single_pass_rtp', 'multi_pass_rtp', 'instant_wh'."
)
# ------------------------------------------------------------------
# Sizing results
# ------------------------------------------------------------------
[docs]
def get_sizing_results(self) -> dict:
"""
Return the sizing results from the built DHWSystem.
Returns
-------
dict
Always contains ``'min_capacity_kbtuh'`` and ``'min_storage_storageT_gal'``.
Parallel loop systems also include ``'min_tm_volume_gal'`` and
``'min_tm_capacity_kbtuh'``.
"""
sys = self._dhw_system
result = {
"min_capacity_kbtuh": sys._minimum_capacity_kbtuh,
"min_storage_storageT_gal": sys._minimum_storage_storageT_gal,
}
if hasattr(sys, "_minimum_tm_volume_gal") and sys._minimum_tm_volume_gal is not None:
result["min_tm_volume_gal"] = sys._minimum_tm_volume_gal
result["min_tm_capacity_kbtuh"] = sys._minimum_tm_capacity_kbtuh
return result
# ------------------------------------------------------------------
# Simulation
# ------------------------------------------------------------------
[docs]
def simulate_3day(self, **sim_run_kwargs):
"""
Run a 3-day design-day simulation at 1-minute timesteps.
Parameters
----------
**sim_run_kwargs
Optional keyword arguments forwarded to SimulationRun.__init__(),
e.g. ``outlet_deficit_threshold_f=5.0``, ``outlet_deficit_max_min=10``.
Returns
-------
SimulationRun
"""
return _simulate_3day(self._dhw_system, self._building, **sim_run_kwargs)
[docs]
def simulate_annual(self, **sim_run_kwargs):
"""
Run a full annual simulation at 10-minute timesteps.
Parameters
----------
**sim_run_kwargs
Optional keyword arguments forwarded to SimulationRun.__init__().
Returns
-------
SimulationRun
"""
return _simulate_annual(self._dhw_system, self._building, **sim_run_kwargs)
# ------------------------------------------------------------------
# Output helpers
# ------------------------------------------------------------------
[docs]
def get_simulation_summary(self, simulation_run) -> dict:
"""
Return a summary dict from a completed SimulationRun.
Parameters
----------
simulation_run : SimulationRun
Returns
-------
dict
Keys: 'successful', 'total_outage_min', 'total_energy_kwh',
'peak_demand_kw', 'num_steps_recorded', 'stopped_early'.
"""
return simulation_run.get_summary()
[docs]
def get_annual_cost_estimate(self, simulation_run) -> dict:
"""
Compute annual utility cost from a completed annual SimulationRun.
Parameters
----------
simulation_run : SimulationRun
Returns
-------
dict
Monthly energy breakdown (kWh) by month.
"""
return {"monthly_energy_kwh": simulation_run.get_monthly_energy_kwh()}
[docs]
def get_hw_magnitude(self) -> float:
"""
Return the total daily DHW usage for the building [gallons at supply temperature].
Returns
-------
float
"""
return self._building.daily_dhw_use_supplyT_gal
[docs]
def plot_simulation(
self,
title: str = "Simulation Results",
filepath: str | None = None,
include_temperatures: bool = False,
return_as_div: bool = False,
):
"""
Run a 3-day design-day simulation and return a Plotly figure of the results.
Parameters
----------
title : str
Figure title. Default ``"Simulation Results"``.
filepath : str, optional
If provided, write the figure to this path as a self-contained
HTML file. The figure is also returned regardless.
include_temperatures : bool
If True, add a right Y axis showing OAT, inlet water temperature,
and all tank node temperatures. Default False.
return_as_div : bool
If True, return the figure as an HTML ``<div>`` string instead of
a ``plotly.graph_objects.Figure``. Default False.
Returns
-------
plotly.graph_objects.Figure or str
A Plotly figure, or an HTML div string when ``return_as_div=True``.
Raises
------
ImportError
If ``plotly`` is not installed.
"""
sim_run = self.simulate_3day()
fig = sim_run.to_plotly(
title = title,
filepath = filepath,
include_temperatures = include_temperatures,
)
if return_as_div:
return fig.to_html(full_html=False, include_plotlyjs=False)
return fig
[docs]
def plot_sizing_curve(
self,
title: str = "Primary Sizing Curve",
filepath: str | None = None,
strat_slope: float = 2.8,
return_as_div: bool = False,
return_with_x_y_points: bool = False,
):
"""
Return a Plotly figure of the sizing curve for the built system.
For systems without load shifting, produces the primary sizing curve
(storage volume vs. heating capacity as a function of daily run hours).
For systems with load shifting (``load_shift_schedule`` was provided),
produces the load-shift sizing curve (storage volume vs. coverage
percentile), with a slider to explore the capacity/storage trade-off.
The recommended point — corresponding to the ``max_daily_run_hr`` or
``load_shift_percent`` passed to the engine — is highlighted on the
curve at page load.
Parameters
----------
title : str
Figure title. Default ``"Primary Sizing Curve"``.
filepath : str, optional
If provided, write the figure to this path as a self-contained
HTML file. The figure is also returned regardless.
strat_slope : float
Temperature gradient [°F / %-height] for stratification factor
calculation. Default 2.8.
return_as_div : bool
If True, return the figure as an HTML ``<div>`` string instead of
a ``plotly.graph_objects.Figure``. Default False.
return_with_x_y_points : bool
If True, return ``[figure_or_div, x_values, y_values, start_index]``
instead of just the figure/div, where ``x_values`` and ``y_values``
are the plot coordinates and ``start_index`` is the recommended
slider position. Default False.
Returns
-------
plotly.graph_objects.Figure
When both flags are False.
str
HTML div string when ``return_as_div=True`` and
``return_with_x_y_points=False``.
list
``[figure_or_div, x_values, y_values, start_index]`` when
``return_with_x_y_points=True``.
Raises
------
ImportError
If ``plotly`` is not installed.
"""
control_schedule, control_map = self._build_control_map()
is_ls = "shed" in control_map
if is_ls:
curve = self._dhw_system.get_ls_sizing_curve(
self._building,
control_schedule = control_schedule,
control_map = control_map,
strat_slope = strat_slope,
load_shift_percent = self.load_shift_percent,
)
x_vals = [p * 100.0 for p in curve["load_shift_percent"]]
y_vals = curve["storage_storageT_gal"]
start_index = curve["recommended_index"]
else:
curve = self._dhw_system.get_sizing_curve(self._building, strat_slope=strat_slope)
x_vals = curve["storage_storageT_gal"][::-1]
y_vals = curve["capacity_kbtuh"][::-1]
start_index = len(x_vals) - 1 - curve["recommended_index"]
fig = self._dhw_system._build_sizing_curve_figure(curve, is_ls, title, filepath)
result = fig.to_html(full_html=False, include_plotlyjs=False) if return_as_div else fig
if return_with_x_y_points:
return [result, x_vals, y_vals, start_index]
return result