polyline_generator.py

#!/usr/bin/env python3

import os
import time
import requests
import pandas as pd
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
import threading

IN = "output/crm_gtfs_02032026.csv"
OLD = "output/crm_gtfs_02032026.csv"
OUT = "output/crm_gtfs_02032026_polylines.csv"

OSRM = "localhost:5000/route/v1/foot/"
WORKERS = 4
RETRIES = 3
TIMEOUT = 60
BAD_LAT = 12.549663
THRESH = 0.7

session = requests.Session()
lock = threading.Lock()

if not os.path.exists(OUT):
    pd.DataFrame(columns=["route_id", "polyline"]).to_csv(OUT, index=False)

def get_polyline(coords):
    if len(coords) < 2:
        return ""

    coord_str = ";".join([f"{lon},{lat}" for lat, lon in coords])
    url = f"{OSRM}{coord_str}?overview=full&geometries=polyline"

    for _ in range(RETRIES):
        try:
            r = session.get(url, timeout=TIMEOUT)
            if r.status_code == 200:
                js = r.json()
                if js.get("routes"):
                    return js["routes"][0]["geometry"]
        except:
            pass
        time.sleep(1)
    return ""

def route_job(item):
    rid, coords = item
    poly = get_polyline(coords)

    with lock:
        pd.DataFrame([[rid, poly]], columns=["route_id","polyline"]) \
          .to_csv(OUT, mode="a", header=False, index=False)
    return rid

def main():
    old = pd.read_csv(OLD)
    old = old.groupby("Route ID")["Latitude"] \
             .apply(lambda x: (x == BAD_LAT).sum() / len(x)) \
             .reset_index(name="bad_ratio")

    valid_ids = set(old[old.bad_ratio < THRESH]["Route ID"].astype(str))

    df = pd.read_csv(IN)
    df = df[df["Route ID"].astype(str).isin(valid_ids)]
    df = df[df["Latitude"] != BAD_LAT]
    df = df.sort_values(["Route ID", "Sequence"])

    route_coords = defaultdict(list)
    for _, r in df.iterrows():
        route_coords[str(r["Route ID"])].append(
            (float(r["Latitude"]), float(r["Longitude"]))
        )

    items = list(route_coords.items())

    with ThreadPoolExecutor(max_workers=WORKERS) as ex:
        futures = [ex.submit(route_job, it) for it in items]
        for _ in tqdm(as_completed(futures), total=len(futures)):
            pass

if __name__ == "__main__":
    main()