feat: raw ELM327 console — сырой слой, API, интерактивная консоль

obd/raw_console.py

@@ -0,0 +1,251 @@
+"""
+obd/raw_console.py — Сырой слой ELM327 (без стейт-машины).
+
+НИКАКОЙ логики протокола:
+  - Нет state machine (State)
+  - Нет классификации ответов (Rsp)
+  - Нет адаптивных таймингов (AdaptiveTiming)
+  - Нет ретраев
+  - Нет хендлеров ошибок
+
+ТОЛЬКО:
+  - send(cmd) → отправляет команду + CR
+  - read(timeout) → читает ВСЁ до '>' или таймаута, байт-за-байтом
+  - drain() → очищает входной буфер
+  - available() → сколько байт ждёт в буфере
+
+ПРЕДНАЗНАЧЕНИЕ:
+  Изучение реального поведения ELM327.
+  «Почему статика работает, а динамика ломается?»
+  Ответ — в сырых байтах.
+"""
+
+import logging
+import time
+
+logger = logging.getLogger("elm.raw")
+
+# ── Конфигурация по умолчанию (можно переопределить) ──
+DEFAULT_TIMEOUT = 500     # мс
+INTER_CMD_DELAY = 0.05    # с — пауза между командой и чтением
+
+
+class RawELM:
+    """Сырой слой ELM327 — только send/read/drain, без протокольной логики."""
+
+    def __init__(self, transport):
+        """
+        Args:
+            transport: объект с методами .write(str) и .read(timeout_ms) → str
+                       (обычно SerialTransport из obd.connection)
+        """
+        self._t = transport
+        self._timeout = DEFAULT_TIMEOUT
+        self._inter_delay = INTER_CMD_DELAY
+        self._log: list[dict] = []  # история команд
+
+    # ── Настройка ────────────────────────────────────
+
+    @property
+    def timeout(self) -> int:
+        return self._timeout
+
+    @timeout.setter
+    def timeout(self, ms: int):
+        self._timeout = ms
+
+    @property
+    def inter_delay(self) -> float:
+        return self._inter_delay
+
+    @inter_delay.setter
+    def inter_delay(self, sec: float):
+        self._inter_delay = sec
+
+    # ── Основные операции ────────────────────────────
+
+    def send(self, cmd: str, timeout: int | None = None) -> dict:
+        """Отправить команду и прочитать сырой ответ.
+
+        Args:
+            cmd: команда (без \r, добавится автоматически)
+            timeout: таймаут в мс (None = использовать self.timeout)
+
+        Returns:
+            {
+                "cmd": str,         # что отправили
+                "raw": str,         # сырой ответ (без '>')
+                "prompt": bool,     # получен ли '>'
+                "elapsed_ms": int,  # сколько мс заняло
+                "bytes": int,       # сколько байт в ответе
+                "error": str|None,  # ошибка если есть
+            }
+
+        Не бросает исключений — всегда возвращает dict с полем error.
+        """
+        tmo = timeout if timeout is not None else self._timeout
+        entry = {"cmd": cmd, "timeout_ms": tmo, "ts": time.time()}
+
+        try:
+            # 1. Отправить
+            self._t.write(cmd)
+
+            # 2. Пауза (ELM начинает отвечать не мгновенно)
+            if self._inter_delay > 0:
+                time.sleep(self._inter_delay)
+
+            # 3. Прочитать
+            t0 = time.monotonic()
+            raw, prompt, nbytes = self._read_raw(tmo)
+            elapsed = int((time.monotonic() - t0) * 1000)
+
+            entry.update({
+                "raw": raw,
+                "prompt": prompt,
+                "elapsed_ms": elapsed,
+                "bytes": nbytes,
+                "error": None,
+            })
+        except TimeoutError:
+            entry.update({
+                "raw": "",
+                "prompt": False,
+                "elapsed_ms": tmo,
+                "bytes": 0,
+                "error": f"timeout {tmo}ms",
+            })
+        except Exception as e:
+            entry.update({
+                "raw": "",
+                "prompt": False,
+                "elapsed_ms": 0,
+                "bytes": 0,
+                "error": str(e),
+            })
+
+        self._log.append(entry)
+        return entry
+
+    def drain(self) -> dict:
+        """Очистить входной буфер. Возвращает что было выброшено.
+
+        Returns:
+            {"drained": str, "bytes": int}
+        """
+        t0 = time.monotonic()
+        drained = []
+        total = 0
+        dl = t0 + 0.5  # 500 мс максимум на дренаж
+        while time.monotonic() < dl:
+            try:
+                ch = self._read_byte(0.05)
+                if ch is not None:
+                    drained.append(chr(ch))
+                    total += 1
+                else:
+                    break  # буфер пуст
+            except Exception:
+                break
+        elapsed = int((time.monotonic() - t0) * 1000)
+        result = {"drained": "".join(drained), "bytes": total, "elapsed_ms": elapsed}
+        if total > 0:
+            logger.info(f"RawELM: drained {total} bytes: {result['drained']!r}")
+        return result
+
+    def available(self) -> int:
+        """Сколько байт ждёт во входном буфере (0 = пусто)."""
+        try:
+            return self._t._ser.in_waiting
+        except Exception:
+            return -1
+
+    # ── Лог ──────────────────────────────────────────
+
+    @property
+    def log(self) -> list[dict]:
+        """История всех команд."""
+        return self._log
+
+    def clear_log(self):
+        """Очистить историю."""
+        self._log.clear()
+
+    def last(self) -> dict | None:
+        """Последняя команда."""
+        return self._log[-1] if self._log else None
+
+    # ── Приватные ────────────────────────────────────
+
+    def _read_raw(self, timeout_ms: int) -> tuple[str, bool, int]:
+        """Читает байт-за-байтом до '>' или таймаута.
+
+        Returns:
+            (raw_text, got_prompt, byte_count)
+        """
+        dl = time.monotonic() + timeout_ms / 1000.0
+        lines, cur = [], []
+        got_prompt = False
+
+        while time.monotonic() < dl:
+            ch = self._read_byte(0.05)
+            if ch is None:
+                continue
+            cp = ch
+
+            if cp == 62:          # '>' — промпт ELM327
+                if cur:
+                    lines.append("".join(cur))
+                    cur.clear()
+                got_prompt = True
+                break
+            elif cp == 13:        # CR — конец строки
+                if cur:
+                    lines.append("".join(cur))
+                    cur.clear()
+            elif cp in (10, 32):  # LF и пробел — игнорируем
+                pass
+            else:
+                cur.append(chr(cp))
+
+        if cur:
+            lines.append("".join(cur))
+
+        return ("\n".join(lines), got_prompt, sum(len(s) for s in lines))
+
+    def _read_byte(self, timeout_s: float) -> int | None:
+        """Прочитать один байт с таймаутом. None = таймаут/нет данных."""
+        import serial
+        try:
+            if self._t._ser.in_waiting > 0:
+                b = self._t._ser.read(1)
+                return b[0] if b else None
+            else:
+                time.sleep(0.001)  # поллинг 1мс
+                return None
+        except serial.SerialException:
+            return None
+
+
+# ── Хелпер ───────────────────────────────────────────
+
+def format_response(entry: dict) -> str:
+    """Форматирует ответ RawELM.send() для вывода в консоль."""
+    lines = [
+        f"→ {entry['cmd']}",
+        f"← {entry['raw']!r}" if entry["raw"] else "← (пусто)",
+    ]
+    if entry["prompt"]:
+        lines.append("   prompt: ✅ >")
+    else:
+        lines.append("   prompt: ❌")
+    lines.append(f"   time: {entry['elapsed_ms']}ms, bytes: {entry['bytes']}")
+    if entry["error"]:
+        lines.append(f"   ⚠️  {entry['error']}")
+    return "\n".join(lines)
+
+
+def format_log(entries: list[dict]) -> str:
+    """Форматирует всю историю команд."""
+    return "\n" + "─" * 50 + "\n" + \
+           "\n".join(format_response(e) for e in entries) + \
+           "\n" + "─" * 50