#!/usr/bin/env python3 """ Build COMPACT GTFS database for aNavigator. ALL trips, ALL stops - deduplicated. - sl: stops library [[name, lat, lon], ...] - sh: shapes library [[[lat,lon], ...], ...] - lines: { "23": { name: "...", patterns: [{ d: direction, h: headsign, s: service (F/D/S), sh: shape_index, sr: [stop_indices], so: [[arr_offset,dep_offset], ...], # offsets in min from departure per stop t: ["HH:MM", ...] # departure times }]}} """ import csv, json, sys, os from collections import defaultdict, OrderedDict GTFS_DIR = sys.argv[1] if len(sys.argv) > 1 else "/tmp/gtfs_tper" OUTPUT = sys.argv[2] if len(sys.argv) > 2 else "/home/alina/aNavigator/src/gtfs_db.json" print(f"Reading GTFS from {GTFS_DIR}...") # 1. Routes routes = {} with open(os.path.join(GTFS_DIR, "routes.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): routes[row["route_id"]] = row # 2. Stops - global library print("Loading stops...") stops_lib = [] stop_sid_to_idx = {} with open(os.path.join(GTFS_DIR, "stops.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): name = row["stop_name"].strip() sid = row["stop_id"] stop_sid_to_idx[sid] = len(stops_lib) stops_lib.append([name, round(float(row["stop_lat"]), 6), round(float(row["stop_lon"]), 6)]) print(f" {len(stops_lib)} unique stops") # 3. Shapes - global library print("Loading shapes...") raw_shapes = defaultdict(list) with open(os.path.join(GTFS_DIR, "shapes.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): raw_shapes[row["shape_id"]].append([round(float(row["shape_pt_lat"]), 6), round(float(row["shape_pt_lon"]), 6)]) def simplify_shape(coords, max_pts=150): if len(coords) <= max_pts: return coords step = len(coords) / max_pts res = [coords[min(int(i * step), len(coords)-1)] for i in range(max_pts)] if res[-1] != coords[-1]: res[-1] = coords[-1] return res shapes_lib = [] shape_sid_to_idx = {} for sid in sorted(raw_shapes.keys()): shape_sid_to_idx[sid] = len(shapes_lib) shapes_lib.append(simplify_shape(raw_shapes[sid])) print(f" {len(shapes_lib)} unique shapes") # 4. Stop times print("Loading stop_times...") stop_times_by_trip = defaultdict(list) with open(os.path.join(GTFS_DIR, "stop_times.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): stop_times_by_trip[row["trip_id"]].append({ "s": row["stop_id"], "seq": int(row["stop_sequence"]), "a": row.get("arrival_time", ""), "d": row.get("departure_time", "") }) print(f" {len(stop_times_by_trip)} trips") # 5. Calendar + Calendar Dates (per _getActiveServices in JS) print("Loading calendar...") cal = {} with open(os.path.join(GTFS_DIR, "calendar.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): days = "" for d in ['monday','tuesday','wednesday','thursday','friday','saturday','sunday']: days += "1" if row.get(d) == "1" else "0" cal[row["service_id"]] = { "start": row["start_date"], "end": row["end_date"], "days": days, "exc": [] } # Calendar dates exceptions (service_id→[[date, exception_type], ...]) with open(os.path.join(GTFS_DIR, "calendar_dates.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): svc = row["service_id"] if svc in cal: cal[svc]["exc"].append([row["date"], row["exception_type"]]) # Build compact calendar (formato atteso da _getActiveServices in JS) cal_compact = {} for svc_id, c in cal.items(): cal_compact[svc_id] = [c["start"], c["end"], c["days"], c["exc"]] print(f" {len(cal_compact)} calendar entries") def service_label(pattern): if not pattern: return "" if "mon" in pattern: return "F" if "tue" in pattern: return "F" if "wed" in pattern: return "F" if "thu" in pattern: return "F" if "fri" in pattern: return "F" if pattern == "sat": return "S" if pattern == "sun": return "D" if "sat" in pattern and "sun" in pattern: return "F" if "sat" in pattern: return "S" if "sun" in pattern: return "D" return "" def time_to_min(t): """Convert HH:MM to minutes. Handles 24:xx+ as continuation.""" if not t or ":" not in t: return 0 h, m = int(t[:2]), int(t[3:5]) if h >= 24: h -= 24 return h * 60 + m def min_to_time(m): """Convert minutes to HH:MM string.""" m = int(m) h = m // 60 mi = m % 60 if h >= 24: h -= 24 return f"{h:02d}:{mi:02d}" # 6. Load trips print("Loading trips...") trips_by_route = defaultdict(list) with open(os.path.join(GTFS_DIR, "trips.txt"), encoding="utf-8") as f: for row in csv.DictReader(f): trips_by_route[row["route_id"]].append(row) # 7. Build compact database print("Building compact database...") db = {"sl": stops_lib, "sh": shapes_lib, "ca": cal_compact, "lines": OrderedDict()} sorted_routes = sorted(routes.values(), key=lambda r: ( int(r["route_short_name"]) if r["route_short_name"].strip().isdigit() else 999, r["route_short_name"] )) for route in sorted_routes: short_name = route["route_short_name"].strip() route_id = route["route_id"] long_name = route["route_long_name"].strip() trip_rows = trips_by_route.get(route_id, []) if not trip_rows: continue # Group by shape_id + direction_id + stop_sequence pattern # Group by shape_id + direction_id + service_id (NON per service_label) pattern_groups = defaultdict(list) for t in trip_rows: sid = t["shape_id"] dir_id = int(t["direction_id"]) svc = t["service_id"] trip_stops = stop_times_by_trip.get(t["trip_id"], []) trip_stops.sort(key=lambda x: x["seq"]) stop_seq = [s["s"] for s in trip_stops] stop_key = ",".join(stop_seq[:8]) pat_key = (sid, dir_id, svc, stop_key) # usa service_id, non label! pattern_groups[pat_key].append((t, trip_stops)) line_key = short_name if line_key not in db["lines"]: db["lines"][line_key] = {"name": long_name, "patterns": []} for pat_key, trips_data in pattern_groups.items(): sid, dir_id, pat, _ = pat_key t_first, trip_stops_first = trips_data[0] # First trip departure minutes dep_first = trip_stops_first[0].get("d", trip_stops_first[0].get("a", "00:00")) dep_first_min = time_to_min(dep_first) # Build stop offsets and indices stop_refs = [] stop_offsets = [] # [[arr_offset, dep_offset], ...] for s in trip_stops_first: idx = stop_sid_to_idx.get(s["s"], 0) stop_refs.append(idx) arr_min = time_to_min(s["a"]) - dep_first_min dep_min = time_to_min(s["d"]) - dep_first_min stop_offsets.append([arr_min, dep_min]) headsign = t_first.get("trip_headsign", "").strip() # Extract pattern letter from trip_short_name (es. "11A" → "A", "11/" → "") trip_short = t_first.get("trip_short_name", "").strip() pattern_letter = "" if trip_short: base = route["route_short_name"].strip() if trip_short.startswith(base): suffix = trip_short[len(base):].strip() if suffix and suffix != "/": pattern_letter = suffix if not headsign and trip_stops_first: first_sid = trip_stops_first[0]["s"] last_sid = trip_stops_first[-1]["s"] if first_sid in stop_sid_to_idx and last_sid in stop_sid_to_idx: fs = stops_lib[stop_sid_to_idx[first_sid]] ls = stops_lib[stop_sid_to_idx[last_sid]] headsign = f"{fs[0]} → {ls[0]}" shape_idx = shape_sid_to_idx.get(sid, 0) # Collect departure times for all trips (deduplicate per minuto) dep_times = set() for t, ts in trips_data: dep = ts[0].get("d", ts[0].get("a", "")) if dep and ":" in dep: dep_times.add(dep[:5]) # solo HH:MM dep_times = sorted(dep_times) if not dep_times: continue db["lines"][line_key]["patterns"].append({ "d": dir_id, "h": headsign, "rl": pattern_letter, "svc": pat, "sh": shape_idx, "sr": stop_refs, "so": stop_offsets, "t": dep_times }) # 8. Write with open(OUTPUT, "w", encoding="utf-8") as f: json.dump(db, f, ensure_ascii=False, separators=(",", ":")) size_mb = os.path.getsize(OUTPUT) / (1024*1024) print(f"Written to {OUTPUT} ({size_mb:.1f} MB)") print("Done!")