aboutsummaryrefslogtreecommitdiff
path: root/proxy_rt_renfe/main.py
diff options
context:
space:
mode:
Diffstat (limited to 'proxy_rt_renfe/main.py')
-rw-r--r--proxy_rt_renfe/main.py71
1 files changed, 0 insertions, 71 deletions
diff --git a/proxy_rt_renfe/main.py b/proxy_rt_renfe/main.py
deleted file mode 100644
index 89c55a0..0000000
--- a/proxy_rt_renfe/main.py
+++ /dev/null
@@ -1,71 +0,0 @@
-# /// script
-# requires-python = ">=3.13"
-# dependencies = [
-# "flask",
-# "protobuf",
-# "requests",
-# ]
-# ///
-
-from flask import Flask, Response
-import requests
-
-from gtfsrt_pb2 import FeedEntity, FeedHeader, FeedMessage, TripDescriptor, TripUpdate
-
-
-def download_rt_json():
- resp = requests.get("https://gtfsrt.renfe.com/trip_updates_LD.json")
-
- return resp.json()
-
-
-def rt_json_to_proto(rt_json):
- header = FeedHeader(
- gtfs_realtime_version=rt_json["header"]["gtfsRealtimeVersion"],
- timestamp=int(rt_json["header"]["timestamp"]),
- )
-
- entities = []
- for entity_json in rt_json["entity"]:
- json_trip_update = entity_json.get("tripUpdate")
- if not json_trip_update:
- print(f"Skipping entity without tripUpdate: {entity_json['id']}")
- continue
-
- if not json_trip_update.get("delay"):
- print(f"Skipping entity without delay: {entity_json['id']}")
- continue
-
- trip_descriptor = TripDescriptor(
- trip_id=json_trip_update["trip"]["tripId"],
- schedule_relationship=TripDescriptor.SCHEDULED,
- )
- stu = TripUpdate.StopTimeUpdate(
- stop_sequence=1,
- arrival=TripUpdate.StopTimeEvent(delay=int(json_trip_update["delay"])),
- departure=TripUpdate.StopTimeEvent(delay=int(json_trip_update["delay"])),
- )
- trip_update = TripUpdate(
- trip=trip_descriptor,
- stop_time_update=[stu],
- delay=int(json_trip_update.get("delay", "0")),
- )
- entity = FeedEntity(id=entity_json["id"], trip_update=trip_update)
- entities.append(entity)
-
- feed = FeedMessage(header=header, entity=entities)
-
- return feed.SerializeToString()
-
-
-app = Flask(__name__)
-
-
-@app.route("/proto", methods=["GET"])
-def proto():
- rt_json = download_rt_json()
- proto_data = rt_json_to_proto(rt_json)
- return Response(proto_data, mimetype="application/x-protobuf")
-
-if __name__ == "__main__":
- app.run(host="localhost", port=5000)