209 lines
8.7 KiB
Python
209 lines
8.7 KiB
Python
"""Связь с ELM327 через Bluetooth SPP (pyserial)."""
|
||
|
||
import re
|
||
import time
|
||
|
||
import serial
|
||
|
||
|
||
class ELM327:
|
||
"""Работа с ELM327 по последовательному порту (Bluetooth SPP)."""
|
||
|
||
# Стандартные PID'ы для чтения
|
||
DEFAULT_PIDS = {
|
||
"0105": ("coolant_temp", "°C"), # температура ОЖ
|
||
"010C": ("rpm", "об/мин"), # обороты двигателя
|
||
"010D": ("speed", "км/ч"), # скорость
|
||
"0111": ("throttle_pos", "%"), # положение дросселя
|
||
"010B": ("map", "кПа"), # давление впуска (MAP)
|
||
"010F": ("iat", "°C"), # температура впуска
|
||
"011F": ("runtime_since_start", "с"), # время с запуска
|
||
"0104": ("engine_load", "%"), # нагрузка двигателя
|
||
"0106": ("stft_b1", "%"), # краткосрочный fuel trim bank 1
|
||
"0107": ("ltft_b1", "%"), # долгосрочный fuel trim bank 1
|
||
}
|
||
|
||
def __init__(self, port: str, baudrate: int = 38400, timeout: float = 5.0):
|
||
self.port = port
|
||
self.ser = serial.Serial(
|
||
port=port,
|
||
baudrate=baudrate,
|
||
timeout=timeout,
|
||
bytesize=serial.EIGHTBITS,
|
||
parity=serial.PARITY_NONE,
|
||
stopbits=serial.STOPBITS_ONE,
|
||
)
|
||
|
||
# ── низкоуровневые команды ────────────────────────────
|
||
|
||
def _cmd(self, cmd: str, wait: float = 0.2) -> str:
|
||
"""Отправляет AT/OBD команду, возвращает сырой ответ."""
|
||
self.ser.reset_input_buffer()
|
||
self.ser.write((cmd + "\r").encode())
|
||
time.sleep(wait)
|
||
lines = []
|
||
while True:
|
||
line = self.ser.readline().decode("utf-8", errors="ignore").strip()
|
||
if not line or line == ">":
|
||
break
|
||
lines.append(line)
|
||
return "\n".join(lines)
|
||
|
||
# ── инициализация ─────────────────────────────────────
|
||
|
||
def init(self) -> bool:
|
||
"""Сброс и настройка ELM327. Возвращает True если OK."""
|
||
resp = self._cmd("ATZ", wait=1.0) # сброс
|
||
if "ELM" not in resp:
|
||
return False
|
||
self._cmd("ATE0") # выкл эхо
|
||
self._cmd("ATL0") # выкл перевод строки
|
||
self._cmd("ATSP0") # авто-протокол
|
||
self._cmd("ATH1") # вкл заголовки
|
||
return True
|
||
|
||
# ── чтение VIN ────────────────────────────────────────
|
||
|
||
def read_vin(self) -> str | None:
|
||
"""Читает VIN (режим 09 PID 02). Возвращает VIN или None."""
|
||
resp = self._cmd("0902", wait=1.5)
|
||
# Формат: 014 0: 49 02 01 57 56 57 ...
|
||
# Ищем строку с байтами после 49 02
|
||
match = re.search(r"49\s*02\s*(.+)", resp.replace("\n", " ").replace(":", ""))
|
||
if not match:
|
||
return None
|
||
# Собираем HEX байты, переводим в ASCII
|
||
hex_bytes = match.group(1).strip().split()
|
||
vin = ""
|
||
for h in hex_bytes:
|
||
h = h.strip()
|
||
if len(h) == 2:
|
||
try:
|
||
vin += chr(int(h, 16))
|
||
except ValueError:
|
||
pass
|
||
return vin if len(vin) == 17 else None
|
||
|
||
# ── чтение ошибок ─────────────────────────────────────
|
||
|
||
def read_dtc_codes(self, mode: str = "03") -> list[dict]:
|
||
"""Читает коды ошибок.
|
||
mode: '03' — сохранённые, '07' — ожидающие.
|
||
Возвращает [{"code": "P0301", "description": "", "status": "stored"}, ...].
|
||
"""
|
||
resp = self._cmd(mode, wait=1.0)
|
||
codes = []
|
||
|
||
# Пример ответа: 43 01 33 00 00 00 00
|
||
for line in resp.split("\n"):
|
||
line = line.strip()
|
||
if not line or "NO DATA" in line.upper():
|
||
continue
|
||
|
||
# Ищем HEX-байты после 43 (mode 03 response) или 47 (mode 07)
|
||
match = re.search(r"4[37]\s*(.+)", line.replace(":", ""))
|
||
if not match:
|
||
continue
|
||
|
||
hex_bytes = match.group(1).strip().split()
|
||
# Парсим по 2 байта на код (первые два байта — количество кодов)
|
||
i = 1 # пропускаем байт количества
|
||
while i + 1 < len(hex_bytes):
|
||
dtc_raw = _decode_dtc(hex_bytes[i], hex_bytes[i + 1])
|
||
if dtc_raw and dtc_raw != "P0000":
|
||
codes.append({
|
||
"code": dtc_raw,
|
||
"description": "",
|
||
"status": "stored" if mode == "03" else "pending",
|
||
})
|
||
i += 2
|
||
|
||
return codes
|
||
|
||
# ── чтение параметров ─────────────────────────────────
|
||
|
||
def read_pid(self, pid: str) -> float | None:
|
||
"""Читает один PID, возвращает числовое значение или None."""
|
||
resp = self._cmd(pid, wait=0.3)
|
||
# Ищем строку ответа: 41 XX YY ZZ ...
|
||
match = re.search(r"4[12]\s*" + pid[2:4] + r"\s*(.+)", resp.replace(":", ""))
|
||
if not match:
|
||
return None
|
||
|
||
hex_bytes = match.group(1).strip().split()
|
||
if not hex_bytes:
|
||
return None
|
||
|
||
# Формулы для стандартных PID (SAE J1979)
|
||
formulas = {
|
||
"05": lambda b: int(b[0], 16) - 40, # coolant °C
|
||
"0C": lambda b: (int(b[0], 16) * 256 + int(b[1], 16)) / 4, # RPM
|
||
"0D": lambda b: int(b[0], 16), # speed km/h
|
||
"11": lambda b: int(b[0], 16) * 100 / 255, # throttle %
|
||
"0B": lambda b: int(b[0], 16), # MAP kPa
|
||
"0F": lambda b: int(b[0], 16) - 40, # IAT °C
|
||
"1F": lambda b: int(b[0], 16) * 256 + int(b[1], 16), # runtime sec
|
||
"04": lambda b: int(b[0], 16) * 100 / 255, # load %
|
||
"06": lambda b: (int(b[0], 16) - 128) * 100 / 128, # STFT %
|
||
"07": lambda b: (int(b[0], 16) - 128) * 100 / 128, # LTFT %
|
||
}
|
||
|
||
pid_short = pid[2:4]
|
||
if pid_short in formulas:
|
||
try:
|
||
return round(formulas[pid_short](hex_bytes), 1)
|
||
except (ValueError, IndexError):
|
||
return None
|
||
|
||
# Generic: первый байт как raw
|
||
try:
|
||
return int(hex_bytes[0], 16)
|
||
except (ValueError, IndexError):
|
||
return None
|
||
|
||
def read_all_pids(self, pids: dict | None = None) -> list[dict]:
|
||
"""Читает все PID'ы из словаря {pid: (name, unit)}.
|
||
Возвращает [{"pid_code": "...", "name": "...", "value": ..., "unit": "..."}, ...].
|
||
"""
|
||
if pids is None:
|
||
pids = self.DEFAULT_PIDS
|
||
|
||
results = []
|
||
for pid_code, (name, unit) in pids.items():
|
||
try:
|
||
value = self.read_pid(pid_code)
|
||
if value is not None:
|
||
results.append({
|
||
"pid_code": pid_code,
|
||
"name": name,
|
||
"value": value,
|
||
"unit": unit,
|
||
})
|
||
except Exception:
|
||
continue
|
||
return results
|
||
|
||
def close(self):
|
||
self._cmd("ATZ", wait=0.5)
|
||
self.ser.close()
|
||
|
||
|
||
def _decode_dtc(b1: str, b2: str) -> str | None:
|
||
"""Декодирует два HEX-байта в код ошибки вида P0301."""
|
||
try:
|
||
a, b = int(b1, 16), int(b2, 16)
|
||
except ValueError:
|
||
return None
|
||
|
||
# Первые 2 бита первого байта — тип:
|
||
types = {0: "P", 1: "C", 2: "B", 3: "U"}
|
||
prefix = types.get(a >> 6, "?")
|
||
|
||
# Оставшиеся биты
|
||
d1 = str((a >> 4) & 0x03) # вторая цифра
|
||
d2 = str(a & 0x0F) # третья цифра
|
||
d3 = f"{(b >> 4) & 0x0F:X}" # четвёртая цифра (hex!)
|
||
d4 = f"{b & 0x0F:X}" # пятая цифра (hex!)
|
||
|
||
return f"{prefix}{d1}{d2}{d3}{d4}"
|