# Scheduling COVID-19 testing in a hospital using a heuristic

I recently came across a problem scheduling Coronavirus testing at a hospital; the testing capacity needed to be allocated to:

• high risk wards (combining many factors)
• ones which hadn't been tested recently

This presents a really tricky problem when scheduling, because as well as complexity in combining many properties of the ward to understand its risk factor, there is a knock on effect where the position of a ward in the schedule dictates its probability of coming up again soon.

Coming back into the programming realm, I wanted to do some kind of weighted average of different factors to compare Wards for "priority", and overload __gt__ to allow for comparison. The problem is I can't directly compare priority of 2 wards to sort the list and create a schedule; ward A and ward B may have exactly the same properties - size, risk factor etc. but if ward B was tested more recently then it has a lower priority.

What I understood is that I can't compare wards, but I can compare different schedules. That is, I can compare timelines to see which is more optimal, and then try and sort a random list in a way that guides it towards a more optimal sorting. That's what I mean by "sorting using a heuristic". I hope that's reasonably clear.

How can I sort a list using a heuristic? I have this base class:

from __future__ import annotations
import numpy as np
from typing import Sequence, Callable, Tuple, Optional
import pprint
import string

class SequenceItemBase:
"""Class that wraps a value and the list that contains it
and overrides normal value comparison with a heuristic for guiding swaps in the list
"""

def __init__(
self,
parent: Sequence[SequenceItemBase],
heuristic: Callable[[Sequence[SequenceItemBase], Tuple[int, int]]],
):
self.parent = parent
self._heuristic = heuristic
def __gt__(self, other):
"An item should be placed higher in the list if doing so would increase the value of the heuristic"
# store a copy of the current list state so we can "imagine" what effect
# swapping self and other would have on the heuristic
after_change = self.parent.copy()
self_index = self.parent.index(self)
other_index = self.parent.index(other)

swap_indecies = sorted((self_index, other_index))

after_change[self_index], after_change[other_index] = after_change[other_index], after_change[self_index]

# whether the swap improved our heuristic
positive_delta_h = self._heuristic(
after_change, swap_indecies
) > self._heuristic(self.parent, swap_indecies)

# if self greater than other, then 1 of 2 things happens:
#     when self is earlier in the list, the swap will happen because we are going ascending
#     vice-versa when self is later in the list
# so if the swap is encouraged by our heuristic, then we must mark self as greater than other
#     only when it is earlier in the list
# and when it is later in the list, then only when our heuristic discourages swapping places
return (self_index < other_index and positive_delta_h) or (
self_index > other_index and not positive_delta_h
)


I've added a few explanatory comments, but essentially what it does is to override the comparison operator which is called at every step of the sorting process, and replace it with one that looks at the current state of the list, imagines swapping the items being compared to see what effect that would have on the list, and if swapping would be good, then make __gt__ return whatever it has to say "the later thing should be earlier in the schedule".

So when asked "Is A greater than B", instead of something like

Is the value of A > the value of B

it says

If I swapped A and B around, would that make the list have a better sorting? If so then yes, A is greater/less than B :)

Bit of playing because we don't know if self or other will be earlier in the list.

This base class can be inherited from to define a sortable class that provides any data the heuristic might need. For example, this one just wraps a value that the heuristic function can access.

class ValueItem(SequenceItemBase):
def __init__(self, value, parent=None, heuristic=None):
self.value = value
super().__init__(parent, heuristic)
def __repr__(self):
return str(self.value)

def prefer_sequences_in_ascending_order_heuristic(
intermediate_state: Sequence[ValueItem],
swap_indecies: Optional[Tuple[int, int]] = None,
):
"This heuristic will return a larger number when the list is sorted in ascending order"
return sum(index * item.value for index, item in enumerate(intermediate_state))


Here the heuristic is equivalent to just doing ascending order. You can see this here:

random_list_of_nums = []
source_nums = np.random.randint(1, 100, 100)
heuristic = prefer_sequences_in_ascending_order_heuristic

# wrap the random numbers in classes that all hold a reference to the containing list
# so that they can be sorted using the heuristic
for i in source_nums:
random_list_of_nums.append(ValueItem(i, random_list_of_nums, heuristic))
before = random_list_of_nums.copy()
perfect = [ValueItem(value, None) for value in sorted(source_nums)]

print(f"{heuristic(before)/heuristic(perfect):0.0%}", before)

selection_sort(random_list_of_nums)
after = random_list_of_nums

print(f"{heuristic(after)/heuristic(perfect):0.0%}", after)


The list is sorted perfectly by value and the heuristic is maximised.

For a more applicable problem, there is a method in scheduling called "minimize average tardiness"; meaning for some tasks each with a duration and due-date, what ordering minimises the average lateness/tardiness:

class TardinessItem(SequenceItemBase):
def __init__(self, duration, due_date, parent=None, heuristic=None):
self.duration = duration
self._due_date = due_date
super().__init__(parent, heuristic)
def tardiness(self, start_date):
return max(0, start_date + self.duration - self._due_date)
def __repr__(self):
return f"{self.name}: duration {self.duration} day{'s' if self.duration > 1 else ''} - due in {self._due_date}"

def tardiness_values(sequence: Sequence[TardinessItem]):
running_date_total = 0
for item in sequence:
yield item.tardiness(running_date_total)
running_date_total += item.duration

def minimising_average_tardiness_heuristic(
intermediate_state: Sequence[TardinessItem],
swap_indecies: Optional[Tuple[int, int]] = None,
):
#negative so that maximising this heuristic will minimise total tardiness
return sum(-tardiness for tardiness in tardiness_values(intermediate_state))


Example:

timeline = []
# source_nums = list(zip(np.random.randint(1,10,10),np.random.randint(20,40,10)))
source_nums = zip([2, 7, 3, 8, 4, 4, 6, 8, 5], [5, 10, 15, 22, 23, 24, 25, 30, 33])

heuristic = minimising_average_tardiness_heuristic

for i, (duration, date) in enumerate(source_nums):
timeline.append(
TardinessItem(duration, date, timeline, minimising_average_tardiness_heuristic)
)
timeline[-1].name = string.ascii_uppercase[i]
pprint.pprint(timeline)
print(
f"Average Tardiness: {np.average(list(tardiness_values(timeline)))}, Heuristic: {heuristic(timeline)}"
)

for _ in range(10):
selection_sort(timeline)
after = timeline

pprint.pprint(after)

print(
f"Average Tardiness: {np.average(list(tardiness_values(timeline)))}, Heuristic: {heuristic(timeline)}"
)


prints

[A: duration 2 days - due in 5,
B: duration 7 days - due in 10,
C: duration 3 days - due in 15,
D: duration 8 days - due in 22,
E: duration 4 days - due in 23,
F: duration 4 days - due in 24,
G: duration 6 days - due in 25,
H: duration 8 days - due in 30,
I: duration 5 days - due in 33]
Average Tardiness: 4.444444444444445, Heuristic: -40

[A: duration 2 days - due in 5,
B: duration 7 days - due in 10,
C: duration 3 days - due in 15,
D: duration 8 days - due in 22,
E: duration 4 days - due in 23,
F: duration 4 days - due in 24,
I: duration 5 days - due in 33,
G: duration 6 days - due in 25,
H: duration 8 days - due in 30]
Average Tardiness: 4.0, Heuristic: -36


which is the same output as MDD gives (another heuristic way to approach minimum tardiness scheduling).

NB

This is designed to be used with an in-place sort because parent effectively holds a live view of the intermediate steps when sorting and at the moment selection_sort is used because I think it reflects the idea of swapping elements as a measure of progress, but I'm open to suggestions...

def selection_sort(nums):
# This value of i corresponds to how many values were sorted
for i in range(len(nums)):
# We assume that the first item of the unsorted segment is the smallest
lowest_value_index = i
# This loop iterates over the unsorted items
for j in range(i + 1, len(nums)):
if nums[j] < nums[lowest_value_index]:
lowest_value_index = j
# Swap values of the lowest unsorted element with the first unsorted
# element
nums[i], nums[lowest_value_index] = nums[lowest_value_index], nums[i]

• Rather than being called a "heuristic", this is called a "metric", "reward function", or "cost". There is a standard body of literature about how to minimize/maximize a cost, once you can evaluate it. Using the standard language this will help you find relevant stuff and other people find it for you. – Zachary Vance Nov 17 '20 at 6:06

swap_indecies -> swap_indices

Also, you have inconsistent type hinting on your methods: this one is complete -

def __init__(
self,
parent: Sequence[SequenceItemBase],
heuristic: Callable[[Sequence[SequenceItemBase], Tuple[int, int]]],
):


but these are not:

def __gt__(self, other):

def __init__(self, value, parent=None, heuristic=None):

def __init__(self, duration, due_date, parent=None, heuristic=None):


The latter suggests that your original hints are incorrect and should be wrapped in Optional.