"""Model for the `Path` relationship between two `Address` objects.""" from __future__ import annotations import functools import itertools import json from typing import List import folium import googlemaps as gm import ordered_set import sqlalchemy as sa from geopy import distance as geo_distance from sqlalchemy import orm from sqlalchemy.dialects import postgresql from urban_meal_delivery import config from urban_meal_delivery import db from urban_meal_delivery.db import meta from urban_meal_delivery.db import utils class Path(meta.Base): """Path between two `Address` objects. Models the path between two `Address` objects, including directions for a `Courier` to get from one `Address` to another. As the couriers are on bicycles, we model the paths as a symmetric graph (i.e., same distance in both directions). Implements an association pattern between `Address` and `Address`. Further info: https://docs.sqlalchemy.org/en/stable/orm/basic_relationships.html#association-object # noqa:E501 """ __tablename__ = 'addresses_addresses' # Columns first_address_id = sa.Column(sa.Integer, primary_key=True) second_address_id = sa.Column(sa.Integer, primary_key=True) city_id = sa.Column(sa.SmallInteger, nullable=False) # Distances are measured in meters. air_distance = sa.Column(sa.Integer, nullable=False) bicycle_distance = sa.Column(sa.Integer, nullable=True) # The duration is measured in seconds. bicycle_duration = sa.Column(sa.Integer, nullable=True) # An array of latitude-longitude pairs approximating a courier's way. _directions = sa.Column('directions', postgresql.JSON, nullable=True) # Constraints __table_args__ = ( # The two `Address` objects must be in the same `.city`. sa.ForeignKeyConstraint( ['first_address_id', 'city_id'], ['addresses.id', 'addresses.city_id'], onupdate='RESTRICT', ondelete='RESTRICT', ), sa.ForeignKeyConstraint( ['second_address_id', 'city_id'], ['addresses.id', 'addresses.city_id'], onupdate='RESTRICT', ondelete='RESTRICT', ), # Each `Address`-`Address` pair only has one distance. sa.UniqueConstraint('first_address_id', 'second_address_id'), sa.CheckConstraint( 'first_address_id < second_address_id', name='distances_are_symmetric_for_bicycles', ), sa.CheckConstraint( '0 <= air_distance AND air_distance < 20000', name='realistic_air_distance', ), sa.CheckConstraint( 'bicycle_distance < 25000', # `.bicycle_distance` may not be negative name='realistic_bicycle_distance', # due to the constraint below. ), sa.CheckConstraint( 'air_distance <= bicycle_distance', name='air_distance_is_shortest', ), sa.CheckConstraint( '0 <= bicycle_duration AND bicycle_duration <= 3600', name='realistic_bicycle_travel_time', ), ) # Relationships first_address = orm.relationship( 'Address', foreign_keys='[Path.first_address_id, Path.city_id]', ) second_address = orm.relationship( 'Address', foreign_keys='[Path.second_address_id, Path.city_id]', overlaps='first_address', ) @classmethod def from_addresses( cls, *addresses: db.Address, google_maps: bool = False, ) -> List[Path]: """Calculate pair-wise paths for `Address` objects. This is the main constructor method for the class. It handles the "sorting" of the `Address` objects by `.id`, which is the logic that enforces the symmetric graph behind the paths. Args: *addresses: to calculate the pair-wise paths for; must contain at least two `Address` objects google_maps: if `.bicycle_distance` and `._directions` should be populated with a query to the Google Maps Directions API; by default, only the `.air_distance` is calculated with `geopy` Returns: paths """ paths = [] # We consider all 2-tuples of `Address`es. The symmetric graph is ... for first, second in itertools.combinations(addresses, 2): # ... implicitly enforced by a precedence constraint for the `.id`s. first, second = ( # noqa:WPS211 (first, second) if first.id < second.id else (second, first) ) # If there is no `Path` object in the database ... path = ( db.session.query(db.Path) .filter(db.Path.first_address == first) .filter(db.Path.second_address == second) .first() ) # ... create a new one. if path is None: air_distance = geo_distance.great_circle( first.location.lat_lng, second.location.lat_lng, ) path = cls( first_address=first, second_address=second, air_distance=round(air_distance.meters), ) db.session.add(path) db.session.commit() paths.append(path) if google_maps: for path in paths: # noqa:WPS440 path.sync_with_google_maps() return paths @classmethod def from_order(cls, order: db.Order, google_maps: bool = False) -> Path: """Calculate the path for an `Order` object. The path goes from the `Order.pickup_address` to the `Order.delivery_address`. Args: order: to calculate the path for google_maps: if `.bicycle_distance` and `._directions` should be populated with a query to the Google Maps Directions API; by default, only the `.air_distance` is calculated with `geopy` Returns: path """ return cls.from_addresses( order.pickup_address, order.delivery_address, google_maps=google_maps, )[0] def sync_with_google_maps(self) -> None: """Fill in `.bicycle_distance` and `._directions` with Google Maps. `._directions` will NOT contain the coordinates of `.first_address` and `.second_address`. This uses the Google Maps Directions API. Further info: https://developers.google.com/maps/documentation/directions """ # To save costs, we do not make an API call # if we already have data from Google Maps. if self.bicycle_distance is not None: return client = gm.Client(config.GOOGLE_MAPS_API_KEY) response = client.directions( origin=self.first_address.location.lat_lng, destination=self.second_address.location.lat_lng, mode='bicycling', alternatives=False, ) # Without "alternatives" and "waypoints", the `response` contains # exactly one "route" that consists of exactly one "leg". # Source: https://developers.google.com/maps/documentation/directions/get-directions#Legs # noqa:E501 route = response[0]['legs'][0] self.bicycle_distance = route['distance']['value'] # noqa:WPS601 self.bicycle_duration = route['duration']['value'] # noqa:WPS601 # Each route consists of many "steps" that are instructions as to how to # get from A to B. As a step's "start_location" may equal the previous step's # "end_location", we use an `OrderedSet` to find the unique latitude-longitude # pairs that make up the path from `.first_address` to `.second_address`. steps = ordered_set.OrderedSet() for step in route['steps']: steps.add( # noqa:WPS221 (step['start_location']['lat'], step['start_location']['lng']), ) steps.add( # noqa:WPS221 (step['end_location']['lat'], step['end_location']['lng']), ) steps.discard(self.first_address.location.lat_lng) steps.discard(self.second_address.location.lat_lng) self._directions = json.dumps(list(steps)) # noqa:WPS601 db.session.add(self) db.session.commit() @property # pragma: no cover def map(self) -> folium.Map: # noqa:WPS125 """Convenience property to obtain the underlying `City.map`.""" return self.first_address.city.map @functools.cached_property def waypoints(self) -> List[utils.Location]: """The couriers' route from `.first_address` to `.second_address`. The returned `Location`s all relate to `.first_address.city.southwest`. Implementation detail: This property is cached as none of the underlying attributes (i.e., `._directions`) are to be changed. """ points = [utils.Location(*point) for point in json.loads(self._directions)] for point in points: point.relate_to(self.first_address.city.southwest) return points def draw( # noqa:WPS211 self, *, reverse: bool = False, start_tooltip: str = 'Start', end_tooltip: str = 'End', start_color: str = 'green', end_color: str = 'red', path_color: str = 'black', ) -> folium.Map: # pragma: no cover """Draw the `.waypoints` from `.first_address` to `.second_address`. Args: reverse: by default, `.first_address` is used as the start; set to `False` to make `.second_address` the start start_tooltip: text shown on marker at the path's start end_tooltip: text shown on marker at the path's end start_color: `folium` color for the path's start end_color: `folium` color for the path's end path_color: `folium` color along the path, which is the line between the `.waypoints` Returns: `.map` for convenience in interactive usage """ # Without `self._directions` synced from Google Maps, # the `.waypoints` are not available. self.sync_with_google_maps() # First, plot the couriers' path between the start and # end locations, so that it is below the `folium.Circle`s. line = folium.PolyLine( locations=( self.first_address.location.lat_lng, *(point.lat_lng for point in self.waypoints), self.second_address.location.lat_lng, ), color=path_color, weight=2, ) line.add_to(self.map) # Draw the path's start and end locations, possibly reversed, # on top of the couriers' path. if reverse: start, end = self.second_address, self.first_address else: start, end = self.first_address, self.second_address start.draw( radius=5, color=start_color, fill_color=start_color, fill_opacity=1, tooltip=start_tooltip, ) end.draw( radius=5, color=end_color, fill_color=end_color, fill_opacity=1, tooltip=end_tooltip, ) return self.map