Modules

Second-Pass TOU Scheduling Decision Model for HPWHs in OCHRE

This module implements the heuristic-based decision model for consumer response to time-of-use (TOU) electricity rates in residential building simulations.

evaluate_value_learning_decision(controller, default_monthly_bill, tou_monthly_bill, default_monthly_unmet_demand, tou_monthly_unmet_demand)

Evaluate value learning decisions for each month.

Parameters:

Name	Type	Description	Default
controller	ValueLearningController	ValueLearningController instance	required
default_monthly_bill	list[float]	List of default monthly bills	required
tou_monthly_bill	list[float]	List of TOU monthly bills	required
default_monthly_unmet_demand	list[float]	List of default monthly unmet demand in kWh	required
tou_monthly_unmet_demand	list[float]	List of TOU monthly unmet demand in kWh	required

Returns:

Type	Description
list[str]	Tuple of (monthly_decisions, states, learning_metrics_history)
list[str]	monthly_decisions: List of "switch" or "stay" for each month
list[dict]	states: List of "default" or "tou" for each month
tuple[list[str], list[str], list[dict]]	learning_metrics_history: List of learning metrics for each month

Source code in rate_design_platform/second_pass.py

def evaluate_value_learning_decision(
    controller: ValueLearningController,
    default_monthly_bill: list[float],
    tou_monthly_bill: list[float],
    default_monthly_unmet_demand: list[float],
    tou_monthly_unmet_demand: list[float],
) -> tuple[list[str], list[str], list[dict]]:
    """
    Evaluate value learning decisions for each month.

    Args:
        controller: ValueLearningController instance
        default_monthly_bill: List of default monthly bills
        tou_monthly_bill: List of TOU monthly bills
        default_monthly_unmet_demand: List of default monthly unmet demand in kWh
        tou_monthly_unmet_demand: List of TOU monthly unmet demand in kWh

    Returns:
        Tuple of (monthly_decisions, states, learning_metrics_history)
        - monthly_decisions: List of "switch" or "stay" for each month
        - states: List of "default" or "tou" for each month
        - learning_metrics_history: List of learning metrics for each month
    """
    decisions = []
    states = []
    learning_metrics_history = []

    print("Starting value learning simulation...")
    print(f"Initial state: {controller.get_current_state()}")
    print(f"Initial learning state: {controller.get_learning_state()}")

    # Get building-specific beta for comfort penalty calculation
    building_beta = controller.state_manager.state.beta
    print(f"Using building-specific beta for comfort penalties: {building_beta:.4f}")

    for i in range(len(default_monthly_bill)):
        # Record current state before decision
        current_state = controller.get_current_state()
        states.append(current_state)

        # Calculate comfort penalties on-the-fly using building-specific beta
        default_comfort_penalty = building_beta * default_monthly_unmet_demand[i]
        tou_comfort_penalty = building_beta * tou_monthly_unmet_demand[i]

        # Make value learning decision
        decision, learning_metrics = controller.evaluate_TOU(
            default_monthly_bill[i],
            tou_monthly_bill[i],
            default_comfort_penalty,
            tou_comfort_penalty,
        )

        decisions.append(decision)
        learning_metrics_history.append(learning_metrics)

        # Log monthly progress
        print(
            f"Month {i + 1}: {current_state} -> {decision} "
            f"(exploration rate: {learning_metrics['epsilon_m']:.3f}, "
            f"value diff: ${learning_metrics['value_difference']:.2f})"
        )

    # Print final learning state
    final_state = controller.get_learning_state()
    print("\nFinal learning state:")
    print(f"  V_default: ${final_state['v_default']:.2f}")
    print(f"  V_tou: ${final_state['v_tou']:.2f}")
    print(f"  Final state: {final_state['current_state']}")
    print(f"  Final exploration rate: {final_state['epsilon_m']:.3f}")

    return decisions, states, learning_metrics_history

run_value_learning_simulation(TOU_params, house_args, building_xml_path=None)

Run complete value learning TOU HPWH simulation.

Parameters:

Name	Type	Description	Default
TOU_params	TOUParameters	TOU parameters	required
house_args	dict	Base house arguments dictionary	required
building_xml_path	Optional[str]	Path to building XML file (for building characteristics)	None

Returns:

Type	Description
tuple[list, dict[str, float], list[dict]]	Tuple of (monthly_results, annual_metrics, learning_metrics_history)

Source code in rate_design_platform/second_pass.py

def run_value_learning_simulation(
    TOU_params: TOUParameters, house_args: dict, building_xml_path: Optional[str] = None
) -> tuple[list, dict[str, float], list[dict]]:
    """
    Run complete value learning TOU HPWH simulation.

    Args:
        TOU_params: TOU parameters
        house_args: Base house arguments dictionary
        building_xml_path: Path to building XML file (for building characteristics)

    Returns:
        Tuple of (monthly_results, annual_metrics, learning_metrics_history)
    """
    from rate_design_platform.utils.building_characteristics import (
        enrich_building_characteristics,
        parse_building_xml,
    )
    from rate_design_platform.utils.prior_initialization import initialize_value_learning_controller_with_priors
    from rate_design_platform.utils.value_learning_params import ValueLearningParameters

    # Use default parameters if None provided
    if TOU_params is None:
        TOU_params = TOUParameters()

    # Parse building characteristics
    if building_xml_path:
        building_chars = parse_building_xml(building_xml_path)
        building_chars = enrich_building_characteristics(building_chars)
    else:
        # Use default building characteristics for the XML file in house_args
        hpxml_file = house_args.get("hpxml_file")
        if hpxml_file:
            building_chars = parse_building_xml(hpxml_file)
            building_chars = enrich_building_characteristics(building_chars)
        else:
            # Fallback to default characteristics
            from rate_design_platform.utils.building_characteristics import BuildingCharacteristics

            building_chars = BuildingCharacteristics(
                state_code="CO",
                zip_code="80202",
                year_built=1980,
                n_residents=2.5,
                n_bedrooms=3,
                conditioned_floor_area=1500.0,
                residential_facility_type="single-family detached",
                climate_zone="5B",
                ami=1.0,
                wh_type="storage",
            )

    # 1. Initialize ValueLearning Parameters
    params = ValueLearningParameters()
    print(f"Initialized value learning parameters with beta_base: {params.beta_base}")

    # 2. Run Year-Long Simulations to get bills and unmet demand
    print("Running year-long pre-simulations to get bills and unmet demand...")
    default_monthly_bills, default_monthly_unmet_demand = simulate_full_cycle("default", TOU_params, house_args)
    tou_monthly_bills, tou_monthly_unmet_demand = simulate_full_cycle("tou", TOU_params, house_args)

    # 3. Initialize Priors (bills only)
    print("Initializing controller with priors (bills only)...")
    controller = initialize_value_learning_controller_with_priors(
        params, building_chars, default_monthly_bills, tou_monthly_bills
    )

    # 4. Run Value Learning Simulation
    print("Starting value learning simulation with on-the-fly comfort penalty calculation...")

    # Extract monthly bill data
    default_monthly_bill = [result.bill for result in default_monthly_bills]
    tou_monthly_bill = [result.bill for result in tou_monthly_bills]

    # Run value learning decision process with unmet demand for on-the-fly comfort penalty calculation
    monthly_decisions, states, learning_metrics_history = evaluate_value_learning_decision(
        controller, default_monthly_bill, tou_monthly_bill, default_monthly_unmet_demand, tou_monthly_unmet_demand
    )

    # Calculate simulation year and months
    simulation_year_months = calculate_simulation_months(house_args)

    # Calculate comfort penalties for final metrics using building-specific beta
    building_beta = controller.state_manager.state.beta
    default_monthly_comfort_penalty = [building_beta * unmet for unmet in default_monthly_unmet_demand]
    tou_monthly_comfort_penalty = [building_beta * unmet for unmet in tou_monthly_unmet_demand]

    # Calculate value learning monthly metrics
    monthly_results = calculate_value_learning_monthly_metrics(
        simulation_year_months,
        monthly_decisions,
        states,
        default_monthly_bill,
        tou_monthly_bill,
        default_monthly_comfort_penalty,
        tou_monthly_comfort_penalty,
        learning_metrics_history,
    )

    # Calculate value learning annual metrics
    annual_metrics = calculate_value_learning_annual_metrics(monthly_results)

    return monthly_results, annual_metrics, learning_metrics_history

simulate_full_cycle(simulation_type, TOU_params, house_args)

Simulate complete annual cycle to get bills and unmet demand for prior initialization

Parameters:

Name	Type	Description	Default
simulation_type	str	Type of simulation to run ("default" or "tou")	required
TOU_params	TOUParameters	TOU parameters (uses default if None)	required
house_args	dict	Base house arguments dictionary	required

Returns:

Type	Description
list[MonthlyMetrics]	Tuple of (monthly_bills, monthly_unmet_demand_totals)
list[float]	monthly_bills: List of MonthlyMetrics with bill amounts only
tuple[list[MonthlyMetrics], list[float]]	monthly_unmet_demand_totals: List of monthly unmet demand totals in kWh

Source code in rate_design_platform/second_pass.py

def simulate_full_cycle(
    simulation_type: str, TOU_params: TOUParameters, house_args: dict
) -> tuple[list[MonthlyMetrics], list[float]]:
    """
    Simulate complete annual cycle to get bills and unmet demand for prior initialization

    Args:
        simulation_type: Type of simulation to run ("default" or "tou")
        TOU_params: TOU parameters (uses default if None)
        house_args: Base house arguments dictionary

    Returns:
        Tuple of (monthly_bills, monthly_unmet_demand_totals)
        - monthly_bills: List of MonthlyMetrics with bill amounts only
        - monthly_unmet_demand_totals: List of monthly unmet demand totals in kWh
    """
    if TOU_params is None:
        TOU_params = TOUParameters()

    if house_args is None:
        house_args = HOUSE_ARGS

    # Create separate output directories for each simulation type
    base_path = os.path.abspath(os.path.join(os.path.dirname(__file__)))
    simulation_output_path = os.path.join(base_path, "outputs", f"{simulation_type}_simulation")

    # Create the directory if it doesn't exist
    os.makedirs(simulation_output_path, exist_ok=True)

    # Update house_args with the new output path
    house_args.update({"output_path": simulation_output_path})
    dwelling = Dwelling(**house_args)

    start_time = house_args["start_time"]
    end_time = house_args["end_time"]
    time_step = house_args["time_res"]
    monthly_rate_structure = calculate_monthly_intervals(start_time, end_time, time_step)

    if simulation_type == "default":
        operation_schedule = create_operation_schedule("default", monthly_rate_structure, TOU_params, time_step)
    else:
        operation_schedule = create_operation_schedule("tou", monthly_rate_structure, TOU_params, time_step)

    simulation_results = run_ochre_wh_dynamic_control(dwelling, operation_schedule, time_step)

    monthly_rates = create_tou_rates(simulation_results.Time, time_step, TOU_params)

    # Combine monthly interals and rates into a single list
    monthly_rates = [
        MonthlyRateStructure(
            year=monthly_rate_structure.year,
            month=monthly_rate_structure.month,
            intervals=monthly_rate_structure.intervals,
            rates=monthly_rate_structure.rates,
        )
        for monthly_rate_structure in monthly_rates
    ]

    # Calculate monthly bills only
    monthly_bills = calculate_monthly_bill(simulation_results, monthly_rates)

    # Extract monthly unmet demand totals for later comfort penalty calculation
    monthly_unmet_demand = []
    time_stamps = simulation_results.Time
    unmet_demand_kWh = simulation_results.D_unmet_mt

    # Group by month
    current_month = None
    month_start_idx = 0

    for i, timestamp in enumerate(time_stamps):
        try:
            month = timestamp.month
        except AttributeError:
            month = timestamp.astype("datetime64[M]").astype(int) % 12 + 1

        if current_month is None:
            current_month = month
        elif month != current_month:
            # Month changed, calculate total unmet demand for the previous month
            month_intervals = i - month_start_idx
            month_demand = unmet_demand_kWh[month_start_idx : month_start_idx + month_intervals]
            monthly_unmet_demand.append(float(np.sum(month_demand)))

            # Start new month
            current_month = month
            month_start_idx = i

    # Handle the last month
    if month_start_idx < len(time_stamps):
        month_intervals = len(time_stamps) - month_start_idx
        month_demand = unmet_demand_kWh[month_start_idx : month_start_idx + month_intervals]
        monthly_unmet_demand.append(float(np.sum(month_demand)))

    return monthly_bills, monthly_unmet_demand