Heat Pump Intelligence
& Power Validation

Phase 1 Must Be Complete

New Capabilities on Top of Phase 1

CN105 Adapter — ESPHome Local Control

The CN105 port is a 5-pin JST connector on every Mitsubishi mini-split — it's the same port the Kumo Cloud WiFi adapter uses. An ESP32 running the SwiCago HeatPump ESPHome component replaces it entirely, giving fully local control with zero cloud dependency.

Hardware options

CN105 pinout

ESPHome config for CN105

# Add to a new yaml file: mitsubishi_hp.yaml
# Same secrets.yaml, same MQTT broker as Phase 1

external_components:
  - source:
      type: git
      url: https://github.com/SwiCago/HeatPump
    components: [mitsubishi_heatpump]

uart:
  id: hp_uart
  tx_pin: GPIO17
  rx_pin: GPIO18
  baud_rate: 2400

climate:
  - platform: mitsubishi_heatpump
    name: "Hyper Heat"
    uart_id: hp_uart

sensor:
  - platform: template
    name: "HP Outdoor Temp"         # exposed by component
  - platform: template
    name: "HP Compressor Frequency"    # 0 = off, >0 = running

mqtt:
  broker: 192.168.1.XXX
  topic_prefix: "heatpump/unit1"

Per-Circuit Watts as Ground Truth

The Emporia Vue doesn't care what the CN105 says. It sees electrons on the wire. This is what makes it the right validation layer — it's completely independent of the heat pump's self-reported state.

Emporia polling — add to erv_monitor.py

# pip install pyemvue
import pyemvue
from pyemvue.enums import Scale, Unit

HP_IDLE_WATTS   = 50    # fan only / standby
HP_ACTIVE_WATTS = 800   # compressor running threshold

async def poll_emporia_loop(mqtt_client):
    vue = pyemvue.PyEmVue()
    vue.login(username='you@email.com', password='yourpassword',
              token_storage_file='emporia_keys.json')

    while True:
        channels = vue.get_recent_usage(
            scale=Scale.MINUTE.value, unit=Unit.WATTS.value
        )
        for ch in channels:
            mqtt_client.publish(
                f"power/{ch.name.lower().replace(' ','_')}/watts",
                ch.usage, qos=1
            )
        await asyncio.sleep(60)  # Emporia API: max 1/min

Emporia as the Independent Enforcer

The full control + validation flow for every heat pump command. The CN105 tells you what the unit thinks it's doing. The Emporia tells you what it's actually drawing. The gap between those two is where failures live.

Validation implementation

# In erv_monitor.py — add to automation engine

async def enforce_hp_command(unit_id: str, mode: str, setpoint: float = None):
    for attempt in range(3):
        # Send command via CN105 MQTT topic
        payload = {"mode": mode}
        if setpoint: payload["temperature"] = setpoint
        mqtt.publish(f"heatpump/{unit_id}/climate/set", json.dumps(payload))

        await asyncio.sleep(90)  # wait for compressor ramp-down

        cn105_mode  = await get_mqtt_state(f"heatpump/{unit_id}/mode")
        emporia_w   = await get_latest_watts("heat_pump_compressor")

        if mode == "off":
            confirmed = (cn105_mode in ("off", "fan_only")) \
                        and (emporia_w < HP_IDLE_WATTS)
        else:
            confirmed = cn105_mode == mode \
                        and emporia_w > HP_IDLE_WATTS

        if confirmed:
            await log_event("HP_COMMAND_CONFIRMED",
                            {"mode": mode, "attempts": attempt+1,
                             "watts": emporia_w})
            return True

        await log_event("HP_COMMAND_RETRY",
                        {"attempt": attempt+1, "cn105": cn105_mode,
                         "emporia_w": emporia_w})

    # Three attempts failed — alert and log
    await send_alert(f"HP {unit_id} not responding to {mode} after 3 attempts")
    return False


async def watchdog_loop():
    """60-second loop: if state drifts from intended, re-issue command."""
    while True:
        for unit_id in HP_UNITS:
            intended = INTENDED_STATE.get(unit_id)
            if not intended: continue

            cn105_mode = await get_mqtt_state(f"heatpump/{unit_id}/mode")
            emporia_w  = await get_latest_watts("heat_pump_compressor")

            # Detect unexpected power draw
            hp_should_be_off = intended["mode"] == "off"
            hp_drawing_power = emporia_w > HP_ACTIVE_WATTS

            if hp_should_be_off and hp_drawing_power:
                await log_event("HP_UNEXPECTED_POWER",
                                {"watts": emporia_w, "cn105": cn105_mode})
                await enforce_hp_command(unit_id, "off")

            # Detect state drift from external input
            elif cn105_mode != intended["mode"]:
                await log_event("HP_STATE_DRIFT",
                                {"actual": cn105_mode, "intended": intended})
                await enforce_hp_command(unit_id, **intended)

        await asyncio.sleep(60)

Forecast-Driven COP Scheduling

This is the highest-value part of Phase 2. The Hyper Heat's efficiency (COP) drops significantly as outdoor temperature falls. A pre-heat decision made 6–12 hours in advance — based on the Open-Meteo forecast already running from Phase 1 — can shift expensive low-COP runtime to cheap high-COP windows without reducing comfort at all.

Hyper Heat approximate COP by outdoor temperature

# Real COP curve query — run after 2+ weeks of data
SELECT
  ROUND(hp_outdoor_temp / 5) * 5     AS temp_bucket_f,
  AVG(hp_power_watts)                AS avg_watts,
  AVG(hp_power_watts) / 1000.0
    * 0.14                            AS cost_per_hour,
  -- replace 0.14 with your actual $/kWh
  COUNT(*)                            AS samples
FROM hp_readings h
JOIN power_readings p ON h.ts = p.ts
WHERE hp_mode = 'heat'
  AND hp_outdoor_temp IS NOT NULL
GROUP BY temp_bucket_f
ORDER BY temp_bucket_f DESC;

Pre-heat scheduling logic

# Pre-heat scheduler — runs every 5 minutes
async def hp_scheduling_loop():
    while True:
        weather      = await get_latest_weather()
        indoor_temps = await get_indoor_temps()   # from Aqara via MQTT
        avg_indoor   = sum(indoor_temps.values()) / len(indoor_temps)

        now_f        = weather["outdoor_temp_f"]
        forecast_8h  = weather["temp_in_8h"]           # from Open-Meteo (Phase 1)
        forecast_4h  = weather["temp_in_4h"]

        current_sp   = await get_hp_setpoint("unit1")
        base_sp      = 68.0  # your normal setpoint

        if now_f > 40 and forecast_8h < 20:
            target_sp = base_sp + 2.0
            reason    = f"pre-heat: {now_f}°F now → {forecast_8h}°F in 8h"

        elif now_f < 5 and avg_indoor >= base_sp - 1:
            target_sp = base_sp - 1.0
            reason    = "extreme cold: drop setpoint, use stored heat"

        elif forecast_4h > now_f + 15:
            target_sp = base_sp
            reason    = "warming coming — hold normal setpoint"

        else:
            target_sp = base_sp
            reason    = "normal operation"

        if target_sp != current_sp:
            await enforce_hp_command("unit1", "heat", setpoint=target_sp)
            await log_event("HP_SETPOINT_CHANGE",
                            {"from": current_sp, "to": target_sp, "reason": reason})

        await asyncio.sleep(300)  # 5 minutes

New Dashboard Panels

Recommended Order