yaris-obd/modules/app_obd.py

from datetime import datetime
from threading import Event, Lock, Thread
import glob
import os
import time

import obd

from modules import app_events, app_gps


RECONNECT_INTERVAL = 10
SENSOR_POLL_INTERVAL = 5

status = None
speed = 0
speed_updated_at = None
rpm = None
coolant_temp = None
fuel_level = None
throttle_position = None
mil_on = None
dtc_codes = []
odometer = None
mil_updated_at = None
dtc_updated_at = None


class OBD:
    def __init__(self):
        self.status = status
        self.rpm = rpm
        self.speed = speed


obd_status = ["Disconnected", "Connected", "Connecting"]
obd_speed_lock = Lock()
obd_rpm_lock = Lock()
obd_sensor_lock = Lock()
_connection_lock = Lock()
_connect_lock = Lock()
_query_lock = Lock()
thread_speed = None
thread_rpm = None
thread_sensors = None
_reconnect_thread = None
stop_speed = Event()
stop_rpm = Event()
stop_sensors = Event()

connection = None


def _decode_odometer(messages):
    try:
        data = messages[0].data[2:]
        if len(data) != 4:
            return None
        return int.from_bytes(bytes(data), byteorder="big") / 10
    except Exception:
        return None


ODOMETER_COMMAND = obd.OBDCommand(
    "ODOMETER",
    "Vehicle odometer",
    b"01A6",
    6,
    _decode_odometer,
)


def _find_obd_port():
    """Locate the USB-serial OBD-II adapter."""
    env_port = os.environ.get("OBD_PORT")
    if env_port:
        return env_port
    ports = sorted(glob.glob("/dev/ttyUSB*"))
    return ports[0] if ports else None


def _connection_status(candidate):
    try:
        return candidate.status()
    except Exception:
        return obd.OBDStatus.NOT_CONNECTED


def _is_car_connected(candidate=None):
    if candidate is None:
        with _connection_lock:
            candidate = connection
    return (
        candidate is not None
        and _connection_status(candidate) == obd.OBDStatus.CAR_CONNECTED
    )


def _emit_disconnected():
    app_events.socketio.emit("obd_status", obd_status[0])
    app_events.socketio.emit("obd_speed", "-")
    app_events.socketio.emit("obd_error", "-")
    app_events.socketio.emit("obd_rpm", "-")


def _close_connection(candidate):
    try:
        candidate.close()
    except Exception as error:
        print("Failed to close OBD connection:", str(error))


def _mark_connection_lost(candidate, operation, error):
    global connection
    with _connection_lock:
        if connection is not candidate:
            return
        connection = None
    print(f"OBD {operation} failed; reconnecting: {error}")
    _close_connection(candidate)
    _emit_disconnected()


def _query(command, *, force=False):
    with _connection_lock:
        candidate = connection
    if candidate is None:
        return None
    if not _is_car_connected(candidate):
        _mark_connection_lost(
            candidate,
            "connection check",
            f"status changed to {_connection_status(candidate)}",
        )
        return None

    try:
        with _query_lock:
            return candidate.query(command, force=force)
    except Exception as error:
        _mark_connection_lost(candidate, command.name, error)
        return None


def _log_connection_diagnostics(candidate):
    firmware = "unknown"
    try:
        with _query_lock:
            response = candidate.query(obd.commands.ELM_VERSION)
        if not response.is_null():
            firmware = str(response.value)
    except Exception as error:
        print("Unable to read ELM327 firmware version:", str(error))

    print(
        "OBD connected:"
        f" port={candidate.port_name()},"
        f" ELM327={firmware},"
        f" protocol={candidate.protocol_id()} ({candidate.protocol_name()})"
    )


def connect_obd():
    """Connect to the adapter and ECU, with actionable failure diagnostics."""
    global connection
    with _connect_lock:
        if _is_car_connected():
            return connection

        with _connection_lock:
            stale_connection = connection
            connection = None
        if stale_connection is not None:
            _close_connection(stale_connection)

        port = _find_obd_port()
        detected_ports = [port] if port else obd.scan_serial()
        if not detected_ports:
            print(
                "No OBD-II adapter detected. Check USB power/cabling and the "
                "OBD_PORT setting."
            )
            with _connection_lock:
                connection = None
            _emit_disconnected()
            return None

        protocol = os.environ.get("OBD_PROTOCOL") or None
        # Some counterfeit ELM327 chips report bad/zero voltage even when the
        # connection is working; set OBD_SKIP_VOLTAGE_CHECK=1 if the adapter
        # fails to connect and logs a voltage-related error.
        check_voltage = os.environ.get("OBD_SKIP_VOLTAGE_CHECK") != "1"
        selected_port = port or detected_ports[0]
        print(
            f"{obd_status[2]} to {selected_port}"
            + (f" using protocol {protocol}" if protocol else " using auto protocol")
            + ("" if check_voltage else " (voltage check disabled)")
            + "..."
        )

        try:
            candidate = obd.OBD(portstr=selected_port, protocol=protocol, check_voltage=check_voltage)
        except Exception as error:
            print(
                f"OBD adapter detected at {selected_port}, but the ELM327 "
                f"handshake failed: {error}"
            )
            with _connection_lock:
                connection = None
            _emit_disconnected()
            return None

        candidate_status = _connection_status(candidate)
        if candidate_status == obd.OBDStatus.NOT_CONNECTED:
            print(
                f"OBD adapter detected at {selected_port}, but the ELM327 "
                "handshake failed."
            )
            _close_connection(candidate)
            candidate = None
        elif candidate_status != obd.OBDStatus.CAR_CONNECTED:
            print(
                f"ELM327 handshake succeeded at {selected_port}, but the car ECU "
                f"did not respond (status: {candidate_status}). Check ignition."
            )
            _close_connection(candidate)
            candidate = None

        with _connection_lock:
            connection = candidate

        if candidate is None:
            _emit_disconnected()
            return None

        # Reset stale sensor values from the previous ignition cycle so the
        # UI doesn't briefly display outdated readings before the first polls.
        global rpm, coolant_temp, fuel_level, throttle_position, mil_on, dtc_codes, odometer
        global mil_updated_at, dtc_updated_at, speed_updated_at
        rpm = None
        coolant_temp = None
        fuel_level = None
        throttle_position = None
        mil_on = None
        dtc_codes = []
        odometer = None
        mil_updated_at = None
        dtc_updated_at = None
        speed_updated_at = None

        _log_connection_diagnostics(candidate)
        app_events.socketio.emit("obd_status", obd_status[1])
        _start_polling_threads()
        return candidate


def get_speedtime():
    time_now = str(datetime.now().strftime("%H:%M:%S"))
    return [time_now, speed]


def get_fresh_speed(max_age=5):
    if speed_updated_at is None or time.monotonic() - speed_updated_at > max_age:
        return None
    return speed


def obd_connection():
    """Emit current state for a newly connected browser and ensure OBD is running."""
    start_obd()
    if _is_car_connected():
        app_events.socketio.emit("obd_status", obd_status[1])
        app_events.socketio.emit("obd_speed", get_speedtime())
        app_events.socketio.emit("obd_rpm", rpm if rpm is not None else "-")
    else:
        _emit_disconnected()


def get_fresh_mil(max_age=SENSOR_POLL_INTERVAL * 3):
    if mil_updated_at is None or time.monotonic() - mil_updated_at > max_age:
        return None
    return mil_on


def get_fresh_dtc_codes(max_age=SENSOR_POLL_INTERVAL * 3):
    if dtc_updated_at is None or time.monotonic() - dtc_updated_at > max_age:
        return []
    return list(dtc_codes)


def obd_speed():
    global speed, speed_updated_at
    with obd_speed_lock:
        while not stop_speed.is_set() and _is_car_connected():
            response = _query(obd.commands.SPEED)
            if response is None:
                break
            if response.is_null():
                print("No speed data received. #" + str(get_speedtime()))
                print(app_gps.get_gps())
            else:
                speed = round(response.value.to("kph").magnitude, 1)
                speed_updated_at = time.monotonic()
                print("Speed:", speed, "kph")

            app_events.socketio.emit("obd_speed", get_speedtime())
            stop_speed.wait(2)


def obd_rpm():
    global rpm
    with obd_rpm_lock:
        while not stop_rpm.is_set() and _is_car_connected():
            response = _query(obd.commands.RPM)
            if response is None:
                break
            if response.is_null():
                print("No RPM data received. #" + str(get_speedtime()))
                app_events.socketio.emit("obd_rpm", "-")
            else:
                rpm = round(response.value.magnitude)
                print("RPM:", rpm)
                app_events.socketio.emit("obd_rpm", rpm)

            stop_rpm.wait(1)


def _quantity_value(command, unit=None):
    response = _query(command)
    if response is None or response.is_null():
        return None
    value = response.value.to(unit) if unit else response.value
    return round(value.magnitude, 1)


def _poll_diagnostics():
    global coolant_temp, fuel_level, throttle_position
    global mil_on, dtc_codes, odometer, mil_updated_at, dtc_updated_at

    coolant_temp = _quantity_value(obd.commands.COOLANT_TEMP, "degC")
    fuel_level = _quantity_value(obd.commands.FUEL_LEVEL, "percent")
    throttle_position = _quantity_value(obd.commands.THROTTLE_POS, "percent")

    status_response = _query(obd.commands.STATUS)
    if status_response is not None and not status_response.is_null():
        mil_on = bool(status_response.value.MIL)
        mil_updated_at = time.monotonic()
        app_events.socketio.emit(
            "obd_mil",
            {"on": mil_on, "dtc_count": status_response.value.DTC_count},
        )

    dtc_response = _query(obd.commands.GET_DTC)
    if dtc_response is not None and not dtc_response.is_null():
        dtc_codes = [
            {"code": code, "description": description}
            for code, description in dtc_response.value
        ]
        dtc_updated_at = time.monotonic()
        app_events.socketio.emit("obd_dtc", dtc_codes)

    odometer_response = _query(ODOMETER_COMMAND, force=True)
    if odometer_response is not None and not odometer_response.is_null():
        odometer = round(odometer_response.value, 1)

    app_events.socketio.emit("obd_coolant_temp", coolant_temp)
    app_events.socketio.emit("obd_fuel_level", fuel_level)
    app_events.socketio.emit("obd_throttle", throttle_position)
    app_events.socketio.emit("obd_odometer", odometer)


def obd_sensors():
    with obd_sensor_lock:
        while not stop_sensors.is_set() and _is_car_connected():
            _poll_diagnostics()
            stop_sensors.wait(SENSOR_POLL_INTERVAL)


def _start_polling_threads():
    global thread_speed, thread_rpm, thread_sensors

    thread_specs = (
        ("speed", obd_speed, stop_speed),
        ("rpm", obd_rpm, stop_rpm),
        ("sensors", obd_sensors, stop_sensors),
    )
    threads = {
        "speed": thread_speed,
        "rpm": thread_rpm,
        "sensors": thread_sensors,
    }

    for name, target, stop_event in thread_specs:
        thread = threads[name]
        if thread is None or not thread.is_alive():
            stop_event.clear()
            thread = Thread(target=target, name=f"obd-{name}", daemon=True)
            thread.start()
            if name == "speed":
                thread_speed = thread
            elif name == "rpm":
                thread_rpm = thread
            else:
                thread_sensors = thread
            print(f"OBD {name} thread started, alive: {thread.is_alive()}")


def _reconnect_loop():
    while True:
        if _is_car_connected():
            _start_polling_threads()
        else:
            connect_obd()
        time.sleep(RECONNECT_INTERVAL)


def start_obd():
    """Start the browser-independent OBD reconnect loop once."""
    global _reconnect_thread
    if _reconnect_thread is None or not _reconnect_thread.is_alive():
        _reconnect_thread = Thread(
            target=_reconnect_loop,
            name="obd-reconnect",
            daemon=True,
        )
        _reconnect_thread.start()
        print(f"OBD reconnect thread started, alive: {_reconnect_thread.is_alive()}")
    return OBD()


def kill_obd(obd_type):
    if obd_type == "speed":
        stop_speed.set()
    elif obd_type == "rpm":
        stop_rpm.set()
    elif obd_type == "sensors":
        stop_sensors.set()
    elif obd_type == "all":
        stop_speed.set()
        stop_rpm.set()
        stop_sensors.set()