Add OrderHistory.first/last_order() methods

- get the `datetime` of the first or last order within a pixel
- unify some fixtures in `tests.forecasts.timify.conftest`
This commit is contained in:
Alexander Hess 2021-01-31 21:24:48 +01:00
parent 7b824a4a12
commit 47ef1f8759
Signed by: alexander
GPG key ID: 344EA5AB10D868E0
5 changed files with 173 additions and 52 deletions

View file

@ -152,6 +152,74 @@ class OrderHistory:
return data.reindex(index, fill_value=0) return data.reindex(index, fill_value=0)
def first_order_at(self, pixel_id: int) -> dt.datetime:
"""Get the time step with the first order in a pixel.
Args:
pixel_id: pixel for which to get the first order
Returns:
minimum "start_at" from when orders take place
Raises:
LookupError: `pixel_id` not in `grid`
# noqa:DAR401 RuntimeError
"""
try:
intra_pixel = self.totals.loc[pixel_id]
except KeyError:
raise LookupError('The `pixel_id` is not in the `grid`') from None
first_order = intra_pixel[intra_pixel['n_orders'] > 0].index.min()
# Sanity check: without an `Order`, the `Pixel` should not exist.
if first_order is pd.NaT: # pragma: no cover
raise RuntimeError('no orders in the pixel')
# Return a proper `datetime.datetime` object.
return dt.datetime(
first_order.year,
first_order.month,
first_order.day,
first_order.hour,
first_order.minute,
)
def last_order_at(self, pixel_id: int) -> dt.datetime:
"""Get the time step with the last order in a pixel.
Args:
pixel_id: pixel for which to get the last order
Returns:
maximum "start_at" from when orders take place
Raises:
LookupError: `pixel_id` not in `grid`
# noqa:DAR401 RuntimeError
"""
try:
intra_pixel = self.totals.loc[pixel_id]
except KeyError:
raise LookupError('The `pixel_id` is not in the `grid`') from None
last_order = intra_pixel[intra_pixel['n_orders'] > 0].index.max()
# Sanity check: without an `Order`, the `Pixel` should not exist.
if last_order is pd.NaT: # pragma: no cover
raise RuntimeError('no orders in the pixel')
# Return a proper `datetime.datetime` object.
return dt.datetime(
last_order.year,
last_order.month,
last_order.day,
last_order.hour,
last_order.minute,
)
def make_horizontal_ts( # noqa:WPS210 def make_horizontal_ts( # noqa:WPS210
self, pixel_id: int, predict_at: dt.datetime, train_horizon: int, self, pixel_id: int, predict_at: dt.datetime, train_horizon: int,
) -> Tuple[pd.Series, int, pd.Series]: ) -> Tuple[pd.Series, int, pd.Series]:

View file

@ -67,4 +67,4 @@ def grid(city):
@pytest.fixture @pytest.fixture
def pixel(grid): def pixel(grid):
"""The `Pixel` in the lower-left corner of the `grid`.""" """The `Pixel` in the lower-left corner of the `grid`."""
return db.Pixel(grid=grid, n_x=0, n_y=0) return db.Pixel(id=1, grid=grid, n_x=0, n_y=0)

View file

@ -0,0 +1,54 @@
"""Fixture for testing the `urban_meal_delivery.forecast.timify` module."""
import pandas as pd
import pytest
from tests import config as test_config
from urban_meal_delivery import config
from urban_meal_delivery.forecasts import timify
@pytest.fixture
def good_pixel_id(pixel):
"""A `pixel_id` that is on the `grid`."""
return pixel.id # `== 1`
@pytest.fixture
def order_totals(good_pixel_id):
"""A mock for `OrderHistory.totals`.
To be a bit more realistic, we sample two pixels on the `grid`.
Uses the LONG_TIME_STEP as the length of a time step.
"""
pixel_ids = [good_pixel_id, good_pixel_id + 1]
gen = (
(pixel_id, start_at)
for pixel_id in pixel_ids
for start_at in pd.date_range(
test_config.START, test_config.END, freq=f'{test_config.LONG_TIME_STEP}T',
)
if config.SERVICE_START <= start_at.hour < config.SERVICE_END
)
# Re-index `data` filling in `0`s where there is no demand.
index = pd.MultiIndex.from_tuples(gen)
index.names = ['pixel_id', 'start_at']
df = pd.DataFrame(data={'n_orders': 1}, index=index)
# Sanity check: n_pixels * n_time_steps_per_day * n_weekdays * n_weeks.
assert len(df) == 2 * 12 * (7 * 2 + 1)
return df
@pytest.fixture
def order_history(order_totals, grid):
"""An `OrderHistory` object that does not need the database."""
oh = timify.OrderHistory(grid=grid, time_step=test_config.LONG_TIME_STEP)
oh._data = order_totals
return oh

View file

@ -11,51 +11,6 @@ import pytest
from tests import config as test_config from tests import config as test_config
from urban_meal_delivery import config from urban_meal_delivery import config
from urban_meal_delivery.forecasts import timify
@pytest.fixture
def good_pixel_id():
"""A `pixel_id` that is on the `grid`."""
return 1
@pytest.fixture
def order_totals(good_pixel_id):
"""A mock for `OrderHistory.totals`.
To be a bit more realistic, we sample two pixels on the `grid`.
"""
pixel_ids = [good_pixel_id, good_pixel_id + 1]
gen = (
(pixel_id, start_at)
for pixel_id in pixel_ids
for start_at in pd.date_range(
test_config.START, test_config.END, freq=f'{test_config.LONG_TIME_STEP}T',
)
if config.SERVICE_START <= start_at.hour < config.SERVICE_END
)
# Re-index `data` filling in `0`s where there is no demand.
index = pd.MultiIndex.from_tuples(gen)
index.names = ['pixel_id', 'start_at']
df = pd.DataFrame(data={'n_orders': 0}, index=index)
# Sanity check: n_pixels * n_time_steps_per_day * n_weekdays * n_weeks.
assert len(df) == 2 * 12 * (7 * 2 + 1)
return df
@pytest.fixture
def order_history(order_totals, grid):
"""An `OrderHistory` object that does not need the database."""
oh = timify.OrderHistory(grid=grid, time_step=test_config.LONG_TIME_STEP)
oh._data = order_totals
return oh
@pytest.fixture @pytest.fixture

View file

@ -1,17 +1,13 @@
"""Test the basic functionalities in the `OrderHistory` class.""" """Test the basic functionalities in the `OrderHistory` class."""
import datetime as dt
import pytest import pytest
from tests import config as test_config from tests import config as test_config
from urban_meal_delivery.forecasts import timify from urban_meal_delivery.forecasts import timify
@pytest.fixture
def order_history(grid):
"""An `OrderHistory` object."""
return timify.OrderHistory(grid=grid, time_step=test_config.LONG_TIME_STEP)
class TestSpecialMethods: class TestSpecialMethods:
"""Test the special methods in `OrderHistory`.""" """Test the special methods in `OrderHistory`."""
@ -32,12 +28,29 @@ class TestProperties:
assert result == time_step assert result == time_step
def test_totals(self, order_history, order_totals):
"""Test `OrderHistory.totals` property.
The result of the `OrderHistory.aggregate_orders()` method call
is cached in the `OrderHistory.totals` property.
Note: `OrderHistory.aggregate_orders()` is not called as
`OrderHistory._data` is already set in the `order_history` fixture.
"""
result = order_history.totals
assert result is order_totals
def test_totals_is_cached(self, order_history, monkeypatch): def test_totals_is_cached(self, order_history, monkeypatch):
"""Test `OrderHistory.totals` property. """Test `OrderHistory.totals` property.
The result of the `OrderHistory.aggregate_orders()` method call The result of the `OrderHistory.aggregate_orders()` method call
is cached in the `OrderHistory.totals` property. is cached in the `OrderHistory.totals` property.
Note: We make `OrderHistory.aggregate_orders()` return a `sentinel`
that is cached into `OrderHistory._data`, which must be unset first.
""" """
monkeypatch.setattr(order_history, '_data', None)
sentinel = object() sentinel = object()
monkeypatch.setattr(order_history, 'aggregate_orders', lambda: sentinel) monkeypatch.setattr(order_history, 'aggregate_orders', lambda: sentinel)
@ -46,3 +59,34 @@ class TestProperties:
assert result1 is result2 assert result1 is result2
assert result1 is sentinel assert result1 is sentinel
class TestMethods:
"""Test various methods in `OrderHistory`."""
def test_first_order_at_existing_pixel(self, order_history, good_pixel_id):
"""Test `OrderHistory.first_order_at()` with good input."""
result = order_history.first_order_at(good_pixel_id)
assert result == test_config.START
def test_first_order_at_non_existing_pixel(self, order_history, good_pixel_id):
"""Test `OrderHistory.first_order_at()` with bad input."""
with pytest.raises(
LookupError, match='`pixel_id` is not in the `grid`',
):
order_history.first_order_at(-1)
def test_last_order_at_existing_pixel(self, order_history, good_pixel_id):
"""Test `OrderHistory.last_order_at()` with good input."""
result = order_history.last_order_at(good_pixel_id)
one_time_step = dt.timedelta(minutes=test_config.LONG_TIME_STEP)
assert result == test_config.END - one_time_step
def test_last_order_at_non_existing_pixel(self, order_history, good_pixel_id):
"""Test `OrderHistory.last_order_at()` with bad input."""
with pytest.raises(
LookupError, match='`pixel_id` is not in the `grid`',
):
order_history.last_order_at(-1)