diff --git a/README.md b/README.md index 145705e..c2a9aa9 100644 --- a/README.md +++ b/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. diff --git a/arduino_bus_display/arduino_bus_display.ino b/arduino_bus_display/arduino_bus_display.ino new file mode 100644 index 0000000..cb70816 --- /dev/null +++ b/arduino_bus_display/arduino_bus_display.ino @@ -0,0 +1,89 @@ +#include +#include +#include + +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(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(); +} diff --git a/bus_monitor.py b/bus_monitor.py new file mode 100644 index 0000000..290af6e --- /dev/null +++ b/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' 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()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7ad05ef --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +pyserial>=3.5