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

update from last round

Browse Source
This commit is contained in:
Ross
2024-12-23 16:56:53 +00:00
parent 51b4e89601
commit 70dc279a02
13 changed files with 161 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rota/shifts.py
+76 -37
View File
@@ -1,6 +1,5 @@
from calendar import week from calendar import week
import datetime import datetime
from distutils.log import debug
import itertools import itertools
from typing import Dict, Iterable, List, Sequence, Tuple, Set, Any, Type, Literal from typing import Dict, Iterable, List, Sequence, Tuple, Set, Any, Type, Literal
import time import time
@@ -9,10 +8,12 @@ from pydantic import BaseModel, Extra, constr
from pathlib import Path from pathlib import Path
import uuid
import datetime import datetime
from pyomo.environ import * from pyomo.environ import *
from pyomo.opt import SolverFactory from pyomo.opt import SolverFactory, TerminationCondition, SolverStatus
import urllib.parse import urllib.parse
from rota.workers import Worker from rota.workers import Worker
@@ -211,7 +212,7 @@ class RotaBuilder(object):
# self.night_blocks = ["weekday", "weekend", "none"] # self.night_blocks = ["weekday", "weekend", "none"]
self.sites = None self.sites: set = set()
self.balance_offset_modifier = balance_offset_modifier self.balance_offset_modifier = balance_offset_modifier
self.ltft_balance_offset = ltft_balance_offset self.ltft_balance_offset = ltft_balance_offset
@@ -246,6 +247,12 @@ class RotaBuilder(object):
"avoid_shifts_by_worker_names": [], "avoid_shifts_by_worker_names": [],
} }
self.terminate_on_warning = [
"Worker/duplicate id",
"Worker/duplicate name",
"Worker/no valid shifts",
]
self.results = None self.results = None
self.warnings: list[tuple[str, str]] = [] self.warnings: list[tuple[str, str]] = []
@@ -294,13 +301,14 @@ class RotaBuilder(object):
self.unavailable_to_work_reason = {} self.unavailable_to_work_reason = {}
self.pref_not_to_work = {} self.pref_not_to_work = {}
self.pref_not_to_work_reason = {} self.pref_not_to_work_reason = {}
self.work_requests: set[Tuple[str, WeekInt, DayStr, ShiftName]] = set() self.work_requests: set[
Tuple[str | uuid.UUID | int, WeekInt, DayStr, ShiftName]
] = set()
self.work_requests_map = {} self.work_requests_map = {}
def solve_model( def solve_model(
self, self,
solver: str = "cbc", solver: str = "appsi_highs",
use_neos: bool = False,
options: dict = {}, options: dict = {},
debug_if_fail: bool = False, debug_if_fail: bool = False,
): ):
@@ -314,41 +322,46 @@ class RotaBuilder(object):
if "threads" in options: if "threads" in options:
options.pop("threads") options.pop("threads")
self.opt = SolverFactory(solver, executable="scip") self.opt = SolverFactory(solver, executable="scip")
elif solver == "appsi_highs":
self.opt = SolverFactory(solver)
if "seconds" in options:
options["time_limit"] = options.pop("seconds")
if "ratio" in options:
options["mip_rel_gap"] = options.pop("ratio")
else: else:
self.opt = SolverFactory(solver) self.opt = SolverFactory(solver)
try: try:
console.print("Solving") console.print("Solving")
console.print(f"Options: {options}")
log_file = f"logs/{self.name}_{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}.log" log_file = f"logs/{self.name}_{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}.log"
if use_neos: results = self.opt.solve(
solver_manager = SolverManagerFactory("neos") # Solve using neos server self.model,
# results = solver_manager.solve(Rota.model, opt=opt, logfile="test.log") tee=True,
results = solver_manager.solve( options=options,
self.model, # options={
keepfiles=True, # "threads": 10,
tee=True, # },
opt=self.opt, # logfile=log_file,
logfile=log_file, keepfiles=True,
) load_solutions=False,
else: )
results = self.opt.solve(
self.model,
tee=True,
options=options,
# options={
# "threads": 10,
# },
logfile=log_file,
keepfiles=True,
)
except KeyboardInterrupt: except KeyboardInterrupt:
return return
pass pass
if (
not results.solver.termination_condition == TerminationCondition.infeasible
#and not results.solver.status == SolverStatus.aborted
):
self.model.solutions.load_from(results)
self.results = results self.results = results
console.print(f"Complete - outcome: {results.solver.status}") console.print(f"Complete - outcome: {results.solver.status}")
console.print(f"Termination condition: {results.solver.termination_condition}")
if results.solver.status != "ok" and debug_if_fail: if results.solver.status != "ok" and debug_if_fail:
console.print(f"Attempting each shift individually") console.print(f"Attempting each shift individually")
@@ -417,7 +430,7 @@ class RotaBuilder(object):
solve=True, solve=True,
debug_if_fail: bool = False, debug_if_fail: bool = False,
export=False, export=False,
solver="cbc", solver="appsi_highs",
): ):
self.run_start_time = datetime.datetime.now() self.run_start_time = datetime.datetime.now()
@@ -1305,7 +1318,15 @@ class RotaBuilder(object):
if self.use_shift_balance_extra: if self.use_shift_balance_extra:
if shift.name in worker.shift_balance_extra: if shift.name in worker.shift_balance_extra:
extra = worker.shift_balance_extra[shift.name] extra = worker.shift_balance_extra[shift.name]
target_shifts = target_shifts + extra
# TODO look at how this affects allocation (how does it affect fte)
match extra:
case "double":
target_shifts = target_shifts * 2
case "half":
target_shifts = target_shifts / 2
case _:
target_shifts = target_shifts + extra
# print(worker.name, shift.name, target_shifts) # print(worker.name, shift.name, target_shifts)
worker.shift_target_number[shift.name] = target_shifts worker.shift_target_number[shift.name] = target_shifts
@@ -2461,6 +2482,15 @@ class RotaBuilder(object):
print(f"[bold red]WARNING:[/bold red] {warning_type} - {message}") print(f"[bold red]WARNING:[/bold red] {warning_type} - {message}")
self.warnings.append((warning_type, message)) self.warnings.append((warning_type, message))
if warning_type in self.terminate_on_warning:
raise WarningTermination(warning_type)
def get_warnings(self, warning_type: None | str = None):
if warning_type is None:
return self.warnings
else:
return [warning for warning in self.warnings if warning[0] == warning_type]
def add_worker(self, worker: Worker) -> None: def add_worker(self, worker: Worker) -> None:
"""Add a worker to the rota """Add a worker to the rota
@@ -2491,22 +2521,23 @@ class RotaBuilder(object):
self.workers_name_map: dict[str, Worker] = {} self.workers_name_map: dict[str, Worker] = {}
for worker in track(self.workers, description="Building workers"): for worker in track(self.workers, description="Building workers"):
print(worker.name)
worker.load_rota(self) worker.load_rota(self)
wid = worker.id wid = worker.id
if wid in self.workers_id_map: if wid in self.workers_id_map:
raise ValueError(f"Worker with id '{wid}' has been added twice") message = f"Worker with id '{wid}' has been added twice"
self.add_warning("Worker/duplicate id", message)
self.workers_id_map[wid] = worker self.workers_id_map[wid] = worker
if worker.name in self.workers_name_map: if worker.name in self.workers_name_map:
raise ValueError( message = f"Worker with name '{worker.name}' has been added twice"
f"Worker with name '{worker.name}' has been added twice" self.add_warning("Worker/duplicate name", message)
)
self.workers_name_map[worker.name] = worker self.workers_name_map[worker.name] = worker
if worker.site not in self.sites: if worker.site not in self.sites:
raise ValueError( message = f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})"
f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})" self.add_warning("Worker/no valid shifts", message)
)
self.workers = sorted(self.workers) self.workers = sorted(self.workers)
@@ -3128,7 +3159,7 @@ class RotaBuilder(object):
# Loop through all the days possible shifts and see # Loop through all the days possible shifts and see
# if the worker has been assigned # if the worker has been assigned
for shift in self.get_shift_names_by_week_day(week, day): for shift in self.get_shift_names_by_week_day(week, day):
if model.works[worker.id, week, day, shift].value > 0: if model.works[worker.id, week, day, shift].value > 0.5:
shifts.append(shift) shifts.append(shift)
a = shift[0] a = shift[0]
shift_name = shift shift_name = shift
@@ -3187,6 +3218,7 @@ class RotaBuilder(object):
shift_diff_dict[shift.name] = diff shift_diff_dict[shift.name] = diff
worker_td = """<td title='Site: {site}' class='worker {site}' worker_td = """<td title='Site: {site}' class='worker {site}'
data-worker-id='{worker_id}'
data-nwds='{nwds}' data-site='{site}' data-worker='{name}' data-nwds='{nwds}' data-site='{site}' data-worker='{name}'
data-fte='{fte}' data-fte_adj='{fte_adj}' data-fte='{fte}' data-fte_adj='{fte_adj}'
data-start_date='{start_date}' data-start_date='{start_date}'
@@ -3204,6 +3236,7 @@ class RotaBuilder(object):
<span class='name' title='{name}'>{name}</span> ({grade}) [{fte}]</td>""".format( <span class='name' title='{name}'>{name}</span> ({grade}) [{fte}]</td>""".format(
site=worker.site, site=worker.site,
nwds=nwds, nwds=nwds,
worker_id=worker.id,
name=worker.name, name=worker.name,
fte=worker.fte, fte=worker.fte,
fte_adj=worker.fte_adj, fte_adj=worker.fte_adj,
@@ -3465,3 +3498,9 @@ class InvalidShift(Exception):
"""Raised when there are no active sites""" """Raised when there are no active sites"""
pass pass
class WarningTermination(Exception):
"""Raised when a warning in the termination group is raised"""
pass
rota/workers.py
+21 -18
View File
@@ -68,7 +68,7 @@ class Worker(BaseModel):
# rules is easier. # rules is easier.
grade: int grade: int
# We can either have a user generated ID # We can either have a user generated ID
id: Optional[int| uuid.UUID] = None id: Optional[int| uuid.UUID| str] = None
fte: int = 100 fte: int = 100
nwds: list[NonWorkingDays] = [] nwds: list[NonWorkingDays] = []
start_date: datetime.date | None = None start_date: datetime.date | None = None
@@ -81,7 +81,7 @@ class Worker(BaseModel):
previous_shifts: dict = {} previous_shifts: dict = {}
shift_balance_extra: dict = {} shift_balance_extra: dict = {}
bank_holiday_extra: int = 0 bank_holiday_extra: int = 0
pair: str | None = None pair: int | str | None = None
class Config: class Config:
extra = Extra.allow extra = Extra.allow
@@ -161,6 +161,9 @@ class Worker(BaseModel):
(self.id, week, day) (self.id, week, day)
] = f"END DATE: {self.calculated_end_date}" ] = f"END DATE: {self.calculated_end_date}"
if not self.not_available_to_work:
Rota.add_warning("Worker/No unavailability", f"{self.name} [{self.id}] has no unavailabilities (leave)")
for item in self.oop: for item in self.oop:
start_oop = item.start_date start_oop = item.start_date
end_oop = item.end_date end_oop = item.end_date
@@ -210,8 +213,8 @@ class Worker(BaseModel):
days_to_end = (self.calculated_end_date - Rota.start_date).days days_to_end = (self.calculated_end_date - Rota.start_date).days
# loop throught dates converting to week / day combination # loop throught dates converting to week / day combination
for item in self.pref_not_to_work: for preference in self.pref_not_to_work:
days_from_start = (item.date - Rota.start_date).days days_from_start = (preference.date - Rota.start_date).days
# Ignore dates past the end of the rota (or end date) # Ignore dates past the end of the rota (or end date)
if days_from_start < days_to_end: if days_from_start < days_to_end:
week = days_from_start // 7 + 1 week = days_from_start // 7 + 1
@@ -222,32 +225,32 @@ class Worker(BaseModel):
Rota.pref_not_to_work[(self.id, week, day)] = 1 / ( Rota.pref_not_to_work[(self.id, week, day)] = 1 / (
len(self.pref_not_to_work) + 1 len(self.pref_not_to_work) + 1
) )
Rota.pref_not_to_work_reason[(self.id, week, day)] = item.reason Rota.pref_not_to_work_reason[(self.id, week, day)] = preference.reason
# print(not_available_to_work) # print(not_available_to_work)
# loop throught dates converting to week / day combination # loop throught dates converting to week / day combination
unavailable_set = set() unavailable_set = set()
for item in self.not_available_to_work: for unavalability in self.not_available_to_work:
days_from_start = (item.date - Rota.start_date).days days_from_start = (unavalability.date - Rota.start_date).days
# Ignore dates past the end of the rota (or end date) # Ignore dates past the end of the rota (or end date)
if days_from_start < days_to_end: if days_from_start < days_to_end:
week = days_from_start // 7 + 1 week = days_from_start // 7 + 1
day = Rota.days[(days_from_start % 7)] day = Rota.days[(days_from_start % 7)]
Rota.unavailable_to_work.add((self.id, week, day)) Rota.unavailable_to_work.add((self.id, week, day))
Rota.unavailable_to_work_reason[(self.id, week, day)] = item.reason Rota.unavailable_to_work_reason[(self.id, week, day)] = unavalability.reason
unavailable_set.add((week, day)) unavailable_set.add((week, day))
for item in self.work_requests: for request in self.work_requests:
days_from_start = (item.date - Rota.start_date).days days_from_start = (request.date - Rota.start_date).days
week = days_from_start // 7 + 1 week = days_from_start // 7 + 1
day = Rota.days[(days_from_start % 7)] day = Rota.days[(days_from_start % 7)]
if item.shift == "*": if request.shift == "*":
for shift in Rota.get_shifts_for_worker_site(self.site): for shift in Rota.get_shifts_for_worker_site(self.site):
Rota.work_requests.add((self.id, week, day, shift.name)) Rota.work_requests.add((self.id, week, day, shift.name))
else: else:
Rota.work_requests.add((self.id, week, day, item.shift)) Rota.work_requests.add((self.id, week, day, request.shift))
# Calculate the proportion of the rota that is being worked # Calculate the proportion of the rota that is being worked
self.proportion_rota_to_work = days_to_work / Rota.rota_days_length self.proportion_rota_to_work = days_to_work / Rota.rota_days_length
@@ -263,15 +266,15 @@ class Worker(BaseModel):
# TODO: this has already been validated, consider moving # TODO: this has already been validated, consider moving
self.non_working_day_list = [] self.non_working_day_list = []
for item in self.nwds: for non_working_day in self.nwds:
start_date = Rota.start_date start_date = Rota.start_date
end_date = Rota.rota_end_date end_date = Rota.rota_end_date
if item.start_date is not None: if non_working_day.start_date is not None:
start_date = item.start_date start_date = non_working_day.start_date
if item.end_date is not None: if non_working_day.end_date is not None:
end_date = item.end_date end_date = non_working_day.end_date
self.non_working_day_list.append((item.day, start_date, end_date)) self.non_working_day_list.append((non_working_day.day, start_date, end_date))
if days_to_work < 1: if days_to_work < 1:
self.fte_adj = 0 self.fte_adj = 0
test/rota_constraints/test_avoid_grades_by_week.py
+2 -2
View File
@@ -145,7 +145,7 @@ class TestWorkerRequests:
Rota.add_grade_constraint_by_week([3], [1,2,3], ["a"]) Rota.add_grade_constraint_by_week([3], [1,2,3], ["a"])
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_avoid_grade_constraint_multiple_shifts(self): def test_avoid_grade_constraint_multiple_shifts(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -192,4 +192,4 @@ class TestWorkerRequests:
Rota.add_grade_constraint_by_week([1], [7,8,9], ["a"]) Rota.add_grade_constraint_by_week([1], [7,8,9], ["a"])
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
test/rota_constraints/test_avoid_workers_by_week.py
+2 -2
View File
@@ -146,7 +146,7 @@ class TestWorkerRequests:
Rota.add_worker_name_constraint_by_week(["worker3"], [1,2,3], ["a"]) Rota.add_worker_name_constraint_by_week(["worker3"], [1,2,3], ["a"])
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_avoid_worker_constraint_multiple_shifts(self): def test_avoid_worker_constraint_multiple_shifts(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -193,4 +193,4 @@ class TestWorkerRequests:
Rota.add_worker_name_constraint_by_week(["worker1"], [7,8,9], ["a"]) Rota.add_worker_name_constraint_by_week(["worker1"], [7,8,9], ["a"])
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
test/test_general_balancing.py
+3 -3
View File
@@ -75,7 +75,7 @@ class TestBalancing:
name="a", name="a",
length=12.5, length=12.5,
days=days, days=days,
balance_weighting=5, balance_weighting=10,
workers_required=1, workers_required=1,
force_as_block=False, force_as_block=False,
constraint=[{"name": "pre","options": "2"}, {"name": "post","options": "2"}], constraint=[{"name": "pre","options": "2"}, {"name": "post","options": "2"}],
@@ -98,7 +98,7 @@ class TestBalancing:
), ),
) )
Rota.build_and_solve(options={"ratio": 0.001}) Rota.build_and_solve(options={"ratio": 0.0001})
Rota.export_rota_to_html("basic_balancing_weighted_shifts") Rota.export_rota_to_html("basic_balancing_weighted_shifts")
shift_summary = Rota.get_shift_summary_dict() shift_summary = Rota.get_shift_summary_dict()
@@ -198,7 +198,7 @@ class TestBalancing:
worker = Rota.get_worker_by_name(worker_name) worker = Rota.get_worker_by_name(worker_name)
# assert worker_shifts["a"] in (46,47,48) # assert worker_shifts["a"] in (46,47,48)
assert worker_shifts["c"] in (7, 8) assert worker_shifts["c"] in (approx(7), approx(8))
#assert worker_shifts["b"] in (53, 54) #assert worker_shifts["b"] in (53, 54)
def test_weighted_shift_balancing4(self): def test_weighted_shift_balancing4(self):
test/test_nights2.py
+3 -3
View File
@@ -146,7 +146,7 @@ class TestNightShifts:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_nights2(self): def test_nights2(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -315,7 +315,7 @@ class TestNightShifts:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("night") Rota.export_rota_to_html("night")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_nights_max_frequency3(self): def test_nights_max_frequency3(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -352,7 +352,7 @@ class TestNightShifts:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("night") Rota.export_rota_to_html("night")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_nights_max_frequency4(self): def test_nights_max_frequency4(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
test/test_nwds.py
+5 -5
View File
@@ -90,7 +90,7 @@ class TestDemoRota:
self.Rota.export_rota_to_html("nwd") self.Rota.export_rota_to_html("nwd")
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
def test_nwd_partial_rota(self): def test_nwd_partial_rota(self):
# Set up rota # Set up rota
@@ -188,7 +188,7 @@ class TestDemoRota:
self.Rota.build_and_solve(options={"ratio": 0.00}) self.Rota.build_and_solve(options={"ratio": 0.00})
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
def test_nwd_force_as_block(self): def test_nwd_force_as_block(self):
# Set up rota # Set up rota
@@ -273,21 +273,21 @@ class TestDemoRota:
self.Rota.add_shifts( self.Rota.add_shifts(
SingleShift( SingleShift(
sites=("group1",), name="d", length= 12.5, days=days[:5], sites=("group1",), name="d", length= 12.5, days=days[:5],
balance_offset=10, #balance_offset=50,
workers_required=2, workers_required=2,
force_as_block_unless_nwd=True, force_as_block_unless_nwd=True,
assign_as_block=True, assign_as_block=True,
), ),
SingleShift( SingleShift(
sites=("group1",), name="w", length= 12.5, days=days[5:], sites=("group1",), name="w", length= 12.5, days=days[5:],
balance_offset=10, #balance_offset=50,
workers_required=4, workers_required=4,
force_as_block_unless_nwd=True, force_as_block_unless_nwd=True,
assign_as_block=True, assign_as_block=True,
), ),
) )
self.Rota.build_and_solve(options={"ratio": 0.00}) self.Rota.build_and_solve()
self.Rota.export_rota_to_html("nwd_block_force_split") self.Rota.export_rota_to_html("nwd_block_force_split")
assert self.Rota.results.solver.status == "ok" assert self.Rota.results.solver.status == "ok"
test/test_remote.py
+1 -1
View File
@@ -40,7 +40,7 @@ class TestRemoteRotas:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_remote_pass_week(self): def test_remote_pass_week(self):
Rota = setup_basic_rota() Rota = setup_basic_rota()
test/test_rota.py
+27 -14
View File
@@ -2,7 +2,8 @@ from copy import deepcopy
from re import S from re import S
from black import main from black import main
import pytest import pytest
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days from pytest import approx
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WarningTermination, days
import datetime import datetime
@@ -285,14 +286,14 @@ class TestDemoRotaNights:
worker = self.Rota.get_worker_by_name(worker_name) worker = self.Rota.get_worker_by_name(worker_name)
if worker.fte == 40: if worker.fte == 40:
assert worker_shifts["night_weekday"] in (0, 4) assert worker_shifts["night_weekday"] in (approx(0), approx(4))
assert worker_shifts["night_weekend"] in (0, 3) assert worker_shifts["night_weekend"] in (approx(0), approx(3))
elif worker.fte == 60: elif worker.fte == 60:
assert worker_shifts["night_weekday"] in (0, 4, 8) assert worker_shifts["night_weekday"] in (approx(0), approx(4), approx(8))
assert worker_shifts["night_weekend"] in (0, 3, 6) assert worker_shifts["night_weekend"] in (approx(0), approx(3), approx(6))
else: else:
assert worker_shifts["night_weekday"] in (4, 8) assert worker_shifts["night_weekday"] in (approx(4), approx(8))
assert worker_shifts["night_weekend"] in (3, 6) assert worker_shifts["night_weekend"] in (approx(3), approx(6))
class TestDemoRotaClear: class TestDemoRotaClear:
@@ -326,7 +327,7 @@ class TestDemoRotaClear:
name="a", name="a",
length=12.5, length=12.5,
days=days[:5], days=days[:5],
balance_offset=40, #balance_offset=40,
workers_required=2, workers_required=2,
constraint=[{"name": "pre", "options": 1}], constraint=[{"name": "pre", "options": 1}],
), ),
@@ -335,7 +336,7 @@ class TestDemoRotaClear:
name="b", name="b",
length=12.5, length=12.5,
days=days[:5], days=days[:5],
balance_offset=40, #balance_offset=40,
workers_required=2, workers_required=2,
), ),
SingleShift( SingleShift(
@@ -343,7 +344,7 @@ class TestDemoRotaClear:
name="c", name="c",
length=12.5, length=12.5,
days=days[3], days=days[3],
balance_offset=40, #balance_offset=40,
workers_required=1, workers_required=1,
constraint=[{"name": "pre", "options": 1}, {"name": "post", "options": 1}], constraint=[{"name": "pre", "options": 1}, {"name": "post", "options": 1}],
), ),
@@ -481,7 +482,7 @@ class TestDemoRotaShiftConstraints:
def test_max_shifts_fail(self): def test_max_shifts_fail(self):
self.Rota.constraint_options["max_shifts_per_month"] = 13 self.Rota.constraint_options["max_shifts_per_month"] = 13
self.Rota.build_and_solve() self.Rota.build_and_solve()
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
assert self.Rota.results.solver.termination_condition == "infeasible" assert self.Rota.results.solver.termination_condition == "infeasible"
def test_max_shifts_extra_worker(self): def test_max_shifts_extra_worker(self):
@@ -495,7 +496,7 @@ class TestDemoRotaShiftConstraints:
self.Rota.constraint_options["max_shifts_per_week"] = 3 self.Rota.constraint_options["max_shifts_per_week"] = 3
self.Rota.build_and_solve() self.Rota.build_and_solve()
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
assert self.Rota.results.solver.termination_condition == "infeasible" assert self.Rota.results.solver.termination_condition == "infeasible"
def test_max_shifts_per_week_pass(self): def test_max_shifts_per_week_pass(self):
@@ -550,7 +551,17 @@ class TestDemoRotaBalanceShiftSites:
self.Rota.constraint_options["balance_nights_across_sites"] = False self.Rota.constraint_options["balance_nights_across_sites"] = False
self.Rota.constraint_options["balance_nights"] = False self.Rota.constraint_options["balance_nights"] = False
self.Rota.constraint_options["constrain_time_off_after_nights"] = False self.Rota.constraint_options["constrain_time_off_after_nights"] = False
# Worker 5 and 6 don't have any valid shifts so this should fail
#with pytest.raises(WarningTermination):
# self.Rota.build_and_solve(options={"ratio": 0.0})
# Unless we remove the warning
self.Rota.terminate_on_warning.remove("Worker/no valid shifts")
self.Rota.build_and_solve(options={"ratio": 0.0}) self.Rota.build_and_solve(options={"ratio": 0.0})
assert len(self.Rota.get_warnings("Worker/no valid shifts")) == 2
self.Rota.export_rota_to_html("test5") self.Rota.export_rota_to_html("test5")
print(self.Rota.get_workers_by_group()) print(self.Rota.get_workers_by_group())
@@ -586,6 +597,8 @@ class TestDemoRotaBalanceShiftSites:
self.Rota.constraint_options["balance_nights_across_sites"] = False self.Rota.constraint_options["balance_nights_across_sites"] = False
self.Rota.constraint_options["balance_nights"] = False self.Rota.constraint_options["balance_nights"] = False
self.Rota.constraint_options["constrain_time_off_after_nights"] = False self.Rota.constraint_options["constrain_time_off_after_nights"] = False
self.Rota.terminate_on_warning.remove("Worker/no valid shifts")
self.Rota.build_and_solve(options={"ratio": 0.1}) self.Rota.build_and_solve(options={"ratio": 0.1})
group_workers = self.Rota.get_workers_by_group() group_workers = self.Rota.get_workers_by_group()
@@ -1151,7 +1164,7 @@ class TestNightUnavailable:
self.Rota.build_and_solve(options={"ratio": 0.00}) self.Rota.build_and_solve(options={"ratio": 0.00})
self.Rota.export_rota_to_html("testnight_unavail") self.Rota.export_rota_to_html("testnight_unavail")
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
assert self.Rota.results.solver.termination_condition == "infeasible" assert self.Rota.results.solver.termination_condition == "infeasible"
def test_assign_non_night_prior_to_unavailablity(self): def test_assign_non_night_prior_to_unavailablity(self):
@@ -1252,7 +1265,7 @@ class TestNightUnavailable:
# #
# self.Rota.build_and_solve(options={"ratio": 0.00}) # self.Rota.build_and_solve(options={"ratio": 0.00})
# #
# assert self.Rota.results.solver.status == "warning" # assert self.Rota.results.solver.status in ("warning", "error")
if __name__ == "__main__": if __name__ == "__main__":
t = TestLimitConstraints() t = TestLimitConstraints()
test/test_shifts.py
+5 -5
View File
@@ -142,7 +142,7 @@ class TestShiftConstraints:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("max_shifts_per_week_block") Rota.export_rota_to_html("max_shifts_per_week_block")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_max_shifts_per_week_block_shift_number(self): def test_max_shifts_per_week_block_shift_number(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -172,7 +172,7 @@ class TestShiftConstraints:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("max_shifts_per_week_block") Rota.export_rota_to_html("max_shifts_per_week_block")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_pre_shift_constraint(self): def test_pre_shift_constraint(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -226,7 +226,7 @@ class TestShiftConstraints:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_pre_shift_constraint2(self): def test_pre_shift_constraint2(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -269,7 +269,7 @@ class TestShiftConstraints:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("pre") Rota.export_rota_to_html("pre")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
def test_post_shift_constraint(self): def test_post_shift_constraint(self):
Rota = generate_basic_rota() Rota = generate_basic_rota()
@@ -311,4 +311,4 @@ class TestShiftConstraints:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("post") Rota.export_rota_to_html("post")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
test/test_weekend_padding.py
+2 -2
View File
@@ -210,7 +210,7 @@ class TestDemoRota:
self.Rota.build_and_solve(options={"ratio": 0.00}) self.Rota.build_and_solve(options={"ratio": 0.00})
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
def test_post_mon(self): def test_post_mon(self):
# Set up rota # Set up rota
@@ -343,4 +343,4 @@ class TestDemoRota:
self.Rota.export_rota_to_html("weekend_padding") self.Rota.export_rota_to_html("weekend_padding")
assert self.Rota.results.solver.status == "warning" assert self.Rota.results.solver.status in ("warning", "error")
test/test_worker_requests.py
+5 -4
View File
@@ -1,5 +1,6 @@
import datetime import datetime
import pytest import pytest
from pytest import approx
from rota.shifts import InvalidShift, NoWorkers, RotaBuilder, SingleShift, days from rota.shifts import InvalidShift, NoWorkers, RotaBuilder, SingleShift, days
from rota.workers import Worker from rota.workers import Worker
@@ -179,7 +180,7 @@ class TestWorkerRequests:
worker = Rota.get_worker_by_name(worker_name) worker = Rota.get_worker_by_name(worker_name)
assert worker_shifts["a"] in (46, 47) assert worker_shifts["a"] in (approx(46), approx(47))
# shift_string = Rota.get_worker_shift_list_string(worker) # shift_string = Rota.get_worker_shift_list_string(worker)
@@ -216,7 +217,7 @@ class TestWorkerRequests:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("unavailable_to_work2") Rota.export_rota_to_html("unavailable_to_work2")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
assert Rota.results.solver.termination_condition == "infeasible" assert Rota.results.solver.termination_condition == "infeasible"
def test_unavailable_to_work3(self): def test_unavailable_to_work3(self):
@@ -304,7 +305,7 @@ class TestWorkerRequests:
Rota.build_and_solve(options={"ratio": 0.000}) Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("unavailable_to_work4") Rota.export_rota_to_html("unavailable_to_work4")
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
assert Rota.results.solver.termination_condition == "infeasible" assert Rota.results.solver.termination_condition == "infeasible"
def test_work_requests(self): def test_work_requests(self):
@@ -523,7 +524,7 @@ class TestWorkerRequests:
worker = Rota.get_worker_by_name(worker_name) worker = Rota.get_worker_by_name(worker_name)
assert worker_shifts["a"] in (26, 27) assert worker_shifts["a"] in (26, 27)
assert worker_shifts["b"] == 10 assert worker_shifts["b"] == approx(10)
shift_string = Rota.get_worker_shift_list_string(worker) shift_string = Rota.get_worker_shift_list_string(worker)
test/test_workers.py
+9 -4
View File
@@ -1,5 +1,5 @@
import pytest import pytest
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WarningTermination, days
import datetime import datetime
from rota.workers import Worker from rota.workers import Worker
@@ -64,7 +64,7 @@ class TestWorkers:
), ),
) )
with pytest.raises(ValueError): with pytest.raises(WarningTermination):
Rota.build_and_solve(solve=False) Rota.build_and_solve(solve=False)
def test_worker_ftes(self): def test_worker_ftes(self):
@@ -143,6 +143,10 @@ class TestWorkers:
), ),
) )
with pytest.raises(WarningTermination):
Rota.build_and_solve()
Rota.terminate_on_warning.remove("Worker/no valid shifts")
Rota.build_and_solve() Rota.build_and_solve()
# All workers should have an adjusted fte of 50 % # All workers should have an adjusted fte of 50 %
@@ -298,6 +302,7 @@ class TestWorkers:
), ),
) )
Rota.terminate_on_warning.remove("Worker/no valid shifts")
Rota.build_and_solve() Rota.build_and_solve()
# All workers should have an adjusted fte of 50 % # All workers should have an adjusted fte of 50 %
@@ -353,7 +358,7 @@ class TestWorkers:
Rota.build_and_solve() Rota.build_and_solve()
# With two paired workers we shouldn't be able to solve # With two paired workers we shouldn't be able to solve
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
worker3 = Worker(name="worker3", site="group1", grade=1, fte=100) worker3 = Worker(name="worker3", site="group1", grade=1, fte=100)
Rota.add_worker(worker3) Rota.add_worker(worker3)
@@ -374,7 +379,7 @@ class TestWorkers:
Rota.build_and_solve() Rota.build_and_solve()
assert Rota.results.solver.status == "warning" assert Rota.results.solver.status in ("warning", "error")
worker3.pair = 0 worker3.pair = 0