obdash/obdcore/transport.py

"""Byte transports for ElmLink: serial (USB / classic-Bluetooth SPP), TCP
(WiFi ELM327), and BLE (Bluetooth Low Energy ELM327).

Each transport presents the same tiny synchronous interface so ElmLink's
command loop doesn't care how the bytes move:

    write(data: bytes) -> None
    read(n: int)       -> bytes   (returns b"" on timeout)
    reset_input_buffer() -> None
    close()            -> None

Classic Bluetooth ELM327 pair as a serial port (COMx / /dev/cu.* / rfcomm), so
they use SerialTransport. Only WiFi (TCP) and BLE need dedicated transports.
"""
import socket
import threading
import time


class SerialTransport:
    def __init__(self, port, baud=38400):
        import serial
        self.ser = serial.Serial(port, baud, timeout=0.2)

    def write(self, data):
        self.ser.write(data)

    def read(self, n):
        return self.ser.read(n)

    def reset_input_buffer(self):
        self.ser.reset_input_buffer()

    def close(self):
        try:
            self.ser.close()
        except Exception:
            pass


class TcpTransport:
    """WiFi ELM327 — a raw TCP socket (commonly 192.168.0.10:35000)."""

    def __init__(self, host, port=35000, connect_timeout=5.0, read_timeout=0.2):
        self._rt = read_timeout
        self.sock = socket.create_connection((host, int(port)), timeout=connect_timeout)
        self.sock.settimeout(read_timeout)

    def write(self, data):
        self.sock.sendall(data)

    def read(self, n):
        try:
            data = self.sock.recv(n)
        except socket.timeout:
            return b""                        # no data yet -- normal
        except OSError as e:
            raise IOError(f"WiFi connection lost: {e}") from e
        if data == b"":                       # peer closed the connection
            raise IOError("WiFi connection closed by adapter")
        return data

    def reset_input_buffer(self):
        # drain pending bytes, but never spin forever if data keeps arriving
        self.sock.settimeout(0.02)
        try:
            for _ in range(64):
                if not self.sock.recv(4096):
                    break
        except socket.timeout:
            pass
        except OSError:
            pass
        finally:
            try:
                self.sock.settimeout(self._rt)
            except OSError:
                pass

    def close(self):
        try:
            self.sock.close()
        except Exception:
            pass


# Common BLE UART characteristic pairs used by cheap ELM327 dongles
# (write-char, notify-char). We try each until one has both properties.
_BLE_UART_PAIRS = [
    ("0000fff2-0000-1000-8000-00805f9b34fb", "0000fff1-0000-1000-8000-00805f9b34fb"),
    ("0000ffe1-0000-1000-8000-00805f9b34fb", "0000ffe1-0000-1000-8000-00805f9b34fb"),
    ("6e400002-b5a3-f393-e0a9-e50e24dcca9e", "6e400003-b5a3-f393-e0a9-e50e24dcca9e"),
]


class BleTransport:
    """Experimental BLE ELM327 transport (needs `bleak`). Runs an asyncio loop
    on a background thread and buffers notifications so read()/write() stay
    synchronous. Untested against every dongle -- BLE ELM327 GATT layouts vary."""

    def __init__(self, address, connect_timeout=15.0):
        try:
            import bleak  # noqa: F401
        except ImportError as e:
            raise RuntimeError("BLE support needs the 'bleak' package "
                               "(pip install bleak)") from e
        self.address = address
        self._buf = bytearray()
        self._lock = threading.Lock()
        self._loop = None
        self._client = None
        self._write_char = None
        self._ready = threading.Event()
        self._err = None
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()
        if not self._ready.wait(connect_timeout) or self._err:
            err = self._err or "timeout"
            self.close()                      # don't leak the client + event-loop thread
            raise RuntimeError(f"BLE connect failed: {err}")

    def _run(self):
        import asyncio
        from bleak import BleakClient
        self._loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._loop)

        async def setup():
            self._client = BleakClient(self.address)
            await self._client.connect()
            notify_char = None
            for svc in self._client.services:
                for ch in svc.characteristics:
                    props = set(ch.properties)
                    if {"write", "write-without-response"} & props and self._write_char is None:
                        self._write_char = ch.uuid
                    if {"notify", "indicate"} & props and notify_char is None:
                        notify_char = ch.uuid
            # prefer a known UART pair if present
            for w, n in _BLE_UART_PAIRS:
                if any(c.uuid == w for s in self._client.services for c in s.characteristics):
                    self._write_char = w
                    notify_char = n
                    break
            if not self._write_char or not notify_char:
                raise RuntimeError("no writable/notify characteristic found")

            def on_notify(_h, data):
                with self._lock:
                    self._buf += bytes(data)
                    if len(self._buf) > 65536:        # cap: never grow unbounded
                        del self._buf[:-65536]

            await self._client.start_notify(notify_char, on_notify)
            self._ready.set()

        try:
            self._loop.run_until_complete(setup())
            self._loop.run_forever()
        except Exception as e:
            self._err = e
            self._ready.set()

    def write(self, data):
        import asyncio
        fut = asyncio.run_coroutine_threadsafe(
            self._client.write_gatt_char(self._write_char, data, response=False), self._loop)
        fut.result(timeout=3.0)

    def read(self, n):
        deadline = time.time() + 0.2
        while time.time() < deadline:
            with self._lock:
                if self._buf:
                    out = bytes(self._buf[:n]); del self._buf[:n]
                    return out
            time.sleep(0.005)
        return b""

    def reset_input_buffer(self):
        with self._lock:
            self._buf.clear()

    def close(self):
        import asyncio
        loop, client = self._loop, self._client
        if loop is None:
            return
        if client is not None:
            try:
                asyncio.run_coroutine_threadsafe(client.disconnect(), loop).result(timeout=3.0)
            except Exception:
                pass
        try:                                  # stop the loop even if disconnect failed,
            loop.call_soon_threadsafe(loop.stop)   # so the background thread exits
        except Exception:
            pass


def ble_scan(timeout=6.0):
    """Return [(address, name), ...] of nearby BLE devices (needs bleak)."""
    import asyncio
    from bleak import BleakScanner
    devs = asyncio.run(BleakScanner.discover(timeout=timeout))
    return [(d.address, d.name or "?") for d in devs]