numberous fixes

This commit is contained in:
Ross
2024-01-07 17:52:36 +00:00
parent 5d55ed77a8
commit a7e1fa7171
4 changed files with 115 additions and 83 deletions
+24
View File
@@ -547,3 +547,27 @@ function hsv2rgb(h, s, v) {
return ("0" + Math.round(x * 255).toString(16)).slice(-2);
}).join('');
};
//$("#workers-json-target").append(syntaxHighlight(JSON.stringify(JSON.parse(($("#worker_details").data("workers"))))))
function syntaxHighlight(json) {
if (typeof json != 'string') {
json = JSON.stringify(json, undefined, 2);
}
json = json.replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;');
return json.replace(/("(\\u[a-zA-Z0-9]{4}|\\[^u]|[^\\"])*"(\s*:)?|\b(true|false|null)\b|-?\d+(?:\.\d*)?(?:[eE][+\-]?\d+)?)/g, function (match) {
var cls = 'number';
if (/^"/.test(match)) {
if (/:$/.test(match)) {
cls = 'key';
} else {
cls = 'string';
}
} else if (/true|false/.test(match)) {
cls = 'boolean';
} else if (/null/.test(match)) {
cls = 'null';
}
return '<span class="' + cls + '">' + match + '</span>';
});
}
+8
View File
@@ -0,0 +1,8 @@
# Load uk based bank holidays
from govuk_bank_holidays.bank_holidays import BankHolidays
bank_holidays = BankHolidays()
bank_holiday_map = {}
for bank_holiday in bank_holidays.get_holidays(division="england-and-wales"):
bank_holiday_map[bank_holiday["date"]] = bank_holiday["title"]
+68 -71
View File
@@ -13,9 +13,11 @@ import datetime
from pyomo.environ import *
from pyomo.opt import SolverFactory
import urllib.parse
from rota.workers import Worker
from rota.console import console
from rota.bank_holidays import bank_holiday_map
import json
@@ -29,6 +31,7 @@ import sys
from rich.pretty import pprint
from rich.progress import track
from rich.panel import Panel
from rich import print
import logging
@@ -47,16 +50,6 @@ days = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
full_weekend_count = 2
# Load uk based bank holidays
from govuk_bank_holidays.bank_holidays import BankHolidays
bank_holidays = BankHolidays()
bank_holiday_map = {}
logging.debug("Creating bank holiday map")
for bank_holiday in bank_holidays.get_holidays(division="england-and-wales"):
bank_holiday_map[bank_holiday["date"]] = bank_holiday["title"]
logging.debug(bank_holiday_map)
SHIFT_BOUNDS = {
"bank_holiday": (0, 9),
@@ -192,8 +185,8 @@ class RotaBuilder(object):
use_bank_holiday_extra: bool = False,
SHIFT_BOUNDS=SHIFT_BOUNDS,
name: str = "",
bank_holidays: dict[datetime.date, str] = bank_holiday_map,
):
self.name = name
console.print(
@@ -207,7 +200,6 @@ class RotaBuilder(object):
justify="left",
)
self.SHIFT_BOUNDS = SHIFT_BOUNDS
self.shifts: List[SingleShift] = [] # type List[SingleShift]
@@ -249,7 +241,6 @@ class RotaBuilder(object):
"prevent_monday_and_tuesday_after_full_weekends": [],
"prevent_fridays_before_full_weekends": [],
"prevent_thursdays_before_full_weekends": [],
"avoid_st2_first_month": False,
"hard_constrain_pair_separation": False,
"avoid_shifts_by_grades": [],
"avoid_shifts_by_worker_names": [],
@@ -257,6 +248,8 @@ class RotaBuilder(object):
self.results = None
self.warnings: list[tuple[str, str]] = []
self.ignore_valid_shifts = False
self.set_rota_dates(start_date, weeks_to_rota)
@@ -264,8 +257,9 @@ class RotaBuilder(object):
self.run_start_time: datetime.datetime | None = None
self.run_end_time: datetime.datetime | None = None
def set_rota_dates(self, start_date: datetime.date, weeks_to_rota: int):
self.bank_holidays = bank_holidays
def set_rota_dates(self, start_date: datetime.date, weeks_to_rota: int):
self.weeks_to_rota = weeks_to_rota
self.days = days
@@ -290,7 +284,7 @@ class RotaBuilder(object):
# Generate a map for week, day combinations to a given date
self.week_day_date_map = {}
n = 0
n: int = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
self.week_day_date_map[(week, day)] = d
@@ -423,9 +417,8 @@ class RotaBuilder(object):
solve=True,
debug_if_fail: bool = False,
export=False,
solver="cbc"
solver="cbc",
):
self.run_start_time = datetime.datetime.now()
self.build_shifts()
@@ -433,7 +426,9 @@ class RotaBuilder(object):
self.build_model()
if solve:
self.solve_model(options=options, debug_if_fail=debug_if_fail, solver=solver)
self.solve_model(
options=options, debug_if_fail=debug_if_fail, solver=solver
)
self.run_end_time = datetime.datetime.now()
@@ -1206,21 +1201,24 @@ class RotaBuilder(object):
except ValueError:
# Occurs if there are no shifts within a defined block
# TODO: test if this breaks (and we should check rathar than except)
logging.debug(
self.add_warning(
"max shifts per month constraint"
f"Failed to constrain max_shifts_per_month for worker: {worker.name}"
)
pass
for constraint_dict in self.constraint_options["avoid_shifts_by_worker_names"]:
for constraint_dict in self.constraint_options[
"avoid_shifts_by_worker_names"
]:
if invalid_shifts := set(constraint_dict["shifts"]).difference(
set(self.get_shift_names())
):
message = f"avoid shifts by worker constraint contains a non existent shift: {invalid_shifts}"
if self.ignore_valid_shifts:
self.add_warning("Non existent shift", message)
continue # Skip if we don't want to raise an error
else:
raise InvalidShift(
f"avoid shifts by grade constraint contains a non existent shift: {invalid_shifts}"
)
raise InvalidShift(message)
if worker.name in constraint_dict["names"]:
self.model.constraints.add(
@@ -1237,18 +1235,18 @@ class RotaBuilder(object):
if invalid_shifts := set(constraint_dict["shifts"]).difference(
set(self.get_shift_names())
):
message = f"avoid shifts by grade constraint contains a non existent shift: {invalid_shifts}"
if self.ignore_valid_shifts:
self.add_warning("Non existent shift", message)
continue # Skip if we don't want to raise an error
else:
raise InvalidShift(
f"avoid shifts by grade constraint contains a non existent shift: {invalid_shifts}"
)
raise InvalidShift(message)
if invalid_grades := set(constraint_dict["grades"]).difference(
set(self.get_worker_grades())
):
raise ValueError(
f"avoid shifts by grade constraint contains a non existent worker grade: {invalid_grades}"
)
message = f"avoid shifts by grade constraint contains a non existent worker grade: {invalid_grades}"
self.add_warning("Non existent shift", message)
raise ValueError(message)
if worker.grade in constraint_dict["grades"]:
self.model.constraints.add(
@@ -1261,24 +1259,6 @@ class RotaBuilder(object):
)
)
# sys.exit(0)
if self.constraint_options["avoid_st2_first_month"] and worker.grade == 2:
# Avoid ST1s on the first month
try:
self.model.constraints.add(
0
== sum(
self.model.works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
if week in (1, 2, 3, 4)
and shift in ("night_weekday", "night_weekend")
)
)
except ValueError as e:
print("Failure setting constraint", "avoid_st2_first_month")
print(e)
# Count number of weekends an worker works
# if self.constraint_options["balance_weekends"]:
self.model.constraints.add(
@@ -1301,7 +1281,6 @@ class RotaBuilder(object):
if (
worker.site in shift.sites
): # Each site specfies which sites self.workers can fullfill it
# calaculate the total number of shifts that need assigning over the rota
# total_shifts = (len(self.weeks) * len(shift.days) *
total_shifts = (
@@ -1458,7 +1437,6 @@ class RotaBuilder(object):
)
if self.constraint_options["balance_bank_holidays"]:
extra_bank_holiday = 0
if self.use_bank_holiday_extra:
extra_bank_holiday = worker.bank_holiday_extra
@@ -1468,7 +1446,7 @@ class RotaBuilder(object):
== sum(
self.model.works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
if self.week_day_date_map[(week, day)] in bank_holiday_map
if self.week_day_date_map[(week, day)] in self.bank_holidays
)
+ extra_bank_holiday
)
@@ -1569,7 +1547,6 @@ class RotaBuilder(object):
worker.weekend_shift_target_number = weekend_shift_target_number
if self.constraint_options["balance_weekends"]:
if weekend_shift_target_number > 0:
self.model.constraints.add(
self.model.weekend_shift_count_t1[worker.id]
@@ -1929,7 +1906,6 @@ class RotaBuilder(object):
for n in track(
range(len(weeks_days)), description="Generate week/day constraints"
):
week, day = weeks_days[n]
pre_map = []
@@ -2100,7 +2076,6 @@ class RotaBuilder(object):
workers = worker_pairs
if self.get_shift_names_by_week_day(week, day):
# Unable to work (hard constraint not preference)
self.model.constraints.add(
self.model.available[worker.id, week, day]
@@ -2309,7 +2284,6 @@ class RotaBuilder(object):
def define_objectives(self):
# Define an objective function with model as input, to pass later
def obj_rule(m):
# c = len(workers)
balance_modifier_constant = 1
@@ -2483,6 +2457,10 @@ class RotaBuilder(object):
# add objective function to the model. rule (pass function) or expr (pass expression directly)
self.model.obj = Objective(rule=obj_rule, sense=minimize)
def add_warning(self, warning_type: str, message: str):
print(f"[bold red]WARNING:[/bold red] {warning_type} - {message}")
self.warnings.append((warning_type, message))
def add_worker(self, worker: Worker) -> None:
"""Add a worker to the rota
@@ -2509,8 +2487,8 @@ 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 = {}
self.workers_id_map: dict[str, Worker] = {}
self.workers_name_map: dict[str, Worker] = {}
for worker in track(self.workers, description="Building workers"):
worker.load_rota(self)
@@ -2525,6 +2503,11 @@ class RotaBuilder(object):
)
self.workers_name_map[worker.name] = worker
if worker.site not in self.sites:
raise ValueError(
f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})"
)
self.workers = sorted(self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers)
@@ -2655,7 +2638,7 @@ class RotaBuilder(object):
self.week_day_shifts_dict[(week, day)] = set()
for s in self.shifts:
if s.bank_holidays_only:
if self.week_day_date_map[(week, day)] in bank_holiday_map:
if self.week_day_date_map[(week, day)] in self.bank_holidays:
print(s.name, self.week_day_date_map[(week, day)])
self.week_day_shift_product.append((week, day, s.name))
self.week_day_shiftclass_product.append((week, day, s))
@@ -2870,28 +2853,31 @@ class RotaBuilder(object):
s.extend(self.shifts_to_force_as_blocks())
return s
def get_all_workers(self) -> List[Worker]:
return self.workers
def get_worker_by_id(self, worker_id: str) -> Worker:
return self.workers_id_map[worker_id]
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(list)
def get_workers_by_group(self) -> dict[str, list[Worker]]:
group_workers: dict[str, list[Worker]] = defaultdict(list)
for worker in self.workers:
group_workers[worker.site].append(worker)
return group_workers
def get_workers_by_remote_group(self) -> Dict[str, Worker]:
group_workers = defaultdict(list)
def get_workers_by_remote_group(self) -> dict[str, list[Worker]]:
group_workers: dict[str, list[Worker]] = defaultdict(list)
for worker in self.workers:
group_workers[worker.remote_site].append(worker)
return group_workers
def get_workers_by_grade(self) -> Dict[str, Worker]:
group_workers = defaultdict(list)
def get_workers_by_grade(self) -> Dict[int, list[Worker]]:
group_workers: dict[int, list[Worker]] = defaultdict(list)
for worker in self.workers:
group_workers[worker.grade].append(worker)
@@ -2904,7 +2890,6 @@ class RotaBuilder(object):
return self.full_time_equivalent
def get_worker_details(self):
w = defaultdict(list)
for worker in self.workers:
# print(worker)
@@ -2926,18 +2911,19 @@ class RotaBuilder(object):
return [
(week, day)
for week, day in self.get_week_day_combinations()
if self.week_day_date_map[(week, day)] in bank_holiday_map
if self.week_day_date_map[(week, day)] in self.bank_holidays
]
def get_week_start_date(self, week: int):
return self.start_date + datetime.timedelta(weeks=week)
# RESULTS
def export_rota_to_html(self, filename: str = "rota", folder=None, timestamp_filename=True):
def export_rota_to_html(
self, filename: str = "rota", folder=None, timestamp_filename=True
):
if timestamp_filename:
filename = f"{filename}_{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}"
if folder is not None:
output_file = Path("output", folder, f"{filename}.html")
else:
@@ -2947,6 +2933,8 @@ class RotaBuilder(object):
output_file.parent.mkdir(parents=True, exist_ok=True)
with open(output_file, "w") as f:
f.write(self.get_worker_timetable_html(True))
url = urllib.parse.quote(str(output_file.resolve()))
print(f"Rota exported [bold blue]('[link url={url}]{url}')[/bold blue]")
def export_rota_to_csv(self, filename: str = "rota"):
output_file = Path("output", f"{filename}.csv")
@@ -3094,7 +3082,7 @@ class RotaBuilder(object):
d = self.start_date + datetime.timedelta(n)
th_class = "date"
if d in bank_holiday_map:
if d in self.bank_holidays:
th_class = th_class + " bank-holiday"
date_row.append(
@@ -3105,7 +3093,7 @@ class RotaBuilder(object):
current_site = ""
for worker in self.workers:
for worker in self.get_all_workers():
if worker.site != current_site:
try:
timetable.append(
@@ -3165,10 +3153,10 @@ class RotaBuilder(object):
title = f"{title} / {unavailable_reason}"
bank_holiday = ""
if d in bank_holiday_map:
if d in self.bank_holidays:
title = f"{title} [Bank Holiday]"
css_class = " ".join((css_class, "bank-holiday"))
bank_holiday = f" data-bank-holiday='{bank_holiday_map[d]}'"
bank_holiday = f" data-bank-holiday='{self.bank_holidays[d]}'"
requests = ""
if (worker.id, week, day) in self.work_requests_map:
@@ -3203,6 +3191,7 @@ class RotaBuilder(object):
data-fte='{fte}' data-fte_adj='{fte_adj}'
data-start_date='{start_date}'
data-end_date='{end_date}'
data-oops='{oops}'
data-worker-targets='{targets}' data-shift-counts='{worker_shift_counts}'
data-weekend-target='{weekend_target}'
data-remote-site='{remote_site}'
@@ -3220,6 +3209,7 @@ class RotaBuilder(object):
fte_adj=worker.fte_adj,
start_date=worker.calculated_start_date,
end_date=worker.calculated_end_date,
oops=json.dumps(json.dumps(worker.oop, default=str)),
targets=worker_targets,
worker_shift_counts=json.dumps(shift_count_dict),
weekend_target=worker.weekend_shift_target_number,
@@ -3302,6 +3292,13 @@ class RotaBuilder(object):
</div>
</details>
<details>
<summary><h2>Worker details</h2></summary>
<div id="worker_details" data-workers='{json.dumps([json.loads(worker.model_dump_json()) for worker in self.get_all_workers()])}'>
<span id="workers-json-target"></span>
</div>
</details>
<details>
<summary><h2>Output</h2></summary>
<pre>
{result_string}
+14 -11
View File
@@ -68,7 +68,7 @@ class Worker(BaseModel):
# rules is easier.
grade: int
# We can either have a user generated ID
id: Optional[int] = None
id: Optional[int| uuid.UUID] = None
fte: int = 100
nwds: list[NonWorkingDays] = []
start_date: datetime.date | None = None
@@ -108,30 +108,32 @@ class Worker(BaseModel):
#
# # days_to_work = Rota.rota_days_length
def load_rota(self, Rota):
self.proportion_rota_to_work = 1
self.shift_target_number = defaultdict(int)
def load_rota(self, Rota: "RotaBuilder"):
self.proportion_rota_to_work: float = 1
self.shift_target_number: dict[ str, int ] = defaultdict(int)
if self.id is None:
self.id = uuid.uuid4()
if self.start_date is not None and self.end_date is not None:
if self.start_date >= self.end_date:
raise ValueError("End date must be after start date")
if self.name == "Steph bailey":
print(self.start_date, self.end_date)
print(self.oop)
Rota.add_warning("Worker/Early end date", f"{self.name} [{self.id}] end date is before rota start date: {self.end_date}")
raise ValueError(f"End date must be after start date: {self.name}")
# if no start date default to the start of the rota
if self.start_date is None:
self.calculated_start_date = Rota.start_date
else:
self.calculated_start_date = self.start_date
# ? test if start date is valid
if self.start_date < Rota.start_date:
self.calculated_start_date = self.start_date
# If worker start date is before rota start date use the rota start date
self.calculated_start_date = Rota.start_date
elif self.start_date >= Rota.rota_end_date:
# If it is after add a warning
Rota.add_warning("Worker/Late start date", f"{self.name} [{self.id}] start date is after rota end date: {self.start_date}")
else:
self.calculated_start_date = self.start_date
pass
if self.end_date is None:
self.calculated_end_date = Rota.rota_end_date
@@ -256,6 +258,7 @@ class Worker(BaseModel):
if self.fte_adj > 100:
# Shouldn't happen ? bug if it does
Rota.add_warning("FTE > 100%", f"{self.name} [{self.id}] FTE = {self.fte_adj}")
raise ValueError("{} : fte_ajd = {}".format(self.name, self.fte_adj))
# TODO: this has already been validated, consider moving