Сегодня, когда в очередной раз я вижу 0 сообщений в телеграм канале, который должен предупреждать о приближающихся автобусах, будет достаточно символично написать эту статью. Дело началось в тот момент, когда..
Дисклеймер: при возникновении странных ощущений при прочтении просто вспомните, что конкретно этот текст полностью написан живым человеком, а не нейронкой (кода, впрочем, это не касается). Don't panic, it's organic!
.. моя девушка снова сидела на кухне и грустила, что опоздала на работу. Для кого-то это пустяк, кому-то это давит на мозги потом целый день. Пазл сложился сраху же, когда из Храма Памяти выстрелило замечательное видео про бостонское метро и как чел сделал себе и друзьям девайс на Raspberry Pi, помогающий ему выйти из дома ровно за столько, сколько нужно, чтобы спокойно дойти до остановки. Явно не только мне хочется идеально приходить на остановку и тут же садиться в бусик :) А ещё, конечно же, хотелось решительно нанести реальную пользу своим кодом хоть одному человеку и взять паузу в бесконечном думскроллинге и откликам на hh.
Начало
После кастдева оформилась цель — орать в телеграм-канал сообщением LAST CHANCE!!!, когда уже стопроцентно надо начинать спокойно одеваться и спокойно идти до остановки. Ключевое тут спокойно, так как оказалось, что конечный пользователь хочет:
\ни в коем случае не бежать к автобусу
\не стоять долго на остановке
\время на одежду, закрывание дверей и лифт должно учитываться в расчётах
Юзкейс: сидя на кухне на смарт-часы приходит мессадж от тг бота ровно за то время, которое нужно на все действия с этого момента, чтобы одеться и прийти на остановку бусика, при этом бусик уже гарантированно будет в пути.
Поиски привели к сайту Цифрового Петербурга и сервису с названием «Данные портала общественного транспорта Санкт-Петербурга». Радости были полные штаны — всё красиво, выглядит современно, сваггер прямо на вкладке с документацией, OAS3 спека. Промпт в чатжпт был простым: «help me create a python app that gets the list of all the bus stops. it's OAS3 <содержимое спеки>». Код выглядел неплохо, но почему-то не работал :) Спустя пару часов стало понятно, проблема не в коде, а в spb-transport.gate.petersburg.ru.
До сих пор не знаю почему гейт не принимал токен, тщательно скопированный из личного кабинета. Не сразу догадался проверить, а работает ли вообще хоть одна ручка в сваггере. Ошибка была ровно та же прямо на сайте.
Запустил сейчас свой старый код и он всё так же отдаёт 401 {"error":"invalid_grant","error_description":"Invalid bearer token"}. Даже если это был мой косяк и вставлять Bearer-токен надо по-другому (научите, пожалуйста!), это был крайне важный момент — глаз зацепился за ещё одну ссылку «Источник статических данных GTFS об общественном транспорте». На тот момент мне это ни о чём не говорило, я просто хотел получить список бусиков и по моим подсчётам на это должно было уйти 5 минут, а не 2 часа.
Оказалось, что GTFS ака General Transit Feed Specification это изначально гугловая спецификация, созданная в 2006 году, и потихоньку ставшая международным стандартом. Параллельно я нашёл отличную статью Егора по решению практически такой же проблемы и всё стало значительно понятнее.
GTFS Feed (он же static GTFS) — это архив, в котором кроме прочего внутри routes.txt можно найти route_id нужного маршрута, а в stops.txt — stop_id нужной остановки.
GTFS Realtime — набор ручек по получению реального положения транспорта с координатами.
route_id,agency_id,route_short_name,route_long_name,route_type,transport_type,circular,urban,night 3812,orgp,193,"ПР. КУЛЬТУРЫ - ЖК ""ЦВЕТНОЙ ГОРОД""",3,bus,0,1,0 3918,orgp,351Б,Ж.-Д. СТАНЦИЯ НОВЫЙ ПЕТЕРГОФ - Ж.-Д. СТАНЦИЯ СТАРЫЙ ПЕТЕРГОФ,3,bus,0,1,0 4055,orgp,194,"АВТОЗАВОД ""АГР"" - СТАНЦИЯ МЕТРО ""КОМЕНДАНТСКИЙ ПРОСПЕКТ""",3,bus,0,1,0
stop_id,stop_code,stop_name,stop_lat,stop_lon,location_type,wheelchair_boarding,transport_type 17609,17609,"ПР. АВИАКОНСТРУКТОРОВ, 49",60.027450,30.223463,0,2,bus 17611,17611,"ПР. АВИАКОНСТРУКТОРОВ, 49",60.026383,30.224832,0,2,bus 17629,17629,Ж.-Д. ПЛАТФОРМА ПОНТОННАЯ ,59.785045,30.627652,0,2,bus 17631,17631,"ОТРАДНОЕ, СУДОРЕМОНТНЫЕ МАСТЕРСКИЕ, ПО ТРЕБОВАНИЮ",59.760553,30.765983,0,2,bus 17632,17632,"ОТРАДНОЕ, ДОРОГА В НИКОЛЬСКОЕ",59.763355,30.774362,0,2,bus
Соответственно, если Feed можно скачать себе, найти нужные айдишники и забыть (маршрут меня интересует только один и крайне вряд ли эти айдишники будут часто меняться), то Realtime это как раз то, что нам надо кверить постоянно для решения нашей задачи. Для того, чтобы понять, как вообще работает система и какие есть поля, мне нужен был код, который выплёвывает максимально сырые данные из системы. Наивный я думал, что достаточно получить предикшены/forecast для конкретной остановки и дело в шляпе, через час будет уже готовый бот и полагаться только на ручку vehicle для получения инфы по вообще всем бусикам мне будет не нужно…
Создаём код
Нейронке хватило такого запроса для написания полностью рабочего кода, от которого я в дальнейшем и отталкивался:
Write python code that gets GTFS Realtime data for all the buses with specific route id, parses it and gives me information about the specific bus on a specific bus stop
Route ID: 3219
Stop ID: 30130
Main URL is: https://transport.orgp.spb.ru/Portal/transport/internalapi/gtfs/realtime/Here are the examples: <копипаста примеров вызовов ручек с сайта>
Флоу данных:
> из vehicle получаем список всех бусов на конкретном маршруте
> из vehicletrips получаем детали трипов конкретных бусов из предыдущего шага
> если stop_id в трипе совпадает с айди интересующей остановки, добавляем в список трипов текущий
> выводим сырое содержимое списка трипов
Что полезного можно взять из ручек vehicle и vehicletrips:
На всякий уточню, что ручка stopforecast дублирует информацию, которую можно получить из vehicletrips. Рабочий базовый код на питоне для получения предикшенов для конкретной остановки и маршрута выглядит так, токен/ключ для API не нужен, просто запускаем:
Скрытый текст
не забываем создать .venv и установить нужные запчасти
python3 -m venv venv source .venv/bin/activate pip install requests gtfs-realtime-bindings
import requests import urllib3 from google.transit import gtfs_realtime_pb2 from datetime import datetime, timezone from zoneinfo import ZoneInfo BASE_URL = "https://transport.orgp.spb.ru/Portal/transport/internalapi/gtfs/realtime" ROUTE_ID = "3219" STOP_ID = "30130" TZ = ZoneInfo("Europe/Moscow") urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def fetch_vehicle_positions(route_id: str): """ Step 1: Get all vehicles currently running on a specific route. """ url = f"{BASE_URL}/vehicle" params = { "transports": "bus", "routeIDs": route_id } response = requests.get(url, params=params, timeout=15, verify=False) response.raise_for_status() feed = gtfs_realtime_pb2.FeedMessage() feed.ParseFromString(response.content) vehicles = [] for entity in feed.entity: if entity.HasField("vehicle"): vehicle = entity.vehicle vehicles.append({ "vehicle_id": vehicle.vehicle.id, "route_id": vehicle.trip.route_id, "trip_id": vehicle.trip.trip_id, "latitude": vehicle.position.latitude, "longitude": vehicle.position.longitude, "timestamp": vehicle.timestamp }) return vehicles def fetch_trip_updates(vehicle_ids): """ Step 2: For a list of vehicle IDs, get detailed stop-by-stop predictions. """ if not vehicle_ids: return [] url = f"{BASE_URL}/vehicletrips" params = { "vehicleIDs": ",".join(vehicle_ids) } response = requests.get(url, params=params, timeout=15, verify=False) response.raise_for_status() feed = gtfs_realtime_pb2.FeedMessage() feed.ParseFromString(response.content) trip_updates = [] for entity in feed.entity: if entity.HasField("trip_update"): trip_updates.append(entity.trip_update) return trip_updates def find_bus_at_stop(route_id: str, stop_id: str): """ Step 3: Combine vehicle positions + trip updates and extract arrival/departure info for a specific stop. """ vehicles = fetch_vehicle_positions(route_id) vehicle_ids = [v["vehicle_id"] for v in vehicles] trip_updates = fetch_trip_updates(vehicle_ids) results = [] for update in trip_updates: vehicle_id = update.vehicle.id trip_id = update.trip.trip_id for stop_time in update.stop_time_update: if stop_time.stop_id == stop_id: arrival = stop_time.arrival.time if stop_time.HasField("arrival") else None results.append({ "vehicle_id": vehicle_id, "trip_id": trip_id, "route_id": route_id, "stop_id": stop_id, "arrival_time_local": ( datetime.fromtimestamp(arrival, tz=timezone.utc) .astimezone(TZ) .isoformat() if arrival else None ) }) return results if __name__ == "__main__": buses = find_bus_at_stop(ROUTE_ID, STOP_ID) if not buses: print("No buses found for this route at this stop.") else: for bus in buses: print("----- BUS FOUND -----") for key, value in bus.items(): print(f"{key}: {value}")
Чтобы не мёрзнуть в -20 на улице, принял решение полагаться на отображение транспорта с яндекс карт. Также нашлась статья с интересными деталями отрисовки транспорта на них. Оказалось, что ребята не полагаются на список бусов по конкретному маршруту, а забирают все существующие бусы, совмещают их с историческими данными (на каком маршруте был бус в предыдущие дни) и смотрят по координатам куда вообще этот бус реально едет и на основе этого уже определяют номер маршрута. Плюс то, что мы видим на картах, это ближайшее будущее или прогнозирование, так как в лучшем случае каждый бус присылает координаты с интервалом в 1 минуту и за это время можно доехать до следующей остановки.
Сам столкнулся с тем, что на линии работают бусы, которые через GTFS не отдаются и на остановку может приехать автобус, которого вообще нет в системе на нужном маршруте. Полагаю, что водители забывают переключить маршрут со старого на новый, включить/настроить транспортный информатор (отличная статья от Льва про них) или просто проблемы с GPS/ГЛОНАСС и инетом.
Ещё из нюансов: очень повезло, что интересующий меня маршрут делает большой крюк до ближайшей станции метро и это занимает чуть больше 15 минут, которые и нужны моей девушке, чтобы с максимальным комфортом отправиться в вояж до остановки. Также за это время иногда успевает «проснуться» система предсказаний и выдать довольно точные данные о приезде буса. Но часто этих предсказаний либо вообше нет, либо они приходят за 4-6 минут до прихода буса на остановку и этим можно пользоваться только уже выйдя из дома. Мы находимся рядом с автобусным кольцом, поэтому, если бы надо было предсказывать отправление автобусов прямо с кольца, реального решения этой проблемы я не вижу без использования расписания из статического GTFS. А верить расписаниям автобусов ещё со времён школы совсем не хочется, хочется оперировать максимально боевыми данными.
Что было найдено при отладке:
/бус может выехать с кольца и наглухо пропасть с радаров. вот он пришёл на первые 3 остановки, а до 4 уже не доехал и потом вдруг оказался на остановке сильно дальше (если совсем не пропал)
/думал использовать предикшены для одной из начальных остановок, чтобы как только приходил предикшен> значит бус едет> отправляем сообщение. но оказалось, что самая ранняя остановка, для которой они появляются в системе, где-то в 12 минутах от нужной и это сильно меньше требуемых 15 минут
/предикшены остаются в системе и выводятся в ответах ещё какое-то время даже после прохождения бусом остановки и их надо фильтровать относительно текущего времени
/предикшен может появиться в системе не для ближайшего к остановке автобуса, а для следующего уже после него
Как промежуточный итог: предикшены невозможно использовать в моём случае, но они могут быть полезными, поэтому решил отправлять их в тг канал отдельными сообщениями, рассказав нюансы работы этого механизма конечному пользователю (да, было сложно).
После такой неприятной неконсистентности в данных сразу же пришло понимание почему яндекс карты буквально полагаются только на координаты автобусов, а не на что-либо ещё. Случайно глаз лёг на параметр bbox в документации к ручке vehicle и оказалось, что речь идёт о bounding box или области на карте и пазл наконец-то сложился :)
Финальная логика работы
Да, у ручки vehicle можно указать координаты bbox и получить все автобусы, находящиеся в этой области. Но снова надеяться на то, что это будет нормально работать абсолютно не хотелось, было решено собирать координаты всех автобусов на линии и сверять их координаты с координатами выделенной области. Спасибо OSM и bboxfinder.com, получить нужные координаты оказалось очень просто. Только пришлось немного посидеть с секундомером и якартами, чтобы понять координаты какой остановки станут тем самым последним шансом выйти из дома, чтобы успеть дойти за 15 минут.
В спеке GTFS есть такая вещь как direction_id у трипа, 0 в одну сторону, 1 в другую. Повезло, что в петербургском GTFS данное значение у трипов всегда 0, поэтому мыслей полагаться на это даже не было. Все надежды легли на bearing/азимут и что эти данные точно будут валидны и точны. Гипотезу проверил генерацией карты с автобусами и их направлением, спасибо geopandas и matplotlib, всё выглядело хорошо. В очередной раз удивился, что бесплатный чатжпт с третьего промпта моментально выдал рабочий код, который не нужно было никак исправлять.
Скрытый текст
write a function that gets all the gps coordinates of buses using fetch_vehicle_positions and plots them on a map (make a picture) using geopandas.
take bearing from the vehicle positions and add direction on a picture. use the previous version of function without direction data from direction_id
В GTFS принято, что 0° = север, 90° = восток, 180° = юг, 270° = запад. Внутри 15-минутного bounding box автобус двигается только на юго-восток и юго-запад, это позволило сделать следующую логику: вначале проверяем, что координаты автобуса находятся внутри bbox, затем — проверка азимута и, если всё хорошо, отправляем данные дальше.
def is_in_bounding_box(lat: float, lon: float, bbox: tuple) -> bool: min_lon, min_lat, max_lon, max_lat = bbox return min_lat <= lat <= max_lat and min_lon <= lon <= max_lon def bearing_to_direction(bearing: float) -> str | None: """ Returns direction name if bearing matches required sectors. GTFS bearing is defined as: 0° = North 90° = East 180° = South 270° = West """ bearing = bearing % 360 if 67.5 <= bearing < 112.5: return "E" if 112.5 <= bearing < 157.5: return "SE" if 157.5 <= bearing < 202.5: return "S" if 202.5 <= bearing < 247.5: return "SW" return None def get_vehicles_in_bbox_heading_target_directions( route_id: str, bbox: tuple, vehicles_snapshot: list[dict] | None = None ) -> List[Dict]: if vehicles_snapshot is None: vehicles = fetch_vehicle_positions(route_id) else: vehicles = vehicles_snapshot result = [] for v in vehicles: if not is_in_bounding_box(v["latitude"], v["longitude"], bbox): continue direction = bearing_to_direction(v["bearing"]) if direction is None: continue v_filtered = v.copy() v_filtered["direction"] = direction result.append(v_filtered) return result
Тут мы приходим к core loop, внутри которой и будут забираться все автобусы нужного маршрута, проверяться их геолокация, азимут и дальше данные бусика будут отправляться прямиком в телеграм канал.
У класса MonitorState есть метод run, внутри которого и происходит поллинг ручек GTFS, сверка координат автобусов с bbox, отправка сообщений в тг канал, логов в stdout.
Из интересного:
/как только бус входит в 15-минутную зону, сразу же посылаем сообщение в тг, далее при выходе из зоны посылаем ещё одно сообщение, обычно это где-то 2 минуты нахождения внутри зоны
/проверяем действительно каждые 2 секунды
cycle_start = time.time() <..> elapsed = time.time() - cycle_start time.sleep(max(0, CHECK_INTERVAL - elapsed))
/в active_buses храним сам бус и время его добавления, чтобы можно было его удалить при залипании дольше чем на 5 минут
/каждый предикшен записываем в sent_arrival_predictions, а время его появления в last_prediction_ts, чтобы не отправлять повторно
/пришлось ввести зону BOUNDING_BOX_BIG, которая значительно больше 15-минутной зоны, чтобы попробовать отлавливать бусы-призраки, которые почему-то минуют вход в 15-минутную зону, но потом могут появиться уже дальше по маршруту
Скрытый текст
/чтобы не спамить постоянно в логи, пишем в stdout только при каждой отправке в телегу, этих данных хватает для дебага
print(f"[{self.format_arrival_hhmm(cycle_start)}] Buses total: {len(all_vehicles)}") print(f"[{self.format_arrival_hhmm(cycle_start)}] Buses in bbox: {len(bbox_vehicles)}") print(f"[{self.format_arrival_hhmm(cycle_start)}] Buses in big bbox: {len(big_bbox_labels)}. Bus labels: {big_bbox_labels}") print(f"[{self.format_arrival_hhmm(cycle_start)}] {bus}\n")
/при отправке предикшенов добавляется общее их количество, включая прошедшие, чтобы было видно насколько много просроченных предсказаний может выдавать GTFS
print(f"[{self.format_arrival_hhmm(cycle_start)}] GTFS arrivals count: {len(predictions)}") print(f"[{self.format_arrival_hhmm(cycle_start)}] sent_arrival_predictions: {len(self.sent_arrival_predictions)}")
/телеграм очень не любит неэкранированные символы вроде точки и слэша в MarkdownV2, поэтому советую сообщения посылать простым курлом перед тем как вставлять их в код. они просто могут не приходить, а вы не будете понимать почему их нет
❯ curl -X POST \ -H 'Content-Type: application/json' \ -d '{"chat_id": "-231..<ваш чат айди тг канала>", "text": "Чот случилось с бусиком 😥\n\nКод бусика: 2438\nНомерной знак: *Е803УТ198*", "parse_mode": "MarkdownV2"}' \ https://api.telegram.org/bot615..<ваш токен>/sendMessage
Всё это было упаковано в образ python:3.14.2-slim через Dockerfile + docker-compose, тг ключ автоматом подхватывается из .env файла рядом.
# это Dockerfile FROM python:3.14.2-slim WORKDIR /app COPY . . RUN pip install --no-cache-dir -r requirements.txt CMD ["python3", "-u", "gtfs.py"] # а это уже docker-compose.yml services: bus180: container_name: bus180 image: bus180:latest build: context: . dockerfile: Dockerfile volumes: - .:/app restart: unless-stopped environment: TELEGRAM_BOT_TOKEN: "${TELEGRAM_BOT_TOKEN}" TELEGRAM_CHANNEL_ID: "${TELEGRAM_CHANNEL_ID}"
Запуск происходит по крону и тушится всё через час.
00 7 * * MON-FRI cd /root/bus180 && docker compose up -d 59 7 * * MON-FRI cd /root/bus180 && docker compose logs >> bus180.log 00 8 * * MON-FRI cd /root/bus180 && docker compose down -t 15
Aftermath
Получать кастомную инфу об автобусах это действительно удобно. Особенно, когда GTFS работает :) Хочется сделать девайс типа пейджера, который будет пиликать и моргать экраном, когда надо выходить, чтобы даже смарт-часы были не нужны, но для этого важно, чтобы всё работало стабильно. Текущие проблемы системы пока не позволяют ориентироваться чисто на тг канал с оповещениями.
За время эксплуатации были ситуации, когда:
вставали вообще все автобусы на карте на полдня минимум. относительно приятно, что у твоего кода такие же проблемы как у яндекс карт, но хочется, чтобы моргало не на день, а в пределах 5 минут
всего автобусов на линии было 2 (арифметическое среднее за всё время 7.3) и даже без GTFS было понятно, что всё плохо. это проблема перевозчика, разумеется
были те самые бусы-призраки, которые появляются из ниоткуда или их вообще не видно ни на я.картах, ни через GTFS
Похвалю, что когда система GTFS работает, она действительно быстро и надёжно отдаёт данные. За всё время у меня был разве что один день, когда сам гейт был недоступен, остальное время либо бусы вставали на 1 точку и не двигались, либо их просто не было в системе, если были проблемы. Полагаю, что такие внешние факторы как глушилки GPS/ГЛОНАСС, человеческая забывчивость (трекеры не работают внутри бусов или настроены не на тот маршрут и система их не отдаёт) могут достаточно сильно влиять и ничего сделать с этим мы не сможем.
Конечно же очень приятно и тепло, когда тебе говорят «выбирала между бровями или уже идти и тут п��иходит про бусик оповещение. вышла сразу, в итоге автобус появился сразу как на остановку пришла».
Буду рад, если кто-то изнутри петербургских автопарков и Цифрового Петербурга подсветит проблемы, чтобы хотя бы понимать причины, что же может быть не так. Очень хочется, чтобы 100% бусов были видны в GTFS и хотя бы их координаты были тем надёжным источником данных, на который можно полагаться.
Так как тг канала у меня нет, порекламирую канал, на который подписан сам. Уважающий разумного и думающего читателя подход к постам, интересная статистика, данные всегда с источниками и приглашающими подумать самому мыслями https://t.me/groks
