Вы можете использовать (datetime, increment) кортежи, чтобы отслеживать изменения в доступности. Событие job-start имеет increment = 1, а событие end-end имеет increment = -1. Тогда itertools.accumulate позволяет рассчитать совокупную доступность, когда задания начинаются и заканчиваются со временем. Вот пример реализации:
from datetime import time
import itertools as it
def compute_availability(jobs, opening_hours, capacity):
jobs = [((x, -1), (y, +1)) for x, y in jobs]
opens, closes = opening_hours
events = [[opens, capacity]] + sorted(t for job in jobs for t in job) + [(closes, 0)]
availability = list(it.accumulate(events,
lambda x, y: [y[0], x[1] + y[1]]))
for x, y in zip(availability, availability[1:]):
# If multiple events happen at the same time, only yield the last one.
if y[0] > x[0]:
yield x
Это добавляет искусственные события (opens, capacity) и (closes, 0) для инициализации вычисления. В приведенном выше примере рассматривается один день, но его легко расширить до нескольких дней, создав opens и closes datetime объектов, которые совместно используют день первого и последнего задания соответственно.
Пример
Вот вывод для примера расписания OP:
from pprint import pprint
jobs = [(time(10), time(15)),
(time(9), time(11)),
(time(12, 30), time(16)),
(time(10), time(18))]
availability = list(compute_availability(
jobs, opening_hours=(time(9), time(18)), capacity=3
))
pprint(availability)
, который печатает:
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 0],
[datetime.time(11, 0), 1],
[datetime.time(12, 30), 0],
[datetime.time(15, 0), 1],
[datetime.time(16, 0), 2]]
Первый элемент указывает, когда изменяется доступность и второй элемент обозначает доступность, которая является результатом этого изменения. Например, в 9 утра отправляется одно задание, в результате чего доступность падает с 3 до 2, а затем в 10 утра отправляются еще два задания, пока первое еще работает (следовательно, доступность снижается до 0).
Добавление новых заданий
Теперь, когда мы рассчитали начальную доступность, важным аспектом является ее обновление по мере добавления новых заданий. Здесь желательно не пересчитывать доступность из полного списка заданий, поскольку это может быть дорогостоящим, если отслеживается много заданий. Поскольку availability уже отсортирован, мы можем использовать модуль bisect, чтобы определить соответствующий диапазон обновления в O (log (N)). Затем необходимо выполнить ряд шагов. Допустим, задание запланировано как [x, y], где x, y являются двумя объектами даты и времени.
- Проверьте, является ли доступность в интервале
[x, y] больше нуля (включая событие слева от x (то есть от предыдущего события)). - Уменьшите доступность всех событий в
[x, y] на 1. - Если
x отсутствует в списке события, которые мы должны добавить, в противном случае нам нужно проверить, можем ли мы объединить событие x с тем, которое ему осталось. - Если
y нет в списке событий, нам нужно добавить его .
Вот соответствующий код:
import bisect
def add_job(availability, job, *, weight=1):
"""weight: how many lanes the job requires"""
job = list(job)
start = bisect.bisect(availability, job[:1])
# Emulate a `bisect_right` which doens't work directly since
# we're comparing lists of different length.
if start < len(availability):
start += (job[0] == availability[start][0])
stop = bisect.bisect(availability, job[1:])
if any(slot[1] < weight for slot in availability[start-1:stop]):
raise ValueError('The requested time slot is not available')
for slot in availability[start:stop]:
slot[1] -= weight
if job[0] > availability[start-1][0]:
previous_availability = availability[start-1][1]
availability.insert(start, [job[0], previous_availability - weight])
stop += 1
else:
availability[start-1][1] -= weight
if start >= 2 and availability[start-1][1] == availability[start-2][1]:
del availability[start-1]
stop -= 1
if stop == len(availability) or job[1] < availability[stop][0]:
previous_availability = availability[stop-1][1]
availability.insert(stop, [job[1], previous_availability + weight])
Пример расписания
Мы можем проверить это, добавив некоторые задания в пример расписания ОП:
for job in [[time(15), time(17)],
[time(11, 30), time(12)],
[time(13), time(14)]]: # this one should raise since availability is zero
print(f'\nAdding {job = }')
add_job(availability, job)
pprint(availability)
который выводит:
Adding job = [datetime.time(15, 0), datetime.time(17, 0)]
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 0],
[datetime.time(11, 0), 1],
[datetime.time(12, 30), 0],
[datetime.time(16, 0), 1],
[datetime.time(17, 0), 2]]
Adding job = [datetime.time(11, 30), datetime.time(12, 0)]
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 0],
[datetime.time(11, 0), 1],
[datetime.time(11, 30), 0],
[datetime.time(12, 0), 1],
[datetime.time(12, 30), 0],
[datetime.time(16, 0), 1],
[datetime.time(17, 0), 2]]
Adding job = [datetime.time(13, 0), datetime.time(14, 0)]
Traceback (most recent call last):
[...]
ValueError: The requested time slot is not available
Блокировка ночных часов
Мы также можем использовать этот интерфейс для блокировки всех полос в часы, когда услуга недоступна (например, с 18:00 до 9:00). на следующий день). Просто отправьте задание с weight=capacity на этот промежуток времени:
add_job(availability,
[datetime(2020, 3, 14, 18), datetime(2020, 3, 15, 9)]
weight=3)
Создание полного расписания с нуля
Мы также можем использовать add_job для создания полного расписания с нуля:
availability = availability = list(compute_availability(
[], opening_hours=(time(9), time(18)), capacity=3
))
print('Initial availability')
pprint(availability)
for job in jobs:
print(f'\nAdding {job = }')
add_job(availability, job)
pprint(availability)
который выводит:
Initial availability
[[datetime.time(9, 0), 3]]
Adding job = (datetime.time(10, 0), datetime.time(15, 0))
[[datetime.time(9, 0), 3],
[datetime.time(10, 0), 2],
[datetime.time(15, 0), 3]]
Adding job = (datetime.time(9, 0), datetime.time(11, 0))
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 1],
[datetime.time(11, 0), 2],
[datetime.time(15, 0), 3]]
Adding job = (datetime.time(12, 30), datetime.time(16, 0))
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 1],
[datetime.time(11, 0), 2],
[datetime.time(12, 30), 1],
[datetime.time(15, 0), 2],
[datetime.time(16, 0), 3]]
Adding job = (datetime.time(10, 0), datetime.time(18, 0))
[[datetime.time(9, 0), 2],
[datetime.time(10, 0), 0],
[datetime.time(11, 0), 1],
[datetime.time(12, 30), 0],
[datetime.time(15, 0), 1],
[datetime.time(16, 0), 2],
[datetime.time(18, 0), 3]]