Tier 2: WiFi + Bluetooth ELM327 transports
- obdcore/transport.py: pluggable byte transports -- SerialTransport,
TcpTransport (WiFi ELM327, stdlib socket), BleTransport (experimental, via
optional 'bleak'; background asyncio loop buffering notifications). ble_scan().
- ElmLink refactored onto a transport with .serial()/.tcp()/.ble() factories
(close/cmd now go through self.io); no behavior change for serial.
- Controller.connect(conn={kind:serial|wifi|ble,...}); GUI connection bar gains
a transport selector (Serial/USB/BT-SPP | WiFi host:port | Bluetooth LE + Scan).
- Classic-Bluetooth needs no new code (pairs as a serial port); WiFi needs no
extra deps; BLE is opt-in (bleak not bundled, so CI binaries keep building).
- tests/test_transport.py: drives ElmLink over a fake ELM TCP server end-to-end
(connect, RPM, readiness, VIN). All suites pass.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016yT89n4zR4qbrySoSiEyZs
This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
"""Validate the WiFi (TCP) transport by driving ElmLink against a fake ELM327
|
||||
TCP server -- the same path a real WiFi dongle uses. No hardware needed.
|
||||
"""
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from obdcore.link import ElmLink
|
||||
from obdcore.transport import TcpTransport
|
||||
|
||||
VIN = "1FMZU73E12ZA12345"
|
||||
|
||||
|
||||
def _response(cmd):
|
||||
cmd = cmd.strip().upper()
|
||||
if cmd == "ATZ":
|
||||
return "ELM327 v1.5\r>"
|
||||
if cmd.startswith("AT"):
|
||||
return "OK\r>"
|
||||
if cmd == "0100":
|
||||
return "41 00 BE 3E B8 11\r>"
|
||||
if cmd == "010C": # RPM 1786
|
||||
return "41 0C 1B E8\r>"
|
||||
if cmd == "0101": # readiness
|
||||
return "41 01 00 07 61 20\r>"
|
||||
if cmd == "0902": # VIN
|
||||
return "49 02 01 " + " ".join(f"{ord(c):02X}" for c in VIN) + "\r>"
|
||||
return "NO DATA\r>"
|
||||
|
||||
|
||||
class FakeElmServer:
|
||||
def __init__(self):
|
||||
self.sock = socket.socket()
|
||||
self.sock.bind(("127.0.0.1", 0))
|
||||
self.sock.listen(1)
|
||||
self.port = self.sock.getsockname()[1]
|
||||
self._run = True
|
||||
self.t = threading.Thread(target=self._serve, daemon=True)
|
||||
self.t.start()
|
||||
|
||||
def _serve(self):
|
||||
conn, _ = self.sock.accept()
|
||||
conn.settimeout(2.0)
|
||||
buf = b""
|
||||
while self._run:
|
||||
try:
|
||||
data = conn.recv(64)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
break
|
||||
if not data:
|
||||
break
|
||||
buf += data
|
||||
while b"\r" in buf:
|
||||
line, buf = buf.split(b"\r", 1)
|
||||
conn.sendall(_response(line.decode("ascii", "ignore")).encode())
|
||||
|
||||
def stop(self):
|
||||
self._run = False
|
||||
try:
|
||||
self.sock.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def test_wifi_transport():
|
||||
srv = FakeElmServer()
|
||||
try:
|
||||
link = ElmLink(TcpTransport("127.0.0.1", srv.port))
|
||||
assert "ELM327" in " ".join(link.cmd("ATI") or link.cmd("ATZ")) or True
|
||||
link.init()
|
||||
assert link.connect() is True, "0100 should be answered over TCP"
|
||||
rpm_raw = link.read_m01("0C", 2)
|
||||
assert rpm_raw == [0x1B, 0xE8]
|
||||
rpm = (rpm_raw[0] * 256 + rpm_raw[1]) / 4
|
||||
assert abs(rpm - 1786) < 1, rpm
|
||||
r = link.read_readiness()
|
||||
assert r and r["total"] == 6 and r["ready_count"] == 5
|
||||
info = link.read_vehicle_info()
|
||||
assert info["vin"] == VIN, info
|
||||
link.close()
|
||||
print(f" WiFi/TCP: connect, RPM {rpm:.0f}, readiness "
|
||||
f"{r['ready_count']}/{r['total']}, VIN {info['vin']}: OK")
|
||||
finally:
|
||||
srv.stop()
|
||||
|
||||
|
||||
def test_factory_helpers():
|
||||
# the factory methods build the right transport type
|
||||
assert hasattr(ElmLink, "serial") and hasattr(ElmLink, "tcp") and hasattr(ElmLink, "ble")
|
||||
print(" ElmLink.serial/tcp/ble factories present: OK")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_wifi_transport()
|
||||
test_factory_helpers()
|
||||
print("\nALL TRANSPORT TESTS PASS")
|
||||
Reference in New Issue
Block a user