update

README.md

@@ -1,2 +1,69 @@
-# bus-arduino
+# Bus LCD for Arduino
 
+Этот набор состоит из двух частей:
+
+- `bus_monitor.py` получает координаты автобусов `94` и `96` с `https://rnd.don.su/get_bus`, считает примерное время до заданных точек и формирует две строки по 16 символов.
+- `arduino_bus_display/arduino_bus_display.ino` принимает эти строки по `Serial` и выводит их на LCD 16x2 через `LiquidCrystal`.
+
+Используются точки:
+
+- `94`: `47.291609, 39.694856`
+- `96`: `47.220350, 39.632114`
+
+На экране выводится короткая ASCII-строка, например:
+
+```text
+94 #8 09m 2.4km
+96 #6 03m 180m
+```
+
+`#8` это сколько автобусов скрипт видит на маршруте в текущем ответе API. `09m` это примерный ETA, `2.4km` или `180m` это расстояние по прямой до нужной точки. Это не маршрутное время, а оценка по текущему положению и скорости автобуса.
+
+## Подключение LCD
+
+Прошивка использует ваш вариант подключения:
+
+- `RS -> D6`
+- `EN -> D7`
+- `DB4 -> D8`
+- `DB5 -> D9`
+- `DB6 -> D10`
+- `DB7 -> D11`
+
+## Что установить на компьютере
+
+```bash
+python3 -m pip install -r requirements.txt
+```
+
+## Как загрузить прошивку
+
+Откройте файл `arduino_bus_display/arduino_bus_display.ino` в Arduino IDE и загрузите его в плату.
+
+Скорость `Serial`: `115200`.
+
+## Как запустить скрипт
+
+Если хотите сначала просто посмотреть вывод в консоль:
+
+```bash
+python3 bus_monitor.py --once
+```
+
+Если Arduino подключена по USB:
+
+```bash
+python3 bus_monitor.py --serial-port /dev/ttyACM0
+```
+
+Полезные параметры:
+
+- `--interval 5` или другое значение, если нужно изменить частоту обновления
+- по умолчанию обновление идет каждые `5` секунд
+- `--baud-rate 115200` скорость `Serial`
+- `--once` один запрос без бесконечного цикла
+
+## Примечания
+
+- Скрипт сам получает свежие `cookie` и `csrf-token` с главной страницы `rnd.don.su`, поэтому не использует одноразовые токены из браузера.
+- Если стандартный LCD 1602 плохо показывает кириллицу, это нормально. Поэтому строки на экране оставлены в ASCII.

arduino_bus_display/arduino_bus_display.ino

@@ -0,0 +1,89 @@
+#include <LiquidCrystal.h>
+#include <stdint.h>
+#include <string.h>
+
+constexpr uint8_t PIN_RS = 6;
+constexpr uint8_t PIN_EN = 7;
+constexpr uint8_t PIN_DB4 = 8;
+constexpr uint8_t PIN_DB5 = 9;
+constexpr uint8_t PIN_DB6 = 10;
+constexpr uint8_t PIN_DB7 = 11;
+
+constexpr unsigned long SERIAL_BAUD = 115200;
+constexpr size_t LCD_COLUMNS = 16;
+constexpr size_t LCD_ROWS = 2;
+constexpr size_t INPUT_BUFFER_SIZE = 40;
+
+LiquidCrystal lcd(PIN_RS, PIN_EN, PIN_DB4, PIN_DB5, PIN_DB6, PIN_DB7);
+
+char line1[LCD_COLUMNS + 1] = "Booting...      ";
+char line2[LCD_COLUMNS + 1] = "Waiting data... ";
+char inputBuffer[INPUT_BUFFER_SIZE];
+size_t inputPos = 0;
+
+void copyLine(char* destination, const char* source) {
+  for (size_t i = 0; i < LCD_COLUMNS; ++i) {
+    if (source[i] == '\0') {
+      for (size_t j = i; j < LCD_COLUMNS; ++j) {
+        destination[j] = ' ';
+      }
+      destination[LCD_COLUMNS] = '\0';
+      return;
+    }
+    destination[i] = source[i];
+  }
+  destination[LCD_COLUMNS] = '\0';
+}
+
+void renderDisplay() {
+  lcd.setCursor(0, 0);
+  lcd.print(line1);
+  lcd.setCursor(0, 1);
+  lcd.print(line2);
+}
+
+void handleMessage(const char* message) {
+  if (strncmp(message, "L1:", 3) == 0) {
+    copyLine(line1, message + 3);
+    renderDisplay();
+    return;
+  }
+
+  if (strncmp(message, "L2:", 3) == 0) {
+    copyLine(line2, message + 3);
+    renderDisplay();
+  }
+}
+
+void readSerial() {
+  while (Serial.available() > 0) {
+    const char incoming = static_cast<char>(Serial.read());
+
+    if (incoming == '\r') {
+      continue;
+    }
+
+    if (incoming == '\n') {
+      inputBuffer[inputPos] = '\0';
+      handleMessage(inputBuffer);
+      inputPos = 0;
+      continue;
+    }
+
+    if (inputPos + 1 < INPUT_BUFFER_SIZE) {
+      inputBuffer[inputPos++] = incoming;
+    } else {
+      inputPos = 0;
+    }
+  }
+}
+
+void setup() {
+  lcd.begin(LCD_COLUMNS, LCD_ROWS);
+  renderDisplay();
+  Serial.begin(SERIAL_BAUD);
+}
+
+void loop() {
+  readSerial();
+}

bus_monitor.py

@@ -0,0 +1,394 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import http.cookiejar
+import json
+import math
+import re
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+from urllib import error, request
+
+API_ROOT = "https://rnd.don.su/"
+GET_BUS_URL = f"{API_ROOT}get_bus"
+USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) "
+    "Gecko/20100101 Firefox/140.0"
+)
+CSRF_RE = re.compile(r'<meta name="csrf-token" content="([^"]+)"')
+ROUTE_REQUEST = ["2_96", "2_94"]
+MAX_BUS_SPEED_KMH = 60.0
+
+
+@dataclass(frozen=True)
+class TargetPoint:
+    route: str
+    lat: float
+    lon: float
+
+
+@dataclass
+class BusEstimate:
+    route: str
+    vehicle_id: str
+    distance_m: float
+    eta_min: float | None
+    speed_kmh: float | None
+    approaching: bool
+    live: bool
+
+
+TARGETS: dict[str, TargetPoint] = {
+    "94": TargetPoint(route="94", lat=47.291609, lon=39.694856),
+    "96": TargetPoint(route="96", lat=47.220350, lon=39.632114),
+}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Gets bus positions from rnd.don.su and sends two LCD lines to Arduino."
+        )
+    )
+    parser.add_argument(
+        "--serial-port",
+        help="Serial port for Arduino, for example /dev/ttyACM0 or COM3.",
+    )
+    parser.add_argument(
+        "--baud-rate",
+        type=int,
+        default=115200,
+        help="Arduino Serial speed. Default: 115200.",
+    )
+    parser.add_argument(
+        "--interval",
+        type=float,
+        default=5.0,
+        help="Polling interval in seconds. Default: 5.",
+    )
+    parser.add_argument(
+        "--once",
+        action="store_true",
+        help="Perform one request and print/send one update.",
+    )
+    return parser.parse_args()
+
+
+def haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    radius = 6_371_000.0
+    d_lat = math.radians(lat2 - lat1)
+    d_lon = math.radians(lon2 - lon1)
+    a = (
+        math.sin(d_lat / 2.0) ** 2
+        + math.cos(math.radians(lat1))
+        * math.cos(math.radians(lat2))
+        * math.sin(d_lon / 2.0) ** 2
+    )
+    return 2.0 * radius * math.asin(math.sqrt(a))
+
+
+def scaled_coord(value: Any) -> float:
+    return float(value) / 1_000_000.0
+
+
+def parse_float(value: Any) -> float | None:
+    if value in (None, ""):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_timestamp(value: Any) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(str(value))
+    except ValueError:
+        return None
+
+
+def current_utc() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def format_distance(distance_m: float) -> str:
+    if distance_m < 1000:
+        return f"{int(round(distance_m))}m"
+    if distance_m < 10_000:
+        return f"{distance_m / 1000:.1f}km"
+    return f"{distance_m / 1000:.0f}km"
+
+
+def format_eta(eta_min: float | None) -> str:
+    if eta_min is None:
+        return "--m"
+    minutes = max(1, int(math.ceil(eta_min)))
+    return f"{minutes:02d}m"
+
+
+def fit_lcd(text: str) -> str:
+    return text[:16].ljust(16)
+
+
+def compute_speed_kmh(item: dict[str, Any]) -> float | None:
+    direct_speed = parse_float(item.get("speed"))
+    if direct_speed is not None and direct_speed > 1.0:
+        return min(direct_speed, MAX_BUS_SPEED_KMH)
+
+    prev = item.get("prev") or {}
+    if not prev:
+        return None
+
+    prev_time = parse_timestamp(prev.get("time"))
+    curr_time = parse_timestamp(item.get("time"))
+    if prev_time is None or curr_time is None:
+        return None
+
+    delta_s = (curr_time - prev_time).total_seconds()
+    if delta_s <= 0 or delta_s > 180:
+        return None
+
+    prev_lat = prev.get("lat")
+    prev_lon = prev.get("lon")
+    if prev_lat is None or prev_lon is None:
+        return None
+
+    delta_m = haversine_m(
+        scaled_coord(prev_lat),
+        scaled_coord(prev_lon),
+        scaled_coord(item["lat"]),
+        scaled_coord(item["lon"]),
+    )
+    if delta_m < 5.0:
+        return None
+
+    return min((delta_m / delta_s) * 3.6, MAX_BUS_SPEED_KMH)
+
+
+def build_estimate(item: dict[str, Any], target: TargetPoint) -> BusEstimate:
+    lat = scaled_coord(item["lat"])
+    lon = scaled_coord(item["lon"])
+    distance_m = haversine_m(lat, lon, target.lat, target.lon)
+
+    prev = item.get("prev") or {}
+    prev_distance = None
+    if prev.get("lat") is not None and prev.get("lon") is not None:
+        prev_distance = haversine_m(
+            scaled_coord(prev["lat"]),
+            scaled_coord(prev["lon"]),
+            target.lat,
+            target.lon,
+        )
+
+    approaching = prev_distance is None or distance_m <= prev_distance + 35.0
+    speed_kmh = compute_speed_kmh(item)
+    effective_speed_kmh = speed_kmh
+
+    if effective_speed_kmh is None and approaching:
+        effective_speed_kmh = 12.0 if distance_m < 500 else 18.0
+    elif effective_speed_kmh is not None and effective_speed_kmh < 4.0 and distance_m > 120:
+        effective_speed_kmh = 4.0
+
+    eta_min = None
+    if effective_speed_kmh is not None:
+        meters_per_minute = effective_speed_kmh * 1000.0 / 60.0
+        eta_min = distance_m / meters_per_minute
+
+    sample_time = parse_timestamp(item.get("time"))
+    age_s = None
+    if sample_time is not None:
+        age_s = (current_utc() - sample_time).total_seconds()
+    live = not bool(item.get("lost")) and (age_s is None or age_s <= 180)
+
+    return BusEstimate(
+        route=target.route,
+        vehicle_id=str(item.get("num", "")),
+        distance_m=distance_m,
+        eta_min=eta_min,
+        speed_kmh=speed_kmh,
+        approaching=approaching,
+        live=live,
+    )
+
+
+def choose_best_estimate(
+    vehicles: list[dict[str, Any]],
+    target: TargetPoint,
+) -> tuple[int, BusEstimate | None]:
+    estimates = [
+        build_estimate(item, target)
+        for item in vehicles
+        if str(item.get("name")) == target.route and str(item.get("type")) == "2"
+    ]
+    if not estimates:
+        return (0, None)
+
+    live_estimates = [estimate for estimate in estimates if estimate.live]
+    approaching_live = [estimate for estimate in live_estimates if estimate.approaching]
+    bus_count = len(live_estimates) if live_estimates else len(estimates)
+
+    def eta_sort_key(estimate: BusEstimate) -> tuple[float, float]:
+        eta = estimate.eta_min if estimate.eta_min is not None else float("inf")
+        return (eta, estimate.distance_m)
+
+    if approaching_live:
+        return (bus_count, min(approaching_live, key=eta_sort_key))
+    if live_estimates:
+        return (bus_count, min(live_estimates, key=lambda estimate: estimate.distance_m))
+    return (bus_count, min(estimates, key=lambda estimate: estimate.distance_m))
+
+
+def build_line(route: str, bus_count: int, estimate: BusEstimate | None) -> str:
+    if estimate is None:
+        return fit_lcd(f"{route} #{bus_count} no signal")
+    text = (
+        f"{route} #{bus_count} "
+        f"{format_eta(estimate.eta_min)} "
+        f"{format_distance(estimate.distance_m)}"
+    )
+    return fit_lcd(text)
+
+
+class BusApiClient:
+    def __init__(self) -> None:
+        self.cookie_jar = http.cookiejar.CookieJar()
+        self.opener = request.build_opener(
+            request.HTTPCookieProcessor(self.cookie_jar)
+        )
+        self.csrf_token: str | None = None
+
+    def refresh_session(self) -> None:
+        req = request.Request(
+            API_ROOT,
+            headers={
+                "User-Agent": USER_AGENT,
+                "Accept": "text/html,application/xhtml+xml",
+            },
+        )
+        with self.opener.open(req, timeout=20) as response:
+            html = response.read().decode("utf-8", errors="replace")
+
+        match = CSRF_RE.search(html)
+        if match is None:
+            raise RuntimeError("CSRF token not found on rnd.don.su main page")
+        self.csrf_token = match.group(1)
+
+    def fetch_vehicles(self) -> list[dict[str, Any]]:
+        if self.csrf_token is None:
+            self.refresh_session()
+
+        payload = json.dumps({"bus": ROUTE_REQUEST}, ensure_ascii=False).encode("utf-8")
+        req = request.Request(
+            GET_BUS_URL,
+            data=payload,
+            method="POST",
+            headers={
+                "User-Agent": USER_AGENT,
+                "Accept": "application/json, text/plain, */*",
+                "Content-Type": "application/json;charset=utf-8",
+                "Origin": API_ROOT.rstrip("/"),
+                "Referer": API_ROOT,
+                "X-CSRF-TOKEN": self.csrf_token or "",
+            },
+        )
+
+        try:
+            with self.opener.open(req, timeout=20) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except error.HTTPError as exc:
+            if exc.code not in (403, 422):
+                raise
+            self.refresh_session()
+            req.remove_header("X-CSRF-TOKEN")
+            req.add_header("X-CSRF-TOKEN", self.csrf_token or "")
+            with self.opener.open(req, timeout=20) as response:
+                return json.loads(response.read().decode("utf-8"))
+
+
+class SerialDisplay:
+    def __init__(self, port: str | None, baud_rate: int) -> None:
+        self.port = port
+        self.baud_rate = baud_rate
+        self._serial = None
+        self._serial_module = None
+
+    def _connect(self) -> None:
+        if self.port is None or self._serial is not None:
+            return
+
+        if self._serial_module is None:
+            try:
+                import serial  # type: ignore
+            except ImportError as exc:
+                raise RuntimeError(
+                    "pyserial is required for --serial-port. Install with: pip install pyserial"
+                ) from exc
+            self._serial_module = serial
+
+        self._serial = self._serial_module.Serial(
+            self.port,
+            self.baud_rate,
+            timeout=1,
+            write_timeout=1,
+        )
+        time.sleep(2.0)
+
+    def send(self, line1: str, line2: str) -> None:
+        if self.port is None:
+            return
+
+        try:
+            self._connect()
+            assert self._serial is not None
+            payload = f"L1:{line1}\nL2:{line2}\n".encode("ascii", errors="replace")
+            self._serial.write(payload)
+            self._serial.flush()
+        except Exception as exc:
+            print(f"Serial send failed: {exc}", file=sys.stderr)
+            if self._serial is not None:
+                try:
+                    self._serial.close()
+                except Exception:
+                    pass
+            self._serial = None
+
+
+def poll_once(client: BusApiClient) -> tuple[str, str]:
+    vehicles = client.fetch_vehicles()
+    lines: list[str] = []
+    for route in ("94", "96"):
+        bus_count, estimate = choose_best_estimate(vehicles, TARGETS[route])
+        lines.append(build_line(route, bus_count, estimate))
+    return lines[0], lines[1]
+
+
+def main() -> int:
+    args = parse_args()
+    client = BusApiClient()
+    display = SerialDisplay(args.serial_port, args.baud_rate)
+
+    while True:
+        try:
+            line1, line2 = poll_once(client)
+            print(line1.rstrip())
+            print(line2.rstrip())
+            print("-" * 16)
+            display.send(line1, line2)
+        except KeyboardInterrupt:
+            return 0
+        except Exception as exc:
+            print(f"Update failed: {exc}", file=sys.stderr)
+
+        if args.once:
+            return 0
+        time.sleep(max(args.interval, 5.0))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

requirements.txt

@@ -0,0 +1 @@
+pyserial>=3.5