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")
|