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

Enhance error handling in leave loading and update date parsing logic; add new site preference for Exeter shifts.

Browse Source
This commit is contained in:
Ross
2026-01-05 20:01:41 +00:00
parent 94b921192f
commit b98c28401f
2 changed files with 119 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gen_proc.py
+17 -13
View File
@@ -138,6 +138,7 @@ def main(
"exeter twilights and weekends", "exeter twilights and weekends",
"exeter no nights", "exeter no nights",
"exeter weekends", "exeter weekends",
"exeter twilights",
), ),
name="weekend_exeter", name="weekend_exeter",
length=12.5, length=12.5,
@@ -303,13 +304,18 @@ def main(
site = w["site"] site = w["site"]
except KeyError: except KeyError:
print(f"Worker {worker} has no site") print(f"Worker {worker} has no site")
raise KeyError print(w)
sys.exit(1)
grade = w["grade"] grade = w["grade"]
try: try:
fte = float(w["fte"]) * 100 fte = float(w["fte"]) * 100
except ValueError: except ValueError:
print(f"{worker} has invalid fte: {fte}") print(f"{worker} has invalid fte: {fte}")
raise ValueError raise ValueError
except KeyError:
print(f"Worker {worker} has no FTE")
print(w)
sys.exit(1)
nwd = w["nwd"] nwd = w["nwd"]
end_date = w["end_date"] end_date = w["end_date"]
@@ -374,10 +380,10 @@ def main(
formatted_oops.append( formatted_oops.append(
{ {
"start_date": datetime.datetime.strptime( "start_date": datetime.datetime.strptime(
s.strip(), "%d/%m/%y" s.strip(), "%d/%m/%Y"
).date(), ).date(),
"end_date": datetime.datetime.strptime( "end_date": datetime.datetime.strptime(
e.strip(), "%d/%m/%y" e.strip(), "%d/%m/%Y"
).date(), ).date(),
} }
) )
@@ -425,21 +431,19 @@ def main(
shift_fte_overrides = {} shift_fte_overrides = {}
#if worker_name == "Ben Kemp": if worker_name == "Hadi Mohamed":
# shift_fte_overrides = { shift_fte_overrides = {
# "plymouth_twilight": 100, "weekend_exeter": 50,
# "weekend_plymouth1": 50, }
# "weekend_plymouth2": 50,
# }
#elif worker_name == "Joel Lim": #elif worker_name == "Joel Lim":
# shift_fte_overrides = { # shift_fte_overrides = {
# "plymouth_twilight": 100, # "plymouth_twilight": 100,
# "weekend_exeter": 50, # "weekend_exeter": 50,
# } # }
if worker_name == "Nang Thiriphoo": #if worker_name == "Nang Thiriphoo":
shift_fte_overrides = { # shift_fte_overrides = {
"weekend_exeter": 40, # "weekend_exeter": 40,
} # }
w = Worker( w = Worker(
name=worker_name, name=worker_name,
leave.py
+102 -37
View File
@@ -1,41 +1,49 @@
import csv import csv
from datetime import datetime from datetime import datetime
import re import re
import sys
from requests import Session from requests import Session
from loguru import logger from loguru import logger
from rich.pretty import pprint from rich.pretty import pprint
from rota_generator.workers import Worker, NotAvailableToWork, NonWorkingDays, WorkRequests, PreferenceNotToWork, OutOfProgramme from rota_generator.workers import (
Worker,
NotAvailableToWork,
NonWorkingDays,
WorkRequests,
PreferenceNotToWork,
OutOfProgramme,
)
date_re = r"[\d]{1,2}\/[\d]{1,2}\/[\d]{2}" date_re = r"[\d]{1,2}\/[\d]{1,2}\/[\d]{2}"
live_rota = True live_rota = True
def load_leave(Rota):
def load_leave(Rota):
with Session() as s: with Session() as s:
shifts = Rota.get_shift_names() shifts = Rota.get_shift_names()
workers = {} workers = {}
headers = { headers = {"Cache-Control": "no-cache", "Pragma": "no-cache"}
"Cache-Control": "no-cache",
"Pragma": "no-cache"
}
if live_rota: if live_rota:
download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vQtNrp9F1fjtAh8vFX-W_R3RlD9H-2P_vCLMoHrVKR0e24yabbsdrD65VjwxoHlV1Qnatmw1NeghQHG/pub?gid=1199196967&single=true&output=csv", headers=headers) download = s.get(
#download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vSRx9VWXSlRubPyA0RhiI-Oqf5eHNYYEc6rFzlraDbR5_8qqr5g13-4uV-gn4u-TjZxiSMv1fBUaESq/pub?gid=814517272&single=true&output=csv", headers=headers) "https://docs.google.com/spreadsheets/d/e/2PACX-1vQtNrp9F1fjtAh8vFX-W_R3RlD9H-2P_vCLMoHrVKR0e24yabbsdrD65VjwxoHlV1Qnatmw1NeghQHG/pub?gid=1199196967&single=true&output=csv",
decoded_content = download.content.decode('utf-8') headers=headers,
)
# download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vSRx9VWXSlRubPyA0RhiI-Oqf5eHNYYEc6rFzlraDbR5_8qqr5g13-4uV-gn4u-TjZxiSMv1fBUaESq/pub?gid=814517272&single=true&output=csv", headers=headers)
decoded_content = download.content.decode("utf-8")
reader = csv.reader(decoded_content.splitlines(), delimiter=',') reader = csv.reader(decoded_content.splitlines(), delimiter=",")
else: else:
with open("/home/ross/Downloads/PROC Draft RK version.csv") as f: with open("/home/ross/Downloads/PROC Draft RK version.csv") as f:
reader = csv.reader(f.read().splitlines(), delimiter=',') reader = csv.reader(f.read().splitlines(), delimiter=",")
n = 0 n = 0
leave = {} leave = {}
@@ -48,7 +56,7 @@ def load_leave(Rota):
for row in reader: for row in reader:
if n < 10: if n < 10:
print(row) print(row)
row_title = row[1] row_title = row[0]
print(row_title) print(row_title)
row_date = row[0] row_date = row[0]
r = row[2:] r = row[2:]
@@ -80,44 +88,81 @@ def load_leave(Rota):
worker = workers[a] worker = workers[a]
lower_item = r[i].lower() lower_item = r[i].lower().strip()
if lower_item == "derriford": if lower_item == "derriford":
lower_item = "plymouth" lower_item = "plymouth"
if lower_item == "derriford twilights": if lower_item == "derriford twilights":
lower_item = "plymouth_twilights" lower_item = "plymouth_twilights"
if "PROC site" in row_title or row_title == "PROC nights from" or "proc nights" in row_title.lower(): if (
"PROC site" in row_title
or row_title == "PROC nights from"
or "proc nights" in row_title.lower()
):
worker["site_pref"] = lower_item worker["site_pref"] = lower_item
elif row_title in ("Rotation", "Site", "Placement", "Placement location"): elif row_title in (
"Rotation",
"Site",
"Placement",
"Placement location",
"Placement location",
):
worker["site"] = lower_item worker["site"] = lower_item
elif row_title in ("Grade", "Grade (ST)") or "Year of training" in row_title: elif (
row_title in ("Grade", "Grade (ST)")
or "Year of training" in row_title
):
worker["grade"] = lower_item.lstrip("ST") worker["grade"] = lower_item.lstrip("ST")
elif row_title in ("%FTE", "%FTE on-call", "FTE", "FTE on-call") or "on-call commitment" in row_title: elif (
if lower_item.endswith("%"): row_title
lower_item = float(lower_item.rstrip("%")) / 100 in (
worker["fte"] = lower_item "%FTE",
"%FTE on-call",
"FTE",
"FTE on-call",
"Full-time equivalent (FTE)",
)
or "on-call commitment" in row_title
):
if lower_item:
if lower_item.endswith("%"):
lower_item = float(lower_item.rstrip("%")) / 100
worker["fte"] = lower_item
elif "NWD" in row_title: elif (
"NWD" in row_title or row_title == "If <100% FTE, specify days off:"
):
worker["nwd"] = lower_item worker["nwd"] = lower_item
elif row_title == "Flexible NWD": elif row_title == "Flexible NWD":
worker["flexible_nwd"] = lower_item worker["flexible_nwd"] = lower_item
elif row_title in ("End Date", "CCT date"): elif row_title in ("End Date", "CCT date", "CCT Date"):
if lower_item: if lower_item:
try: try:
date = datetime.strptime(lower_item, "%d/%m/%y").date() date = datetime.strptime(lower_item, "%d/%m/%y").date()
except ValueError: except ValueError:
date = datetime.strptime(lower_item, "%d/%m/%Y").date() try:
date = datetime.strptime(lower_item, "%d-%m-%y").date()
except ValueError:
try:
date = datetime.strptime(
lower_item, "%d/%m/%Y"
).date()
except ValueError:
print(
f"Cannot parse date: '{lower_item}' (n: {n}, row: {row})"
)
sys.exit(1)
worker["end_date"] = date worker["end_date"] = date
else: else:
worker["end_date"] = None worker["end_date"] = None
elif row_title in ("Start date",): elif row_title in ("Start date",):
if lower_item: if lower_item:
try: try:
@@ -125,13 +170,22 @@ def load_leave(Rota):
except ValueError: except ValueError:
date = datetime.strptime(lower_item, "%d/%m/%Y").date() date = datetime.strptime(lower_item, "%d/%m/%Y").date()
except ValueError: except ValueError:
logger.warning(f"Cannot parse date: {lower_item} (n: {n}, row: {row})") logger.warning(
f"Cannot parse date: {lower_item} (n: {n}, row: {row})"
)
raise ValueError(f"Cannot parse date: {lower_item}") raise ValueError(f"Cannot parse date: {lower_item}")
worker["start_date"] = date worker["start_date"] = date
else: else:
worker["start_date"] = None worker["start_date"] = None
elif "OOP" in row_title or "out of programme" in row_title.lower(): elif "OOP" in row_title or "out of programme" in row_title.lower():
if lower_item in (
"OOPT 1 START",
"OOPT 1 END",
"OOPT 2 START",
"OOPT 2 END",
):
continue
worker["oop"] = lower_item worker["oop"] = lower_item
elif row_title == "Group": elif row_title == "Group":
@@ -157,39 +211,50 @@ def load_leave(Rota):
worker["bank_holiday_extra"] = int(lower_item) worker["bank_holiday_extra"] = int(lower_item)
elif re.match(date_re, row_date) is not None: elif re.match(date_re, row_date) is not None:
if lower_item != "": if lower_item != "":
try: try:
try: try:
date = datetime.strptime(row_date, "%d/%m/%y").date() date = datetime.strptime(row_date, "%d/%m/%y").date()
except ValueError: except ValueError:
date = datetime.strptime(row_date, "%d/%m/%Y").date() date = datetime.strptime(row_date, "%d/%m/%Y").date()
# This may be easier to do as a dict # This may be easier to do as a dict
if lower_item in shifts: if lower_item in shifts:
worker["requests"].append(WorkRequests(date=date, shift=lower_item)) worker["requests"].append(
WorkRequests(date=date, shift=lower_item)
)
elif lower_item in ("nights only"): elif lower_item in ("nights only"):
if date.weekday() < 5: if date.weekday() < 5:
worker["requests"].append(WorkRequests(date=date, shift="night_weekday")) worker["requests"].append(
WorkRequests(date=date, shift="night_weekday")
)
else: else:
worker["requests"].append(WorkRequests(date=date, shift="night_weekend")) worker["requests"].append(
WorkRequests(date=date, shift="night_weekend")
elif lower_item in ("work_request", "happy to work", "offered", "may be", "volunteered"): )
worker["requests"].append(WorkRequests(date=date, shift="*"))
elif lower_item in (
"work_request",
"happy to work",
"offered",
"may be",
"volunteered",
):
worker["requests"].append(
WorkRequests(date=date, shift="*")
)
else: else:
worker["leave"].append(NotAvailableToWork(date=date, reason=lower_item)) worker["leave"].append(
NotAvailableToWork(date=date, reason=lower_item)
)
except ValueError as e: except ValueError as e:
print(f"Error with: {lower_item}") print(f"Error with: {lower_item}")
print(f"{row=}") print(f"{row=}")
raise e raise e
n = n + 1 n = n + 1
return workers return workers