Turn by turn directions

Makefile

@@ -1,5 +1,5 @@
 .PHONY: all
-all: db/import/mellowroute.fixture db/import/chicago.table
+all: db/import/mellowroute.fixture db/import/chicago.table db/import/chicago_parks.table
 
 db/import/%.fixture: app/mbm/fixtures/%.json
 	(cd app && python manage.py loaddata $*) && touch $@
@@ -14,6 +14,12 @@ db/import/chicago.table: db/raw/chicago-filtered.osm
 		AND osm_ways.tags @> 'oneway:bicycle => no'" && \
 	touch $@
 
+db/import/chicago_parks.table: db/chicago_parks.geojson
+	ogr2ogr -f "PostgreSQL" PG:"dbname=mbm user=postgres password=postgres host=postgres" \
+	        $< -nln chicago_parks -overwrite -lco GEOMETRY_NAME=wkb_geometry && \
+	PGPASSWORD=postgres psql -U postgres -h postgres -d mbm -f /app/db/label-parks.sql && \
+	touch $@
+
 
 db/raw/chicago-filtered.osm: db/raw/chicago.osm
 	osmconvert $< --drop-author --drop-version --out-osm -o="$@"

README.md

@@ -39,7 +39,9 @@ The app will be available on http://localhost:8000.
 
 ### Testing
 
-To run backend tests:
+#### Automated
+
+To run the Python test suite:
 
 ```
 docker compose run --rm app sh -c "python -m pytest /app/tests"
@@ -51,6 +53,19 @@ To run frontend tests:
 docker compose run --rm app sh -c "npm test"
 ```
 
+#### Manual
+
+To generate directions on the command line:
+
+docker compose run --rm app sh -c "python manage.py generate_route --sourceCoordinates 'x, y' --targetCoordinates 'x, y'"
+
+To fuzz turn-by-turn directions (to find routes that don't work, tag sets that look iffy for bikes, and streets that show up as unnamed in directions):
+
+```
+docker compose run --rm app sh -c "python /app/scripts/fuzz_directions.py --runs 5"
+```
+
+
 ## Mapping
 
 To begin mapping, make sure you've created an admin user with

app/mbm/directions.py

@@ -0,0 +1,245 @@
+from typing import TypedDict, Dict, List, Optional, Any
+from mbm.types import Route, RouteProperties
+
+class DirectionSegment(TypedDict):
+    gid: Optional[int]
+    osmData: Dict[str, Any]
+    maneuver: str
+    cardinal: str
+    distance: float
+    name: Optional[str]
+    effectiveName: str
+    featureIndex: int
+
+class Direction(TypedDict):
+    directionSegments: List[DirectionSegment]
+    name: Optional[str]
+    effectiveName: str
+    distance: float
+    maneuver: str
+    heading: float
+    cardinal: str
+    type: Optional[str]
+    osmData: Dict[str, Any]
+    featureIndices: List[int]
+    directionText: str
+
+def directions_list(features: Route) -> List[Direction]:
+    directions: List[Direction] = []
+    previous_heading = None
+    previous_effective_name = None
+
+    for i, feature in enumerate(features):
+        props: RouteProperties = feature['properties']
+        name = props.get('name')
+        heading: float = props.get('heading', 0.0) or 0.0
+        distance: float = props.get('distance', 0)
+        route_type: Optional[str] = props.get('type')
+        osm_tags: Optional[Dict[str, str]] = props.get('osm_tags')
+        park_name: Optional[str] = props.get('park_name')
+
+        osm_data = {
+            'gid': props.get('gid'),
+            'osm_id': props.get('osm_id'),
+            'tag_id': props.get('tag_id'),
+            'oneway': props.get('oneway'),
+            'rule': props.get('rule'),
+            'priority': props.get('priority'),
+            'maxspeed_forward': props.get('maxspeed_forward'),
+            'maxspeed_backward': props.get('maxspeed_backward'),
+            'length_m': distance,
+            'osm_tags': osm_tags,
+            'park_name': park_name,
+        }
+
+        maneuver_info = _heading_to_english_maneuver(heading, previous_heading)
+        maneuver = maneuver_info['maneuver']
+        cardinal = maneuver_info['cardinal']
+
+        effective_name = name or _describe_unnamed_street(osm_tags, park_name)
+
+        direction_segment: DirectionSegment = {
+            'gid': props.get('gid'),
+            'osmData': osm_data,
+            'maneuver': maneuver,
+            'cardinal': cardinal,
+            'distance': distance,
+            'name': name,
+            'effectiveName': effective_name,
+            'featureIndex': i,
+        }
+
+        direction: Direction = {
+            'directionSegments': [direction_segment],
+            'name': name,
+            'effectiveName': effective_name,
+            'distance': distance,
+            'maneuver': maneuver,
+            'heading': heading,
+            'cardinal': cardinal,
+            'type': props.get('type'),
+            'osmData': osm_data,
+            'featureIndices': [i],
+            'directionText': '',
+        }
+
+        if _should_merge_segments(maneuver, effective_name, previous_effective_name, name):
+            _merge_with_previous_direction(directions, direction_segment)
+        else:
+            directions.append(direction)
+
+        previous_heading = heading
+        previous_effective_name = effective_name
+
+    for index, direction in enumerate(directions):
+        street_name = direction.get('effectiveName') or direction.get('name') or 'an unknown street'
+        distance_text = _format_distance(direction['distance'])
+        if index == 0:
+            direction_text = f"Head {direction['cardinal']} on {street_name} for {distance_text}"
+        else:
+            direction_text = (
+                f"{direction['maneuver']} onto {street_name} and head "
+                f"{direction['cardinal']} for {distance_text}"
+            )
+
+        if index == len(directions) - 1:
+            direction_text += ' until you reach your destination'
+
+        direction['directionText'] = direction_text
+
+    return directions
+
+def _nearest_45(x: float) -> int:
+    """Round an angle to the nearest 45-degree increment (0-360)."""
+    return (round(x / 45) * 45) % 360
+
+def _format_distance(meters: float) -> str:
+    meters_per_mile = 1609.344
+    meters_per_foot = 0.3048
+    miles = meters / meters_per_mile
+    def round_half_up(value: float) -> int:
+        return int(value + 0.5)
+
+    if miles < 0.09:
+        feet = round_half_up(meters / meters_per_foot)
+        unit = 'foot' if feet == 1 else 'feet'
+        return f"{feet} {unit}"
+
+    rounded_miles = round_half_up(miles * 10) / 10
+    unit = 'mile' if rounded_miles == 1 else 'miles'
+    return f"{rounded_miles} {unit}"
+
+# Relevant docs:
+# https://wiki.openstreetmap.org/wiki/Key:highway
+# https://wiki.openstreetmap.org/wiki/Tag:highway
+def _describe_unnamed_street(
+    osm_tags: Optional[Dict[str, str]], 
+    park_name: Optional[str]
+) -> str:
+    if not osm_tags or not isinstance(osm_tags, dict):
+        return 'an unknown street'
+
+    description = ''
+
+    if osm_tags.get('highway') == 'service' and osm_tags.get('service') == 'alley':
+        description = 'an alley'
+    elif osm_tags.get('highway') == 'service':
+        description = 'an access road'
+    elif osm_tags.get('footway') == 'crossing':
+        description = 'a crosswalk'
+    elif osm_tags.get('highway') == 'cycleway':
+        description = 'a bike path'
+    elif osm_tags.get('bicycle') == 'designated':
+        description = 'a bike path'
+    elif osm_tags.get('highway') == 'pedestrian':
+        description = 'a pedestrian path'
+    elif osm_tags.get('highway') == 'footway' or osm_tags.get('footway') == 'sidewalk':
+        description = 'a sidewalk'
+    elif osm_tags.get('highway') == 'pedestrian' and osm_tags.get('bicycle') == 'yes':
+        description = 'a mixed-use path'
+    elif osm_tags.get('highway') == 'path' and osm_tags.get('bicycle') == 'permissive':
+        description = 'a mixed-use path'
+    elif park_name:
+        description = 'a path'
+    else:
+        description = 'an unknown street'
+
+    if park_name:
+        description += f' inside {park_name}'
+
+    return description
+
+
+def _heading_to_english_maneuver(
+    heading: float, 
+    previous_heading: Optional[float]
+) -> Dict[str, str]:
+    degrees = {
+        0: {'maneuver': 'Continue', 'cardinal': 'north'},
+        45: {'maneuver': 'Turn slightly to the right', 'cardinal': 'northeast'},
+        90: {'maneuver': 'Turn right', 'cardinal': 'east'},
+        135: {'maneuver': 'Take a sharp right turn', 'cardinal': 'southeast'},
+        180: {'maneuver': 'Turn around', 'cardinal': 'south'},
+        225: {'maneuver': 'Take a sharp left turn', 'cardinal': 'southwest'},
+        270: {'maneuver': 'Turn left', 'cardinal': 'west'},
+        315: {'maneuver': 'Turn slightly to the left', 'cardinal': 'northwest'},
+    }
+
+    if previous_heading is not None:
+        angle = _nearest_45(((heading - previous_heading) + 360) % 360)
+        maneuver = degrees.get(angle, {}).get('maneuver', 'Continue')
+    else:
+        maneuver = 'Continue'
+
+    cardinal = degrees.get(_nearest_45(heading), {}).get('cardinal', 'north')
+
+    return {'maneuver': maneuver, 'cardinal': cardinal}
+
+
+def _should_merge_segments(
+    maneuver: str,
+    new_segment_effective_name: str,
+    previous_segment_effective_name: Optional[str],
+    name: Optional[str],
+) -> bool:
+    # Merge continues and slight turns when two segments have the same street name
+    is_slight_turn = (maneuver == 'Turn slightly to the left' or 
+                     maneuver == 'Turn slightly to the right' or maneuver == 'Continue')
+    is_named_street = bool(name)
+    same_named_street = new_segment_effective_name == previous_segment_effective_name
+    if is_slight_turn and is_named_street and same_named_street:
+        return True
+
+    # Merge continues on unnamed streets
+    if maneuver == 'Continue' and not is_named_street:
+        return True
+
+    return False
+
+
+def _merge_with_previous_direction(
+    directions: List[Direction],
+    direction_segment: DirectionSegment,
+) -> None:
+    if not directions:
+        return
+
+    previous_direction = directions[-1]
+    previous_direction['distance'] += direction_segment['distance']
+    previous_direction['directionSegments'].append(direction_segment)
+    previous_direction['featureIndices'].append(direction_segment['featureIndex'])
+    name = direction_segment.get('name')
+    effective_name = direction_segment['effectiveName']
+    osm_data = direction_segment['osmData']
+
+    # Sometimes only some chicago_ways of a street are named, so check if the
+    # previous chicago_way is named and backfill the name if not
+    if name and not previous_direction['name']:
+        previous_direction['name'] = name
+        previous_direction['effectiveName'] = effective_name
+
+    # For unnamed streets, preserve OSM data from the current chicago_way if previous doesn't have a name
+    if not name and osm_data and not previous_direction['name']:
+        previous_direction['osmData'] = osm_data
+        previous_direction['effectiveName'] = effective_name
+

app/mbm/management/commands/generate_directions.py

@@ -0,0 +1,95 @@
+import json
+from typing import List
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import connection
+
+from mbm.directions import directions_list
+from mbm.models import fetchall
+from mbm.routing import calculate_route
+
+
+def parse_coordinate_param(value: str) -> List[float]:
+    if not value:
+        raise CommandError("Coordinate value is required.")
+    parts = value.split(",")
+    if len(parts) != 2:
+        raise CommandError("Coordinate must be in 'lat,lng' format.")
+    try:
+        lat = float(parts[0])
+        lng = float(parts[1])
+    except ValueError as exc:
+        raise CommandError("Coordinate must be in 'lat,lng' format.") from exc
+    return [lat, lng]
+
+
+def get_nearest_vertex_id(coord: List[float]) -> int:
+    with connection.cursor() as cursor:
+        cursor.execute(
+            """
+            SELECT vert.id
+            FROM chicago_ways_vertices_pgr AS vert
+            ORDER BY vert.the_geom <-> ST_SetSRID(
+                ST_MakePoint(%s, %s),
+                4326
+            )
+            LIMIT 1
+            """,
+            [coord[1], coord[0]],  # ST_MakePoint expects lng,lat
+        )
+        rows = fetchall(cursor)
+    if rows:
+        return rows[0]["id"]
+    raise CommandError(f"No vertex found near point {coord[0]},{coord[1]}.")
+
+
+class Command(BaseCommand):
+    help = "Generate directions from coordinate inputs."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--sourceCoordinates",
+            required=True,
+            help="Source coordinate in 'lat,lng' format.",
+        )
+        parser.add_argument(
+            "--targetCoordinates",
+            required=True,
+            help="Target coordinate in 'lat,lng' format.",
+        )
+        parser.add_argument(
+            "--enable-v2",
+            action="store_true",
+            help="Enable the v2 routing costs.",
+        )
+
+    def handle(self, *_args, **options):
+        source_coord = parse_coordinate_param(options["sourceCoordinates"])
+        target_coord = parse_coordinate_param(options["targetCoordinates"])
+        enable_v2 = options["enable_v2"]
+
+        source_vertex_id = get_nearest_vertex_id(source_coord)
+        target_vertex_id = get_nearest_vertex_id(target_coord)
+
+        features, _, _ = calculate_route(
+            source_vertex_id,
+            target_vertex_id,
+            enable_v2,
+        )
+        directions = directions_list(features)
+
+        lines = []
+        for direction in directions:
+            direction_text = direction.get("directionText", "")
+            lines.append(direction_text)
+
+            for segment in direction.get("directionSegments", []):
+                feature_index = segment.get("featureIndex")
+                gid = segment.get("gid")
+                name = segment.get("name") or segment.get("effectiveName") or "an unknown street"
+                distance = segment.get("distance", 0)
+                lines.append(
+                    f"  way {feature_index}: {name} (gid: {gid}; distance: {distance}m)"
+                )
+
+        self.stdout.write("\n".join(lines))