299 lines
12 KiB
Python
Executable File
299 lines
12 KiB
Python
Executable File
#!/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 <ms> — установить таймаут чтения."""
|
||
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 <sec> — пауза между командой и чтением."""
|
||
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 <file.json> — сохранить лог в 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()
|