def conflict_map(feed: "Feed", zoom_start: int = 12) -> folium.Map:
"""Builds the `direction_id` conflict inspector map for `feed`.
Args:
feed: A `Feed` whose `feed.shapes.stop_shapes` has already been
through `assign_direction_ids` (this happens automatically
during `Feed` construction whenever `shapes.txt`-derived
geometry is available).
zoom_start: Initial Leaflet zoom level.
"""
stop_shapes_df = (
feed.shapes.stop_shapes.select(
[
"shape_id",
"stop_id",
"stop_sequence",
"shape_pt_lat",
"shape_pt_lon",
"shape_total_distance",
"route_id",
"direction_id",
"direction_conflict",
"shape_direction",
"shape_direction_backwards",
]
)
.drop_nulls(["shape_pt_lat", "shape_pt_lon"])
.collect()
)
parent_by_stop = feed.stops.lf.select(["stop_id", "parent_station"]).collect()
stop_shapes_df = stop_shapes_df.join(parent_by_stop, on="stop_id", how="left").with_columns(
pl.col("parent_station").fill_null(pl.col("stop_id")).alias("station")
)
station_pos = stop_shapes_df.group_by("station").agg(
pl.col("shape_pt_lat").mean().alias("lat"),
pl.col("shape_pt_lon").mean().alias("lon"),
)
conflict_per_station = stop_shapes_df.group_by("station").agg(
pl.col("direction_conflict").any().alias("conflict")
)
stations_meta = station_pos.join(conflict_per_station, on="station")
# The same per-(route_id, station) widest-gap split boundary
# `_assign_direction_ids_for_route` computes internally, recomputed
# here from the same forward/backward-forced-180 bearings -- pooled
# across the station's platform stop_ids, same as the real assignment
# -- so a clicked stop marker can show *why* it got its direction_id,
# not just what it is. `None` when there was nothing to split (e.g.
# every shape here has an identical bearing, or only one shape serves
# this station).
split_angle_by_route_station: dict[tuple, float | None] = {}
for (route_id, station), sub in stop_shapes_df.group_by(["route_id", "station"], maintain_order=True):
shape_angles = {}
for shape_id, fwd, bwd in zip(sub["shape_id"], sub["shape_direction"], sub["shape_direction_backwards"]):
angles = _reconcile_fwd_bwd(fwd, bwd)
if angles:
shape_angles[shape_id] = angles
split_angle_by_route_station[(route_id, station)] = _split_angle(shape_angles) if shape_angles else None
def _round_or_none(v):
return None if v is None or (isinstance(v, float) and v != v) else round(float(v), 1)
def _shape_stops(shape_id: str, route_id: str) -> list[dict]:
shape_full = stop_shapes_df.filter(
(pl.col("shape_id") == shape_id) & (pl.col("route_id") == route_id)
).sort("stop_sequence")
out = []
for r in shape_full.iter_rows(named=True):
corrected = _reconcile_fwd_bwd(r["shape_direction"], r["shape_direction_backwards"])
corrected_fwd = corrected[0] if len(corrected) >= 1 else None
corrected_bwd = corrected[1] if len(corrected) >= 2 else None
# `direction_id` is constant across every stop of a shape (see
# step 8 of `_assign_direction_ids_for_route`); `conflict`
# marks the specific stop(s) where that reported value
# disagreed with the local geometry. Since direction_id is
# always binary, the disagreeing stop's *actual* local reading
# is simply the other value -- surfaced here as
# `real_direction_id` so the map can show it without
# recomputing anything.
conflict = bool(r["direction_conflict"])
real_direction_id = (1 - r["direction_id"]) if (conflict and r["direction_id"] is not None) else r["direction_id"]
out.append(
{
"stop_id": r["stop_id"],
"stop_sequence": r["stop_sequence"],
"lat": r["shape_pt_lat"],
"lon": r["shape_pt_lon"],
"direction_id": r["direction_id"],
"conflict": conflict,
"real_direction_id": real_direction_id,
"split_angle": _round_or_none(split_angle_by_route_station.get((route_id, r["station"]))),
"fwd_raw": _round_or_none(r["shape_direction"]),
"bwd_raw": _round_or_none(r["shape_direction_backwards"]),
"fwd_corrected": _round_or_none(corrected_fwd),
"bwd_corrected": _round_or_none(corrected_bwd),
}
)
return out
conflicts = {}
for station in stations_meta.filter(pl.col("conflict"))["station"]:
sub = stop_shapes_df.filter(pl.col("station") == station)
route_ids = sub.filter(pl.col("direction_conflict"))["route_id"].unique().to_list()
route_entries = {}
for route_id in route_ids:
route_sub = sub.filter(pl.col("route_id") == route_id)
conflicting_shape_ids = sorted(
route_sub.filter(pl.col("direction_conflict"))["shape_id"].unique().to_list()
)
conflicting = {
shape_id: {"stops": _shape_stops(shape_id, route_id)}
for shape_id in conflicting_shape_ids
}
# Every conflicting shape_id's reported `direction_id` is constant
# (see step 8 of `_assign_direction_ids_for_route`), so its
# "real" direction -- the one its own geometry actually
# supports, at least at this flagged station -- is simply the
# other binary value. These get appended to that direction's
# dropdown too (after the genuinely non-conflicting shapes),
# so a conflicting shape can be visually compared against the
# direction its geometry suggests it actually belongs to; the
# `conflict: True` flag lets the UI render that entry's label
# in red as a reminder it's not an unconditional match.
conflict_real_direction = {
shape_id: 1 - route_sub.filter(pl.col("shape_id") == shape_id)["direction_id"][0]
for shape_id in conflicting_shape_ids
}
ok_by_direction = {}
for dir_value in (0, 1):
candidates = (
route_sub.filter(
(~pl.col("direction_conflict")) & (pl.col("direction_id") == dir_value)
)
.select(["shape_id", "shape_total_distance"])
.unique()
.sort("shape_total_distance", descending=True)
)
entries = [
{
"shape_id": r["shape_id"],
"length": None if r["shape_total_distance"] is None else round(float(r["shape_total_distance"]), 1),
"stops": _shape_stops(r["shape_id"], route_id),
"conflict": False,
}
for r in candidates.iter_rows(named=True)
]
conflicting_candidates = sorted(
(
shape_id
for shape_id in conflicting_shape_ids
if conflict_real_direction[shape_id] == dir_value
),
key=lambda shape_id: route_sub.filter(pl.col("shape_id") == shape_id)["shape_total_distance"][0] or 0,
reverse=True,
)
entries += [
{
"shape_id": shape_id,
"length": (
None
if route_sub.filter(pl.col("shape_id") == shape_id)["shape_total_distance"][0] is None
else round(float(route_sub.filter(pl.col("shape_id") == shape_id)["shape_total_distance"][0]), 1)
),
"stops": _shape_stops(shape_id, route_id),
"conflict": True,
}
for shape_id in conflicting_candidates
]
if entries:
ok_by_direction[str(dir_value)] = entries
route_entries[route_id] = {"conflicting": conflicting, "ok": ok_by_direction}
conflicts[station] = route_entries
data = {
"stations": [
{"station": r["station"], "lat": r["lat"], "lon": r["lon"], "conflict": bool(r["conflict"])}
for r in stations_meta.iter_rows(named=True)
],
"conflicts": conflicts,
}
mean_lat = stations_meta["lat"].mean()
mean_lon = stations_meta["lon"].mean()
m = folium.Map(location=[mean_lat, mean_lon], zoom_start=zoom_start, tiles=None)
add_basemap(m)
panels_html = '<div id="conflict-panel-routes"></div><div id="conflict-panel-shapes"></div>'
data_json = json.dumps(data, separators=(",", ":")).replace("</", "<\\/")
js_code = (
_read_static("conflict_map.js")
.replace("__MAP_VAR__", m.get_name())
.replace("__DATA_JSON__", data_json)
)
css_code = _read_static("conflict_map.css")
m.get_root().html.add_child(folium.Element(panels_html))
m.get_root().html.add_child(folium.Element(f"<style>{css_code}</style>"))
m.get_root().script.add_child(folium.Element(js_code))
return m