aio_overpass.pt

  1"""Classes and queries specialized on public transportation routes."""
  2
  3from collections import Counter
  4from collections.abc import Generator
  5from dataclasses import dataclass
  6from enum import Enum, auto
  7from typing import Any, Final, TypeGuard
  8
  9from aio_overpass._dist import fast_distance
 10from aio_overpass.element import Bbox, Element, Node, Relation, Relationship, Way, collect_elements
 11from aio_overpass.ql import one_of_filter, poly_clause
 12from aio_overpass.query import Query
 13from aio_overpass.spatial import GeoJsonDict, Spatial
 14
 15import shapely.ops
 16from shapely.geometry import GeometryCollection, Point, Polygon
 17
 18
 19__docformat__ = "google"
 20__all__ = (
 21    "collect_routes",
 22    "RouteQuery",
 23    "SingleRouteQuery",
 24    "RoutesWithinQuery",
 25    "Route",
 26    "Vehicle",
 27    "Stop",
 28    "Connection",
 29    "RouteScheme",
 30)
 31
 32
 33class RouteQuery(Query):
 34    """
 35    Base class for queries that produce ``Route`` and ``RouteSegment`` objects.
 36
 37    Be aware that to build full ``RouteSegment`` objects with tags and geometry, this query loads:
 38     - every route member
 39     - every stop area any stop on the route is related to
 40     - every stop in one of those stop areas
 41
 42    Args:
 43        input_code: A query that puts the desired routes into the ``.routes`` set,
 44                     and optionally, its route masters into the ``.masters`` set
 45                     (for example by recursing up from ``.routes``).
 46    """
 47
 48    __slots__ = ()
 49
 50    def __init__(
 51        self,
 52        input_code: str,
 53        **kwargs: Any,  # noqa: ANN401
 54    ) -> None:
 55        input_code = f"""
 56            {input_code}
 57            .routes >> -> .route_members;
 58            way.route_members -> .route_ways;
 59
 60            (
 61                node.route_members[highway=bus_stop];
 62                node.route_members[public_transport];
 63                way .route_members[public_transport];
 64                rel .route_members[public_transport];
 65            ) -> .route_pt_members;
 66
 67            .route_pt_members <;
 68            rel._[public_transport=stop_area]->.stop_areas;
 69            node(r.stop_areas:"stop")[public_transport=stop_position]->.stop_area_stops;
 70
 71            .masters out;
 72            .routes out geom;
 73            .route_ways out tags;
 74            .route_pt_members out geom;
 75            .stop_areas out;
 76            .stop_area_stops out;
 77        """
 78
 79        super().__init__(input_code, **kwargs)
 80
 81
 82class SingleRouteQuery(RouteQuery):
 83    """
 84    A query that produces a single ``Route`` object.
 85
 86    Args:
 87        relation_id: the desired route's relation ID
 88    """
 89
 90    __slots__ = ("relation_id",)
 91
 92    def __init__(
 93        self,
 94        relation_id: int,
 95        **kwargs: Any,  # noqa: ANN401
 96    ) -> None:
 97        self.relation_id = relation_id
 98
 99        input_code = f"""
100            rel({self.relation_id});
101            rel._[type=route]->.routes;
102            .routes <<;
103            rel._[type=route_master]->.masters;
104        """
105
106        super().__init__(input_code, **kwargs)
107
108
109class RoutesWithinQuery(RouteQuery):
110    """
111    A query that produces ``Route`` objects for any route within the exterior of a polygon.
112
113    Args:
114        polygon: Any route that has at least one member element within this shape
115                 will be in the result set of this query. Note that the route members
116                 are not limited to this polygon - the majority of a route may in fact
117                 be outside of it. This shape should be simplified, since a larger number
118                 of coordinates on the exterior will slow down the query.
119        vehicles: A list of transportation modes to filter by, or an empty list to include
120                  routes of any type. A non-empty list will filter routes by the ``route``
121                  key.
122    """
123
124    __slots__ = (
125        "polygon",
126        "vehicles",
127    )
128
129    def __init__(
130        self,
131        polygon: Polygon,
132        vehicles: list["Vehicle"] | None = None,
133        **kwargs: Any,  # noqa: ANN401
134    ) -> None:
135        if not vehicles:
136            vehicles = list(Vehicle)
137
138        self.polygon = polygon
139        self.vehicles = vehicles
140
141        region_clause = poly_clause(self.polygon)
142        route_filter = one_of_filter("route", *(v.name.lower() for v in vehicles))
143        input_code = f"""
144            rel{region_clause}{route_filter}[type=route]->.routes;
145            rel{region_clause}{route_filter}[type=route_master]->.masters;
146        """
147
148        super().__init__(input_code, **kwargs)
149
150
151class _RouteRole(Enum):
152    """
153    A role in a route relation that is relevant to public transportation.
154
155    References:
156        - https://taginfo.openstreetmap.org/relations/route#roles
157    """
158
159    STOP = auto()
160    PLATFORM = auto()
161    NONE = auto()
162
163
164class Connection(Enum):
165    """
166    Indicates whether you can enter, exit, or do both at a stop on a route.
167
168    References:
169        - https://taginfo.openstreetmap.org/relations/route#roles
170    """
171
172    ENTRY_AND_EXIT = auto()
173    ENTRY_ONLY = auto()
174    EXIT_ONLY = auto()
175
176    @property
177    def entry_possible(self) -> bool:
178        """``True`` if you can enter at this stop on the route."""
179        return self != Connection.EXIT_ONLY
180
181    @property
182    def exit_possible(self) -> bool:
183        """``True`` if you can exit at this stop on the route."""
184        return self != Connection.ENTRY_ONLY
185
186    def __repr__(self) -> str:
187        return f"{type(self).__name__}.{self.name}"
188
189
190@dataclass(kw_only=True, slots=True)
191class Stop(Spatial):
192    """
193    A stop on a public transportation route.
194
195    Typically, a stop is modelled as two members in a relation: a stop_position node with the
196    'stop' role, and a platform with the 'platform' role. These members may be grouped in
197    a common stop_area.
198
199    Attributes:
200        idx: stop index on the route
201        platform: the platform node, way or relation associated with this stop, if any
202        stop_position: the stop position node associated with this stop, if any
203        stop_coords: a point that, compared to ``stop_position``, is guaranteed to be on the
204                     track of the route whenever it is set. Only set if you are collecting
205                     ``RouteSegments``.
206    """
207
208    idx: int
209    platform: Relationship | None
210    stop_position: Relationship | None
211    stop_coords: Node | Point | None
212
213    @property
214    def name(self) -> str | None:
215        """
216        This stop's name.
217
218        If platform and stop position names are the same, that name will be returned.
219        Otherwise, the most common name out of platform, stop position, and all related stop areas
220        will be returned.
221        """
222        stop_pos_name = self.stop_position.member.tag("name") if self.stop_position else None
223        platform_name = self.platform.member.tag("name") if self.platform else None
224
225        if stop_pos_name == platform_name and stop_pos_name is not None:
226            return stop_pos_name
227
228        names = [stop_pos_name, platform_name, *(rel.tag("name") for rel in self.stop_areas)]
229        names = [name for name in names if name]
230
231        if not names:
232            return None
233
234        counter = Counter(names)
235        ((most_common, _),) = counter.most_common(1)
236        return most_common
237
238    @property
239    def connection(self) -> Connection:
240        """Indicates whether you can enter, exit, or do both at this stop."""
241        options = [
242            _connection(relship) for relship in (self.stop_position, self.platform) if relship
243        ]
244        return next(
245            (opt for opt in options if opt != Connection.ENTRY_AND_EXIT), Connection.ENTRY_AND_EXIT
246        )
247
248    @property
249    def stop_areas(self) -> set[Relation]:
250        """Any stop area related to this stop."""
251        return {
252            relship_to_stop_area.relation
253            for relship_to_route in (self.platform, self.stop_position)
254            if relship_to_route
255            for relship_to_stop_area in relship_to_route.member.relations
256            if relship_to_stop_area.relation.tag("public_transport") == "stop_area"
257        }
258
259    @property
260    def geojson(self) -> GeoJsonDict:
261        """A mapping of this object, using the GeoJSON format."""
262        # TODO: Stop geojson
263        raise NotImplementedError
264
265    @property
266    def _stop_point(self) -> Point | None:
267        """This is set if we have a point that is on the track of the route."""
268        if isinstance(self.stop_coords, Node):
269            return self.stop_coords.geometry
270        if isinstance(self.stop_coords, Point):
271            return self.stop_coords
272        return None
273
274    @property
275    def _geometry(self) -> GeometryCollection:
276        """Collection of ``self.platform``, ``self.stop_position`` and  ``self.stop_coords``."""
277        geoms = []
278
279        # geometry can be None if the platform is a relation
280        if self.platform and self.platform.member.base_geometry:
281            geoms.append(self.platform.member.base_geometry)
282
283        if self.stop_position:
284            assert isinstance(self.stop_position.member, Node)
285            geoms.append(self.stop_position.member.geometry)
286
287        if isinstance(self.stop_coords, Point) and self.stop_coords not in geoms:
288            geoms.append(self.stop_coords)
289
290        return GeometryCollection(geoms)
291
292    def __repr__(self) -> str:
293        if self.stop_position:
294            elem = f"stop_position={self.stop_position.member}"
295        elif self.platform:
296            elem = f"platform={self.platform.member}"
297        else:
298            raise AssertionError
299
300        return f"{type(self).__name__}({elem}, name='{self.name}')"
301
302
303class Vehicle(Enum):
304    """
305    Most common modes of public transportation.
306
307    References:
308        - https://wiki.openstreetmap.org/wiki/Relation:route#Public_transport_routes
309    """
310
311    # motor vehicles
312    BUS = auto()
313    TROLLEYBUS = auto()
314    MINIBUS = auto()
315    SHARE_TAXI = auto()
316
317    # railway vehicles
318    TRAIN = auto()
319    LIGHT_RAIL = auto()
320    SUBWAY = auto()
321    TRAM = auto()
322
323    # boats
324    FERRY = auto()
325
326    def __repr__(self) -> str:
327        return f"{type(self).__name__}.{self.name}"
328
329
330class RouteScheme(Enum):
331    """
332    Tagging schemes for public transportation routes.
333
334    References:
335        - https://wiki.openstreetmap.org/wiki/Public_transport#Different_tagging_schemas
336        - https://wiki.openstreetmap.org/wiki/Key:public_transport:version
337    """
338
339    EXPLICIT_V1 = auto()
340    EXPLICIT_V2 = auto()
341
342    ASSUME_V1 = auto()
343    ASSUME_V2 = auto()
344
345    OTHER = auto()
346
347    @property
348    def version_number(self) -> int | None:
349        """Public transport tagging scheme."""
350        match self:
351            case RouteScheme.EXPLICIT_V1 | RouteScheme.ASSUME_V1:
352                return 1
353            case RouteScheme.EXPLICIT_V2 | RouteScheme.ASSUME_V2:
354                return 2
355            case RouteScheme.OTHER:
356                return None
357            case _:
358                raise AssertionError
359
360    def __repr__(self) -> str:
361        return f"{type(self).__name__}.{self.name}"
362
363
364@dataclass(kw_only=True, slots=True)
365class Route(Spatial):
366    """
367    A public transportation service route, e.g. a bus line.
368
369    Instances of this class are meant to represent OSM routes using the
370    'Public Transport Version 2' (PTv2) scheme. Compared to PTv1, this means f.e. that routes
371    with multiple directions (forwards, backwards) are represented by multiple route elements.
372
373    Attributes:
374        relation: the underlying relation that describes the path taken by a route,
375                  and places where people can embark and disembark from the transit service
376        scheme: The tagging scheme that was either assumed, or explicitly set on the route relation.
377        stops: a sorted list of stops on this route as they would appear on its timetable,
378               which was derived from the ``relation``
379
380    References:
381        - https://wiki.openstreetmap.org/wiki/Public_transport
382        - https://wiki.openstreetmap.org/wiki/Relation:route#Public_transport_routes
383    """
384
385    relation: Relation
386    scheme: RouteScheme
387    stops: list[Stop]
388
389    @property
390    def id(self) -> int:
391        """Route relation ID."""
392        return self.relation.id
393
394    @property
395    def tags(self) -> dict[str, str]:
396        """
397        Tags of the route relation.
398
399        Some tags can be inherited from a route master, if not set already.
400        """
401        from_relation = self.relation.tags or {}
402        from_master = {}
403
404        masters = self.masters
405        master = masters[0] if len(masters) == 1 else None
406
407        if master and master.tags is not None:
408            from_master = {k: v for k, v in master.tags.items() if k in _TAGS_FROM_ROUTE_MASTER}
409
410        return from_master | from_relation
411
412    def tag(self, key: str, default: str | None = None) -> str | None:
413        """
414        Get the tag value for the given key.
415
416        Some tags can be inherited from a route master, if not set already.
417        """
418        value = self.relation.tag(key, default)
419        if value is not default:
420            return value
421
422        if key not in _TAGS_FROM_ROUTE_MASTER:
423            return default
424
425        masters = self.masters
426        master = masters[0] if len(masters) == 1 else None
427
428        if not master:
429            return default
430
431        return master.tag(key, default)
432
433    @property
434    def ways(self) -> list[Way]:
435        """
436        The ways making up the path of the route.
437
438        Ways may be ordered, and appear multiple times if a way is travelled on more than once.
439        All the ways may be contiguous, but gaps are not uncommon.
440        """
441        return [
442            relship.member
443            for relship in self.relation.members
444            if isinstance(relship.member, Way) and _role(relship) == _RouteRole.NONE
445        ]
446
447    @property
448    def masters(self) -> list[Relation]:
449        """
450        Route master relations this route is a part of.
451
452        By convention, this should be a single relation at most.
453
454        References:
455            - https://wiki.openstreetmap.org/wiki/Relation:route_master
456        """
457        return [
458            relship.relation
459            for relship in self.relation.relations
460            if relship.relation.tag("type") == "route_master"
461        ]
462
463    @property
464    def name_from(self) -> str | None:
465        """
466        Name of the start station.
467
468        This is either the value of the relation's ``from`` key, a name derived
469        from the route's first stop (see ``Stop.name``), or ``None`` if neither
470        is available.
471        """
472        return self.relation.tag(
473            key="from",
474            default=self.stops[0].name if self.stops else None,
475        )
476
477    @property
478    def name_to(self) -> str | None:
479        """
480        Name of the end station.
481
482        This is either the value of the relation's ``to`` key, a name derived
483        from the route's last stop (see ``Stop.name``), or ``None`` if neither
484        is available.
485        """
486        return self.relation.tag(
487            key="to",
488            default=self.stops[-1].name if len(self.stops) > 1 else None,
489        )
490
491    @property
492    def name_via(self) -> str | None:
493        """
494        A name of an important station along the route.
495
496        This is either the value of the relation's ``via`` key, or ``None`` if it is not set.
497        """
498        return self.relation.tag("via")
499
500    @property
501    def name(self) -> str | None:
502        """
503        The name of the route.
504
505        This is either the value relation's ``name`` key, ``{from_} => {via} => {to}`` if at
506        least ``from`` and ``to`` are set, or ``None`` otherwise.
507        """
508        name_tag = self.tag("name")
509        if name_tag:
510            return name_tag
511
512        from_ = self.name_from
513        via = self.name_via
514        to = self.name_to
515
516        if from_ and to and via:
517            return f"{from_} => {via} => {to}"
518
519        if from_ and to:
520            return f"{from_} => {to}"
521
522        return None
523
524    @property
525    def vehicle(self) -> Vehicle:
526        """
527        The mode of transportation used on this route.
528
529        This value corresponds with the value of the relation's ``route`` key.
530        """
531        if not self.relation.tags or "route" not in self.relation.tags:
532            raise AssertionError
533        return Vehicle[self.relation.tags["route"].upper()]
534
535    @property
536    def bounds(self) -> Bbox | None:
537        """The bounding box around all stops of this route."""
538        geom = GeometryCollection([stop._geometry for stop in self.stops if stop._geometry])
539        return geom.bounds or None
540
541    @property
542    def geojson(self) -> GeoJsonDict:
543        """A mapping of this object, using the GeoJSON format."""
544        # TODO: Route geojson
545        raise NotImplementedError
546
547    def __repr__(self) -> str:
548        return f"{type(self).__name__}(id={self.relation.id}, name='{self.name}')"
549
550
551_TAGS_FROM_ROUTE_MASTER: Final[set[str]] = {
552    "colour",
553    "interval",
554    "name",
555    "network",
556    "opening_hours",
557    "operator",
558    "ref",
559    "school",
560    "tourism",
561    "wheelchair",
562}
563"""
564Tags that, by convention, can be applied to route masters instead of their routes,
565in order to avoid duplication.
566
567References:
568    - https://wiki.openstreetmap.org/wiki/Relation:route_master
569"""
570
571
572def collect_routes(query: RouteQuery, perimeter: Polygon | None = None) -> list[Route]:
573    # TODO: the way 'perimeter' works might be confusing
574    """
575    Consumes the result set of a query and produces ``Route`` objects.
576
577    The order of elements is *not* retained.
578
579    The result set in the input query will be empty after calling this function.
580
581    Args:
582        query: The query that produced a result set of route relations.
583        perimeter: If set, ``stops`` at the end will be truncated if they are outside of this
584                   polygon. This means stops outside the polygon will still be in the list as
585                   long as there is at least one following stop that *is* inside the list.
586                   Any relation members in ``members`` will not be filtered.
587
588    Raises:
589        ValueError: if the input query has no result set
590
591    Returns:
592        all routes in the result set of the input query
593    """
594    elements = collect_elements(query)
595    route_rels = [elem for elem in elements if _is_tagged_route_relation(elem)]
596
597    routes = []
598
599    for route_rel in route_rels:
600        # Group route relation members per stop:
601        # f.e. a platform, and a stop position at the same station.
602        stops = list(_stops(route_rel))
603
604        # Filter stops by perimeter: cut off at the last stop that is in the perimeter.
605        if perimeter:
606            while stops and not perimeter.contains(stops[-1]._stop_point):
607                del stops[-1]
608
609        route = Route(
610            relation=route_rel,
611            scheme=_scheme(route_rel),
612            stops=stops,
613        )
614
615        routes.append(route)
616
617    return routes
618
619
620def _is_tagged_route_relation(elem: Element) -> TypeGuard[Relation]:
621    return isinstance(elem, Relation) and elem.tag("type") == "route"
622
623
624def _scheme(route: Relation) -> RouteScheme:
625    """Try to identify a route's tagging scheme."""
626    tagged_version = route.tag("public_transport:version")
627
628    match tagged_version:
629        case "1":
630            return RouteScheme.EXPLICIT_V1
631        case "2":
632            return RouteScheme.EXPLICIT_V2
633        case _ if tagged_version:
634            return RouteScheme.OTHER
635
636    # any directed and/or numbered tags like "forward:stop:1" indicate PTv1
637    directions = {
638        "forward:",
639        "forward_",
640        "backward:",
641        "backward_",
642    }
643    assume_v1 = any(
644        (
645            role and (role.startswith(prefix) or role[-1].isnumeric())
646            for role, _ in route
647            for prefix in directions
648        )
649    )
650
651    return RouteScheme.ASSUME_V1 if assume_v1 else RouteScheme.ASSUME_V2
652
653
654def _stops(route_relation: Relation) -> Generator[Stop, None, None]:
655    """
656    Group route relation members so that each member belongs to the same stop along the route.
657
658    Typically, a stop is represented by a stop position or platform, or both.
659    """
660    idx = 0
661
662    def to_stop(*selected: Relationship) -> Stop:
663        nonlocal idx
664        stop_idx = idx
665        idx += 1
666        return Stop(
667            idx=stop_idx,
668            platform=next(
669                (relship for relship in selected if _role(relship) == _RouteRole.PLATFORM), None
670            ),
671            stop_position=next(
672                (relship for relship in selected if _role(relship) == _RouteRole.STOP), None
673            ),
674            stop_coords=None,  # set later
675        )
676
677    # Consider all route relation members that are tagged or given a role that makes them
678    # relevant for public transportation.
679    route_members = [
680        relship for relship in route_relation.members if _role(relship) is not _RouteRole.NONE
681    ]
682
683    # no more than two elements per group (best case: roles "stop" & "platform")
684    prev: Relationship | None = None
685    for next_ in route_members:
686        if not prev:  # case 1: no two members to compare yet
687            prev = next_
688            continue
689
690        if not _at_same_stop(prev, next_):  # case 2: members are not part of same station
691            yield to_stop(prev)
692            prev = next_
693            continue
694
695        yield to_stop(prev, next_)  # case 3: members are part of same station
696        prev = None
697
698    if prev:
699        yield to_stop(prev)
700
701
702def _connection(relship: Relationship) -> Connection:
703    """Returns the connection at the route member, according to its role."""
704    if relship.role:
705        if relship.role.endswith("_entry_only"):
706            return Connection.ENTRY_ONLY
707
708        if relship.role.endswith("_exit_only"):
709            return Connection.EXIT_ONLY
710
711    return Connection.ENTRY_AND_EXIT
712
713
714def _role(relship: Relationship) -> _RouteRole:
715    """
716    Returns the route member's tagged role, or a fitting fallback.
717
718    If a member is tagged as platform or stop_position, use the role as is.
719    Otherwise, assign the roles that fit the member. This means that if there's
720    a platform in the route relation, we'll recognize it as such, even if it has
721    not been given the platform role (perhaps due to oversight by a contributor).
722    """
723    if relship.role:
724        if relship.role.startswith("platform") or relship.role == "hail_and_ride":
725            return _RouteRole.PLATFORM
726
727        if relship.role.startswith("stop"):
728            return _RouteRole.STOP
729
730    if (
731        relship.member.tag("public_transport") == "platform"
732        or relship.member.tag("highway") == "bus_stop"
733    ):
734        return _RouteRole.PLATFORM
735
736    # The correct tag for this is [public_transport=stop_position], but we'll accept
737    # values like [public_transport=station] as well, as long as the member is a node.
738    # The assumption is that any node that is not representing a platform is *probably*
739    # supposed to represent the stop position.
740    if relship.member.type == "node" and relship.member.tag("public_transport"):
741        return _RouteRole.STOP
742
743    return _RouteRole.NONE
744
745
746def _share_stop_area(a: Relationship, b: Relationship) -> bool:
747    """``True`` if the given route members share least one common stop area."""
748    a_areas = {
749        relship.relation.id
750        for relship in a.member.relations
751        if relship.relation.tag("public_transport") == "stop_area"
752    }
753    b_areas = {
754        relship.relation.id
755        for relship in b.member.relations
756        if relship.relation.tag("public_transport") == "stop_area"
757    }
758    return not a_areas.isdisjoint(b_areas)
759
760
761def _connection_compatible(a: Connection, b: Connection) -> bool:
762    """
763    Check whether two connections are compatible.
764
765    Returns:
766        ``False`` if one of the connections is entry-only, while the other is exit-only,
767        ``True`` otherwise.
768    """
769    return Connection.ENTRY_AND_EXIT in (a, b) or a == b
770
771
772def _at_same_stop(a: Relationship, b: Relationship) -> bool:
773    """
774    Check if two members of a route belong to the same stop in the timetable.
775
776    Looking at the route on a map makes it trivial to group elements as part of the same station,
777    but automating this requires some heuristics: we will look at members' type, role, name, and
778    their proximity to each other.
779
780    If members grouped by this function are far apart, they might be mistagged, i.e. there was a
781    mix-up with either the stop position, platform or order (f.e. having "forward" & "backward"
782    stops listed consecutively in the route, instead of their correct timetable order).
783    """
784    # require not both exit_only & enter_only at same stop
785    if not _connection_compatible(_connection(a), _connection(b)):
786        return False
787
788    # require no duplicate role at same stop
789    if _role(a) == _role(b) and _role(a) is not _RouteRole.NONE:
790        return False
791
792    # same name, assume same stop
793    if a.member.tag("name") and a.member.tag("name") == b.member.tag("name"):
794        return True
795
796    # same stop area, assume same stop
797    if _share_stop_area(a, b):
798        return True
799
800    # assume same stop if close together
801    if not a.member.base_geometry or not b.member.base_geometry:
802        return False
803
804    # euclidean nearest
805    pt_a, pt_b = shapely.ops.nearest_points(a.member.base_geometry, b.member.base_geometry)
806    distance = fast_distance(*pt_a.coords[0], *pt_b.coords[0])
807    return distance <= _MAX_DISTANCE_TO_TRACK
808
809
810_MAX_DISTANCE_TO_TRACK: Final[float] = 30.0  # meters
811"""
812An expectation of the maximum distance between a stop position and its platform.
813
814The two should typically be pretty close to each other. Effectively, this value
815should be lower than any sensible (beeline) distance between two consecutive
816stops on the same route. It should not be used as a hard constraint.
817"""