Alexander Hess
e4e543bd40
- add new ORM models `ReplaySimulation` and `ReplayedOrder`
to store the data generated by the routing simulations
- add migrations script to create the corresponding database tables
+ create "replay_simulations" and "replayed_orders" tables
+ add missing check constraints to "orders" table
+ add unique constraint to "orders" table to enable compound key
+ drop unnecessary check constraints from the "orders" table
- add tests for the new ORM models
+ add `simulation`, `replayed_order`, `make_replay_order()`, and
`pre_order` fixtures
+ add `ReplayedOrderFactor` faker class
- fix some typos
244 lines
8.6 KiB
Python
244 lines
8.6 KiB
Python
"""Provide the ORM's `City` model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import functools
|
|
|
|
import folium
|
|
import sqlalchemy as sa
|
|
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 City(meta.Base):
|
|
"""A city where the UDP operates in."""
|
|
|
|
__tablename__ = 'cities'
|
|
|
|
# Generic columns
|
|
id = sa.Column( # noqa:WPS125
|
|
sa.SmallInteger, primary_key=True, autoincrement=False,
|
|
)
|
|
name = sa.Column(sa.Unicode(length=10), nullable=False)
|
|
kml = sa.Column(sa.UnicodeText, nullable=False)
|
|
|
|
# Google Maps related columns
|
|
center_latitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
center_longitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
northeast_latitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
northeast_longitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
southwest_latitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
southwest_longitude = sa.Column(postgresql.DOUBLE_PRECISION, nullable=False)
|
|
initial_zoom = sa.Column(sa.SmallInteger, nullable=False)
|
|
|
|
# Relationships
|
|
addresses = orm.relationship('Address', back_populates='city')
|
|
grids = orm.relationship('Grid', back_populates='city')
|
|
replays = orm.relationship('ReplaySimulation', back_populates='city')
|
|
|
|
# We do not implement a `.__init__()` method and use SQLAlchemy's default.
|
|
# The uninitialized attribute `._map` is computed on the fly. note:d334120ei
|
|
|
|
def __repr__(self) -> str:
|
|
"""Non-literal text representation."""
|
|
return '<{cls}({name})>'.format(cls=self.__class__.__name__, name=self.name)
|
|
|
|
@functools.cached_property
|
|
def center(self) -> utils.Location:
|
|
"""Location of the city's center.
|
|
|
|
Implementation detail: This property is cached as none of the
|
|
underlying attributes to calculate the value are to be changed.
|
|
"""
|
|
return utils.Location(self.center_latitude, self.center_longitude)
|
|
|
|
@functools.cached_property
|
|
def northeast(self) -> utils.Location:
|
|
"""The city's northeast corner of the Google Maps viewport.
|
|
|
|
Implementation detail: This property is cached as none of the
|
|
underlying attributes to calculate the value are to be changed.
|
|
"""
|
|
return utils.Location(self.northeast_latitude, self.northeast_longitude)
|
|
|
|
@functools.cached_property
|
|
def southwest(self) -> utils.Location:
|
|
"""The city's southwest corner of the Google Maps viewport.
|
|
|
|
Implementation detail: This property is cached as none of the
|
|
underlying attributes to calculate the value are to be changed.
|
|
"""
|
|
return utils.Location(self.southwest_latitude, self.southwest_longitude)
|
|
|
|
@property
|
|
def total_x(self) -> int:
|
|
"""The horizontal distance from the city's west to east end in meters.
|
|
|
|
The city borders refer to the Google Maps viewport.
|
|
"""
|
|
return self.northeast.easting - self.southwest.easting
|
|
|
|
@property
|
|
def total_y(self) -> int:
|
|
"""The vertical distance from the city's south to north end in meters.
|
|
|
|
The city borders refer to the Google Maps viewport.
|
|
"""
|
|
return self.northeast.northing - self.southwest.northing
|
|
|
|
def clear_map(self) -> City: # pragma: no cover
|
|
"""Create a new `folium.Map` object aligned with the city's viewport.
|
|
|
|
The map is available via the `.map` property. Note that it is mutable
|
|
and changed from various locations in the code base.
|
|
|
|
Returns:
|
|
self: enabling method chaining
|
|
""" # noqa:DAR203 note:d334120e
|
|
self._map = folium.Map(
|
|
location=[self.center_latitude, self.center_longitude],
|
|
zoom_start=self.initial_zoom,
|
|
)
|
|
|
|
return self
|
|
|
|
@property # pragma: no cover
|
|
def map(self) -> folium.Map: # noqa:WPS125
|
|
"""A `folium.Map` object aligned with the city's viewport.
|
|
|
|
See docstring for `.clear_map()` for further info.
|
|
"""
|
|
if not hasattr(self, '_map'): # noqa:WPS421 note:d334120e
|
|
self.clear_map()
|
|
|
|
return self._map
|
|
|
|
def draw_restaurants(
|
|
self, order_counts: bool = False, # pragma: no cover
|
|
) -> folium.Map:
|
|
"""Draw all restaurants on the`.map`.
|
|
|
|
Args:
|
|
order_counts: show the number of orders
|
|
|
|
Returns:
|
|
`.map` for convenience in interactive usage
|
|
"""
|
|
# Obtain all primary `Address`es in the city that host `Restaurant`s.
|
|
addresses = (
|
|
db.session.query(db.Address)
|
|
.filter(
|
|
db.Address.id.in_(
|
|
db.session.query(db.Address.primary_id) # noqa:WPS221
|
|
.join(db.Restaurant, db.Address.id == db.Restaurant.address_id)
|
|
.filter(db.Address.city == self)
|
|
.distinct()
|
|
.all(),
|
|
),
|
|
)
|
|
.all()
|
|
)
|
|
|
|
for address in addresses:
|
|
# Show the restaurant's name if there is only one.
|
|
# Otherwise, list all the restaurants' ID's.
|
|
restaurants = (
|
|
db.session.query(db.Restaurant)
|
|
.join(db.Address, db.Restaurant.address_id == db.Address.id)
|
|
.filter(db.Address.primary_id == address.id)
|
|
.all()
|
|
)
|
|
if len(restaurants) == 1:
|
|
tooltip = f'{restaurants[0].name} (#{restaurants[0].id})' # noqa:WPS221
|
|
else:
|
|
tooltip = 'Restaurants ' + ', '.join( # noqa:WPS336
|
|
f'#{restaurant.id}' for restaurant in restaurants
|
|
)
|
|
|
|
if order_counts:
|
|
# Calculate the number of orders for ALL restaurants ...
|
|
n_orders = (
|
|
db.session.query(db.Order.id)
|
|
.join(db.Address, db.Order.pickup_address_id == db.Address.id)
|
|
.filter(db.Address.primary_id == address.id)
|
|
.count()
|
|
)
|
|
# ... and adjust the size of the red dot on the `.map`.
|
|
if n_orders >= 1000:
|
|
radius = 20
|
|
elif n_orders >= 500:
|
|
radius = 15
|
|
elif n_orders >= 100:
|
|
radius = 10
|
|
elif n_orders >= 10:
|
|
radius = 5
|
|
else:
|
|
radius = 1
|
|
|
|
tooltip += f' | n_orders={n_orders}' # noqa:WPS336
|
|
|
|
address.draw(
|
|
radius=radius,
|
|
color=config.RESTAURANT_COLOR,
|
|
fill_color=config.RESTAURANT_COLOR,
|
|
fill_opacity=0.3,
|
|
tooltip=tooltip,
|
|
)
|
|
|
|
else:
|
|
address.draw(
|
|
radius=1, color=config.RESTAURANT_COLOR, tooltip=tooltip,
|
|
)
|
|
|
|
return self.map
|
|
|
|
def draw_zip_codes(self) -> folium.Map: # pragma: no cover
|
|
"""Draw all addresses on the `.map`, colorized by their `.zip_code`.
|
|
|
|
This does not make a distinction between restaurant and customer addresses.
|
|
Also, due to the high memory usage, the number of orders is not calculated.
|
|
|
|
Returns:
|
|
`.map` for convenience in interactive usage
|
|
"""
|
|
# First, create a color map with distinct colors for each zip code.
|
|
all_zip_codes = sorted(
|
|
row[0]
|
|
for row in db.session.execute(
|
|
sa.text(
|
|
f""" -- # noqa:S608
|
|
SELECT DISTINCT
|
|
{config.CLEAN_SCHEMA}.addresses.zip_code
|
|
FROM
|
|
{config.CLEAN_SCHEMA}.addresses AS addresses
|
|
WHERE
|
|
{config.CLEAN_SCHEMA}.addresses.city_id = {self.id};
|
|
""",
|
|
),
|
|
)
|
|
)
|
|
cmap = utils.make_random_cmap(len(all_zip_codes), bright=False)
|
|
colors = {
|
|
code: utils.rgb_to_hex(*cmap(index))
|
|
for index, code in enumerate(all_zip_codes)
|
|
}
|
|
|
|
# Second, draw every address on the `.map.
|
|
for address in self.addresses:
|
|
# Non-primary addresses are covered by primary ones anyway.
|
|
if not address.is_primary:
|
|
continue
|
|
|
|
marker = folium.Circle( # noqa:WPS317
|
|
(address.latitude, address.longitude),
|
|
color=colors[address.zip_code],
|
|
radius=1,
|
|
)
|
|
marker.add_to(self.map)
|
|
|
|
return self.map
|