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

start adding locum availability

Browse Source
This commit is contained in:
Ross
2025-01-20 10:56:47 +00:00
parent 7ce4677921
commit 8459a4b3a9
5 changed files with 429 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rota/shifts.py
+65 -13
View File
@@ -209,6 +209,7 @@ class SingleShift(BaseModel):
return self.workers_required
for wr in self.workers_required:
#print(date, wr)
if wr.start_date <= date < wr.end_date:
return wr.number
@@ -315,6 +316,12 @@ class RotaBuilder(object):
self.exported_rota_file = None
def set_rota_constraint(self, constraint: str, value: Any):
if constraint not in self.constraint_options:
raise ValueError(f"Constraint {constraint} not valid")
self.constraint_options[constraint] = value
def set_rota_dates(self, start_date: datetime.date, weeks_to_rota: int):
self.weeks_to_rota = weeks_to_rota
@@ -354,6 +361,10 @@ class RotaBuilder(object):
Tuple[str | uuid.UUID | int, WeekInt, DayStr, ShiftName]
] = set()
self.work_requests_map = {}
self.locum_availability: set[
Tuple[str | uuid.UUID | int, WeekInt, DayStr, ShiftName]
] = set()
self.locum_availability_map = {}
def solve_model(
self,
@@ -836,6 +847,37 @@ class RotaBuilder(object):
initialize=availability_init,
)
locum_request_sets = set()
for locum_request in self.locum_availability:
if locum_request[3] == "*":
pass
else:
locum_request_sets.add(locum_request[3])
# Validate locum shifts are valid
invalid_shifts = locum_request_sets - set(self.get_shift_names())
if invalid_shifts and not self.ignore_valid_shifts:
raise InvalidShift(
f"Invalid locum shift request [shift(s): ${invalid_shifts}]"
)
def locum_request_init(model, wid, week, day, shift):
if (wid, week, day, shift) in self.locum_availability:
self.locum_availability_map[(wid, week, day)] = shift
return 1
return 0
self.model.locum_availability = Param(
(
(worker.id, week, day, shift)
for worker in self.workers
for week, day, shift in self.get_all_shiftname_combinations()
),
initialize=locum_request_init,
)
work_request_sets = set()
for work_request in self.work_requests:
if work_request[3] == "*":
@@ -845,7 +887,6 @@ class RotaBuilder(object):
work_request_sets.add(work_request[3])
# Validate shifts are valid
# work_request_sets = set([i[3] for i in self.work_requests])
invalid_shifts = work_request_sets - set(self.get_shift_names())
if invalid_shifts and not self.ignore_valid_shifts:
raise InvalidShift(
@@ -1190,6 +1231,7 @@ class RotaBuilder(object):
workers_required = shift.get_worker_requirement_by_date(
self.get_week_start_date(week)
)
print(worker.name, week, workers_required, self.get_week_start_date(week))
if shift.force_as_block_unless_nwd:
workers = self.get_workers_for_shift(shift)
@@ -1376,11 +1418,11 @@ class RotaBuilder(object):
)
full_time_equivalent_joined = sum(
self.full_time_equivalent_sites[i] for i in shift.sites
self.full_time_equivalent_sites[i][shift.name] for i in shift.sites
)
target_shifts = (
total_shifts / full_time_equivalent_joined * worker.fte_adj
total_shifts / full_time_equivalent_joined * worker.get_fte(shift=shift.name)
)
if self.use_previous_shifts:
@@ -1408,7 +1450,7 @@ class RotaBuilder(object):
if shift.hard_constrain_shift:
adjusted_balance_offset = shift.balance_offset
if worker.fte_adj < 100:
if worker.get_fte(shift=shift.name) < 100:
adjusted_balance_offset = (
adjusted_balance_offset * self.ltft_balance_offset
)
@@ -2412,6 +2454,8 @@ class RotaBuilder(object):
# c = len(workers)
balance_modifier_constant = 1
balance_quadratic_shift_modifier_constant = 3
block_shift_balancing_constant = 1
if self.constraint_options["balance_shifts_over_workers"]:
worker_shift_balancing = sum(
@@ -2440,7 +2484,7 @@ class RotaBuilder(object):
if self.constraint_options["balance_shifts_quadratic"]:
quadratic_shift_balancing = sum(
balance_modifier_constant
balance_quadratic_shift_modifier_constant
* (
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
@@ -2550,7 +2594,7 @@ class RotaBuilder(object):
if self.constraint_options["balance_blocks"]:
blocks_balancing = sum(
1 * self.model.blocks_assigned[week, shift]
block_shift_balancing_constant * self.model.blocks_assigned[week, shift]
for week in self.weeks
for shift in self.shifts_to_assign_as_blocks()
)
@@ -2646,15 +2690,19 @@ class RotaBuilder(object):
self.workers = sorted(self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers) # TODO: rewrite as per shift
self.full_time_equivalent_sites = {}
self.workers_at_sites = {}
for site in track(self.sites, description="Generating site workers"):
self.workers_at_sites[site] = [w for w in self.workers if w.site == site]
self.full_time_equivalent_sites[site] = sum(
w.fte_adj for w in self.workers if w.site == site
)
self.full_time_equivalent_sites[site] = {}
for shift in self.get_shifts():
self.full_time_equivalent_sites[site][shift.name] = sum(
w.get_fte(shift=shift.name) for w in self.workers if w.site == site
)
self.worker_pairs = []
pairs = defaultdict(list)
@@ -3131,16 +3179,18 @@ class RotaBuilder(object):
return [worker for worker in self.workers if worker.site in shift.sites]
def get_workers_total_fte(self) -> float:
"""Does not take into account shift adjusted ftes"""
return self.full_time_equivalent
def get_worker_details(self):
w = defaultdict(list)
w: dict[str, Worker] = defaultdict(list)
worker: Worker
for worker in self.workers:
w[worker.site].append(worker)
l = []
for site in w:
l.append(f"{site}: {sum(worker.fte_adj for worker in w[site]) / 100}")
l.append(f"{site}: {sum(worker.get_fte() for worker in w[site]) / 100}")
t = "\n".join(l)
return f"Full time equivalent trainees by site:\n{t}"
@@ -3158,6 +3208,8 @@ class RotaBuilder(object):
]
def get_week_start_date(self, week: int):
"""Week start date is 1 indexed"""
week = week - 1
return self.start_date + datetime.timedelta(weeks=week)
# RESULTS
@@ -3454,7 +3506,7 @@ class RotaBuilder(object):
worker_id=worker.id,
name=worker.name,
fte=worker.fte,
fte_adj=worker.fte_adj,
fte_adj=worker.get_fte(),
start_date=worker.calculated_start_date,
end_date=worker.calculated_end_date,
oops=json.dumps(json.dumps(worker.oop, default=str)),
rota/workers.py
+27 -1
View File
@@ -87,7 +87,7 @@ class Worker(BaseModel):
site: str
# Grade are equivalent to roles, by keeping them integer defining model
# rules is easier.
grade: int
grade: int = 1
# We can either have a user generated ID
id: Optional[int | uuid.UUID | str] = None
fte: int = 100
@@ -98,11 +98,15 @@ class Worker(BaseModel):
not_available_to_work: list[NotAvailableToWork] = []
pref_not_to_work: list[PreferenceNotToWork] = []
work_requests: list[WorkRequests] = []
locum_availability: list[WorkRequests] = []
remote_site: str = "plymouth" # We set a default proc_site
previous_shifts: dict = {}
shift_balance_extra: dict = {}
bank_holiday_extra: int = 0
pair: int | str | None = None
locum: bool = False
shift_fte_overrides: dict[str, int] = {} # Need checks to ensure shifts exist
model_config = ConfigDict(
extra="allow",
@@ -284,6 +288,16 @@ class Worker(BaseModel):
else:
Rota.work_requests.add((self.id, week, day, request.shift))
for request in self.locum_availability:
days_from_start = (request.date - Rota.start_date).days
week = days_from_start // 7 + 1
day = Rota.days[(days_from_start % 7)]
if request.shift == "*":
for shift in Rota.get_shifts_for_worker_site(self.site):
Rota.locum_availability.add((self.id, week, day, shift.name))
else:
Rota.locum_availability.add((self.id, week, day, request.shift))
# Calculate the proportion of the rota that is being worked
self.proportion_rota_to_work = days_to_work / Rota.rota_days_length
self.days_to_work = days_to_work
@@ -291,6 +305,12 @@ class Worker(BaseModel):
# We have to adjust the full time equivalent for people who CCT / leave the rota early
self.fte_adj = self.fte * self.proportion_rota_to_work
self.fte_adj_shifts = {}
if self.shift_fte_overrides:
for shift, fte in self.shift_fte_overrides.items():
self.fte_adj_shifts[shift] = fte * self.proportion_rota_to_work
if self.fte_adj > 100:
# Shouldn't happen ? bug if it does
Rota.add_warning(
@@ -356,3 +376,9 @@ class Worker(BaseModel):
def get_shift_targets(self) -> dict:
return self.shift_target_number
def get_fte(self, shift: str | None=None) -> int:
if shift is not None:
if shift in self.fte_adj_shifts:
return self.fte_adj_shifts[shift]
return self.fte_adj