ironiq.py

# ironiq.py
# Connect to the M-CAN bus of the Hyundai/Kia EVS (pre 2024).
# Pressing the BOT button (left one) will try to manually initate the battery
# preconditioning by spoofing messages from the car navigation to indicate
# that one is close to a charging station.
# To stop the process or terminate the battery preconditioning, press the
# KEY (right one) button.

# technical details:
#  - When BOT is pressed, three can messages are sent with id TOGGLE_PRECOND_FRAME_ID with
#    data in START_PRECOND_PAYLOAD.
#  - Then very time a message with id NAV_FRAME_ID is detected, immediately send a 
#    message with the same frame ID with the data in START_PRECOND_PAYLOAD. This will fool the
#    BMU into believing that the car is close to a charging station and thus start the 
#    preconditioning process.
#
# if everything works correctly the BMU will respond with a message with
# 0x2AD suggesting that it is preparing the preconditioning. After about a minute
# the 2AD message will change to indicate that the preconditioning is active.

import time
from machine import Pin
import CAN  # type: ignore[reportMissingImports]
from micropython import const
import lcd_bus
import lvgl as lv

_BITRATE = 500000
_TX_PIN  = 17
_RX_PIN  = 21
_WIDTH = const(320)
_HEIGHT = const(170)

TOGGLE_PRECOND_FRAME_ID = 0x0C7
START_PRECOND_PAYLOAD = list(bytes.fromhex("FF00004003000000"))
STOP_PRECOND_PAYLOAD = list(bytes.fromhex("FF00000000000000"))

NAV_FRAME_ID = 0x4ED
MAX_NAV_SPOOF_REPEATS = 5000
SPOOF_TIMEOUT = 300  # seconds
NAV_SPOOF_PAYLOAD = bytes.fromhex("10A000") # the last three bytes will be replaced

BMU_FRAME_ID = 0x2AD
BMU_STATUS_INDEX = 1
BMU_PRECOND_INACTIVE_STATUS = 0x01
BMU_PRECOND_CHECKING_STATUS = 0x05
BMU_PRECOND_ACTIVE_STATUS = 0x15

BAT_TEMP_FRAME_ID = 0x152
BAT_TEMP_COLDEST_CELL_INDEX = 0
BAT_TEMP_HOTTEST_CELL_INDEX = 1
BAT_TARGET_TEMP_C = 21

BAT_HEATING_RATE = 15/45 # Based on measurements, the battery heats up from 6C to 21C in about 40 minutes with the preconditioning active


# Buttons
BOT_BUTTON_PIN = 0  # GPIO pin for the BOT button (left one)
KEY_BUTTON_PIN = 14  # GPIO pin for the KEY button (right one)

BOT_BUTTON = Pin(BOT_BUTTON_PIN, Pin.IN, Pin.PULL_UP)
KEY_BUTTON = Pin(KEY_BUTTON_PIN, Pin.IN, Pin.PULL_UP)

import s3lcd_lvgl
_DISPLAY = s3lcd_lvgl.init(rotation=4)

import task_handler
_TASK_HANDLER = task_handler.TaskHandler(duration=5)



def init_ui():
    scr = lv.screen_active()
    scr.set_style_bg_color(lv.color_make(8, 12, 18), 0)
    scr.set_style_pad_all(0, 0)

    content = lv.obj(scr)
    content.set_size(lv.pct(100), lv.pct(100))
    content.set_style_bg_opa(lv.OPA.TRANSP, 0)
    content.set_style_border_width(0, 0)
    content.set_style_pad_top(4, 0)
    content.set_style_pad_bottom(8, 0)
    content.set_style_pad_left(10, 0)
    content.set_style_pad_right(10, 0)
    content.set_style_pad_row(4, 0)
    content.set_layout(lv.LAYOUT.FLEX)
    content.set_flex_flow(lv.FLEX_FLOW.COLUMN)
    content.set_flex_align(lv.FLEX_ALIGN.START, lv.FLEX_ALIGN.CENTER, lv.FLEX_ALIGN.CENTER)

    title_label = lv.label(content)
    title_label.set_style_text_font(lv.font_montserrat_28, 0)
    title_label.set_style_text_color(lv.color_make(255, 255, 255), 0)
    title_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    title_label.set_width(lv.pct(100))
    title_label.set_text("IRONIQ")

    command_title = lv.label(content)
    command_title.set_style_text_font(lv.font_montserrat_16, 0)
    command_title.set_style_text_color(lv.color_make(130, 150, 170), 0)
    command_title.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    command_title.set_width(lv.pct(100))
    command_title.set_text("Command")

    command_label = lv.label(content)
    command_label.set_style_text_font(lv.font_montserrat_24, 0)
    command_label.set_style_text_color(lv.color_make(210, 220, 235), 0)
    command_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    command_label.set_width(lv.pct(100))
    command_label.set_text("Ready")

    temp_title = lv.label(content)
    temp_title.set_style_text_font(lv.font_montserrat_20, 0)
    temp_title.set_style_text_color(lv.color_make(130, 150, 170), 0)
    temp_title.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    temp_title.set_width(lv.pct(100))
    temp_title.set_text("Battery temp")

    temp_label = lv.label(content)
    temp_label.set_style_text_font(lv.font_montserrat_20, 0)
    temp_label.set_style_text_color(lv.color_make(210, 220, 235), 0)
    temp_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    temp_label.set_width(lv.pct(100))
    temp_label.set_text("Waiting...")

    temp_eta_label = lv.label(content)
    temp_eta_label.set_style_text_font(lv.font_montserrat_20, 0)
    temp_eta_label.set_style_text_color(lv.color_make(165, 190, 215), 0)
    temp_eta_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    temp_eta_label.set_width(lv.pct(100))
    temp_eta_label.set_text("Heating ETA...")

    bmu_title = lv.label(content)
    bmu_title.set_style_text_font(lv.font_montserrat_16, 0)
    bmu_title.set_style_text_color(lv.color_make(130, 150, 170), 0)
    bmu_title.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    bmu_title.set_width(lv.pct(100))
    bmu_title.set_text("BMU status")

    bmu_label = lv.label(content)
    bmu_label.set_style_text_font(lv.font_montserrat_20, 0)
    bmu_label.set_style_text_color(lv.color_make(210, 220, 235), 0)
    bmu_label.set_style_text_align(lv.TEXT_ALIGN.CENTER, 0)
    bmu_label.set_width(lv.pct(100))
    bmu_label.set_text("Waiting...")

    bot_indicator = lv.label(scr)
    bot_indicator.set_style_text_font(lv.font_montserrat_20, 0)
    bot_indicator.set_style_text_color(lv.color_make(255, 184, 77), 0)
    bot_indicator.set_text("B")
    bot_indicator.align(lv.ALIGN.BOTTOM_LEFT, 8, -8)

    return command_label, temp_label, temp_eta_label, bmu_label


def set_status(label, text, color):
    label.set_style_text_color(color, 0)
    label.set_text(text)


def partial(func, *fixed_args, **fixed_kwargs):
    def _bound(*args, **kwargs):
        merged_kwargs = {}
        merged_kwargs.update(fixed_kwargs)
        merged_kwargs.update(kwargs)
        return func(*(fixed_args + args), **merged_kwargs)

    return _bound


def _to_signed32(value):
    value &= 0xFFFFFFFF
    if value & 0x80000000:
        return value - 0x100000000
    return value


def _to_bytes(payload):
	if isinstance(payload, bytes):
		return payload
	if isinstance(payload, bytearray):
		return bytes(payload)
	return bytes(payload)


def _build_reply_payload(payload):
	data = bytearray(_to_bytes(payload))
	while len(data) < 3:
		data.append(0x00)
	data[-3:] = b"\x10\xA0\x00"
	return data

def _configure_filter(dev, filt_id, filt_mask):
    # FILTER_RAW_SINGLE expects raw TWAI acceptance register values.
    raw_code = (filt_id & 0x7FF) << 21
    id_mask = (~filt_mask) & 0x7FF
    raw_mask = (id_mask << 21) | 0x001FFFFF
    args = [_to_signed32(raw_code), _to_signed32(raw_mask)]
    try:
        dev.set_filters(0, CAN.FILTER_RAW_SINGLE, args)
    except TypeError:
        dev.set_filters(0, CAN.FILTER_RAW_SINGLE, args, False, False)


def _configure_dual_filter(dev, id1, id2):
    # FILTER_RAW_DUAL packs two 11-bit exact-match filters into one 32-bit pair.
    # SJA1000/TWAI dual-filter layout for standard frames:
    #   code bits [31:24] = ID1[10:3], [23:16] = ID1[2:0] << 5
    #   code bits [15:8]  = ID2[10:3], [7:0]   = ID2[2:0] << 5
    #   mask: 0 = compare bit, 1 = don't care;
    #         lower 5 bits of each byte are RTR/data bits -> always don't care (0x1F)
    raw_code = (
        (((id1 & 0x7FF) >> 3) & 0xFF) << 24 |
        ((id1 & 0x07) << 5) << 16 |
        (((id2 & 0x7FF) >> 3) & 0xFF) << 8 |
        ((id2 & 0x07) << 5)
    )
    raw_mask = 0x001F001F  # exact match on both ID fields; RTR/data bytes don't care
    args = [_to_signed32(raw_code), _to_signed32(raw_mask)]
    try:
        dev.set_filters(0, CAN.FILTER_RAW_DUAL, args)
    except TypeError:
        dev.set_filters(0, CAN.FILTER_RAW_DUAL, args, False, False)


def init_can():
    dev = CAN(
        0,
        extframe=False,
        tx=_TX_PIN,
        rx=_RX_PIN,
        mode=CAN.NORMAL,
        bitrate=_BITRATE,
        auto_restart=False,
    )
    _configure_dual_filter(dev, BMU_FRAME_ID, NAV_FRAME_ID)
    return dev


def _send_ok(status):
    # Different driver builds return either None/bool or esp_err_t style ints.
    # Treat None, True, and numeric 0 (ESP_OK) as success.
    return status is None or status is True or status == 0


def _bmu_requests_stop(status_byte):
    return status_byte in (
        BMU_PRECOND_CHECKING_STATUS,
        BMU_PRECOND_ACTIVE_STATUS,
    )


def _bmu_is_not_active(status_byte):
    return status_byte not in (
        BMU_PRECOND_CHECKING_STATUS,
        BMU_PRECOND_ACTIVE_STATUS,
    )

def precond_toggle(payload, frame_id, pending_text, sent_text, can_dev, command_label):
    set_status(
        command_label,
        pending_text,
        lv.color_make(255, 184, 77),
    )
    send_status = None
    for _ in range(3):
        send_status = can_dev.send(payload, frame_id)
        time.sleep_ms(20)
    if _send_ok(send_status):
        set_status(
            command_label,
            sent_text,
            lv.color_make(120, 200, 255),
        )
        print("Precond toggle message sent successfully")
    else:
        set_status(
            command_label,
            "Command send\nfailed",
            lv.color_make(255, 96, 96),
        )
        print("Precond toggle send failed with status:", send_status)

start_precond = partial(precond_toggle, 
    START_PRECOND_PAYLOAD, 
    TOGGLE_PRECOND_FRAME_ID, 
    "Sending START",
    "START sent")
stop_precond = partial(precond_toggle, 
    STOP_PRECOND_PAYLOAD, 
    TOGGLE_PRECOND_FRAME_ID, 
    "Sending STOP",
    "STOP sent")

def update_precond_status(payload, status_label):
    if len(payload) < BMU_STATUS_INDEX + 1:
        print("Received BMU message with insufficient length:", list(bytes(payload)))
        return None
    status_byte = payload[BMU_STATUS_INDEX]
    if status_byte == BMU_PRECOND_INACTIVE_STATUS:
        set_status(
            status_label,
            "Precond\nnot active",
            lv.color_make(165, 190, 215),
        )
        return status_byte
    elif status_byte == BMU_PRECOND_CHECKING_STATUS:
        set_status(
            status_label,
            "Checking",
            lv.color_make(80, 205, 255),
        )
        return status_byte
    elif status_byte == BMU_PRECOND_ACTIVE_STATUS:
        set_status(
            status_label,
            "Active",
            lv.color_make(80, 235, 120),
        )
        return status_byte
    else:
        set_status(
            status_label,
            "Unknown",
            lv.color_make(255, 150, 90),
        )
        print("Received BMU message with unknown precond status: 0x{:02X}".format(status_byte))
        return status_byte


def update_battery_temp(payload, temp_label, temp_eta_label):
    if len(payload) < 2:
        print("Received battery temperature message with insufficient length:", list(bytes(payload)))
        return

    coldest_cell_temp = payload[BAT_TEMP_COLDEST_CELL_INDEX]
    hottest_cell_temp = payload[BAT_TEMP_HOTTEST_CELL_INDEX]
    set_status(
        temp_label,
        "{}C / {}C".format(coldest_cell_temp, hottest_cell_temp),
        lv.color_make(255, 184, 77),
    )
    if coldest_cell_temp >= BAT_TARGET_TEMP_C:
        set_status(
            temp_eta_label,
            "At or above {}C".format(BAT_TARGET_TEMP_C),
            lv.color_make(80, 235, 120),
        )
    else:
        minutes_to_target = int((BAT_TARGET_TEMP_C - coldest_cell_temp) / BAT_HEATING_RATE + 0.5)
        set_status(
            temp_eta_label,
            "~{} min to {}C".format(minutes_to_target, BAT_TARGET_TEMP_C),
            lv.color_make(165, 190, 215),
        )
    print(
        "Battery temperatures - Coldest cell: {}C, Hottest cell: {}C".format(
            coldest_cell_temp,
            hottest_cell_temp,
        )
    )


def main():
    can = init_can()
    command_label, temp_label, temp_eta_label, bmu_label = init_ui()
    set_status(
        command_label,
        "Start precond\nPress B",
        lv.color_make(210, 220, 235),
    )
    set_status(
        temp_label,
        "Waiting...",
        lv.color_make(165, 190, 215),
    )
    set_status(
        temp_eta_label,
        "Waiting...",
        lv.color_make(165, 190, 215),
    )
    set_status(
        bmu_label,
        "Waiting...",
        lv.color_make(165, 190, 215),
    )
    print("CAN initialized, waiting for messages...")
    bot_was_pressed = False

    precond_starting = False
    bmu_status = None
    nav_bmu_filter = True
    switch_filter_timeout = 5 # seconds 
    t = time.time()
    while True:
        bot_is_pressed = BOT_BUTTON.value() == 0
        if bot_is_pressed and not bot_was_pressed:
            if _bmu_requests_stop(bmu_status):
                stop_precond(can, command_label)
                precond_starting = False
                print("BOT button pressed stopping precond")
            else:
                precond_starting = True
                start_precond(can, command_label)
                print("BOT button pressed starting precond")
        bot_was_pressed = bot_is_pressed

        if time.time() - t > switch_filter_timeout:
            if nav_bmu_filter:
                _configure_filter(can, BAT_TEMP_FRAME_ID, 0x7FF)  # switch to single filter for battery temp
                print("Switched CAN filter to listen for battery temperature frames")
                nav_bmu_filter = False
                switch_filter_timeout = 1
            else:
                _configure_dual_filter(can, BMU_FRAME_ID, NAV_FRAME_ID)  # ensure we're listening for BMU and NAV frames during precond process
                print("Using dual CAN filter for BMU and NAV frames")
                nav_bmu_filter = True
                switch_filter_timeout = 5
            t = time.time()

        if can.any():
            frame_id, is_extended, is_rtr, payload = can.recv()
            print("Received CAN message: ID={:03X} ext={} rtr={} data={}".format(
                frame_id, is_extended, is_rtr, list(bytes(payload))))
            if frame_id == BMU_FRAME_ID:
                bmu_status = update_precond_status(payload, bmu_label)
                if _bmu_requests_stop(bmu_status):
                    set_status(
                        command_label,
                        "Press B to stop precond",
                        lv.color_make(210, 220, 235),
                    )
            if precond_starting and frame_id == NAV_FRAME_ID and _bmu_is_not_active(bmu_status):
                # Spoof NAV messages to keep preconditioning active
                spoof_payload = bytearray(payload)
                spoof_payload[-3:] = bytearray(NAV_SPOOF_PAYLOAD)
                can.send(list(spoof_payload), NAV_FRAME_ID)
                print("Sent spoofed NAV message to maintain precond")
            if frame_id == BAT_TEMP_FRAME_ID:
                update_battery_temp(payload, temp_label, temp_eta_label)
                # only need one temp frame to update, switch back to BMU/NAV filter to reduce noise
                _configure_dual_filter(can, BMU_FRAME_ID, NAV_FRAME_ID)
                print("Switched CAN filter back to BMU/NAV after receiving battery temp frame")
                nav_bmu_filter = True
                switch_filter_timeout = 5
                t = time.time()
        time.sleep_ms(1)


if __name__ == "__main__":
    main()