Source Copenhagen transit data from Rejseplanen GTFS
- Replace OSM Overpass transit data with the Rejseplanen GTFS feed (routes keyed by (agency, short name); styles from modes.yaml) - Draw one shape per (style, ref, direction), choosing the shape that serves the most in-area stops so lines pass the stops we show - Collapse stops by names (verified unambiguous); prune stops whose serving refs have no drawn line within 300 m - Patch Københavns Havn + Nordhavn into the area polygon so ferry routes and sub-harbour metro tunnels survive clipping; 100 m buffer closes relation boundary slivers - Drop OSM download pipeline; keep tiled basemap
This commit is contained in:
+42
-176
@@ -1,14 +1,19 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Render the Copenhagen transit map to PNG/SVG/PDF.
|
||||
|
||||
Reads data/processed/master.gpkg (layers: lines, stations) + area.gpkg +
|
||||
config/styling.yaml, applies CLI filters, and composes a printable map:
|
||||
- Carto Positron (no labels) basemap via contextily (EPSG:3857)
|
||||
Reads data/processed/master.gpkg (layers: lines, stops, stations — all
|
||||
GTFS-derived by prepare.py) + area.gpkg + config/styling.yaml, applies CLI
|
||||
filters, and composes a printable map:
|
||||
- Carto/Esri grey basemap via contextily (EPSG:3857)
|
||||
- two-tone lines: dark casing + colour body, per route
|
||||
- z-order: bus (bottom) -> s_tog -> light_rail -> regional -> metro (top)
|
||||
- shapeburst fade mask outside the City Pass area
|
||||
- station labels (metro + S-tog) via adjustText
|
||||
|
||||
All transit data comes from the Rejseplanen GTFS feed. Stop/station layers
|
||||
are pre-collapsed (one point per stop name / station) by prepare.py, so
|
||||
rendering is pure plotting — no clustering or merging here.
|
||||
|
||||
Usage:
|
||||
uv run python render.py # default: all modes, PNG, markers only
|
||||
uv run python render.py --labels # add station name labels
|
||||
@@ -25,6 +30,7 @@ import matplotlib
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import yaml
|
||||
from affine import Affine
|
||||
from adjustText import adjust_text
|
||||
@@ -81,15 +87,11 @@ def filter_lines(lines, args):
|
||||
refs = list(dict.fromkeys(bus["ref"].dropna()))
|
||||
keep_refs = set(refs[:args.max_bus_routes])
|
||||
bus = bus[bus["ref"].isin(keep_refs)]
|
||||
out = gpd.GeoDataFrame(pd_concat([other, bus]), crs=out.crs)
|
||||
out = gpd.GeoDataFrame(pd.concat([other, bus], ignore_index=True),
|
||||
crs=out.crs)
|
||||
return out
|
||||
|
||||
|
||||
def pd_concat(frames):
|
||||
import pandas as pd
|
||||
return pd.concat(frames, ignore_index=True)
|
||||
|
||||
|
||||
def add_basemap(ax, styling, zoom=None):
|
||||
if zoom is None:
|
||||
zoom = styling["basemap"]["zoom"]
|
||||
@@ -158,60 +160,23 @@ def plot_lines(ax, lines, styling):
|
||||
capstyle="round", joinstyle="round")
|
||||
|
||||
|
||||
RAIL_STYLES = ("metro", "s_tog", "light_rail", "regional")
|
||||
|
||||
|
||||
def classify_station(r):
|
||||
"""Return the style key for a station row, or None."""
|
||||
if r.get("subway") == "yes" or r.get("station") == "subway":
|
||||
return "metro"
|
||||
if r.get("light_rail") == "yes" or r.get("station") == "light_rail":
|
||||
return "s_tog"
|
||||
if r.get("railway") == "station" or r.get("train") == "yes":
|
||||
return "regional"
|
||||
return None
|
||||
|
||||
|
||||
DISTANCE_CRS = "EPSG:25832" # UTM 32N — accurate meters for Copenhagen
|
||||
|
||||
|
||||
def cluster_stops(coords, threshold_m):
|
||||
"""Cluster (x, y) coordinates in EPSG:3857 within threshold_m (real meters).
|
||||
|
||||
Reprojects to UTM 32N for accurate distance computation, clusters, then
|
||||
returns centroids in EPSG:3857 for plotting.
|
||||
"""
|
||||
from pyproj import Transformer
|
||||
from scipy.cluster.hierarchy import fcluster, linkage
|
||||
from scipy.spatial.distance import pdist
|
||||
|
||||
coords = np.asarray(coords)
|
||||
if len(coords) <= 1:
|
||||
return coords.tolist() if len(coords) else []
|
||||
|
||||
# reproject to UTM for accurate distances
|
||||
to_utm = Transformer.from_crs(TARGET_CRS, DISTANCE_CRS, always_xy=True)
|
||||
back = Transformer.from_crs(DISTANCE_CRS, TARGET_CRS, always_xy=True)
|
||||
utm = np.array([to_utm.transform(x, y) for x, y in coords])
|
||||
|
||||
dists = pdist(utm)
|
||||
links = linkage(dists, method="single")
|
||||
labels = fcluster(links, t=threshold_m, criterion="distance")
|
||||
|
||||
centroids = []
|
||||
for label in set(labels):
|
||||
members = utm[labels == label]
|
||||
ux, uy = members.mean(axis=0)
|
||||
cx, cy = back.transform(ux, uy)
|
||||
centroids.append((cx, cy))
|
||||
return centroids
|
||||
"""Style key for a station row, or None. Stations are pre-classified."""
|
||||
s = r.get("style")
|
||||
return s if s in RAIL_STYLES else None
|
||||
|
||||
|
||||
def plot_stations(ax, stations, styling, active_styles):
|
||||
"""Draw station markers. Input is pre-collapsed: one row per (name, style)."""
|
||||
if stations is None or stations.empty:
|
||||
return
|
||||
cfg = styling.get("stations", {})
|
||||
if not cfg.get("show", True):
|
||||
return
|
||||
marker_styles = set(cfg.get("styles", ["metro", "s_tog", "regional"]))
|
||||
cluster_cfg = cfg.get("stop_cluster_m", {})
|
||||
mk = cfg.get("marker", {})
|
||||
sizes = mk.get("size", {})
|
||||
shape = mk.get("shape", "circle")
|
||||
@@ -220,37 +185,25 @@ def plot_stations(ax, stations, styling, active_styles):
|
||||
lw = mk.get("linewidth", 0.8)
|
||||
marker = "o" if shape == "circle" else "s"
|
||||
|
||||
# group coordinates by style, then by name for clustering
|
||||
pts_by_style = {}
|
||||
names_by_style = {}
|
||||
for _, r in stations.iterrows():
|
||||
s = classify_station(r)
|
||||
if s is None or s not in marker_styles or s not in active_styles:
|
||||
continue
|
||||
coord = (r.geometry.x, r.geometry.y)
|
||||
name = r.get("name") if isinstance(r.get("name"), str) else None
|
||||
pts_by_style.setdefault(s, []).append(coord)
|
||||
names_by_style.setdefault(s, {}).setdefault(name, []).append(coord)
|
||||
pts_by_style.setdefault(s, []).append((r.geometry.x, r.geometry.y))
|
||||
|
||||
zord = styling["zorder"]
|
||||
top_z = max(zord.values()) + 1 # all markers above all lines
|
||||
for s in sorted(pts_by_style, key=lambda k: zord.get(k, 0)):
|
||||
threshold = cluster_cfg.get(s, 50) if isinstance(cluster_cfg, dict) else cluster_cfg
|
||||
# cluster same-name stations to one marker; keep unnamed as-is
|
||||
centroids = []
|
||||
for name, coords in names_by_style[s].items():
|
||||
if name and len(coords) > 1:
|
||||
centroids.extend(cluster_stops(coords, threshold))
|
||||
else:
|
||||
centroids.extend(coords)
|
||||
pts = pts_by_style[s]
|
||||
sz = sizes.get(s, 3.0)
|
||||
ax.scatter([p[0] for p in centroids], [p[1] for p in centroids],
|
||||
ax.scatter([p[0] for p in pts], [p[1] for p in pts],
|
||||
s=sz ** 2, marker=marker, c=fill, edgecolors=edge,
|
||||
linewidths=lw, zorder=top_z, alpha=1.0)
|
||||
|
||||
|
||||
def rail_station_names(stations):
|
||||
"""Return the set of names for rail stations (metro/s_tog/regional)."""
|
||||
"""Set of names of rail stations (metro/s_tog/light_rail/regional)."""
|
||||
if stations is None or stations.empty:
|
||||
return set()
|
||||
names = set()
|
||||
@@ -262,109 +215,34 @@ def rail_station_names(stations):
|
||||
return names
|
||||
|
||||
|
||||
def group_by_route_membership(rows, cluster_m):
|
||||
"""Group stop rows by (name, shared route_ref) into distinct stops.
|
||||
|
||||
Two same-name stops that share at least one route_ref are the same stop.
|
||||
Stops on disjoint routes are different stops. Within each group,
|
||||
platform duplicates are merged by distance clustering.
|
||||
Returns a list of (x, y) centroids.
|
||||
"""
|
||||
if rows is None or rows.empty:
|
||||
return []
|
||||
|
||||
coords = [(r.geometry.x, r.geometry.y) for _, r in rows.iterrows()]
|
||||
route_sets = []
|
||||
for _, r in rows.iterrows():
|
||||
rr = r.get("route_refs")
|
||||
if isinstance(rr, str) and rr:
|
||||
route_sets.append(set(rr.split(";")))
|
||||
else:
|
||||
route_sets.append(set())
|
||||
|
||||
n = len(coords)
|
||||
# union-find: stops sharing a route are connected
|
||||
parent = list(range(n))
|
||||
|
||||
def find(x):
|
||||
while parent[x] != x:
|
||||
parent[x] = parent[parent[x]]
|
||||
x = parent[x]
|
||||
return x
|
||||
|
||||
def union(a, b):
|
||||
ra, rb = find(a), find(b)
|
||||
if ra != rb:
|
||||
parent[ra] = rb
|
||||
|
||||
for i in range(n):
|
||||
for j in range(i + 1, n):
|
||||
if route_sets[i] & route_sets[j]:
|
||||
union(i, j)
|
||||
|
||||
# group indices by connected component
|
||||
components = {}
|
||||
for i in range(n):
|
||||
root = find(i)
|
||||
components.setdefault(root, []).append(i)
|
||||
|
||||
centroids = []
|
||||
for indices in components.values():
|
||||
comp_coords = [coords[i] for i in indices]
|
||||
comp_routes = [route_sets[i] for i in indices]
|
||||
has_routes = any(rs for rs in comp_routes)
|
||||
if len(comp_coords) == 1:
|
||||
centroids.append(comp_coords[0])
|
||||
elif has_routes:
|
||||
# stops sharing a route are the same stop; merge to centroid
|
||||
centroids.append(tuple(np.mean(comp_coords, axis=0)))
|
||||
else:
|
||||
# no route info (orphans); fall back to distance clustering
|
||||
centroids.extend(cluster_stops(comp_coords, cluster_m))
|
||||
return centroids
|
||||
|
||||
|
||||
def plot_bus_stops(ax, stops, styling, active_styles, rail_names=None):
|
||||
"""Draw small markers for bus stops from the stops layer.
|
||||
"""Draw small markers for bus stops (pre-collapsed: one point per name).
|
||||
|
||||
Bus stops whose name matches a rail station are skipped — the rail
|
||||
station marker represents that stop. Stops are grouped by shared
|
||||
route membership: same-name stops on disjoint routes are distinct.
|
||||
A bus stop whose name exactly matches a rail station is skipped — the
|
||||
station marker represents it.
|
||||
"""
|
||||
if stops is None or stops.empty or "bus" not in active_styles:
|
||||
if stops is None or stops.empty or not ({"bus", "ferry"} & active_styles):
|
||||
return
|
||||
cfg = styling.get("stations", {})
|
||||
if not cfg.get("show", True) or "bus" not in set(cfg.get("styles", [])):
|
||||
return
|
||||
bus = stops[stops["style"] == "bus"]
|
||||
bus = stops[stops["style"].isin(["bus", "ferry"])]
|
||||
if bus.empty:
|
||||
return
|
||||
|
||||
if rail_names is None:
|
||||
rail_names = set()
|
||||
|
||||
cluster_cfg = cfg.get("stop_cluster_m", {})
|
||||
cluster_m = cluster_cfg.get("bus", 50) if isinstance(cluster_cfg, dict) else cluster_cfg
|
||||
rail_names = rail_names or set()
|
||||
mk = cfg.get("marker", {})
|
||||
sizes = mk.get("size", {})
|
||||
sz = sizes.get("bus", 1.5)
|
||||
sz = mk.get("size", {}).get("bus", 1.5)
|
||||
fill = mk.get("fill", "white")
|
||||
edge = mk.get("edge", "#2b2b2b")
|
||||
lw = mk.get("linewidth", 0.8)
|
||||
shape = mk.get("shape", "circle")
|
||||
marker = "o" if shape == "circle" else "s"
|
||||
marker = "o" if mk.get("shape", "circle") == "circle" else "s"
|
||||
top_z = max(styling["zorder"].values()) + 1
|
||||
|
||||
# group by name, then by route membership; skip rail station names
|
||||
centroids = []
|
||||
skipped = 0
|
||||
for name, grp in bus.groupby("name"):
|
||||
if name and name in rail_names:
|
||||
skipped += len(grp)
|
||||
continue
|
||||
centroids.extend(group_by_route_membership(grp, cluster_m))
|
||||
|
||||
ax.scatter([p[0] for p in centroids], [p[1] for p in centroids],
|
||||
pts = [(r.geometry.x, r.geometry.y)
|
||||
for _, r in bus.iterrows()
|
||||
if not (isinstance(r.get("name"), str) and r.get("name") in rail_names)]
|
||||
ax.scatter([p[0] for p in pts], [p[1] for p in pts],
|
||||
s=sz ** 2, marker=marker, c=fill, edgecolors=edge,
|
||||
linewidths=lw, zorder=top_z, alpha=0.8)
|
||||
|
||||
@@ -379,33 +257,21 @@ def label_stations(ax, stations, styling, active_styles):
|
||||
fmin = cfg.get("min_fontsize", 5)
|
||||
fmax = cfg.get("max_fontsize", 9)
|
||||
|
||||
def is_metro(r):
|
||||
return classify_station(r) == "metro"
|
||||
|
||||
def is_stog(r):
|
||||
return classify_station(r) == "s_tog"
|
||||
|
||||
pts = []
|
||||
# stations are unique per (name, style); dedupe by name for labelling
|
||||
seen = set()
|
||||
uniq = []
|
||||
for _, r in stations.iterrows():
|
||||
s = classify_station(r)
|
||||
if s not in label_styles:
|
||||
continue
|
||||
name = r.get("name")
|
||||
if not isinstance(name, str) or not name.strip():
|
||||
if not isinstance(name, str) or not name.strip() or name in seen:
|
||||
continue
|
||||
pts.append((r.geometry.x, r.geometry.y, name, s == "metro"))
|
||||
seen.add(name)
|
||||
uniq.append((r.geometry.x, r.geometry.y, name, s == "metro"))
|
||||
|
||||
if not pts:
|
||||
if not uniq:
|
||||
return
|
||||
# dedupe by name (keep first location)
|
||||
seen = {}
|
||||
uniq = []
|
||||
for x, y, name, is_m in pts:
|
||||
if name in seen:
|
||||
continue
|
||||
seen[name] = True
|
||||
uniq.append((x, y, name, is_m))
|
||||
|
||||
texts = []
|
||||
for x, y, name, is_m in uniq:
|
||||
fs = fmax if is_m else fmin
|
||||
@@ -485,7 +351,7 @@ def main():
|
||||
fontsize=fig_w * 1.1, fontweight="bold", color="#222",
|
||||
bbox=dict(boxstyle="round,pad=0.3", fc="white", ec="none", alpha=0.7))
|
||||
ax.text(0.01, 0.01,
|
||||
"Data: © OpenStreetMap contributors (ODbL) · Base: Esri, HERE",
|
||||
"Transit: Rejseplanen GTFS · Area: © OpenStreetMap contributors (ODbL) · Base: Esri, HERE",
|
||||
transform=ax.transAxes, ha="left", va="bottom",
|
||||
fontsize=max(5, fig_w * 0.5), color="#666")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user