From a7e1fa71716f6e6c4f5ae2e7d41b07b0154f088c Mon Sep 17 00:00:00 2001 From: Ross Date: Sun, 7 Jan 2024 17:52:36 +0000 Subject: [PATCH] numberous fixes --- output/timetable.js | 26 +++++++- rota/bank_holidays.py | 8 +++ rota/shifts.py | 139 +++++++++++++++++++++--------------------- rota/workers.py | 25 ++++---- 4 files changed, 115 insertions(+), 83 deletions(-) create mode 100644 rota/bank_holidays.py diff --git a/output/timetable.js b/output/timetable.js index a980ff9..4a5c27f 100644 --- a/output/timetable.js +++ b/output/timetable.js @@ -546,4 +546,28 @@ function hsv2rgb(h, s, v) { return '#' + rgb.map(function (x) { return ("0" + Math.round(x * 255).toString(16)).slice(-2); }).join(''); -}; \ No newline at end of file +}; + +//$("#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, '&').replace(//g, '>'); + 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 '' + match + ''; + }); +} \ No newline at end of file diff --git a/rota/bank_holidays.py b/rota/bank_holidays.py new file mode 100644 index 0000000..6bcd715 --- /dev/null +++ b/rota/bank_holidays.py @@ -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"] diff --git a/rota/shifts.py b/rota/shifts.py index 7b29cf1..07b4951 100644 --- a/rota/shifts.py +++ b/rota/shifts.py @@ -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):
+

Worker details

 + +
+ +
+
+

Output

 
         {result_string}
diff --git a/rota/workers.py b/rota/workers.py
index 2468cf9..3bda9ff 100644
--- a/rota/workers.py
+++ b/rota/workers.py
@@ -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