394 lines
11 KiB
Python
394 lines
11 KiB
Python
#!/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())
|