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

start working on shift paring balancing

Browse Source
This commit is contained in:
Ross
2024-12-26 16:39:04 +00:00
parent 1ed5f4fbcd
commit f0b9ee0ebc
6 changed files with 741 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gen.py
+470 -335
View File
@@ -3,6 +3,10 @@ import os
import sys
import time
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, days
import typer
import subprocess
app = typer.Typer()
sites = (
@@ -15,6 +19,8 @@ sites = (
"proc",
)
NIGHT_REGISTRAR_NUMBER = 4
from rota.workers import (
Worker,
NotAvailableToWork,
@@ -24,355 +30,484 @@ from rota.workers import (
OutOfProgramme,
)
@app.command()
def main(
suspend_on_finish: bool = False,
solve: bool = True,
time_to_run: int = 60 * 60,
ratio: float = 0.001,
start_date: datetime.datetime = "2025-03-03",
weeks: int = 26,
bom: int = 1,
):
suspend_on_finish = False
solve = True
# time_to_run = 508500
# time_to_run = 2660000
time_to_run = 60 * 60 * 100
#time_to_run = 60 * 300
# allow = 5000
ratio = 0.5
#ratio = 0.9
rota_start_date = start_date.date()
Rota = RotaBuilder(
rota_start_date, weeks_to_rota=weeks, balance_offset_modifier=bom, name="proc_rota"
)
# Rota = RotaBuilder(start_date, weeks_to_rota=20, balance_offset_modifier=1)
Rota.constraint_options["balance_weekends"] = True
Rota.constraint_options["max_night_frequency"] = 3 # 3
Rota.constraint_options["max_night_frequency_week_exclusions"] = []
Rota.constraint_options["max_weekend_frequency"] = 2
Rota.constraint_options["hard_constrain_pair_separation"] = True
# Rota.constraint_options["avoid_st2_first_month"] = True
start_date = datetime.date(2024, 3, 4)
# start_date = datetime.date(2023, 5, 1)
# start_date = start_date + datetime.timedelta(weeks=13)
# start_date = datetime.date(2022, 10, 10)
Rota.add_shifts(
SingleShift(
sites=(
"exeter",
"exeter twilights and weekends",
"exeter no nights",
"exeter twilights",
),
name="exeter_twilight",
length=12.5,
days=days[:5],
balance_offset=4,
constraint=[
{
"name": "max_shifts_per_week",
"options": 2,
}
],
),
SingleShift(
sites=(
"truro",
"truro no nights",
"truro twilights",
"truro twilights and weekends",
"truro twilights and weekend nights",
"truro twilights, weekends and weekend nights",
),
name="truro_twilight",
length=12.5,
days=days[:4],
balance_offset=4,
),
SingleShift(
sites=("torbay", "torbay twilights", "torbay twilights and weekends"),
name="torbay_twilight",
length=12.5,
days=days[:4],
balance_offset=4,
assign_as_block=True,
),
SingleShift(
sites=(
"plymouth",
"plymouth twilights",
"plymouth twilights and weekends",
"plymouth twilights, weekends and weekend nights",
),
name="plymouth_twilight",
length=12.5,
days=days[:5],
balance_offset=4,
workers_required=1,
constraint=[{"name": "max_shifts_per_week", "options": 2}],
),
SingleShift(
sites=(
"exeter",
"exeter twilights and weekends",
"exeter no nights",
"exeter weekends",
),
name="weekend_exeter",
length=12.5,
days=days[5:],
balance_offset=3,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "post", "options": 2}, {"name": "pre", "options": 2}],
# constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2},],
),
SingleShift(
sites=(
"truro",
"truro no nights",
"truro twilights and weekends",
"truro weekends",
"truro twilights, weekends and weekend nights",
),
name="weekend_truro",
length=12.5,
days=days[4:],
balance_offset=3,
#rota_on_nwds=True,
#force_as_block=True,
assign_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
# force_as_block_unless_nwd=True
),
SingleShift(
sites=("torbay", "torbay twilights and weekends"),
name="weekend_torbay",
length=12.5,
days=days[4:],
balance_offset=3,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
# force_as_block_unless_nwd=True
),
SingleShift(
sites=(
"plymouth",
"plymouth twilights and weekends",
"plymouth first on weekends only",
"plymouth twilights, weekends and weekend nights",
),
name="weekend_plymouth1",
length=8,
days=days[5:],
balance_offset=2.9,
workers_required=1,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
),
SingleShift(
sites=(
"plymouth",
"plymouth twilights and weekends",
"plymouth twilights, weekends and weekend nights",
),
name="weekend_plymouth2",
length=8,
days=days[5:],
balance_offset=2.9,
workers_required=1,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
),
SingleShift(
sites=(
"plymouth",
"plymouth twilights",
"plymouth twilights and weekends",
"plymouth twilights, weekends and weekend nights",
),
name="plymouth_bank_holidays",
length=8,
days=days[5:],
balance_offset=1,
workers_required=1,
rota_on_nwds=True,
force_as_block=False,
bank_holidays_only=True,
),
SingleShift(
sites=sites,
name="night_weekday",
length=12.25,
days=days[:4],
balance_offset=3.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=3,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 2},
{
"name": "require_remote_site_presence_week",
"options": ("plymouth", 1),
},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
end_date=datetime.datetime.strptime("2025-06-01", "%Y-%m-%d").date(),
),
SingleShift(
sites=[
*sites,
"plymouth twilights, weekends and weekend nights",
"truro twilights and weekend nights",
"weekend nights",
"truro twilights, weekends and weekend nights",
],
name="night_weekend",
length=12.25,
days=days[4:],
balance_offset=2.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=3,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 3},
{
"name": "require_remote_site_presence_week",
"options": ("plymouth", 1),
},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
end_date=datetime.datetime.strptime("2025-06-01", "%Y-%m-%d").date(),
),
SingleShift(
sites=sites,
name="night_weekday4",
length=12.25,
days=days[:4],
balance_offset=3.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=NIGHT_REGISTRAR_NUMBER,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 2},
{
"name": "require_remote_site_presence_week",
"options": ("plymouth", 1),
},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
start_date=datetime.datetime.strptime("2025-06-02", "%Y-%m-%d").date(),
),
SingleShift(
sites=[
*sites,
"plymouth twilights, weekends and weekend nights",
"truro twilights and weekend nights",
"weekend nights",
"truro twilights, weekends and weekend nights",
],
name="night_weekend4",
length=12.25,
days=days[4:],
balance_offset=2.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=NIGHT_REGISTRAR_NUMBER,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 3},
{
"name": "require_remote_site_presence_week",
"options": ("plymouth", 1),
},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
start_date=datetime.datetime.strptime("2025-06-02", "%Y-%m-%d").date(),
),
)
Rota = RotaBuilder(
start_date, weeks_to_rota=8, balance_offset_modifier=1, name="proc rota"
)
# Rota = RotaBuilder(start_date, weeks_to_rota=20, balance_offset_modifier=1)
Rota.constraint_options["balance_weekends"] = True
Rota.constraint_options["max_night_frequency"] = 3 # 3
Rota.constraint_options["max_night_frequency_week_exclusions"] = []
Rota.constraint_options["max_weekend_frequency"] = 2
Rota.constraint_options["hard_constrain_pair_separation"] = True
# Rota.constraint_options["avoid_st2_first_month"] = True
# Rota.add_grade_constraint_by_week([2], [1, 2], ["night_weekday", "night_weekend"])
Rota.add_shifts(
SingleShift(
sites=("exeter", "exeter twilights and weekends", "exeter no nights"),
name="exeter_twilight",
length=12.5,
days=days[:5],
balance_offset=4,
constraint=[
{
"name": "max_shifts_per_week",
"options": 2,
}
],
),
SingleShift(
sites=("truro", "truro twilights", "truro twilights and weekends"),
name="truro_twilight",
length=12.5,
days=days[:4],
balance_offset=4,
),
SingleShift(
sites=("torbay", "torbay twilights", "torbay twilights and weekends"),
name="torbay_twilight",
length=12.5,
days=days[:4],
balance_offset=4,
assign_as_block=True,
),
SingleShift(
sites=("plymouth", "plymouth twilights", "plymouth twilights and weekends", "plymouth twilights, weekends and weekend nights"),
name="plymouth_twilight",
length=12.5,
days=days[:5],
balance_offset=4,
workers_required=1,
constraint=[{"name": "max_shifts_per_week", "options": 2}],
),
SingleShift(
sites=("exeter", "exeter twilights and weekends", "exeter no nights"),
name="weekend_exeter",
length=12.5,
days=days[5:],
balance_offset=3,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "post", "options": 2}, {"name": "pre", "options": 2}],
# constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2},],
),
SingleShift(
sites=("truro", "truro twilights and weekends"),
name="weekend_truro",
length=12.5,
days=days[4:],
balance_offset=3,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
# force_as_block_unless_nwd=True
),
SingleShift(
sites=("torbay", "torbay twilights and weekends"),
name="weekend_torbay",
length=12.5,
days=days[4:],
balance_offset=3,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
# force_as_block_unless_nwd=True
),
SingleShift(
sites=("plymouth", "plymouth twilights and weekends", "plymouth first on weekends only", "plymouth twilights, weekends and weekend nights"),
name="weekend_plymouth1",
length=8,
days=days[5:],
balance_offset=2.9,
workers_required=1,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
),
SingleShift(
sites=("plymouth", "plymouth twilights and weekends", "plymouth twilights, weekends and weekend nights"),
name="weekend_plymouth2",
length=8,
days=days[5:],
balance_offset=2.9,
workers_required=1,
rota_on_nwds=True,
force_as_block=True,
constraint=[{"name": "pre", "options": 2}, {"name": "post", "options": 2}],
),
SingleShift(
sites=("plymouth", "plymouth twilights", "plymouth twilights and weekends", "plymouth twilights, weekends and weekend nights"),
name="plymouth_bank_holidays",
length=8,
days=days[5:],
balance_offset=1,
workers_required=1,
rota_on_nwds=True,
force_as_block=False,
bank_holidays_only=True,
),
SingleShift(
sites=sites,
name="night_weekday",
length=12.25,
days=days[:4],
balance_offset=3.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=3,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 2},
{"name": "require_remote_site_presence_week", "options": ("plymouth", 1)},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
),
SingleShift(
sites=[*sites, "plymouth twilights, weekends and weekend nights"],
name="night_weekend",
length=12.25,
days=days[4:],
balance_offset=2.9,
balance_weighting=1,
# hard_constrain_shift=False,
workers_required=3,
force_as_block=True,
rota_on_nwds=True,
constraint=[
{"name": "night"},
{"name": "pre", "options": 2},
{"name": "post", "options": 3},
{"name": "require_remote_site_presence_week", "options": ("plymouth", 1)},
{"name": "limit_grade_number", "options": {2: 1}},
{"name": "minimum_grade_number", "options": (4, 1)},
],
),
)
load_leave = True
Rota.build_shifts()
# Rota.add_grade_constraint_by_week([2], [1], Rota.get_shift_names())
# Rota.add_grade_constraint_by_week([2], [1, 2], ["night_weekday", "night_weekend"])
if load_leave:
import leave
load_leave = True
Rota.build_shifts()
# Rota.add_grade_constraint_by_week([2], [1], Rota.get_shift_names())
workers = leave.load_leave(Rota)
if load_leave:
import leave
n = 0
for worker in workers:
n = n + 1
worker_name = worker
w = workers[worker]
site = w["site"]
grade = w["grade"]
try:
fte = float(w["fte"]) * 100
except ValueError:
print(f"{worker} has invalid fte: {fte}")
raise ValueError
workers = leave.load_leave(Rota)
nwd = w["nwd"]
end_date = w["end_date"]
start_date = w["start_date"]
oop = w["oop"]
grade = w["grade"]
pair = w["pair"]
# name, site, grade, fte, nwd, end_date, oop, twi, twi_we, night, night_we, pw1, pw2, extra = row
n = 0
for worker in workers:
n = n + 1
worker_name = worker
w = workers[worker]
site = w["site"]
grade = w["grade"]
try:
fte = float(w["fte"]) * 100
except ValueError:
print(f"{worker} has invalid fte: {fte}")
raise ValueError
# Ignore trainees if fte == 0 (what are they doing here anyway)
if fte == 0 or fte == "0" or fte == "":
continue
nwd = w["nwd"]
end_date = w["end_date"]
start_date = w["start_date"]
oop = w["oop"]
grade = w["grade"]
pair = w["pair"]
# name, site, grade, fte, nwd, end_date, oop, twi, twi_we, night, night_we, pw1, pw2, extra = row
nwds = []
if nwd:
for i in nwd.split(","):
nwd_start_date = Rota.start_date
nwd_end_date = Rota.rota_end_date
# Ignore trainees if fte == 0 (what are they doing here anyway)
if fte == 0 or fte == "0" or fte == "":
continue
if "[" in i:
print("Split nwd", worker_name, i)
a, b = i.split("[")[1][:-1].split("-")
print(f"{a=} {b=}")
nwd_start_date = datetime.datetime.strptime(
a, "%d/%m/%Y"
).date()
nwd_end_date = datetime.datetime.strptime(b, "%d/%m/%Y").date()
nwds = []
if nwd:
for i in nwd.split(","):
nwd_start_date = Rota.start_date
nwd_end_date = Rota.rota_end_date
if "[" in i:
print("Split nwd", worker_name, i)
a, b = i.split("[")[1][:-1].split("-")
print(f"{a=} {b=}")
nwd_start_date = datetime.datetime.strptime(a, "%d/%m/%y").date()
nwd_end_date = datetime.datetime.strptime(b, "%d/%m/%y").date()
nwds.append(
NonWorkingDays(
day=i.split("[")[0].strip()[:3].capitalize(),
start_date=nwd_start_date,
end_date=nwd_end_date,
nwds.append(
NonWorkingDays(
day=i.split("[")[0].strip()[:3].capitalize(),
start_date=nwd_start_date,
end_date=nwd_end_date,
)
)
)
# nwds = [i.strip()[:3].capitalize() for i in nwd.split("/")] if nwd else None
end_date = end_date if end_date else None
# nwds = [i.strip()[:3].capitalize() for i in nwd.split("/")] if nwd else None
end_date = end_date if end_date else None
if oop:
formatted_oops = []
for dates in oop.split(","):
print(dates)
s, e = dates.split("-")
try:
formatted_oops.append(
{
"start_date": datetime.datetime.strptime(
s.strip(), "%d/%m/%y"
).date(),
"end_date": datetime.datetime.strptime(
e.strip(), "%d/%m/%y"
).date(),
}
)
except ValueError as e:
print(e)
print("WORKER", worker)
print("DATES", s, e)
raise
if oop:
formatted_oops = []
for dates in oop.split(","):
print(dates)
if "-" in dates:
s, e = dates.split("-")
elif "to" in dates:
s, e = dates.split(" to ")
else:
raise ValueError(f"Cannot parse OOP dates: {dates}")
try:
formatted_oops.append(
{
"start_date": datetime.datetime.strptime(
s.strip(), "%d/%m/%Y"
).date(),
"end_date": datetime.datetime.strptime(
e.strip(), "%d/%m/%Y"
).date(),
}
)
except ValueError:
try:
formatted_oops.append(
{
"start_date": datetime.datetime.strptime(
s.strip(), "%d/%m/%y"
).date(),
"end_date": datetime.datetime.strptime(
e.strip(), "%d/%m/%y"
).date(),
}
)
except ValueError as e:
print(e)
print("WORKER", worker)
print("DATES", s, e)
raise
oop = formatted_oops
else:
oop = []
print(f"{worker} OOP {oop}")
previous_shifts = {}
# if twi:
# worked = twi.split(" ")[0]
# allocated = twi.split(" ")[1][1:-1]
# previous_shifts["{}_twilight".format(site.lower())] = (worked, allocated)
# if twi_we:
# worked = twi_we.split(" ")[0]
# allocated = twi_we.split(" ")[1][1:-1]
# previous_shifts["weekend_{}".format(site.lower())] = (worked, allocated)
# if night:
# worked = night.split(" ")[0]
# allocated = night.split(" ")[1][1:-1]
# previous_shifts["night_weekday".format(site.lower())] = (worked, allocated)
# if night_we:
# worked = night_we.split(" ")[0]
# allocated = night_we.split(" ")[1][1:-1]
# previous_shifts["night_weekend".format(site.lower())] = (worked, allocated)
# if pw1:
# worked = pw1.split(" ")[0]
# allocated = pw1.split(" ")[1][1:-1]
# previous_shifts["weekend_plymouth1"] = (worked, allocated)
# if pw2:
# worked = pw2.split(" ")[0]
# allocated = pw2.split(" ")[1][1:-1]
# previous_shifts["weekend_plymouth2"] = (worked, allocated)
leave_requests = w["leave"]
work_requests = w["requests"]
site_pref = w["site_pref"]
w = Worker(
name=worker_name,
site=site.lower(),
grade=int(grade[2]),
id=n,
fte=int(fte),
nwds=nwds,
start_date=start_date,
end_date=end_date,
oop=oop,
not_available_to_work=leave_requests,
work_requests=work_requests,
remote_site=site_pref,
previous_shifts=previous_shifts,
pair=pair,
shift_balance_extra=w["shift_balance_extra"],
bank_holiday_extra=w["bank_holiday_extra"],
)
print(w)
Rota.add_worker(w)
# Rota.build_workers()
# Rota.build_model()
solver_options = {"ratio": ratio, "seconds": time_to_run, "threads": 10}
#solver_options = {"seconds": time_to_run, "threads": 10}
# start_time = time.time()
Rota.build_and_solve(solver_options, export=True, solve=solve, solver="appsi_highs")
#Rota.solve_shifts_by_block(solver_options, block_length=13)
#Rota.solve_shifts_individually(solver_options)
# Rota.solve_model(options=solver_options)
# end_time = time.time()
# print(f"Time taken {end_time-start_time}")
# print(Rota.get_worker_details())
# optimizer = SolverFactory('cbc')
# result = optimizer.solve(prob,tee=True)
# result.Solver.Status = SolverStatus.warning
# prob.solutions.load_from(result)
# ResultsHolder = RotaResults(Rota)
# worker_timetable_brief = ResultsHolder.get_worker_timetable_brief(
# show_prefs=False, show_unavailable=False
# )
#
# print(worker_timetable_brief)
# Rota.export_rota_to_html("proc_rota")
# Rota.export_rota_to_csv("rota")
subprocess.run(["scp", Rota.exported_rota_file, "ross@46.101.13.46:proc/proc-rota/output/"])
if suspend_on_finish:
os.system("systemctl suspend")
oop = formatted_oops
else:
oop = []
print(f"{worker} OOP {oop}")
previous_shifts = {}
# if twi:
# worked = twi.split(" ")[0]
# allocated = twi.split(" ")[1][1:-1]
# previous_shifts["{}_twilight".format(site.lower())] = (worked, allocated)
# if twi_we:
# worked = twi_we.split(" ")[0]
# allocated = twi_we.split(" ")[1][1:-1]
# previous_shifts["weekend_{}".format(site.lower())] = (worked, allocated)
# if night:
# worked = night.split(" ")[0]
# allocated = night.split(" ")[1][1:-1]
# previous_shifts["night_weekday".format(site.lower())] = (worked, allocated)
# if night_we:
# worked = night_we.split(" ")[0]
# allocated = night_we.split(" ")[1][1:-1]
# previous_shifts["night_weekend".format(site.lower())] = (worked, allocated)
# if pw1:
# worked = pw1.split(" ")[0]
# allocated = pw1.split(" ")[1][1:-1]
# previous_shifts["weekend_plymouth1"] = (worked, allocated)
# if pw2:
# worked = pw2.split(" ")[0]
# allocated = pw2.split(" ")[1][1:-1]
# previous_shifts["weekend_plymouth2"] = (worked, allocated)
leave_requests = w["leave"]
work_requests = w["requests"]
site_pref = w["site_pref"]
w = Worker(
name=worker_name,
site=site.lower(),
grade=int(grade[2]),
id=n,
fte=int(fte),
nwds=nwds,
start_date=start_date,
end_date=end_date,
oop=oop,
not_available_to_work=leave_requests,
work_requests=work_requests,
remote_site=site_pref,
previous_shifts=previous_shifts,
pair=pair,
shift_balance_extra=w["shift_balance_extra"],
bank_holiday_extra=w["bank_holiday_extra"],
)
print(w)
Rota.add_worker(w)
# Rota.build_workers()
# Rota.build_model()
solver_options = {"ratio": ratio, "seconds": time_to_run, "threads": 10}
#start_time = time.time()
Rota.build_and_solve(solver_options, export=True, solve=True)
# Rota.solve_shifts_by_block(solver_options, block_length=13)
# Rota.solve_shifts_individually(solver_options)
# Rota.solve_model(options=solver_options)
#end_time = time.time()
#print(f"Time taken {end_time-start_time}")
# print(Rota.get_worker_details())
# optimizer = SolverFactory('cbc')
# result = optimizer.solve(prob,tee=True)
# result.Solver.Status = SolverStatus.warning
# prob.solutions.load_from(result)
# ResultsHolder = RotaResults(Rota)
# worker_timetable_brief = ResultsHolder.get_worker_timetable_brief(
# show_prefs=False, show_unavailable=False
# )
#
# print(worker_timetable_brief)
#Rota.export_rota_to_html("proc_rota")
# Rota.export_rota_to_csv("rota")
if suspend_on_finish:
os.system("systemctl suspend")
if __name__ == "__main__":
app()
leave.py
+39 -15
View File
@@ -2,6 +2,9 @@ import csv
from datetime import datetime
import re
from requests import Session
from loguru import logger
from rich.pretty import pprint
from rota.workers import Worker, NotAvailableToWork, NonWorkingDays, WorkRequests, PreferenceNotToWork, OutOfProgramme
@@ -9,6 +12,8 @@ from rota.workers import Worker, NotAvailableToWork, NonWorkingDays, WorkRequest
date_re = r"[\d]{1,2}\/[\d]{1,2}\/[\d]{2}"
live_rota = True
def load_leave(Rota):
with Session() as s:
@@ -21,14 +26,17 @@ def load_leave(Rota):
"Pragma": "no-cache"
}
download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vRYqSKq-OYW1tNnGJtx9LIMo_FXVfcHBRPF4HEMcBxWPfmXimWtr5oEDR_rIUSXrOndb9YjP3iC_4bS/pub?gid=0&single=true&output=csv", headers=headers)
decoded_content = download.content.decode('utf-8')
if live_rota:
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=',')
#with open("/home/ross/Downloads/Draft PROC Rota March 24 - Sept 24 - Sheet1.csv") as f:
# reader = csv.reader(f.read().splitlines(), delimiter=',')
else:
with open("/home/ross/Downloads/PROC Draft RK version.csv") as f:
reader = csv.reader(f.read().splitlines(), delimiter=',')
n = 0
leave = {}
requests = {}
site_prefs = {}
@@ -37,7 +45,11 @@ def load_leave(Rota):
n = 1
for row in reader:
row_title = row[0]
if n < 10:
print(row)
row_title = row[1]
print(row_title)
row_date = row[0]
r = row[2:]
# header, extract names
if n == 1:
@@ -70,16 +82,16 @@ def load_leave(Rota):
if lower_item == "derriford twilights":
lower_item = "plymouth_twilights"
if "PROC site" in row_title:
if "PROC site" in row_title or row_title == "PROC nights from":
worker["site_pref"] = lower_item
if row_title in ("Rotation", "Site"):
elif row_title in ("Rotation", "Site", "Placement", "Placement location"):
worker["site"] = lower_item
elif row_title == "Grade":
elif row_title in ("Grade", "Grade (ST)"):
worker["grade"] = lower_item
elif row_title == "%FTE":
elif row_title in ("%FTE", "%FTE on-call", "FTE", "FTE on-call"):
worker["fte"] = lower_item
elif "NWD" in row_title:
@@ -98,7 +110,13 @@ def load_leave(Rota):
elif row_title in ("Start date",):
if lower_item:
date = datetime.strptime(lower_item, "%d/%m/%y").date()
try:
date = datetime.strptime(lower_item, "%d/%m/%y").date()
except ValueError:
date = datetime.strptime(lower_item, "%d/%m/%Y").date()
except ValueError:
logger.warning(f"Cannot parse date: {lower_item} (n: {n}, row: {row})")
raise ValueError(f"Cannot parse date: {lower_item}")
worker["start_date"] = date
else:
worker["start_date"] = None
@@ -128,19 +146,25 @@ def load_leave(Rota):
if lower_item:
worker["bank_holiday_extra"] = int(lower_item)
elif re.match(date_re, row_title) is not None:
elif re.match(date_re, row_date) is not None:
if lower_item != "":
try:
try:
date = datetime.strptime(row[0], "%d/%m/%y").date()
date = datetime.strptime(row_date, "%d/%m/%y").date()
except ValueError:
date = datetime.strptime(row[0], "%d/%m/%Y").date()
date = datetime.strptime(row_date, "%d/%m/%Y").date()
# This may be easier to do as a dict
if lower_item in shifts:
worker["requests"].append(WorkRequests(date=date, shift=lower_item))
elif lower_item in ("nights only"):
if date.weekday() < 5:
worker["requests"].append(WorkRequests(date=date, shift="night_weekday"))
else:
worker["requests"].append(WorkRequests(date=date, shift="night_weekend"))
elif lower_item in ("work_request", "happy to work", "offered"):
elif lower_item in ("work_request", "happy to work", "offered", "may be", "volunteered"):
worker["requests"].append(WorkRequests(date=date, shift="*"))
output/timetable.js
+3
View File
@@ -149,6 +149,8 @@ function generateExtra() {
bank_holidays_worked = bank_holidays.length;
nights_worked= $(tr).find(".night-shift").length;
shift_diff = JSON.parse(worker_td.attr("data-shift-diff"));
summed_shift_diff = Object.values(shift_diff).reduce((a, b) => a + b);
@@ -161,6 +163,7 @@ function generateExtra() {
worker_td.after(`<div class='worker-summary auto-generated'>
<span>Total shifts: ${total_shifts}, </span>
<span>Weekends: ${weekends_worked}, </span>
<span>Nights: ${nights_worked}, </span>
<span>Bank holidays: ${bank_holidays_worked}, </span>
<span class='shift-diff-span'>Summed shift diff: ${summed_shift_diff.toPrecision(2)}, </span>
rota/shifts.py
+61 -25
View File
@@ -125,6 +125,7 @@ class SingleShift(BaseModel):
constraint: List[ShiftConstraint] = []
start_date: datetime.date | None = None
end_date: datetime.date | None = None
pair_proxy: bool = False
model_config = ConfigDict(
extra = "allow",
@@ -280,6 +281,8 @@ class RotaBuilder(object):
self.paired_shifts = []
self.exported_rota_file = None
def set_rota_dates(self, start_date: datetime.date, weeks_to_rota: int):
self.weeks_to_rota = weeks_to_rota
@@ -1118,23 +1121,26 @@ class RotaBuilder(object):
print(shift_name)
print(week, self.get_week_start_date(week), shift.start_date, shift.end_date)
if self.get_week_start_date(week) <= shift.start_date:
continue
#if self.get_week_start_date(week) <= shift.start_date:
# continue
if self.get_week_start_date(week) > shift.end_date:
continue
#if self.get_week_start_date(week) > shift.end_date:
# continue
for worker in self.get_workers_for_shift(shift):
self.model.constraints.add(
8
* self.model.blocks_worker_shift_assigned[
worker.id, week, shift_name
]
>= sum(
self.model.works[worker.id, week, day, shift_name]
for day in shift.days
try:
self.model.constraints.add(
8
* self.model.blocks_worker_shift_assigned[
worker.id, week, shift_name
]
>= sum(
self.model.works[worker.id, week, day, shift_name]
for day in shift.days
)
)
)
except KeyError:
pass
self.model.constraints.add(
self.model.blocks_assigned[week, shift_name]
== sum(
@@ -1203,7 +1209,7 @@ class RotaBuilder(object):
worker.id, week, shift.name
]
for worker in self.workers
)
)
<= shift.workers_required
)
@@ -2161,15 +2167,23 @@ class RotaBuilder(object):
# shift spans 7 days you may get >7 allocations in a row
# as it only checks for a different shift allocation
for constraint_shift in self.get_shifts_with_constraints(
"pre", week=week
"pre"#, week=week
):
for n in range(0, constraint_shift.constraint_options["pre"]):
if day in constraint_shift.days:
self.model.constraints.add(
1
>= self.model.works[
#week_start_date = self.get_week_start_date(week)
#try:
#if constraint_shift.start_date <= week_start_date < constraint_shift.end_date:
# works = 1
try:
works = self.model.works[
worker.id, week, day, constraint_shift.name
]
except KeyError:
continue
self.model.constraints.add(
1
>= works
+ sum(
pre_map[n][0]
* self.model.works[
@@ -2209,15 +2223,25 @@ class RotaBuilder(object):
# )
for constraint_shift in self.get_shifts_with_constraints(
"post", week=week
"post"#, week=week
):
for n in range(0, constraint_shift.constraint_options["post"]):
if day in constraint_shift.days:
self.model.constraints.add(
1
>= self.model.works[
#week_start_date = self.get_week_start_date(week)
#if constraint_shift.start_date <= week_start_date < constraint_shift.end_date:
#works = 1
#else:
try:
works = self.model.works[
worker.id, week, day, constraint_shift.name
]
except KeyError:
# If we can't find the key the shift is not assigned
# so we don't need to worry about it
continue
self.model.constraints.add(
1
>= works
+ sum(
post_map[n][0]
* self.model.works[
@@ -2230,6 +2254,8 @@ class RotaBuilder(object):
for w in workers
)
)
#except KeyError:
# pass
# for constraint_shift in self.get_shifts_with_constraints(
# "preclear", "preclear2"
@@ -2664,7 +2690,7 @@ class RotaBuilder(object):
def pair_shifts(self, shift1: SingleShift, shift2: SingleShift):
# NOTE: currently designed to pair two shifts
# this could be extended...
self.pair_shifts.append(set(shift1, shift2))
self.paired_shifts.append([shift1, shift2])
def build_shifts(self):
"""
@@ -2684,6 +2710,11 @@ class RotaBuilder(object):
for s in self.shifts:
pass
## For paired shifts we create proxy shifts
#self.paired_shift_map = {}
#for shifts in self.paired_shifts:
for s in self.shifts:
if s.name in self.shift_names:
raise InvalidShift(f"Duplicate shift: {s.name}")
@@ -2746,6 +2777,7 @@ class RotaBuilder(object):
for shift in self.get_shifts_with_constraint("post"):
self.max_post = max(shift.constraint_options["post"], self.max_post)
# # Todo replace with week_day.....
# self.day_shift_product = []
# self.day_shiftclass_product = []
@@ -3038,6 +3070,7 @@ class RotaBuilder(object):
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]")
self.exported_rota_file = output_file
def export_rota_to_csv(self, filename: str = "rota"):
output_file = Path("output", f"{filename}.csv")
@@ -3272,6 +3305,8 @@ class RotaBuilder(object):
shift = self.get_shift_by_name(shift_name)
if "require_remote_site_presence_week" in shift.constraints:
remote_site = f" data-shift-remote-site='{shift.constraint_options['require_remote_site_presence_week'][0]}'"
if "night" in shift.constraints:
css_class = " ".join((css_class, "night-shift"))
shift_tds.append(
f"<td title='{title}' class='rota-day {css_class}' data-shift='{shift_name}' data-available='{available}' data-unavailable_reason='{unavailable_reason}' data-date='{d}' data-week='{week}' data-day='{day}'{remote_site}{requests}{bank_holiday}>{a}</td>"
@@ -3453,7 +3488,7 @@ class RotaBuilder(object):
return timetable
def get_worker_shifts_by_date(self, worker: Worker) -> Dict:
def get_worker_shifts_by_date(self, worker: Worker) -> Dict[datetime.date, str]:
shifts = {}
n = 0
@@ -3472,6 +3507,7 @@ class RotaBuilder(object):
shifts = {}
for worker in self.workers:
print("*", worker.name)
shifts[worker.name] = len(
[i for i in self.get_worker_shift_list(worker) if i != ""]
)
@@ -3487,7 +3523,7 @@ class RotaBuilder(object):
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0:
if self.model.works[worker.id, week, day, shift].value > 0.90:
temp = shift
shifts.append(temp)
test/test_rota.py
+5 -1
View File
@@ -598,7 +598,11 @@ class TestDemoRotaBalanceShiftSites:
self.Rota.constraint_options["balance_nights"] = False
self.Rota.constraint_options["constrain_time_off_after_nights"] = False
self.Rota.terminate_on_warning.remove("Worker/no valid shifts")
try:
self.Rota.terminate_on_warning.remove("Worker/no valid shifts")
except ValueError:
# Seems to happen with sequential test running
pass
self.Rota.build_and_solve(options={"ratio": 0.1})
group_workers = self.Rota.get_workers_by_group()
test/test_shifts.py
+163 -2
View File
@@ -2,7 +2,7 @@ import datetime
import pytest
from rota.shifts import InvalidShift, NoWorkers, RotaBuilder, SingleShift, WarningTermination, days
from rota.workers import Worker
from rota.workers import NotAvailableToWork, WorkRequests, Worker
import itertools
@@ -438,4 +438,165 @@ class TestShiftDates:
for worker in Rota.get_workers():
assert Rota.get_worker_shift_list_string(worker).startswith("-"*7)
assert Rota.get_worker_shift_list_string(worker).endswith("-"*28)
assert Rota.get_worker_shift_list_string(worker).endswith("-"*28)
def test_shift_start_date_end_date_block_more_workers(self):
Rota = generate_basic_rota(workers=10)
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"), name="a", length= 12.5, days=days[:4],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4}],
#start_date=(Rota.start_date + datetime.timedelta(days=7)),
#end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
SingleShift(
sites=("group1", "group2"), name="b", length= 12.5, days=days[4:],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4},
{"name": "pre", "options": 3}, {"name": "post", "options": 3}],
start_date=(Rota.start_date + datetime.timedelta(days=7)),
end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
)
Rota.build_and_solve(options={"ratio": 0.000})
Rota.export_rota_to_html("test_shift_start_date")
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
total_shifts = Rota.get_workers_total_shifts()
for worker in Rota.get_workers():
print(Rota.get_worker_shift_list_string(worker))
assert total_shifts[worker.name] == 22
assert "ab" not in Rota.get_worker_shift_list_string(worker)
assert "ba" not in Rota.get_worker_shift_list_string(worker)
def test_shift_start_date_end_date_block_with_requests(self):
Rota = generate_basic_rota(workers=7)
request_date = Rota.start_date + datetime.timedelta(days=(7*6)+4)
request = [WorkRequests(date=request_date, shift="c")]
Rota.add_workers(
[
Worker(name=f"worker-wr-{i}", site="group2", grade=1, work_requests=request) for i in range(1, 4)
#Worker(name="worker2", site="group1", grade=1),
]
)
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"), name="a", length= 12.5, days=days[:4],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4}],
#start_date=(Rota.start_date + datetime.timedelta(days=7)),
#end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
SingleShift(
sites=("group1", "group2"), name="b", length= 12.5, days=days[4:],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4},
{"name": "pre", "options": 3}, {"name": "post", "options": 3}],
start_date=(Rota.start_date + datetime.timedelta(days=7)),
end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
SingleShift(
sites=("group1", "group2"), name="c", length= 12.5, days=days[4:],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4},
{"name": "pre", "options": 3}, {"name": "post", "options": 3}],
start_date=(Rota.start_date + datetime.timedelta(days=(7*6))),
),
)
Rota.build_and_solve(options={"ratio": 0.1})
Rota.export_rota_to_html("test_shift_start_date")
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
total_shifts = Rota.get_workers_total_shifts()
for worker in Rota.get_workers():
print(Rota.get_worker_shift_list_string(worker))
assert total_shifts[worker.name] in range(24,30)
shift_string = Rota.get_worker_shift_list_string(worker)
assert "ab" not in shift_string
assert "ba" not in shift_string
# Check all has been assigned as block
assert shift_string.count("aaaa") == shift_string.count("a") / 4
assert shift_string.count("bbb") == shift_string.count("b") / 3
assert shift_string.count("ccc") == shift_string.count("c") / 3
# TODO test the request assignment
for worker in Rota.get_workers_by_group()["group2"]:
assert Rota.get_worker_shifts_by_date(worker)[request_date] == "c"
def test_shift_start_date_end_date_block_paired(self):
Rota = generate_basic_rota(workers=7)
request_date = Rota.start_date + datetime.timedelta(days=(7*6)+4)
no_work = [NotAvailableToWork(date=d, shift="*") for d in date_generator(request_date, 18)]
Rota.add_workers(
[
Worker(name=f"worker-wr-{i}", site="group2", grade=1, not_available_to_work=no_work) for i in range(1, 4)
#Worker(name="worker2", site="group1", grade=1),
]
)
a, b, c = (SingleShift(
sites=("group1", "group2"), name="a", length= 12.5, days=days[:4],
workers_required=8,
constraint=[{"name": "max_shifts_per_week", "options": 4}],
end_date=(Rota.start_date + datetime.timedelta(days=7)),
#end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
SingleShift(
sites=("group1", "group2"), name="b", length= 12.5, days=days[4:],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4},
{"name": "pre", "options": 3}, {"name": "post", "options": 3}],
start_date=(Rota.start_date + datetime.timedelta(days=7)),
end_date=(Rota.start_date + datetime.timedelta(days=((7*6)-1))),
),
SingleShift(
sites=("group1", "group2"), name="c", length= 12.5, days=days[4:],
assign_as_block=True,
force_as_block=True,
workers_required=4,
constraint=[{"name": "max_shifts_per_week", "options": 4},
{"name": "pre", "options": 3}, {"name": "post", "options": 3}],
start_date=(Rota.start_date + datetime.timedelta(days=(7*6))),
))
Rota.add_shifts(
a,
b, c
)
Rota.pair_shifts(b, c)
Rota.build_and_solve(options={"ratio": 0.01})
Rota.export_rota_to_html("test_shift_start_date")
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"