From f887257a55193b4260766ffe35f1226c32bbc36f Mon Sep 17 00:00:00 2001 From: Ross Date: Mon, 16 May 2022 19:35:38 +0100 Subject: [PATCH] refactor and add some more tests --- output/timetable.js | 1 + rota/run2.py | 8 -- rota/shifts.py | 53 +++++++- rota/workers.py | 134 +++++++++++++------- test/test_nwds.py | 2 - test/test_workers.py | 285 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 427 insertions(+), 56 deletions(-) create mode 100644 test/test_workers.py diff --git a/output/timetable.js b/output/timetable.js index ec74399..a3a1121 100644 --- a/output/timetable.js +++ b/output/timetable.js @@ -361,6 +361,7 @@ function viewWorker(worker) { dlg = $(`
Site: ${ds.site}
FTE: ${ds.fte} (${ds.fte_adj})
+ Start date: ${ds.start_date}
End date: ${ds.end_date}
Non working days: ${ds.nwds}
`) diff --git a/rota/run2.py b/rota/run2.py index f4559ce..644a7e6 100644 --- a/rota/run2.py +++ b/rota/run2.py @@ -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 diff --git a/rota/shifts.py b/rota/shifts.py index d291115..b2e2917 100644 --- a/rota/shifts.py +++ b/rota/shifts.py @@ -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 = """ 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 = { diff --git a/rota/workers.py b/rota/workers.py index e19770f..9110bac 100644 --- a/rota/workers.py +++ b/rota/workers.py @@ -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 diff --git a/test/test_nwds.py b/test/test_nwds.py index 35ac66e..f102cb5 100644 --- a/test/test_nwds.py +++ b/test/test_nwds.py @@ -1,5 +1,3 @@ -from copy import deepcopy -from black import main import pytest from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days diff --git a/test/test_workers.py b/test/test_workers.py new file mode 100644 index 0000000..0489f4a --- /dev/null +++ b/test/test_workers.py @@ -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")