#!/usr/bin/env python3 """ elm_console.py — Интерактивная консоль ELM327 (сырой режим). НИКАКОЙ автоматики: - Нет init(), probe(), send() со стейт-машиной - Нет классификации ответов - Нет адаптивных таймингов ТОЛЬКО вы вводите команду — ELM отвечает сырыми байтами. ЗАПУСК: python tools/elm_console.py # порт по умолчанию /dev/rfcomm0 python tools/elm_console.py --port /dev/rfcomm0 # явно указать порт python tools/elm_console.py --baud 38400 # другая скорость python tools/elm_console.py --timeout 1000 # таймаут 1с python tools/elm_console.py --no-init # не слать AT-инит КОМАНДЫ КОНСОЛИ: ATZ — отправить "ATZ" в ELM 0105 — отправить "0105" (PID coolant temp) !drain — очистить входной буфер !timeout 2000 — установить таймаут 2000 мс !delay 0.5 — пауза между командой и чтением (сек) !log — показать историю команд !available — сколько байт в буфере !save file.json — сохранить лог в файл !help — справка !quit — выход ЦЕЛЬ: Понять, КАК на самом деле работает ELM327. Почему статика работает, а динамика ломается? Ответ — в сырых байтах. """ import argparse import cmd import json import logging import sys import time from pathlib import Path # Добавляем корень проекта в PYTHONPATH sys.path.insert(0, str(Path(__file__).parent.parent)) from obd.connection import SerialTransport from obd.raw_console import RawELM, format_response, format_log logging.basicConfig( level=logging.DEBUG, format="%(asctime)s [%(name)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("elm.console") class ElmConsole(cmd.Cmd): """Интерактивная консоль ELM327.""" intro = """ ╔══════════════════════════════════════════════════════╗ ║ ELM327 Raw Console ║ ║ Сырое взаимодействие — без стейт-машины ║ ║ Команды: ATZ, 0105, !drain, !help, !quit ║ ╚══════════════════════════════════════════════════════╝ """ prompt = "\nelm> " def __init__(self, port: str, baudrate: int, timeout: int, delay: float, no_init: bool): super().__init__() self._port = port self._baud = baudrate self._no_init = no_init self._transport = None self._elm: RawELM | None = None # ── Подключение ───────────────────────────────── def connect(self): """Открыть порт и создать RawELM.""" print(f"🔌 Подключение к {self._port} @ {self._baud}...") try: self._transport = SerialTransport(self._port, self._baud) self._transport.connect() except Exception as e: print(f"❌ Не удалось открыть порт: {e}") print(" Проверь: bash scripts/setup-bt.sh") return False self._elm = RawELM(self._transport) print(f"✅ Порт открыт. RawELM готов.") print(f" Таймаут: {self._elm.timeout}ms, пауза: {self._elm.inter_delay}s") print(f" Буфер: {self._elm.available()} байт") if not self._no_init: print("\n📡 Быстрая проверка связи (ATZ)...") r = self._elm.send("ATZ", timeout=3000) print(format_response(r)) if r["error"]: print("⚠️ ELM327 не ответил на ATZ. Проверь питание адаптера.") print(" Продолжаем, но команды могут не работать.") return True def close(self): """Закрыть порт.""" if self._transport: self._transport.close() print("🔌 Порт закрыт.") # ── cmd.Cmd overrides ──────────────────────────── def default(self, line: str): """Любая не-! команда = отправить в ELM327.""" cmd_str = line.strip() if not cmd_str: return if cmd_str.startswith("!"): print(f"Неизвестная команда: {cmd_str}. !help для списка.") return # Отправить в ELM result = self._elm.send(cmd_str) print(format_response(result)) def emptyline(self): """Пустая строка — ничего не делаем.""" pass # ── Специальные команды (!) ────────────────────── def do_drain(self, arg): """!drain — очистить входной буфер ELM327.""" r = self._elm.drain() if r["bytes"] > 0: print(f"🗑 Выброшено {r['bytes']} байт: {r['drained']!r}") else: print("✅ Буфер пуст.") def do_timeout(self, arg): """!timeout — установить таймаут чтения.""" try: ms = int(arg.strip()) self._elm.timeout = ms print(f"⏱ Таймаут: {ms}ms") except ValueError: print(f"❌ Нужно число: !timeout 2000") def do_delay(self, arg): """!delay — пауза между командой и чтением.""" try: sec = float(arg.strip()) self._elm.inter_delay = sec print(f"⏱ Пауза: {sec}s") except ValueError: print(f"❌ Нужно число: !delay 0.5") def do_log(self, arg): """!log [N] — показать последние N команд (по умолчанию все).""" entries = self._elm.log if not entries: print("📭 Лог пуст.") return try: n = int(arg.strip()) if arg.strip() else len(entries) except ValueError: n = len(entries) to_show = entries[-n:] if n < len(entries) else entries print(format_log(to_show)) print(f"Всего: {len(entries)} команд.") def do_available(self, arg): """!available — сколько байт в буфере.""" n = self._elm.available() if n < 0: print("⚠️ Не удалось проверить буфер (порт закрыт?).") elif n == 0: print("✅ Буфер пуст.") else: print(f"📥 В буфере: {n} байт.") def do_save(self, arg): """!save — сохранить лог в JSON.""" path = arg.strip() if not path: print("❌ Укажи имя файла: !save log.json") return try: with open(path, "w") as f: json.dump(self._elm.log, f, indent=2, default=str) print(f"💾 Сохранено: {path} ({len(self._elm.log)} команд)") except Exception as e: print(f"❌ Ошибка: {e}") def do_raw(self, arg): """!raw — показать последний ответ в repr (все символы).""" last = self._elm.last() if not last: print("📭 Нет команд.") return print(f"raw = {last['raw']!r}") print(f"prompt = {last['prompt']}") print(f"elapsed = {last['elapsed_ms']}ms") print(f"bytes = {last['bytes']}") def do_help(self, arg): """!help — справка.""" print(""" ╔══════════════════════════════════════════════════════╗ ║ КОМАНДЫ ELM327 (вводи как есть): ║ ║ ATZ — сброс ║ ║ ATI — идентификация ║ ║ ATE0 — echo off ║ ║ ATL0 — linefeeds off ║ ║ ATS0 — spaces off ║ ║ ATH1 — headers on ║ ║ ATSP0 — авто-протокол ║ ║ ATRV — напряжение ║ ║ ATDPN — номер протокола ║ ║ 0105 — температура ОЖ (PID) ║ ║ 010C — обороты ║ ║ 010D — скорость ║ ║ 03 — сохранённые DTC ║ ║ 07 — pending DTC ║ ║ 0902 — VIN ║ ║ ║ ║ КОМАНДЫ КОНСОЛИ (с !): ║ ║ !drain — очистить буфер ║ ║ !timeout N — таймаут (ms) ║ ║ !delay N — пауза перед чтением (s) ║ ║ !log [N] — история команд ║ ║ !raw — последний ответ в repr ║ ║ !available — байт в буфере ║ ║ !save f.json — сохранить лог ║ ║ !help — эта справка ║ ║ !quit — выход ║ ╚══════════════════════════════════════════════════════╝ """) def do_quit(self, arg): """!quit — выход.""" print("👋") self.close() return True def do_exit(self, arg): """!exit — то же что !quit.""" return self.do_quit(arg) # Сокращения do_q = do_quit do_h = do_help do_d = do_drain do_t = do_timeout do_l = do_log do_a = do_available # ── main ───────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="ELM327 Raw Console — интерактивное сырое взаимодействие" ) parser.add_argument("--port", default="/dev/rfcomm0", help="Порт (default: /dev/rfcomm0)") parser.add_argument("--baud", type=int, default=38400, help="Скорость (default: 38400)") parser.add_argument("--timeout", type=int, default=500, help="Таймаут чтения ms (default: 500)") parser.add_argument("--delay", type=float, default=0.05, help="Пауза перед чтением s (default: 0.05)") parser.add_argument("--no-init", action="store_true", help="Не слать ATZ при старте") args = parser.parse_args() console = ElmConsole( port=args.port, baudrate=args.baud, timeout=args.timeout, delay=args.delay, no_init=args.no_init, ) if not console.connect(): sys.exit(1) try: console.cmdloop() except KeyboardInterrupt: print("\n👋") finally: console.close() if __name__ == "__main__": main()