Fix route optimizer and group assignment; add test suite
Route building / optimization (tatami_masterplan.py): - fast_total_time was permutation-invariant: it applied get_courses to group *values* instead of slots and ignored the permutation, so every ordering scored identically and the annealing optimized nothing. It now maps each rotation slot to its assigned group via the permutation. - Replaced the broken next_permutation/simulated_annealing (enumerated n! orderings per iteration, fed unnormalized Boltzmann weights to np.random.choice -> ValueError, and returned the last random sample) with a standard neighbor-swap annealer that tracks and returns the best solution and handles <2 slots. - Convert the reduced Timedelta matrix to float seconds before annealing (np.exp can't operate on Timedelta). Group building (tatami_masterplan.py): - assign_courses set each group's hosts (sorted by course) before all courses were assigned, so hosts whose course was still None got mis-ordered. Assign all courses first, then wire up hosts. - get_masterplan no longer mutates the caller's participant list. classes.py: - Narrow casts on distance-matrix lookups to satisfy the mypy gate (pre-existing failures). Tests: - Add pytest suite (74 tests) covering the rotation topology, route cost and optimization, the domain model, the masterplan pipeline, and the Routes API wrapper (HTTP mocked). The cost cross-check caught the assign_courses ordering bug above.
This commit is contained in:
@@ -0,0 +1,67 @@
|
||||
"""Shared pytest fixtures and test setup for the tatami package.
|
||||
|
||||
``tatami.traveltimes`` raises at import time if ``GOOGLE_MAPS_API_KEY`` is unset,
|
||||
so we set a dummy value here (conftest is imported before any test module, and
|
||||
therefore before the package is imported) to make the modules importable without
|
||||
a real API key. Tests never hit the live API; the HTTP layer is mocked.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
os.environ.setdefault("GOOGLE_MAPS_API_KEY", "test-key")
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
from tatami.classes import Participant
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _seed_rng():
|
||||
"""Make the randomized steps (shuffles, annealing) deterministic per test."""
|
||||
np.random.seed(1234)
|
||||
|
||||
|
||||
def make_participants(
|
||||
n: int, kitchen_sizes: list[float] | None = None
|
||||
) -> list[Participant]:
|
||||
"""Create ``n`` participants with distinct addresses and uuids."""
|
||||
if kitchen_sizes is None:
|
||||
kitchen_sizes = [10.0] * n
|
||||
return [
|
||||
Participant(
|
||||
name=f"P{i}",
|
||||
address=f"address {i}",
|
||||
phone=f"phone {i}",
|
||||
kitchen_size=kitchen_sizes[i],
|
||||
allergies="",
|
||||
)
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
|
||||
def make_timedelta_matrix(
|
||||
participants: list[Participant], seconds: np.ndarray
|
||||
) -> pd.DataFrame:
|
||||
"""Build a uuid-indexed square matrix of ``pd.Timedelta`` from a seconds array.
|
||||
|
||||
Mirrors the shape produced by ``get_participant_distance_matrix``: object dtype
|
||||
cells holding ``pd.Timedelta`` values, indexed and columned by participant uuid.
|
||||
"""
|
||||
uuids = [p.uuid for p in participants]
|
||||
matrix = pd.DataFrame(index=pd.Index(uuids), columns=pd.Index(uuids), dtype=object)
|
||||
for i, u in enumerate(uuids):
|
||||
for j, v in enumerate(uuids):
|
||||
matrix.at[u, v] = pd.to_timedelta(f"{int(seconds[i][j])}s")
|
||||
return matrix
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def participants_factory():
|
||||
return make_participants
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def matrix_factory():
|
||||
return make_timedelta_matrix
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Tests for the core domain model: Participant and Group."""
|
||||
|
||||
import datetime as dt
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from tatami.classes import Group, Participant
|
||||
from conftest import make_participants, make_timedelta_matrix
|
||||
|
||||
|
||||
class TestParticipant:
|
||||
def test_uuid_is_unique(self):
|
||||
a = Participant("A", "addr", "", 5, "")
|
||||
b = Participant("A", "addr", "", 5, "")
|
||||
assert a.uuid != b.uuid
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kitchen_size,minutes",
|
||||
[(10, 0), (7, 9), (0, 30), (5, 15)],
|
||||
)
|
||||
def test_penalty_scales_with_kitchen_size(self, kitchen_size, minutes):
|
||||
p = Participant("A", "addr", "", kitchen_size, "")
|
||||
assert p.get_penalty() == dt.timedelta(minutes=minutes)
|
||||
|
||||
def test_after_party_time_combines_penalty_and_travel(self):
|
||||
people = make_participants(2, kitchen_sizes=[7.0, 10.0])
|
||||
# 600s travel between the two participants.
|
||||
matrix = make_timedelta_matrix(people, np.array([[0, 600], [600, 0]]))
|
||||
after_party = Group(members=[people[1]])
|
||||
result = people[0].get_after_party_time(matrix, after_party)
|
||||
assert result == dt.timedelta(minutes=9) + dt.timedelta(seconds=600)
|
||||
|
||||
def test_dict_roundtrip_fields(self):
|
||||
p = Participant("Alice", "addr", "555", 8, "peanuts")
|
||||
d = p.dict()
|
||||
assert d == {
|
||||
"uuid": p.uuid,
|
||||
"name": "Alice",
|
||||
"address": "addr",
|
||||
"phone": "555",
|
||||
"kitchen_size": 8,
|
||||
"allergies": "peanuts",
|
||||
}
|
||||
|
||||
|
||||
class TestGroup:
|
||||
def test_default_main_member_is_first(self):
|
||||
people = make_participants(3)
|
||||
group = Group(members=people)
|
||||
assert group.main_member is people[0]
|
||||
|
||||
def test_explicit_main_member_index(self):
|
||||
people = make_participants(3)
|
||||
group = Group(members=people, main_member=2)
|
||||
assert group.main_member is people[2]
|
||||
|
||||
def test_add_member_appends_without_changing_main(self):
|
||||
people = make_participants(2)
|
||||
group = Group(members=[people[0]])
|
||||
group.add_member(people[1])
|
||||
assert people[1] in group.members
|
||||
assert group.main_member is people[0]
|
||||
|
||||
def test_add_member_can_promote_to_main(self):
|
||||
people = make_participants(2)
|
||||
group = Group(members=[people[0]])
|
||||
group.add_member(people[1], main_member=True)
|
||||
assert group.main_member is people[1]
|
||||
|
||||
def test_set_hosts_sorts_by_course(self):
|
||||
people = make_participants(3)
|
||||
groups = [Group(members=[p]) for p in people]
|
||||
groups[0].set_course("dessert")
|
||||
groups[1].set_course("starter")
|
||||
groups[2].set_course("main")
|
||||
host = Group(members=[make_participants(1)[0]])
|
||||
host.set_hosts([groups[0], groups[1], groups[2]])
|
||||
assert [g.course for g in host.hosts] == ["starter", "main", "dessert"]
|
||||
|
||||
def test_get_total_time_sums_legs_and_penalties(self):
|
||||
# Three host groups (penalties 0) + after party; verify the summed route.
|
||||
people = make_participants(4, kitchen_sizes=[10, 10, 10, 10])
|
||||
seconds = np.array(
|
||||
[
|
||||
[0, 100, 0, 0, 50],
|
||||
[0, 0, 200, 0, 0],
|
||||
[0, 0, 0, 0, 300],
|
||||
[0, 0, 0, 0, 0],
|
||||
[0, 0, 0, 0, 0],
|
||||
]
|
||||
)
|
||||
# build matrix including an after party participant
|
||||
after_party_p = make_participants(1)[0]
|
||||
all_people = people + [after_party_p]
|
||||
full = make_timedelta_matrix(all_people, seconds)
|
||||
|
||||
starter = Group(members=[people[0]])
|
||||
starter.set_course("starter")
|
||||
main = Group(members=[people[1]])
|
||||
main.set_course("main")
|
||||
dessert = Group(members=[people[2]])
|
||||
dessert.set_course("dessert")
|
||||
after_party = Group(members=[after_party_p])
|
||||
|
||||
visitor = Group(members=[people[3]])
|
||||
visitor.set_hosts([starter, main, dessert])
|
||||
total = visitor.get_total_time(full, after_party)
|
||||
# legs: starter->main (100) + main->dessert (200) + dessert->afterparty (300)
|
||||
assert total == dt.timedelta(seconds=600)
|
||||
|
||||
def test_get_total_time_raises_without_hosts(self):
|
||||
group = Group(members=make_participants(1))
|
||||
after_party = Group(members=make_participants(1))
|
||||
with pytest.raises(ValueError):
|
||||
group.get_total_time(None, after_party)
|
||||
|
||||
def test_get_guests_collects_visiting_members(self):
|
||||
people = make_participants(4)
|
||||
host = Group(members=[people[0]])
|
||||
guest_a = Group(members=[people[1]])
|
||||
guest_b = Group(members=[people[2], people[3]])
|
||||
# get_guests returns early when the host has no hosts of its own, so give it
|
||||
# a (self-)host as it would have in a real plan.
|
||||
host.set_hosts([host])
|
||||
guest_a.set_hosts([host])
|
||||
guest_b.set_hosts([host])
|
||||
guests = host.get_guests([guest_a, guest_b])
|
||||
guest_uuids = {p.uuid for p in guests}
|
||||
assert guest_uuids == {people[1].uuid, people[2].uuid, people[3].uuid}
|
||||
|
||||
def test_dict_serializes_uuids(self):
|
||||
people = make_participants(2)
|
||||
group = Group(members=people)
|
||||
host = Group(members=make_participants(1))
|
||||
host.set_course("starter")
|
||||
group.set_hosts([host])
|
||||
d = group.dict()
|
||||
assert d["main_member"] == people[0].uuid
|
||||
assert d["members"] == [p.uuid for p in people]
|
||||
assert d["hosts"] == [host.uuid]
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Integration tests for the masterplan pipeline (group building + assignment)."""
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from tatami.classes import Group
|
||||
from tatami.traveltimes import reduce_distance_matrix
|
||||
from tatami.tatami_masterplan import (
|
||||
assign_courses,
|
||||
fast_total_time,
|
||||
get_after_party_group,
|
||||
get_courses,
|
||||
get_masterplan,
|
||||
load_csv_to_participants,
|
||||
)
|
||||
from conftest import make_participants, make_timedelta_matrix
|
||||
|
||||
|
||||
def full_matrix(participants, after_party_group, seed=0):
|
||||
"""Synthetic uuid-indexed Timedelta matrix over participants + after party."""
|
||||
all_people = participants + [after_party_group.main_member]
|
||||
rng = np.random.default_rng(seed)
|
||||
n = len(all_people)
|
||||
seconds = rng.integers(60, 1200, size=(n, n)).astype(float)
|
||||
np.fill_diagonal(seconds, 0)
|
||||
return make_timedelta_matrix(all_people, seconds)
|
||||
|
||||
|
||||
class TestGetAfterPartyGroup:
|
||||
def test_single_member_with_no_penalty(self):
|
||||
group = get_after_party_group("party street 1")
|
||||
assert len(group.members) == 1
|
||||
assert group.main_member.address == "party street 1"
|
||||
assert group.main_member.get_penalty().total_seconds() == 0
|
||||
|
||||
|
||||
class TestLoadCsv:
|
||||
def test_parses_tab_separated_file(self, tmp_path):
|
||||
csv = tmp_path / "config.csv"
|
||||
csv.write_text(
|
||||
"name\taddress\tphone\tkitchen_size\tallergies\n"
|
||||
"Alice\tStreet 1\t111\t8\tnone\n"
|
||||
"Bob\tStreet 2\t222\t5\tpeanuts\n"
|
||||
)
|
||||
participants = load_csv_to_participants(str(csv))
|
||||
assert [p.name for p in participants] == ["Alice", "Bob"]
|
||||
assert participants[0].address == "Street 1"
|
||||
assert participants[1].kitchen_size == 5
|
||||
|
||||
|
||||
class TestAssignCourses:
|
||||
def test_assigns_cycling_courses_and_hosts(self):
|
||||
groups = [Group(members=[p]) for p in make_participants(6)]
|
||||
courses = ["starter", "main", "dessert"] * 2
|
||||
assign_courses(groups, courses)
|
||||
assert [g.course for g in groups] == courses
|
||||
for i, group in enumerate(groups):
|
||||
expected_host_uuids = {groups[j].uuid for j in get_courses(i, len(groups))}
|
||||
assert {h.uuid for h in group.hosts} == expected_host_uuids
|
||||
assert group.uuid in {h.uuid for h in group.hosts}
|
||||
|
||||
|
||||
class TestGetMasterplan:
|
||||
@pytest.mark.parametrize("n", [6, 12, 18])
|
||||
def test_group_count_and_coverage(self, n):
|
||||
participants = make_participants(n, kitchen_sizes=list(np.linspace(0, 10, n)))
|
||||
after_party = get_after_party_group("party street")
|
||||
matrix = full_matrix(participants, after_party)
|
||||
|
||||
group_dicts, participant_dicts = get_masterplan(
|
||||
participants, matrix, after_party
|
||||
)
|
||||
|
||||
expected_groups = 3 * (n // 6)
|
||||
assert len(group_dicts) == expected_groups
|
||||
assert len(participant_dicts) == n
|
||||
|
||||
# Every participant ends up in exactly one group.
|
||||
assigned = [uuid for g in group_dicts for uuid in g["members"]]
|
||||
assert len(assigned) == n
|
||||
assert set(assigned) == {p.uuid for p in participants}
|
||||
|
||||
def test_topology_of_output(self):
|
||||
participants = make_participants(18, kitchen_sizes=list(np.linspace(0, 10, 18)))
|
||||
after_party = get_after_party_group("party street")
|
||||
matrix = full_matrix(participants, after_party)
|
||||
|
||||
group_dicts, _ = get_masterplan(participants, matrix, after_party)
|
||||
by_uuid = {g["uuid"]: g for g in group_dicts}
|
||||
|
||||
# Courses are balanced across the three slots.
|
||||
course_counts = {}
|
||||
for g in group_dicts:
|
||||
course_counts[g["course"]] = course_counts.get(g["course"], 0) + 1
|
||||
assert course_counts == {"starter": 3, "main": 3, "dessert": 3}
|
||||
|
||||
for g in group_dicts:
|
||||
host_courses = sorted(by_uuid[h]["course"] for h in g["hosts"])
|
||||
assert host_courses == ["dessert", "main", "starter"]
|
||||
assert g["uuid"] in g["hosts"] # a group hosts its own course
|
||||
|
||||
def test_does_not_mutate_input_participant_order(self):
|
||||
participants = make_participants(12)
|
||||
original_order = list(participants)
|
||||
after_party = get_after_party_group("party street")
|
||||
matrix = full_matrix(participants, after_party)
|
||||
get_masterplan(participants, matrix, after_party)
|
||||
assert participants == original_order
|
||||
|
||||
|
||||
class TestCostConsistency:
|
||||
def test_fast_total_time_matches_group_get_total_time(self):
|
||||
# The optimizer cost (fast_total_time on the penalty-baked reduced matrix)
|
||||
# must equal the sum of per-group route times computed independently by
|
||||
# Group.get_total_time on the raw matrix.
|
||||
n = 6
|
||||
participants = make_participants(n, kitchen_sizes=[10, 8, 6, 4, 2, 0])
|
||||
after_party = get_after_party_group("party street")
|
||||
groups = [Group(members=[p]) for p in participants]
|
||||
courses = ["starter", "main", "dessert"] * (n // 3)
|
||||
assign_courses(groups, courses)
|
||||
|
||||
full = full_matrix(participants, after_party, seed=5)
|
||||
|
||||
reduced = reduce_distance_matrix(full, [*groups, after_party])
|
||||
reduced_seconds = np.array(
|
||||
[[pd_to_seconds(x) for x in row] for row in reduced.to_numpy()],
|
||||
dtype=float,
|
||||
)
|
||||
optimizer_cost = fast_total_time(reduced_seconds, list(range(n)))
|
||||
|
||||
independent_cost = sum(
|
||||
g.get_total_time(full, after_party).total_seconds() for g in groups
|
||||
)
|
||||
assert optimizer_cost == pytest.approx(independent_cost)
|
||||
|
||||
|
||||
def pd_to_seconds(value):
|
||||
import pandas as pd
|
||||
|
||||
return pd.Timedelta(value).total_seconds()
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Tests for the dinner-rotation topology and the route-cost / optimization logic.
|
||||
|
||||
These cover the two areas most prone to subtle bugs: the ``get_courses`` rotation
|
||||
(who hosts whom) and ``fast_total_time`` / ``simulated_annealing`` (the route cost
|
||||
and its optimization).
|
||||
"""
|
||||
|
||||
import itertools
|
||||
from collections import Counter
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from tatami.tatami_masterplan import (
|
||||
fast_total_time,
|
||||
get_courses,
|
||||
simulated_annealing,
|
||||
)
|
||||
|
||||
# Group counts must be multiples of 3 (one slot per course); test a range of sizes.
|
||||
GROUP_COUNTS = [3, 6, 9, 12, 15]
|
||||
|
||||
|
||||
def asymmetric_matrix(n: int, seed: int = 0) -> np.ndarray:
|
||||
"""Random asymmetric (n+1)x(n+1) cost matrix, last index = after party."""
|
||||
rng = np.random.default_rng(seed)
|
||||
matrix = rng.integers(60, 1200, size=(n + 1, n + 1)).astype(float)
|
||||
np.fill_diagonal(matrix, 0)
|
||||
return matrix
|
||||
|
||||
|
||||
class TestGetCourses:
|
||||
@pytest.mark.parametrize("n", GROUP_COUNTS)
|
||||
def test_group_is_its_own_first_host(self, n):
|
||||
# A group always hosts the course it cooks, so it appears in its own host list.
|
||||
for i in range(n):
|
||||
assert i in get_courses(i, n)
|
||||
|
||||
@pytest.mark.parametrize("n", GROUP_COUNTS)
|
||||
def test_each_group_visits_three_distinct_courses(self, n):
|
||||
for i in range(n):
|
||||
hosts = get_courses(i, n)
|
||||
assert len(hosts) == 3
|
||||
assert len(set(hosts)) == 3, "a group must visit three distinct hosts"
|
||||
# The three hosts must cook three different courses (course == index % 3).
|
||||
assert {h % 3 for h in hosts} == {0, 1, 2}
|
||||
|
||||
@pytest.mark.parametrize("n", GROUP_COUNTS)
|
||||
def test_hosts_returned_in_starter_main_dessert_order(self, n):
|
||||
# get_courses sorts by index % 3 -> starter(0), main(1), dessert(2).
|
||||
for i in range(n):
|
||||
a, b, c = get_courses(i, n)
|
||||
assert (a % 3, b % 3, c % 3) == (0, 1, 2)
|
||||
|
||||
@pytest.mark.parametrize("n", GROUP_COUNTS)
|
||||
def test_every_host_receives_exactly_three_groups(self, n):
|
||||
# The defining invariant of a running dinner: every hosting location serves
|
||||
# exactly three groups (itself + two guests) for its course.
|
||||
received = Counter()
|
||||
for i in range(n):
|
||||
for host in get_courses(i, n):
|
||||
received[host] += 1
|
||||
assert set(received.values()) == {3}
|
||||
assert len(received) == n
|
||||
|
||||
@pytest.mark.parametrize("n", GROUP_COUNTS)
|
||||
def test_guests_at_each_host_cook_distinct_courses(self, n):
|
||||
# Invert the relation: the (exactly three) groups that show up at host h for
|
||||
# its course must each be responsible for a different course in the rotation.
|
||||
hosts_of = {i: set(get_courses(i, n)) for i in range(n)}
|
||||
for host in range(n):
|
||||
guests = [g for g in range(n) if host in hosts_of[g]]
|
||||
assert len(guests) == 3
|
||||
assert {g % 3 for g in guests} == {0, 1, 2}
|
||||
|
||||
|
||||
class TestFastTotalTime:
|
||||
def test_matches_hand_computed_value_n3(self):
|
||||
# For n=3 every slot resolves to hosts (0, 1, 2), so the cost is
|
||||
# 3 * (D[s0][s1] + D[s1][s2] + D[s2][afterparty]).
|
||||
matrix = np.array(
|
||||
[
|
||||
[0, 10, 20, 30],
|
||||
[40, 0, 50, 60],
|
||||
[70, 80, 0, 90],
|
||||
[1, 2, 3, 0], # after-party row (unused as origin)
|
||||
],
|
||||
dtype=float,
|
||||
)
|
||||
solution = [0, 1, 2]
|
||||
expected = 3 * (matrix[0][1] + matrix[1][2] + matrix[2][3])
|
||||
assert fast_total_time(matrix, solution) == pytest.approx(expected)
|
||||
|
||||
def test_cost_depends_on_permutation(self):
|
||||
# The whole point of the optimizer: different slot->group assignments must
|
||||
# generally produce different costs on an asymmetric matrix.
|
||||
matrix = asymmetric_matrix(6, seed=2)
|
||||
costs = {
|
||||
fast_total_time(matrix, list(p))
|
||||
for p in itertools.islice(itertools.permutations(range(6)), 50)
|
||||
}
|
||||
assert len(costs) > 1
|
||||
|
||||
def test_returns_float(self):
|
||||
matrix = asymmetric_matrix(6, seed=2)
|
||||
assert isinstance(fast_total_time(matrix, list(range(6))), float)
|
||||
|
||||
|
||||
class TestSimulatedAnnealing:
|
||||
@pytest.mark.parametrize("seed", [0, 1, 2, 3, 4])
|
||||
def test_never_returns_worse_than_optimum(self, seed):
|
||||
# Sanity floor: the cost can never be below the true (brute-force) optimum.
|
||||
n = 6
|
||||
matrix = asymmetric_matrix(n, seed=seed)
|
||||
optimum = min(
|
||||
fast_total_time(matrix, list(p)) for p in itertools.permutations(range(n))
|
||||
)
|
||||
result = simulated_annealing(matrix, list(range(n)), 1000, 0.99, 5000)
|
||||
assert fast_total_time(matrix, result) >= optimum
|
||||
|
||||
@pytest.mark.parametrize("seed", [0, 1, 2, 3, 4])
|
||||
def test_beats_random_baseline(self, seed):
|
||||
# SA should reliably do better than the average random ordering.
|
||||
n = 9
|
||||
matrix = asymmetric_matrix(n, seed=seed)
|
||||
random_costs = []
|
||||
for _ in range(500):
|
||||
perm = list(range(n))
|
||||
np.random.shuffle(perm)
|
||||
random_costs.append(fast_total_time(matrix, perm))
|
||||
result = simulated_annealing(matrix, list(range(n)), 1000, 0.99, 5000)
|
||||
assert fast_total_time(matrix, result) < np.mean(random_costs)
|
||||
|
||||
def test_returns_valid_permutation(self):
|
||||
n = 9
|
||||
matrix = asymmetric_matrix(n, seed=3)
|
||||
result = simulated_annealing(matrix, list(range(n)), 1000, 0.99, 1000)
|
||||
assert sorted(result) == list(range(n))
|
||||
|
||||
def test_handles_single_slot(self):
|
||||
matrix = asymmetric_matrix(1, seed=0)
|
||||
assert simulated_annealing(matrix, [0], 1000, 0.99, 100) == [0]
|
||||
|
||||
def test_handles_empty(self):
|
||||
matrix = np.zeros((1, 1))
|
||||
assert simulated_annealing(matrix, [], 1000, 0.99, 100) == []
|
||||
|
||||
def test_does_not_mutate_input_indices(self):
|
||||
n = 6
|
||||
matrix = asymmetric_matrix(n, seed=1)
|
||||
indices = list(range(n))
|
||||
simulated_annealing(matrix, indices, 1000, 0.99, 100)
|
||||
assert indices == list(range(n))
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Tests for the Google Routes API wrapper (HTTP layer mocked)."""
|
||||
|
||||
import datetime as dt
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
import tatami.traveltimes as tt
|
||||
from tatami.classes import Group
|
||||
from tatami.traveltimes import (
|
||||
get_distance_matrix,
|
||||
get_participant_distance_matrix,
|
||||
reduce_distance_matrix,
|
||||
)
|
||||
from conftest import make_participants, make_timedelta_matrix
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, status_code, json_data=None, text=""):
|
||||
self.status_code = status_code
|
||||
self._json = json_data
|
||||
self.text = text
|
||||
|
||||
def json(self):
|
||||
return self._json
|
||||
|
||||
|
||||
def route_matrix_payload(durations_seconds):
|
||||
"""Build a computeRouteMatrix-style flat list of elements from an n x n array."""
|
||||
payload = []
|
||||
n = len(durations_seconds)
|
||||
for i in range(n):
|
||||
for j in range(n):
|
||||
payload.append(
|
||||
{
|
||||
"originIndex": i,
|
||||
"destinationIndex": j,
|
||||
"duration": f"{int(durations_seconds[i][j])}s",
|
||||
"distanceMeters": int(durations_seconds[i][j]) * 4,
|
||||
}
|
||||
)
|
||||
return payload
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_post(monkeypatch):
|
||||
"""Patch requests.post; returns a setter for the desired response."""
|
||||
state = {}
|
||||
|
||||
def fake_post(url, json=None, headers=None):
|
||||
return state["response"]
|
||||
|
||||
monkeypatch.setattr(tt.requests, "post", fake_post)
|
||||
|
||||
def set_response(response):
|
||||
state["response"] = response
|
||||
|
||||
return set_response
|
||||
|
||||
|
||||
class TestGetDistanceMatrix:
|
||||
def test_builds_duration_matrix(self, mock_post):
|
||||
durations = np.array([[0, 120], [300, 0]])
|
||||
mock_post(FakeResponse(200, route_matrix_payload(durations)))
|
||||
matrix = get_distance_matrix(["a", "b"])
|
||||
assert matrix.shape == (2, 2)
|
||||
assert matrix.loc[0, 1] == pd.to_timedelta("120s")
|
||||
assert matrix.loc[1, 0] == pd.to_timedelta("300s")
|
||||
|
||||
def test_distance_meters_value(self, mock_post):
|
||||
durations = np.array([[0, 120], [300, 0]])
|
||||
mock_post(FakeResponse(200, route_matrix_payload(durations)))
|
||||
matrix = get_distance_matrix(["a", "b"], value="distanceMeters")
|
||||
assert matrix.loc[0, 1] == 480.0
|
||||
|
||||
def test_empty_addresses_raises(self):
|
||||
with pytest.raises(ValueError):
|
||||
get_distance_matrix([])
|
||||
|
||||
def test_non_200_raises(self, mock_post):
|
||||
mock_post(FakeResponse(500, text="boom"))
|
||||
with pytest.raises(Exception, match="boom"):
|
||||
get_distance_matrix(["a", "b"])
|
||||
|
||||
def test_invalid_value_raises(self, mock_post):
|
||||
durations = np.array([[0, 120], [300, 0]])
|
||||
mock_post(FakeResponse(200, route_matrix_payload(durations)))
|
||||
with pytest.raises(ValueError):
|
||||
get_distance_matrix(["a", "b"], value="bogus")
|
||||
|
||||
|
||||
class TestGetParticipantDistanceMatrix:
|
||||
def test_labels_axes_with_uuids(self, mock_post):
|
||||
participants = make_participants(2)
|
||||
durations = np.array([[0, 120], [300, 0]])
|
||||
mock_post(FakeResponse(200, route_matrix_payload(durations)))
|
||||
matrix = get_participant_distance_matrix(participants)
|
||||
uuids = [p.uuid for p in participants]
|
||||
assert list(matrix.index) == uuids
|
||||
assert list(matrix.columns) == uuids
|
||||
|
||||
|
||||
class TestReduceDistanceMatrix:
|
||||
def test_selects_main_members_and_adds_penalty(self):
|
||||
# group 0 main member has kitchen_size 7 -> 9 minute (540s) penalty.
|
||||
people = make_participants(2, kitchen_sizes=[7.0, 10.0])
|
||||
seconds = np.array([[0, 100], [200, 0]])
|
||||
matrix = make_timedelta_matrix(people, seconds)
|
||||
groups = [Group(members=[people[0]]), Group(members=[people[1]])]
|
||||
|
||||
reduced = reduce_distance_matrix(matrix, groups)
|
||||
|
||||
u0, u1 = people[0].uuid, people[1].uuid
|
||||
assert list(reduced.index) == [u0, u1]
|
||||
# Penalty added to every entry of group 0's row.
|
||||
assert reduced.loc[u0, u1] == dt.timedelta(seconds=100) + dt.timedelta(
|
||||
minutes=9
|
||||
)
|
||||
# Group 1 has no penalty (kitchen_size 10).
|
||||
assert reduced.loc[u1, u0] == dt.timedelta(seconds=200)
|
||||
Reference in New Issue
Block a user