implement automatic climb naming

2 files changed, 362 insertions, 0 deletions

diff --git a/fietsboek/geo.py b/fietsboek/geo.py
index ee26424..91da445 100644
--- a/fietsboek/geo.py
+++ b/fietsboek/geo.py
@@ -27,6 +27,8 @@ CLIMB_MIN_ASCENSION = 0.03
 CLIMB_MERGE_THRESHOLD = 50
 """Window for when climbs are merged (in m)."""
 
+NUM_SIMILARITY_SAMPLES = 10
+
 
 @dataclass
 class Waypoint:
@@ -275,6 +277,64 @@ class Path:
             else:
                 i += 1
 
+    def similarity(self, reference: "Path") -> float:
+        start_idx = 0
+        start_dst = float("+inf")
+        end_idx = len(reference.points) - 1
+        end_dst = float("+inf")
+
+        for i, point in enumerate(reference.points):
+            if (dst := self.points[0].flat_distance(point)) < start_dst:
+                start_idx = i
+                start_dst = dst
+            if (dst := self.points[-1].flat_distance(point)) < end_dst:
+                end_idx = i
+                end_dst = dst
+
+        if start_idx < end_idx:
+            reference = Path(reference.points[start_idx:end_idx + 1])
+        else:
+            subpath = reference.points[end_idx:start_idx + 1]
+            subpath.reverse()
+            reference = Path(subpath)
+
+        similarity = 0.0
+        for sample in range(NUM_SIMILARITY_SAMPLES):
+            where = sample / (NUM_SIMILARITY_SAMPLES - 1)
+            here = self.interpolate(where)
+            there = reference.interpolate(where)
+            dst = here.flat_distance(there)
+            similarity += dst
+
+        return similarity / NUM_SIMILARITY_SAMPLES
+
+    def interpolate(self, where: float) -> Point:
+        if len(self.points) == 1:
+            return self.points[0]
+
+        total_length = self._flat_length()
+        target = where * total_length
+        for i, (a, b) in enumerate(self._point_pairs()):
+            dst = a.flat_distance(b)
+            if dst < target and i < len(self.points) - 2:
+                target -= dst
+                continue
+            if dst == 0.0:
+                partial = 0.0
+            else:
+                partial = min(target / dst, 1.0)
+            return Point(
+                longitude=a.longitude + (b.longitude - a.longitude) * partial,
+                latitude=a.latitude + (b.latitude - a.latitude) * partial,
+                elevation=a.elevation + (b.elevation - a.elevation) * partial,
+                time_offset=a.time_offset + (b.time_offset - a.time_offset) * partial,
+            )
+
+    def _flat_length(self):
+        length = 0.0
+        for a, b in self._point_pairs():
+            length += a.flat_distance(b)
+        return length
 
 def gpx_xml(
     title: str | None,
diff --git a/fietsboek/osm.py b/fietsboek/osm.py
new file mode 100644
index 0000000..101bb89
--- /dev/null
+++ b/fietsboek/osm.py
@@ -0,0 +1,302 @@
+"""OSM related utilities for fietsboek."""
+
+import logging
+import pathlib
+from urllib.parse import urljoin
+from xml.etree import ElementTree
+
+import requests
+
+from . import __VERSION__
+from .geo import Path, Point
+
+LOGGER = logging.getLogger(__name__)
+API_URL = "https://api.openstreetmap.org/"
+HEADERS = {
+    "user-agent": f"Fietsboek/{__VERSION__}",
+}
+
+MAX_SIMILARITY_STREETNAME = 100
+MAX_DISTANCE_PEAK = 750
+MAX_DISTANCE_CITY = 750
+MAX_DISTANCE_WATER = 750
+
+
+def find_name(path: Path) -> str:
+    """Attempt to find a name for the given path.
+
+    :param path: Path to find a name for.
+    :return: Name for the path.
+    """
+    LOGGER.debug("Starting naming process")
+    session = requests.Session()
+
+    ((min_lat, min_lon), (max_lat, max_lon)) = path.bounds()
+
+    min_lat -= 0.001
+    min_lon -= 0.0015
+    max_lat += 0.001
+    max_lon += 0.0015
+
+    LOGGER.debug("bbox=%s,%s,%s,%s", min_lon, min_lat, max_lon, max_lat)
+
+    urlpath = f"api/0.6/map?bbox={min_lon},{min_lat},{max_lon},{max_lat}"
+    with requests.get(urljoin(API_URL, urlpath), headers=HEADERS) as response:
+        data = response.content
+
+    tree = ElementTree.fromstring(data)
+
+    name = _name_from_street(tree, path)
+    if name is None:
+        name = _name_from_peak(tree, path)
+    if name is None:
+        name = _name_from_water(tree, path)
+    if name is None:
+        name = _name_from_city(tree, path)
+    if name is None:
+        name = _fallback_name(tree, path)
+    if name is None:
+        name = "Climb"
+
+    LOGGER.debug("Final name: %s", name)
+    return name
+
+
+def _name_from_street(tree: ElementTree.ElementTree, path: Path) -> str | None:
+    LOGGER.debug("Attempting to name by street...")
+    node_to_coords = {}
+    for node in tree.iterfind("./node"):
+        node_id = node.get("id")
+        lat = float(node.get("lat"))
+        lon = float(node.get("lon"))
+        node_to_coords[node_id] = (lat, lon)
+
+    named_streets = {}
+
+    for way in tree.iterfind("./way"):
+        name_tag = way.find("./tag[@k='name']")
+        if name_tag is None:
+            continue
+        name = name_tag.get("v")
+
+        # Generally, highway=* indicates roads
+        highway_elem = way.find("./tag[@k='highway']")
+        if highway_elem is None:
+            continue
+
+        nodes = [nd.get("ref") for nd in way.iterfind("./nd")]
+
+        stretches = named_streets.setdefault(name, [])
+        stretches.append(nodes)
+
+    best_name = None
+    best_score = None
+
+    for name, stretches in named_streets.items():
+        merged_way = stretches[0]
+        remaining = stretches[1:]
+        while remaining:
+            num_remaining = len(remaining)
+            anchor = merged_way[-1]
+            for i, stretch in enumerate(remaining):
+                if stretch[0] == anchor:
+                    merged_way.extend(stretch)
+                    del remaining[i]
+                    break
+                if stretch[-1] == anchor:
+                    stretch.reverse()
+                    merged_way.extend(stretch)
+                    del remaining[i]
+                    break
+            anchor = merged_way[0]
+            for i, stretch in enumerate(remaining):
+                if stretch[-1] == anchor:
+                    stretch.extend(merged_way)
+                    merged_way = stretch
+                    del remaining[i]
+                    break
+                if stretch[0] == anchor:
+                    stretch.reverse()
+                    stretch.extend(merged_way)
+                    merged_way = stretch
+                    del remaining[i]
+                    break
+
+            if num_remaining == len(remaining):
+                # Made no progress, abort
+                break
+
+        points = []
+        for node_id in merged_way:
+            coords = node_to_coords[node_id]
+            point = Point(longitude=coords[1], latitude=coords[0], elevation=0.0, time_offset=0.0)
+            points.append(point)
+        if not points:
+            continue
+
+        reference_path = Path(points)
+        similarity = path.similarity(reference_path)
+
+        if best_score is None or similarity < best_score:
+            best_score = similarity
+            best_name = name
+
+    accepted = best_score is not None and best_score < MAX_SIMILARITY_STREETNAME
+    LOGGER.debug("Best name: %s (score=%s) (accepted=%s)", best_name, best_score, accepted)
+
+    if accepted:
+        return best_name
+    return None
+
+
+def _name_from_peak(tree: ElementTree.ElementTree, path: Path) -> str | None:
+    LOGGER.debug("Attempting to name by peak...")
+    best_name = None
+    best_distance = None
+    for peak in tree.iterfind("./node"):
+        if peak.find("./tag[@k='natural'][@v='peak']") is None:
+            continue
+
+        name_elem = peak.find("./tag[@k='name']")
+        if name_elem is None:
+            continue
+        name = name_elem.get("v")
+        lat = float(peak.get("lat"))
+        lon = float(peak.get("lon"))
+
+        peak_point = Point(longitude=lon, latitude=lat, elevation=0.0, time_offset=0.0)
+        distance = peak_point.flat_distance(path.points[-1])
+
+        if best_distance is None or distance < best_distance:
+            best_distance = distance
+            best_name = name
+
+    accepted = best_distance is not None and best_distance <= MAX_DISTANCE_PEAK
+    LOGGER.debug("Best name: %s (distance=%s) (accepted=%s)", best_name, best_distance, accepted)
+
+    if accepted:
+        return best_name
+    return None
+
+
+def _name_from_water(tree: ElementTree.ElementTree, path: Path) -> str | None:
+    LOGGER.debug("Attempting to name by water...")
+    best_name = None
+    best_distance = None
+    node_to_coord = {}
+    for water in tree.iterfind("./node"):
+        node_id = water.get("id")
+        lat = float(water.get("lat"))
+        lon = float(water.get("lon"))
+        node_to_coord[node_id] = (lat, lon)
+
+        if water.find("./tag[@k='natural'][@v='water']") is None:
+            continue
+
+        name_elem = water.find("./tag[@k='name']")
+        if name_elem is None:
+            continue
+        name = name_elem.get("v")
+
+        water_point = Point(longitude=lon, latitude=lat, elevation=0.0, time_offset=0.0)
+        distance = water_point.flat_distance(path.points[-1])
+
+        if best_distance is None or distance < best_distance:
+            best_distance = distance
+            best_name = name
+
+    for water in tree.iterfind("./way"):
+        if water.find("./tag[@k='natural'][@v='water']") is None:
+            continue
+
+        name_elem = water.find("./tag[@k='name']")
+        if name_elem is None:
+            continue
+        name = name_elem.get("v")
+
+        for node in water.iterfind("nd"):
+            lat, lon = node_to_coord[node.get("ref")]
+            water_point = Point(longitude=lon, latitude=lat, elevation=0.0, time_offset=0.0)
+            distance = water_point.flat_distance(path.points[-1])
+
+            if best_distance is None or distance < best_distance:
+                best_distance = distance
+                best_name = name
+
+
+    accepted = best_distance is not None and best_distance <= MAX_DISTANCE_WATER
+    LOGGER.debug(
+        "Best name: %s (distance=%s) (accepted=%s)",
+        best_name,
+        best_distance,
+        accepted,
+    )
+
+    if accepted:
+        return best_name
+    return None
+
+
+def _name_from_city(tree: ElementTree.ElementTree, path: Path) -> str | None:
+    LOGGER.debug("Attempting to name by city...")
+    best_name = None
+    best_distance = None
+    best_place_type = None
+    for city in tree.iterfind("./node"):
+        place_elem = city.find("./tag[@k='place']")
+        if place_elem is None:
+            continue
+        place = place_elem.get("v")
+        if place not in {"city", "town", "village", "suburb"}:
+            continue
+
+        name_elem = city.find("./tag[@k='name']")
+        if name_elem is None:
+            continue
+        name = name_elem.get("v")
+        lat = float(city.get("lat"))
+        lon = float(city.get("lon"))
+
+        city_point = Point(longitude=lon, latitude=lat, elevation=0.0, time_offset=0.0)
+        distance = city_point.flat_distance(path.points[-1])
+
+        if best_distance is None or distance < best_distance:
+            best_distance = distance
+            best_name = name
+            best_place_type = place
+
+    accepted = best_distance is not None and best_distance <= MAX_DISTANCE_CITY
+    LOGGER.debug(
+        "Best name: %s (place=%s) (distance=%s) (accepted=%s)",
+        best_name,
+        best_place_type,
+        best_distance,
+        accepted,
+    )
+
+    if accepted:
+        return best_name
+    return None
+
+
+def _fallback_name(tree: ElementTree.ElementTree, path: Path) -> str | None:
+    LOGGER.debug("Attempting to find a fallback name...")
+    best_name = None
+    best_distance = None
+    for node in tree.iterfind("./node"):
+        name_elem = node.find("./tag[@k='name']")
+        if name_elem is None:
+            continue
+        name = name_elem.get("v")
+        lat = float(node.get("lat"))
+        lon = float(node.get("lon"))
+
+        point = Point(longitude=lon, latitude=lat, elevation=0.0, time_offset=0.0)
+        distance = point.flat_distance(path.points[-1])
+
+        if best_distance is None or distance < best_distance:
+            best_distance = distance
+            best_name = name
+
+    LOGGER.debug("Fallback name: %s (distance=%s)", best_name, best_distance)
+    return best_name