ross/proc-rota
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor and add some more tests

Browse Source
This commit is contained in:
Ross
2022-05-16 19:35:38 +01:00
parent ff03358119
commit f887257a55
6 changed files with 427 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
output/timetable.js
+1
View File
@@ -361,6 +361,7 @@ function viewWorker(worker) {
dlg = $(`<div class='dialog' title='${ds.worker}'>
Site: ${ds.site}</br>
FTE: ${ds.fte} (${ds.fte_adj})</br>
Start date: ${ds.start_date}</br>
End date: ${ds.end_date}</br>
Non working days: ${ds.nwds}
</div>`)
rota/run2.py
-8
View File
@@ -5,15 +5,7 @@ from workers import Worker
from copy import deepcopy
import math
import scipy.stats as stats
import statistics
import datetime
import itertools
import operator
use_neos = False
rota/shifts.py
+49 -4
View File
@@ -267,12 +267,13 @@ class RotaBuilder(object):
if not results.solver.status:
sys.exit(0)
def build_and_solve(self, options={"ratio": 0.1, "seconds": 1000, "threads": 10}):
def build_and_solve(self, options={"ratio": 0.1, "seconds": 1000, "threads": 10}, solve=True):
self.build_shifts()
self.build_workers()
self.build_model()
self.solve_model(options=options)
if solve:
self.solve_model(options=options)
def build_model(self):
# Initialize model
@@ -1916,7 +1917,7 @@ class RotaBuilder(object):
self.workers.append(worker)
def add_workers(self, workers: List) -> None:
"""Add multiple worker to the rota
"""Add multiple workers to the rota
Args:
workers (List(Worker)):
@@ -1932,6 +1933,21 @@ class RotaBuilder(object):
if not self.workers:
raise NoWorkers("Workers must be added prior to calling build_workers")
self.workers_id_map = {}
self.workers_name_map = {}
for worker in self.workers:
wid = worker.id
if wid in self.workers_id_map:
raise ValueError(f"Worker with id '{wid}' has been added twice")
self.workers_id_map[wid] = worker
if worker.name in self.workers_name_map:
raise ValueError(f"Worker with name '{worker.name}' has been added twice")
self.workers_name_map[worker.name] = worker
self.workers = sorted(self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers)
@@ -2216,6 +2232,9 @@ class RotaBuilder(object):
s.extend(self.shifts_to_force_as_blocks())
return s
def get_worker_by_name(self, name: str) -> Worker:
return self.workers_name_map[name]
def get_workers_by_group(self) -> Dict[str, Worker]:
group_workers = defaultdict(set)
for worker in self.workers:
@@ -2315,6 +2334,7 @@ class RotaBuilder(object):
week_table[week][day][shift].append(worker.get_details())
return week_table
def get_worker_timetable(self):
works = self.model.works
timetable = {
@@ -2511,7 +2531,9 @@ class RotaBuilder(object):
worker_td = """<td title='Site: {site}' class='worker {site}'
data-nwds='{nwds}' data-site='{site}' data-worker='{name}'
data-fte='{fte}' data-fte_adj='{fte_adj}' data-end_date='{end_date}'
data-fte='{fte}' data-fte_adj='{fte_adj}'
data-start_date='{start_date}'
data-end_date='{end_date}'
data-worker-targets='{targets}' data-shift-counts='{worker_shift_counts}'
data-weekend-target='{weekend_target}'
data-remote-site='{remote_site}'
@@ -2527,6 +2549,7 @@ class RotaBuilder(object):
name=worker.name,
fte=worker.fte,
fte_adj=worker.fte_adj,
start_date=worker.start_date,
end_date=worker.end_date,
targets=worker_targets,
worker_shift_counts=json.dumps(shift_count_dict),
@@ -2645,6 +2668,22 @@ class RotaBuilder(object):
return timetable
def get_worker_shifts_by_date(self, worker: Worker) -> Dict:
shifts = {}
n = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
n = n + 1
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0:
shifts[d]=shift
return shifts
def get_worker_shift_list(self, worker: Worker) -> List:
shifts = []
@@ -2660,6 +2699,12 @@ class RotaBuilder(object):
return shifts
def get_worker_shift_list_string(self, worker: Worker) -> str:
shifts = self.get_worker_shift_list(worker)
# Convert shift to a string representation
return "".join([i if i != "" else "-" for i in shifts])
def get_shift_summary(self):
works = self.model.works
timetable = {
rota/workers.py
+92 -42
View File
@@ -2,10 +2,10 @@ import datetime
from collections import defaultdict
from typing import List
#from .shifts import RotaBuilder, days, sites
# from .shifts import RotaBuilder, days, sites
import uuid
#from rota.shifts import RotaBuilder
# from rota.shifts import RotaBuilder
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -21,9 +21,10 @@ class Worker:
grade: int,
id=None,
fte: int = 100,
nwd: list[str] | list[str, datetime.datetime, datetime.datetime]=[],
end_date: datetime.datetime | None =None,
oop=None,
nwd: list[str] | list[str, datetime.datetime, datetime.datetime] = [],
start_date: datetime.datetime | None = None,
end_date: datetime.datetime | None = None,
oop:list[(datetime.datetime, datetime.datetime, str)] | None = None,
not_available_to_work=None,
pref_not_to_work=None,
work_requests=None,
@@ -57,57 +58,106 @@ class Worker:
self.shift_target_number = defaultdict(int)
days_to_work = Rota.rota_days_length
#days_to_work = Rota.rota_days_length
# if no start date default to the start of the rota
if start_date is None:
self.start_date = Rota.start_date
else:
# ? test if start date is valid
self.start_date = start_date
if end_date is None:
self.end_date = None
self.end_date = Rota.rota_end_date
else:
self.end_date = datetime.datetime.strptime(end_date, "%d/%m/%y").date()
if isinstance(end_date, datetime.date):
self.end_date = end_date
else:
self.end_date = datetime.datetime.strptime(end_date, "%d/%m/%y").date()
if self.end_date > Rota.rota_end_date:
self.end_date = Rota.rota_end_date
else:
if self.end_date > Rota.start_date:
days_to_work = (self.end_date - Rota.start_date).days
# add unavalabilities
for weeks_days in Rota.weeks_days_product[days_to_work:]:
week, day = weeks_days
Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[
(self.id, week, day)
] = "END DATE"
#if self.start_date >= self.end_date:
# raise ValueError("End date must be after start date")
days_to_work = (self.end_date - self.start_date).days
for week, day in Rota.weeks_days_product:
date = Rota.week_day_date_map[(week, day)]
if date < self.start_date:
Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[
(self.id, week, day)
] = f"START DATE: {self.start_date}"
if date >= self.end_date:
Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[
(self.id, week, day)
] = f"END DATE: {self.end_date}"
#if self.start_date > Rota.start_date:
# # add unavalabilities
# for weeks_days in Rota.weeks_days_product[:days_to_work]:
# week, day = weeks_days
# Rota.unavailable_to_work.add((self.id, week, day))
# Rota.unavailable_to_work_reason[
# (self.id, week, day)
# ] = f"START DATE: {self.start_date}"
#if self.end_date < Rota.rota_end_date:
# # add unavalabilities
# for weeks_days in Rota.weeks_days_product[days_to_work:]:
# week, day = weeks_days
# Rota.unavailable_to_work.add((self.id, week, day))
# Rota.unavailable_to_work_reason[
# (self.id, week, day)
# ] = f"END DATE: {self.end_date}"
if oop is not None:
start_oop, end_oop = oop
start_oop_date = datetime.datetime.strptime(start_oop, "%d/%m/%y").date()
end_oop_date = datetime.datetime.strptime(end_oop, "%d/%m/%y").date()
# ignore oops if they finish before the rota start date
print(end_oop_date, Rota.start_date)
if end_oop_date > Rota.start_date:
if start_oop_date > Rota.rota_end_date:
pass
for start_oop, end_oop, oop_name in oop:
#start_oop, end_oop = oop
if isinstance(start_oop, datetime.date):
start_oop_date = start_oop
else:
if end_oop_date > Rota.rota_end_date:
end_oop_date = Rota.rota_end_date
start_oop_date = datetime.datetime.strptime(start_oop, "%d/%m/%y").date()
if start_oop_date < Rota.start_date:
start_oop_date = Rota.start_date
if isinstance(end_oop, datetime.date):
end_oop_date = end_oop
else:
end_oop_date = datetime.datetime.strptime(end_oop, "%d/%m/%y").date()
oop_length = (end_oop_date - start_oop_date).days
days_to_work = days_to_work - oop_length
if start_oop_date >= end_oop_date:
raise ValueError("End OOP date must be after start date")
days_until_oop = (start_oop_date - Rota.start_date).days
# ignore oops if they finish before the rota (or worker) start date
if end_oop_date > self.start_date:
if start_oop_date > self.end_date:
pass
else:
if end_oop_date > Rota.rota_end_date:
end_oop_date = Rota.rota_end_date
for weeks_days in Rota.weeks_days_product[
days_until_oop : days_until_oop + oop_length
]:
week, day = weeks_days
Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[
(self.id, week, day)
] = "OOP ()".format(oop)
if start_oop_date < Rota.start_date:
start_oop_date = Rota.start_date
oop_length = (end_oop_date - start_oop_date).days
days_to_work = days_to_work - oop_length
days_until_oop = (start_oop_date - Rota.start_date).days
for weeks_days in Rota.weeks_days_product[
days_until_oop : days_until_oop + oop_length
]:
week, day = weeks_days
Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[
(self.id, week, day)
] = f"OOP ({oop_name})".format(oop)
if pref_not_to_work is not None:
# loop throught dates converting to week / day combination
test/test_nwds.py
-2
View File
@@ -1,5 +1,3 @@
from copy import deepcopy
from black import main
import pytest
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days
test/test_workers.py
+285
View File
@@ -0,0 +1,285 @@
import pytest
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days
import datetime
from rota.workers import Worker
def setup_basic_rota():
weeks_to_rota = 10
start_date = datetime.date(2022, 3, 7)
return RotaBuilder(
start_date,
weeks_to_rota=weeks_to_rota,
)
class TestWorkers:
def test_no_workers(self):
"""Check error raised if attempt to solve with no workers"""
Rota = setup_basic_rota()
Rota.add_shifts(
SingleShift(
("group1",),
"d",
12.5,
days[:5],
balance_offset=10,
workers_required=1,
),
SingleShift(("group1",), "w", 12.5, days[5:], workers_required=1),
)
with pytest.raises(NoWorkers):
Rota.build_and_solve(solve=False)
def test_workers_duplicate_name(self):
Rota = setup_basic_rota()
workers = []
workers.append(Worker(Rota, "worker1", "group1", 1, fte=100))
workers.append(Worker(Rota, "worker1", "group1", 1, fte=100))
workers.append(Worker(Rota, "worker3", "group2", 1, fte=50))
workers.append(Worker(Rota, "worker4", "group2", 1, fte=50))
Rota.add_workers(workers)
Rota.add_shifts(
SingleShift(
("group1",),
"d",
12.5,
days[:5],
balance_offset=10,
workers_required=1,
),
)
with pytest.raises(ValueError):
Rota.build_and_solve(solve=False)
def test_worker_ftes(self):
weeks_to_rota = 10
start_date = datetime.date(2022, 3, 7)
Rota = RotaBuilder(
start_date,
weeks_to_rota=weeks_to_rota,
)
worker1_end_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 2)
worker2_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 2)
worker3_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 4)
worker3_end_date = start_date + datetime.timedelta(weeks=3 * weeks_to_rota / 4)
workers = []
workers.append(
Worker(Rota, "worker1", "group1", 1, fte=100, end_date=worker1_end_date)
)
workers.append(
Worker(Rota, "worker2", "group1", 1, fte=100, start_date=worker2_start_date)
)
workers.append(
Worker(
Rota,
"worker3",
"group2",
1,
fte=100,
start_date=worker3_start_date,
end_date=worker3_end_date,
)
)
workers.append(Worker(Rota, "worker4", "group2", 1, fte=50))
workers.append(Worker(Rota, "worker5", "group2", 1, fte=50, nwd=["Mon", "Tue"]))
workers.append(
Worker(
Rota,
"worker6",
"group3",
1,
fte=100,
start_date=start_date + datetime.timedelta(weeks=weeks_to_rota),
)
)
workers.append(
Worker(Rota, "worker7", "group3", 1, fte=50, end_date=start_date)
)
Rota.add_workers(workers)
Rota.add_shifts(
SingleShift(
("group1", "group2"),
"d",
12.5,
days[:5],
balance_offset=10,
workers_required=1,
),
)
Rota.build_and_solve()
# All workers should have an adjusted fte of 50 %
for worker in Rota.get_workers():
if worker.site == "group3":
assert worker.fte_adj == 0
assert Rota.get_worker_shift_list(worker).count("d") == 0
continue
assert worker.fte_adj == 50
if worker.fte == 100:
assert worker.proportion_rota_to_work == 0.5
assert worker.days_to_work == 35
assert Rota.get_worker_shift_list(worker).count("d") == 10
if worker.name == "worker1":
assert Rota.get_worker_shift_list_string(worker).endswith("-"*(Rota.rota_days_length//2))
if worker.name == "worker2":
assert Rota.get_worker_shift_list_string(worker).startswith("-"*(Rota.rota_days_length//2))
if worker.name == "worker3":
assert Rota.get_worker_shift_list_string(worker).startswith("-"*(Rota.rota_days_length//4))
assert Rota.get_worker_shift_list_string(worker).endswith("-"*(Rota.rota_days_length//4))
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
Rota.export_rota_to_html("worker_ftes")
def test_worker_oops(self):
weeks_to_rota = 10
start_date = datetime.date(2022, 3, 7)
Rota = RotaBuilder(
start_date,
weeks_to_rota=weeks_to_rota,
)
worker1_oop = [
(
start_date,
start_date + datetime.timedelta(weeks=weeks_to_rota / 2),
"First half OOP",
)
]
worker2_oop = [
(
start_date + datetime.timedelta(weeks=weeks_to_rota / 2),
Rota.rota_end_date,
"First half OOP",
)
]
worker3_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 4)
worker3_oop = [
(
start_date + datetime.timedelta(weeks=weeks_to_rota / 2),
start_date + datetime.timedelta(weeks=3 * weeks_to_rota / 4),
"Quarter oop",
)
]
worker4_oop = [
(
start_date,
start_date + datetime.timedelta(weeks=weeks_to_rota / 4),
"Quarter oop",
),
(
start_date + datetime.timedelta(weeks=weeks_to_rota / 2),
start_date
+ datetime.timedelta(weeks=weeks_to_rota / 2)
+ datetime.timedelta(weeks=weeks_to_rota / 4),
"Quarter oop 2",
),
]
workers = []
workers.append(Worker(Rota, "worker1", "group1", 1, fte=100, oop=worker1_oop))
workers.append(Worker(Rota, "worker2", "group1", 1, fte=100, oop=worker2_oop))
workers.append(
Worker(
Rota,
"worker3",
"group2",
1,
fte=100,
start_date=worker3_start_date,
oop=worker3_oop,
)
)
workers.append(Worker(Rota, "worker4", "group2", 1, oop=worker4_oop))
workers.append(Worker(Rota, "worker5", "group2", 1, fte=50, nwd=["Mon", "Tue"]))
worker6_oop = [
(
start_date,
start_date + datetime.timedelta(weeks=weeks_to_rota / 6),
"oop",
),
(
start_date + datetime.timedelta(weeks=weeks_to_rota / 2),
start_date + datetime.timedelta(weeks=weeks_to_rota / 2)
+ datetime.timedelta(weeks=weeks_to_rota / 3),
" oop 2",
),
]
worker7_oop = [
(
start_date-datetime.timedelta(weeks=1),
Rota.rota_end_date,
"oop",
),
]
workers.append(Worker(Rota, "worker6", "group2", 1, fte=100, oop=worker6_oop))
workers.append(Worker(Rota, "worker7", "group3", 1, fte=100, oop=worker7_oop))
Rota.add_workers(workers)
Rota.add_shifts(
SingleShift(
("group1", "group2"),
"d",
12.5,
days[:5],
balance_offset=10,
workers_required=1,
),
)
Rota.build_and_solve()
# All workers should have an adjusted fte of 50 %
for worker in Rota.get_workers():
if worker.site == "group3":
assert worker.fte_adj == 0
continue
assert worker.fte_adj == pytest.approx(50, 2)
if worker.fte == 100:
assert worker.proportion_rota_to_work == pytest.approx(0.5, 0.2)
assert worker.days_to_work == pytest.approx(35, 1)
assert Rota.get_worker_shift_list(worker).count("d") in (8, 9)
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
Rota.export_rota_to_html("worker_oops")