Skip to content

poi

POI-congestion density analysis using OpenStreetMap data via OSMnx.

poi

POI-congestion density analysis.

Downloads Points of Interest from OpenStreetMap via OSMnx, computes POI density around traffic segments, and correlates density with jam factor.

download_pois 

download_pois(city_code, category_name, tags)

Download POIs from OSM for city_code matching tags.

Source code in src/trafficpipeline/poi.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def download_pois(
    city_code: str,
    category_name: str,
    tags: dict,
) -> gpd.GeoDataFrame:
    """Download POIs from OSM for *city_code* matching *tags*."""
    city = CITIES[city_code]
    try:
        pois = ox.features_from_bbox(bbox=city["bbox"], tags=tags)
        if len(pois) > 0:
            pois = pois.copy()
            pois["geometry"] = pois.geometry.centroid
            pois = pois[pois.geometry.type == "Point"]
        return pois
    except Exception:
        return gpd.GeoDataFrame()

compute_poi_density 

compute_poi_density(traffic_gdf, poi_gdf, buffer_distance=0.005)

Count POIs within buffer_distance (degrees) of each traffic centroid.

Source code in src/trafficpipeline/poi.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
def compute_poi_density(
    traffic_gdf: gpd.GeoDataFrame,
    poi_gdf: gpd.GeoDataFrame,
    buffer_distance: float = 0.005,
) -> np.ndarray:
    """Count POIs within *buffer_distance* (degrees) of each traffic centroid."""
    if len(poi_gdf) == 0:
        return np.zeros(len(traffic_gdf))
    centroids = traffic_gdf.geometry.centroid
    counts = []
    for c in centroids:
        counts.append(int(poi_gdf.geometry.within(c.buffer(buffer_distance)).sum()))
    return np.array(counts)

analyze_city 

analyze_city(city_code, base_dir='.', buffer_distance=0.003)

Run POI-congestion analysis for one city.

Returns (summary_results, category_results, traffic_gdf).

Source code in src/trafficpipeline/poi.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def analyze_city(
    city_code: str,
    base_dir: str | Path = ".",
    buffer_distance: float = 0.003,
) -> tuple[dict, list[dict], gpd.GeoDataFrame]:
    """Run POI-congestion analysis for one city.

    Returns ``(summary_results, category_results, traffic_gdf)``.
    """
    city = CITIES[city_code]
    city_name = city["name"]
    print(f"\nAnalyzing {city_name} …")

    folder = Path(base_dir) / city["traffic_output_dir"]
    traffic = gpd.read_file(str(folder / f"evening_peak_{city_code}.gpkg"))
    if traffic.crs is None:
        traffic = traffic.set_crs("EPSG:4326")

    results: dict = {"city": city_name, "n_segments": len(traffic)}
    all_poi_counts = np.zeros(len(traffic))
    cat_results: list[dict] = []

    for cat_name, cat_info in POI_CATEGORIES.items():
        pois = download_pois(city_code, cat_name, cat_info["tags"])
        n_pois = len(pois)
        if n_pois == 0:
            continue

        if pois.crs is None:
            pois = pois.set_crs("EPSG:4326")
        elif pois.crs != traffic.crs:
            pois = pois.to_crs(traffic.crs)

        counts = compute_poi_density(traffic, pois, buffer_distance)
        all_poi_counts += counts

        jf = traffic["jam_factor_mean"].values
        mask = ~(np.isnan(jf) | np.isnan(counts))
        if mask.sum() > 10:
            r, p = stats.pearsonr(jf[mask], counts[mask])
            rho, rho_p = stats.spearmanr(jf[mask], counts[mask])
            cat_results.append({
                "category": cat_name,
                "description": cat_info["description"],
                "n_pois": n_pois,
                "pearson_r": r, "pearson_p": p,
                "spearman_r": rho, "spearman_p": rho_p,
            })

    # Total density correlation
    jf = traffic["jam_factor_mean"].values
    mask = ~(np.isnan(jf) | np.isnan(all_poi_counts))
    if mask.sum() > 10:
        r_t, p_t = stats.pearsonr(jf[mask], all_poi_counts[mask])
        rho_t, rho_p_t = stats.spearmanr(jf[mask], all_poi_counts[mask])
        results.update(total_pois=int(all_poi_counts.sum()),
                       total_pearson_r=r_t, total_pearson_p=p_t,
                       total_spearman_r=rho_t, total_spearman_p=rho_p_t)

    traffic["poi_density"] = all_poi_counts
    return results, cat_results, traffic

plot_scatter 

plot_scatter(traffic, city_code, figures_dir='figures')

Scatter plot: POI density vs jam factor.

Source code in src/trafficpipeline/poi.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def plot_scatter(traffic: gpd.GeoDataFrame, city_code: str, figures_dir: str | Path = "figures") -> Path:
    """Scatter plot: POI density vs jam factor."""
    fig_dir = Path(figures_dir)
    fig_dir.mkdir(exist_ok=True)
    city = CITIES[city_code]

    fig, ax = plt.subplots(figsize=(10, 8))
    x, y = traffic["poi_density"].values, traffic["jam_factor_mean"].values
    mask = ~(np.isnan(x) | np.isnan(y))
    x, y = x[mask], y[mask]
    ax.scatter(x, y, alpha=0.4, s=20, color=city["color"], edgecolors="white", linewidth=0.3)

    slope, intercept, r, p, _ = stats.linregress(x, y)
    xl = np.linspace(x.min(), x.max(), 100)
    ax.plot(xl, slope * xl + intercept, "r-", lw=2, label=f"r={r:.4f} (p={p:.4f})")

    ax.set_xlabel("POI Density (count within 300 m)")
    ax.set_ylabel("Jam Factor")
    ax.set_title(f"{city['name']}: POI Density vs Congestion", fontweight="bold")
    ax.legend()
    ax.grid(alpha=0.3)
    fp = fig_dir / f"{city_code}_poi_congestion_scatter.png"
    plt.savefig(fp, dpi=150, bbox_inches="tight")
    plt.close()
    return fp

plot_category_comparison 

plot_category_comparison(all_cat_results, figures_dir='figures')

Bar chart of per-category Spearman correlations.

Source code in src/trafficpipeline/poi.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def plot_category_comparison(
    all_cat_results: dict[str, list[dict]],
    figures_dir: str | Path = "figures",
) -> Path:
    """Bar chart of per-category Spearman correlations."""
    fig_dir = Path(figures_dir)
    fig_dir.mkdir(exist_ok=True)

    categories = list(POI_CATEGORIES.keys())
    x = np.arange(len(categories))
    w = 0.25

    fig, ax = plt.subplots(figsize=(14, 8))
    for i, (code, cats) in enumerate(all_cat_results.items()):
        vals = []
        for cat in categories:
            row = next((c for c in cats if c["category"] == cat), None)
            vals.append(row["spearman_r"] if row else 0)
        ax.bar(x + i * w, vals, w, label=CITIES[code]["name"], color=CITIES[code]["color"], alpha=0.8)

    ax.axhline(0, color="black", lw=0.5)
    ax.set_xlabel("POI Category")
    ax.set_ylabel("Spearman Correlation")
    ax.set_title("POI-Congestion Correlation by Category", fontweight="bold")
    ax.set_xticks(x + w)
    ax.set_xticklabels([POI_CATEGORIES[c]["description"] for c in categories], rotation=45, ha="right")
    ax.legend()
    ax.grid(axis="y", alpha=0.3)
    plt.tight_layout()
    fp = fig_dir / "poi_category_comparison.png"
    plt.savefig(fp, dpi=150, bbox_inches="tight")
    plt.close()
    return fp

run_analysis 

run_analysis(base_dir='.', figures_dir='figures', output_dir='analysis_results')

Run POI analysis for all cities.

Source code in src/trafficpipeline/poi.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def run_analysis(
    base_dir: str | Path = ".",
    figures_dir: str | Path = "figures",
    output_dir: str | Path = "analysis_results",
) -> None:
    """Run POI analysis for all cities."""
    out = Path(output_dir)
    out.mkdir(exist_ok=True)

    all_results, all_cats = [], {}
    for code in CITIES:
        res, cats, traffic = analyze_city(code, base_dir)
        all_results.append(res)
        all_cats[code] = cats
        plot_scatter(traffic, code, figures_dir)

    plot_category_comparison(all_cats, figures_dir)
    pd.DataFrame(all_results).to_csv(out / "poi_congestion_correlations.csv", index=False)
    print("POI analysis complete.")