Ross
876 lines
40 KiB
Python
876 lines
40 KiB
Python
from django import forms
|
|
from django.forms import widgets
|
|
|
|
from .models import Worker, Leave, RotaSchedule
|
|
import importlib
|
|
import json
|
|
|
|
# Try to import typed helper models from rota_generator to validate complex
|
|
# worker option structures. If unavailable, fall back to permissive behavior.
|
|
try:
|
|
from rota_generator.workers import (
|
|
NonWorkingDays,
|
|
OutOfProgramme,
|
|
NotAvailableToWork,
|
|
PreferenceNotToWork,
|
|
WorkRequests,
|
|
HardDayDependency,
|
|
HardDayExclusion,
|
|
)
|
|
except Exception:
|
|
NonWorkingDays = OutOfProgramme = NotAvailableToWork = PreferenceNotToWork = None
|
|
WorkRequests = HardDayDependency = HardDayExclusion = None
|
|
|
|
|
|
class WorkerForm(forms.ModelForm):
|
|
# Expose core model fields plus a set of richer worker options. Complex
|
|
# structures (lists/dicts) are edited as JSON in textareas; common simple
|
|
# options are provided as dedicated form controls for convenience.
|
|
class Meta:
|
|
model = Worker
|
|
fields = ["name", "email", "site", "grade", "fte", "active"]
|
|
|
|
# Simple option fields
|
|
locum = forms.BooleanField(required=False, label="Locum")
|
|
locum_max_shifts = forms.IntegerField(required=False, label="Locum max shifts")
|
|
locum_max_shifts_per_week = forms.IntegerField(required=False, label="Locum max shifts per week")
|
|
locum_on_nwds = forms.BooleanField(required=False, label="Locum on non-working days")
|
|
remote_site = forms.CharField(required=False, label="Remote site")
|
|
pair = forms.CharField(required=False, label="Pair")
|
|
bank_holiday_extra = forms.IntegerField(required=False, label="Bank holiday extra")
|
|
weekend_shift_target_number = forms.IntegerField(required=False, label="Weekend shift target number")
|
|
prefer_multi_shift_together = forms.IntegerField(required=False, label="Prefer multi-shift together weight")
|
|
max_days_worked_per_week = forms.IntegerField(required=False, label="Max days worked per week")
|
|
|
|
# Complex structured options edited as JSON
|
|
assign_as_block_preferences = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->weight", label="Block preferences (JSON)")
|
|
shift_fte_overrides = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->fte", label="Shift FTE overrides (JSON)")
|
|
exact_shifts = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->exact_count", label="Exact shifts (JSON)")
|
|
groups = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of group names (e.g. [\"group1\", \"group2\"])", label="Groups (JSON)")
|
|
previous_shifts = forms.CharField(required=False, widget=forms.Textarea, help_text="Free JSON structure for previous shifts", label="Previous shifts (JSON)")
|
|
shift_balance_extra = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON structure for extra shift-balance penalties", label="Shift balance extra (JSON)")
|
|
allowed_multi_shift_sets = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of sets (e.g. [[\"a\",\"b\"], ...])", label="Allowed multi-shift sets (JSON)")
|
|
hard_day_dependencies = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of HardDayDependency objects", label="Hard day dependencies (JSON)")
|
|
hard_day_exclusions = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of HardDayExclusion objects", label="Hard day exclusions (JSON)")
|
|
force_assign_with = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping of shift->list", label="Force assign with (JSON)")
|
|
max_shifts_per_week_by_shift_name = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->max", label="Max shifts per week by shift (JSON)")
|
|
forced_assignments = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of tuples (week, day, shift)", label="Forced assignments (JSON)")
|
|
forced_assignments_by_date = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of tuples (date, shift)", label="Forced assignments by date (JSON)")
|
|
|
|
# Availability / request lists
|
|
nwds = forms.CharField(required=False, widget=forms.Textarea, help_text="Non-working days JSON list", label="Non-working days (JSON)")
|
|
oop = forms.CharField(required=False, widget=forms.Textarea, help_text="Out of programme JSON list", label="Out of programme (JSON)")
|
|
not_available_to_work = forms.CharField(required=False, widget=forms.Textarea, help_text="Not-available JSON list", label="Not available to work (JSON)")
|
|
pref_not_to_work = forms.CharField(required=False, widget=forms.Textarea, help_text="Preference not to work JSON list", label="Preferred not to work (JSON)")
|
|
work_requests = forms.CharField(required=False, widget=forms.Textarea, help_text="Work requests JSON list", label="Work requests (JSON)")
|
|
locum_availability = forms.CharField(required=False, widget=forms.Textarea, help_text="Locum availability JSON list", label="Locum availability (JSON)")
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
instance = kwargs.get("instance")
|
|
super().__init__(*args, **kwargs)
|
|
# Populate initial values for the extra fields from instance.options
|
|
initial_opts = {}
|
|
if instance is not None:
|
|
try:
|
|
initial_opts = instance.options or {}
|
|
except Exception:
|
|
initial_opts = {}
|
|
|
|
# Simple field initialisation
|
|
for fld in [
|
|
"locum",
|
|
"locum_max_shifts",
|
|
"locum_max_shifts_per_week",
|
|
"locum_on_nwds",
|
|
"remote_site",
|
|
"pair",
|
|
"bank_holiday_extra",
|
|
"weekend_shift_target_number",
|
|
"prefer_multi_shift_together",
|
|
"max_days_worked_per_week",
|
|
]:
|
|
if fld in self.fields:
|
|
self.fields[fld].initial = initial_opts.get(fld, self.fields[fld].initial)
|
|
|
|
# JSON fields: pretty-print if present
|
|
for json_fld in [
|
|
"assign_as_block_preferences",
|
|
"shift_fte_overrides",
|
|
"exact_shifts",
|
|
"groups",
|
|
"previous_shifts",
|
|
"shift_balance_extra",
|
|
"allowed_multi_shift_sets",
|
|
"hard_day_dependencies",
|
|
"hard_day_exclusions",
|
|
"force_assign_with",
|
|
"max_shifts_per_week_by_shift_name",
|
|
"forced_assignments",
|
|
"forced_assignments_by_date",
|
|
"nwds",
|
|
"oop",
|
|
"not_available_to_work",
|
|
"pref_not_to_work",
|
|
"work_requests",
|
|
"locum_availability",
|
|
]:
|
|
if json_fld in self.fields:
|
|
val = initial_opts.get(json_fld, None)
|
|
if val is None:
|
|
self.fields[json_fld].initial = ""
|
|
else:
|
|
try:
|
|
self.fields[json_fld].initial = json.dumps(val, indent=2, default=str)
|
|
except Exception:
|
|
self.fields[json_fld].initial = str(val)
|
|
|
|
def save(self, commit=True):
|
|
"""Save Worker model and persist extra options into `instance.options`."""
|
|
instance = super().save(commit=False)
|
|
opts = instance.options or {}
|
|
|
|
# Simple options
|
|
simple_opts = [
|
|
"locum",
|
|
"locum_max_shifts",
|
|
"locum_max_shifts_per_week",
|
|
"locum_on_nwds",
|
|
"remote_site",
|
|
"pair",
|
|
"bank_holiday_extra",
|
|
"weekend_shift_target_number",
|
|
"prefer_multi_shift_together",
|
|
"max_days_worked_per_week",
|
|
]
|
|
for k in simple_opts:
|
|
if k in self.cleaned_data:
|
|
opts[k] = self.cleaned_data.get(k)
|
|
|
|
# JSON fields: parse where possible
|
|
json_fields = [
|
|
"assign_as_block_preferences",
|
|
"shift_fte_overrides",
|
|
"exact_shifts",
|
|
"groups",
|
|
"previous_shifts",
|
|
"shift_balance_extra",
|
|
"allowed_multi_shift_sets",
|
|
"hard_day_dependencies",
|
|
"hard_day_exclusions",
|
|
"force_assign_with",
|
|
"max_shifts_per_week_by_shift_name",
|
|
"forced_assignments",
|
|
"forced_assignments_by_date",
|
|
"nwds",
|
|
"oop",
|
|
"not_available_to_work",
|
|
"pref_not_to_work",
|
|
"work_requests",
|
|
"locum_availability",
|
|
]
|
|
for k in json_fields:
|
|
if k in self.cleaned_data:
|
|
val = self.cleaned_data.get(k)
|
|
if val is None or val == "":
|
|
opts[k] = []
|
|
else:
|
|
# If the cleaned value is a JSON string, parse it. If
|
|
# it's already a Python structure (possibly containing
|
|
# pydantic models from `clean()`), keep it but normalize
|
|
# any model instances into serializable dicts.
|
|
if isinstance(val, str):
|
|
try:
|
|
parsed = json.loads(val)
|
|
except Exception:
|
|
parsed = val
|
|
else:
|
|
parsed = val
|
|
|
|
def _normalize(obj):
|
|
# Normalize pydantic BaseModel instances (v2:v1) and recurse
|
|
if hasattr(obj, "model_dump"):
|
|
try:
|
|
res = obj.model_dump()
|
|
return _normalize(res)
|
|
except Exception:
|
|
pass
|
|
if hasattr(obj, "dict") and not isinstance(obj, dict):
|
|
try:
|
|
res = obj.dict()
|
|
return _normalize(res)
|
|
except Exception:
|
|
pass
|
|
# Recurse lists/tuples/sets
|
|
if isinstance(obj, list):
|
|
return [_normalize(i) for i in obj]
|
|
if isinstance(obj, tuple):
|
|
return [_normalize(i) for i in obj]
|
|
if isinstance(obj, set):
|
|
# JSON can't represent sets; convert to list
|
|
return [_normalize(i) for i in obj]
|
|
if isinstance(obj, dict):
|
|
return {k: _normalize(v) for k, v in obj.items()}
|
|
return obj
|
|
|
|
opts[k] = _normalize(parsed)
|
|
|
|
instance.options = opts
|
|
if commit:
|
|
instance.save()
|
|
try:
|
|
self.save_m2m()
|
|
except Exception:
|
|
pass
|
|
return instance
|
|
|
|
def clean(self):
|
|
"""Validate and parse typed JSON worker option fields when possible.
|
|
|
|
This will parse JSON strings into Python structures and, where we can
|
|
import the typed helper models from `rota_generator.workers`, will
|
|
validate each list item against the corresponding Pydantic model.
|
|
"""
|
|
cleaned = super().clean()
|
|
|
|
def _parse_and_validate(field_name, model_cls=None):
|
|
val = cleaned.get(field_name)
|
|
if val is None or val == "":
|
|
return []
|
|
# If already parsed (not a str), accept it
|
|
if not isinstance(val, str):
|
|
parsed = val
|
|
else:
|
|
try:
|
|
parsed = json.loads(val)
|
|
except Exception:
|
|
# leave as raw string to preserve old behaviour
|
|
parsed = val
|
|
|
|
# If a model class is provided, validate each element
|
|
if model_cls and parsed and isinstance(parsed, (list, tuple)):
|
|
validated = []
|
|
for i, item in enumerate(parsed):
|
|
try:
|
|
# If item is already an instance, accept it
|
|
if hasattr(model_cls, "model_dump") or hasattr(model_cls, "__fields__"):
|
|
# pydantic v2 or v1 compatible
|
|
validated.append(model_cls(**item) if isinstance(item, dict) else model_cls.parse_obj(item) if hasattr(model_cls, 'parse_obj') else model_cls(**item))
|
|
else:
|
|
validated.append(item)
|
|
except Exception as e:
|
|
self.add_error(field_name, forms.ValidationError(f"Invalid item in {field_name}[{i}]: {e}"))
|
|
validated.append(item)
|
|
return validated
|
|
|
|
return parsed
|
|
|
|
# Typed list fields
|
|
cleaned["nwds"] = _parse_and_validate("nwds", NonWorkingDays)
|
|
cleaned["oop"] = _parse_and_validate("oop", OutOfProgramme)
|
|
cleaned["not_available_to_work"] = _parse_and_validate("not_available_to_work", NotAvailableToWork)
|
|
cleaned["pref_not_to_work"] = _parse_and_validate("pref_not_to_work", PreferenceNotToWork)
|
|
cleaned["work_requests"] = _parse_and_validate("work_requests", WorkRequests)
|
|
cleaned["locum_availability"] = _parse_and_validate("locum_availability", WorkRequests)
|
|
|
|
# Hard day dependency/exclusion structures
|
|
cleaned["hard_day_dependencies"] = _parse_and_validate("hard_day_dependencies", HardDayDependency)
|
|
cleaned["hard_day_exclusions"] = _parse_and_validate("hard_day_exclusions", HardDayExclusion)
|
|
|
|
# forced assignments: expect list of tuples [week, day, shift]
|
|
fa = cleaned.get("forced_assignments")
|
|
if isinstance(fa, str):
|
|
try:
|
|
fa_parsed = json.loads(fa)
|
|
except Exception:
|
|
fa_parsed = fa
|
|
else:
|
|
fa_parsed = fa
|
|
if isinstance(fa_parsed, list):
|
|
# basic validation
|
|
for i, t in enumerate(fa_parsed):
|
|
if not (isinstance(t, (list, tuple)) and len(t) == 3):
|
|
self.add_error("forced_assignments", forms.ValidationError(f"forced_assignments[{i}] must be a tuple/list of (week, day, shift)"))
|
|
cleaned["forced_assignments"] = fa_parsed
|
|
|
|
# forced_assignments_by_date: expect list of tuples [date, shift]
|
|
fabd = cleaned.get("forced_assignments_by_date")
|
|
if isinstance(fabd, str):
|
|
try:
|
|
fabd_parsed = json.loads(fabd)
|
|
except Exception:
|
|
fabd_parsed = fabd
|
|
else:
|
|
fabd_parsed = fabd
|
|
if isinstance(fabd_parsed, list):
|
|
for i, t in enumerate(fabd_parsed):
|
|
if not (isinstance(t, (list, tuple)) and len(t) == 2):
|
|
self.add_error("forced_assignments_by_date", forms.ValidationError(f"forced_assignments_by_date[{i}] must be a tuple/list of (date, shift)"))
|
|
cleaned["forced_assignments_by_date"] = fabd_parsed
|
|
|
|
return cleaned
|
|
|
|
|
|
class LeaveForm(forms.ModelForm):
|
|
class Meta:
|
|
model = Leave
|
|
fields = ["start_date", "end_date", "reason"]
|
|
widgets = {
|
|
"start_date": widgets.DateInput(attrs={"class": "datepicker input", "autocomplete": "off"}),
|
|
"end_date": widgets.DateInput(attrs={"class": "datepicker input", "autocomplete": "off"}),
|
|
}
|
|
|
|
|
|
class RotaScheduleForm(forms.ModelForm):
|
|
class Meta:
|
|
model = RotaSchedule
|
|
# Do not expose end_date as editable: computed from start_date + weeks
|
|
fields = ["name", "start_date", "description"]
|
|
widgets = {
|
|
"start_date": widgets.DateInput(attrs={"class": "datepicker input", "autocomplete": "off"}),
|
|
}
|
|
|
|
weeks = forms.IntegerField(min_value=1, initial=4, help_text="Number of weeks to generate the rota for")
|
|
end_date = forms.DateField(required=False, disabled=True, widget=widgets.DateInput(attrs={"class": "datepicker input", "autocomplete": "off"}))
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
instance = kwargs.get("instance")
|
|
super().__init__(*args, **kwargs)
|
|
if instance and instance.start_date and instance.end_date:
|
|
delta = instance.end_date - instance.start_date
|
|
weeks = max(1, delta.days // 7)
|
|
self.fields["weeks"].initial = weeks
|
|
self.fields["end_date"].initial = instance.end_date
|
|
# Dynamically add RotaBuilder constraint option fields using the
|
|
# typed `RotaConstraintOptions` metadata (types + descriptions).
|
|
constraint_defaults = {}
|
|
constraint_meta = {}
|
|
try:
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
RotaBuilder = getattr(mod, "RotaBuilder")
|
|
rb = RotaBuilder()
|
|
opts_model = getattr(rb, "constraint_options_model", None)
|
|
if opts_model is not None:
|
|
# default values
|
|
constraint_defaults = opts_model.model_dump()
|
|
# Try to extract field descriptions from Pydantic model_fields
|
|
try:
|
|
mf = opts_model.__class__.model_fields
|
|
for k, info in mf.items():
|
|
desc = None
|
|
# field info may expose 'description' attribute or mapping
|
|
if hasattr(info, "description") and info.description:
|
|
desc = info.description
|
|
else:
|
|
try:
|
|
desc = info.get("description")
|
|
except Exception:
|
|
desc = None
|
|
constraint_meta[k] = desc
|
|
except Exception:
|
|
constraint_meta = {}
|
|
except Exception:
|
|
constraint_defaults = {}
|
|
constraint_meta = {}
|
|
|
|
self._constraint_defaults = constraint_defaults
|
|
self._constraint_meta = constraint_meta
|
|
|
|
# Load existing options from instance if present
|
|
existing_opts = {}
|
|
if instance and getattr(instance, "options", None):
|
|
existing_opts = instance.options or {}
|
|
|
|
for key, default in constraint_defaults.items():
|
|
field_name = f"opt__{key}"
|
|
initial = existing_opts.get(key, default)
|
|
help_text = constraint_meta.get(key) if isinstance(constraint_meta, dict) else None
|
|
label = key.replace("_", " ")
|
|
if isinstance(default, bool):
|
|
self.fields[field_name] = forms.BooleanField(required=False, initial=bool(initial), label=label, help_text=help_text)
|
|
elif isinstance(default, int):
|
|
self.fields[field_name] = forms.IntegerField(required=False, initial=initial, label=label, help_text=help_text)
|
|
elif isinstance(default, float):
|
|
self.fields[field_name] = forms.FloatField(required=False, initial=initial, label=label, help_text=help_text)
|
|
elif isinstance(default, (list, dict)):
|
|
# JSON text area for structured defaults
|
|
self.fields[field_name] = forms.CharField(required=False, initial=json.dumps(initial), widget=forms.Textarea, label=label, help_text=(help_text or "Enter JSON"))
|
|
else:
|
|
self.fields[field_name] = forms.CharField(required=False, initial=initial, label=label, help_text=help_text)
|
|
|
|
# Expose a small set of RotaBuilder constructor arguments on the form
|
|
# so they can be set per-rota. These are stored inside `instance.options`
|
|
# and will be passed through by `RotaSchedule.to_rota_builder()` as
|
|
# legacy builder args (the remaining keys in `options` are treated as
|
|
# constraint configuration for `RotaConstraintOptions`).
|
|
builder_args = {
|
|
"balance_offset_modifier": {"type": "float", "default": 1.0, "help": "Balance offset modifier used by the builder"},
|
|
"ltft_balance_offset": {"type": "float", "default": 1.0, "help": "LTFT balance offset used by the builder"},
|
|
"use_previous_shifts": {"type": "bool", "default": False, "help": "Use previous shifts when building the rota"},
|
|
"use_shift_balance_extra": {"type": "bool", "default": False, "help": "Enable extra shift-balance penalties"},
|
|
"use_bank_holiday_extra": {"type": "bool", "default": False, "help": "Apply bank-holiday balancing extras"},
|
|
"allow_force_assignment_with_leave_conflict": {"type": "bool", "default": False, "help": "Allow force assignments even when leave conflicts exist"},
|
|
}
|
|
|
|
for k, meta in builder_args.items():
|
|
field_name = f"opt__{k}"
|
|
initial = existing_opts.get(k, meta["default"])
|
|
if meta["type"] == "bool":
|
|
self.fields[field_name] = forms.BooleanField(required=False, initial=bool(initial), label=k.replace("_", " "), help_text=meta.get("help"))
|
|
elif meta["type"] == "int":
|
|
self.fields[field_name] = forms.IntegerField(required=False, initial=initial, label=k.replace("_", " "), help_text=meta.get("help"))
|
|
elif meta["type"] == "float":
|
|
self.fields[field_name] = forms.FloatField(required=False, initial=initial, label=k.replace("_", " "), help_text=meta.get("help"))
|
|
else:
|
|
self.fields[field_name] = forms.CharField(required=False, initial=initial, label=k.replace("_", " "), help_text=meta.get("help"))
|
|
|
|
# Expose solver selection and solver options (JSON) so users can tune
|
|
# solver behaviour per-rota. Persisted into `RotaSchedule.options` as
|
|
# keys `solver` (string) and `solver_options` (dict).
|
|
solver_initial = existing_opts.get("solver", "appsi_highs")
|
|
solver_opts_initial = existing_opts.get("solver_options", {})
|
|
try:
|
|
solver_opts_str = json.dumps(solver_opts_initial, indent=2)
|
|
except Exception:
|
|
solver_opts_str = ""
|
|
|
|
self.fields["solver"] = forms.CharField(required=False, initial=solver_initial, label="Solver", help_text="Solver plugin name to use (e.g. appsi_highs, scip, gurobi)")
|
|
# Common solver ratio is surfaced as a dedicated field for convenience
|
|
# (many workflows tune this). It's stored inside the solver_options
|
|
# dict under the key 'ratio'. The free-form JSON textarea remains for
|
|
# advanced options.
|
|
solver_ratio_initial = None
|
|
try:
|
|
if isinstance(solver_opts_initial, dict) and "ratio" in solver_opts_initial:
|
|
solver_ratio_initial = solver_opts_initial.get("ratio")
|
|
else:
|
|
# also accept legacy top-level 'ratio' key in existing_opts
|
|
solver_ratio_initial = existing_opts.get("ratio")
|
|
except Exception:
|
|
solver_ratio_initial = None
|
|
|
|
self.fields["solver"] = forms.CharField(required=False, initial=solver_initial, label="Solver", help_text="Solver plugin name to use (e.g. appsi_highs, scip, gurobi)")
|
|
self.fields["solver_ratio"] = forms.FloatField(required=False, initial=solver_ratio_initial, label="Solver ratio (relative gap)", help_text="Relative MIP gap / ratio (e.g. 0.01 for 1%)")
|
|
self.fields["solver_options"] = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 6}), initial=solver_opts_str, label="Solver options (JSON)", help_text="JSON object of solver-specific options; e.g. {\"seconds\": 60, \"threads\": 4}")
|
|
|
|
def clean_start_date(self):
|
|
start = self.cleaned_data.get("start_date")
|
|
if start is None:
|
|
return start
|
|
if start.weekday() != 0:
|
|
raise forms.ValidationError("Start date must be a Monday")
|
|
return start
|
|
|
|
def clean(self):
|
|
cleaned = super().clean()
|
|
start = cleaned.get("start_date")
|
|
weeks = cleaned.get("weeks")
|
|
if start and weeks:
|
|
import datetime
|
|
|
|
end_date = start + datetime.timedelta(days=weeks * 7)
|
|
cleaned["end_date"] = end_date
|
|
# update the form data/display
|
|
try:
|
|
self.data = self.data.copy()
|
|
self.data["end_date"] = end_date.isoformat()
|
|
except Exception:
|
|
pass
|
|
self.fields["end_date"].initial = end_date
|
|
return cleaned
|
|
|
|
def save(self, commit=True):
|
|
"""Save the RotaSchedule and persist dynamic `opt__*` fields into
|
|
`instance.options`.
|
|
|
|
This persists both typed constraint option fields (created from
|
|
RotaConstraintOptions) and the small set of builder-argument fields
|
|
(created in __init__). For structured fields we attempt to parse JSON
|
|
textareas into Python objects.
|
|
"""
|
|
instance = super().save(commit=False)
|
|
|
|
opts = instance.options or {}
|
|
|
|
# Persist typed constraint option fields using inferred defaults map
|
|
try:
|
|
defaults = getattr(self, "_constraint_defaults", {}) or {}
|
|
except Exception:
|
|
defaults = {}
|
|
|
|
for key, default in defaults.items():
|
|
field_name = f"opt__{key}"
|
|
if field_name in self.cleaned_data:
|
|
val = self.cleaned_data[field_name]
|
|
if isinstance(default, (list, dict)):
|
|
# JSON textarea
|
|
try:
|
|
parsed = json.loads(val) if val is not None and val != "" else []
|
|
except Exception:
|
|
parsed = val
|
|
opts[key] = parsed
|
|
else:
|
|
if val == "" or val is None:
|
|
opts[key] = None
|
|
else:
|
|
opts[key] = val
|
|
|
|
# Persist any remaining opt__ fields (builder args and others)
|
|
for fname, val in self.cleaned_data.items():
|
|
if not fname.startswith("opt__"):
|
|
continue
|
|
key = fname[5:]
|
|
if key in defaults:
|
|
# already handled
|
|
continue
|
|
v = val
|
|
# Try to parse JSON-like strings into structures
|
|
if isinstance(v, str):
|
|
s = v.strip()
|
|
if s.startswith("[") or s.startswith("{"):
|
|
try:
|
|
v = json.loads(v)
|
|
except Exception:
|
|
pass
|
|
opts[key] = v
|
|
|
|
# Persist the top-level 'weeks' value from the form into options so
|
|
# the builder can prefer this explicit value instead of deriving it
|
|
# from end_date. Do not persist end_date itself here.
|
|
try:
|
|
weeks_val = self.cleaned_data.get("weeks")
|
|
if weeks_val is not None:
|
|
opts["weeks"] = int(weeks_val)
|
|
except Exception:
|
|
pass
|
|
|
|
# Persist solver selection and options (if provided). Solver options are
|
|
# stored as a dict under opts['solver_options']. If the user supplied
|
|
# a JSON string, attempt to parse it; otherwise keep raw value.
|
|
try:
|
|
solver_name = self.cleaned_data.get("solver")
|
|
if solver_name:
|
|
opts["solver"] = solver_name
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
so = self.cleaned_data.get("solver_options")
|
|
if so is not None and so != "":
|
|
if isinstance(so, str):
|
|
try:
|
|
parsed = json.loads(so)
|
|
except Exception:
|
|
# keep as raw string if parsing fails (non-fatal)
|
|
parsed = so
|
|
else:
|
|
parsed = so
|
|
opts["solver_options"] = parsed
|
|
except Exception:
|
|
pass
|
|
|
|
# If a dedicated solver_ratio field was provided, inject/override it
|
|
# into the solver_options dict so it is honoured at runtime. This
|
|
# ensures the friendly ratio field wins over any value in the JSON
|
|
# textarea.
|
|
try:
|
|
ratio_val = self.cleaned_data.get("solver_ratio")
|
|
if ratio_val is not None and ratio_val != "":
|
|
so = opts.get("solver_options")
|
|
if so is None or not isinstance(so, dict):
|
|
so = {}
|
|
try:
|
|
so["ratio"] = float(ratio_val)
|
|
except Exception:
|
|
so["ratio"] = ratio_val
|
|
opts["solver_options"] = so
|
|
except Exception:
|
|
pass
|
|
|
|
instance.options = opts
|
|
|
|
if commit:
|
|
instance.save()
|
|
return instance
|
|
|
|
|
|
class WorkerSelfServiceForm(forms.ModelForm):
|
|
"""Small form exposed to workers via tokenized links so they can set
|
|
their FTE and pick non-working days.
|
|
"""
|
|
WEEKDAYS = [
|
|
('0', 'Monday'),
|
|
('1', 'Tuesday'),
|
|
('2', 'Wednesday'),
|
|
('3', 'Thursday'),
|
|
('4', 'Friday'),
|
|
('5', 'Saturday'),
|
|
('6', 'Sunday'),
|
|
]
|
|
|
|
nwds = forms.MultipleChoiceField(choices=WEEKDAYS, required=False, widget=forms.CheckboxSelectMultiple, label="Non-working days")
|
|
|
|
class Meta:
|
|
model = Worker
|
|
fields = ["fte"]
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
instance = kwargs.get('instance')
|
|
initial = kwargs.get('initial', {})
|
|
if instance is not None:
|
|
try:
|
|
opts = instance.options or {}
|
|
nwds = opts.get('nwds', [])
|
|
# store as strings for the MultipleChoiceField
|
|
initial['nwds'] = [str(x) for x in (nwds or [])]
|
|
initial['fte'] = instance.fte
|
|
except Exception:
|
|
pass
|
|
kwargs['initial'] = initial
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def save(self, commit=True):
|
|
"""Persist FTE and non-working days (nwds) into the Worker instance.
|
|
|
|
The form exposes only `fte` and `nwds`. `fte` is handled by the
|
|
ModelForm machinery; we need to ensure `nwds` is persisted inside
|
|
`instance.options['nwds']` as a list of ints.
|
|
"""
|
|
instance = super().save(commit=False)
|
|
|
|
# Persist non-working days into instance.options['nwds'] as list of ints
|
|
opts = instance.options or {}
|
|
nwds = self.cleaned_data.get('nwds') or []
|
|
try:
|
|
opts['nwds'] = [int(x) for x in nwds]
|
|
except Exception:
|
|
# fallback: keep as-is
|
|
opts['nwds'] = nwds
|
|
|
|
instance.options = opts
|
|
|
|
if commit:
|
|
instance.save()
|
|
try:
|
|
self.save_m2m()
|
|
except Exception:
|
|
pass
|
|
return instance
|
|
|
|
|
|
class RotaOptionsForm(forms.Form):
|
|
options_json = forms.CharField(
|
|
widget=forms.Textarea(attrs={"rows": 10, "class": "textarea"}),
|
|
required=False,
|
|
help_text="Edit rota options as JSON. See available constraint keys below.",
|
|
)
|
|
|
|
|
|
class RotaConstraintOptionsForm(forms.Form):
|
|
"""Dynamically generated form exposing typed `RotaConstraintOptions` fields.
|
|
|
|
Fields are created as `opt__{key}` to match existing modal rendering logic
|
|
and to allow reuse of the `rota_options` endpoint/modal. Saving via
|
|
`save_for_rota(rota)` will update only the constraint keys inside
|
|
`RotaSchedule.options`, leaving other keys (e.g. legacy builder args)
|
|
untouched.
|
|
"""
|
|
|
|
def __init__(self, *args, initial_options=None, **kwargs):
|
|
"""initial_options: dict of existing rota.options to populate current values."""
|
|
super().__init__(*args, **kwargs)
|
|
initial_options = initial_options or {}
|
|
|
|
# Try to import the typed model from the rota package. Prefer the
|
|
# RotaConstraintOptions class (no heavy instantiation) to avoid any
|
|
# side-effects of creating a RotaBuilder instance during form init.
|
|
opts_model = None
|
|
try:
|
|
mod = importlib.import_module("rota_generator.shifts")
|
|
RotaConstraintOptions = getattr(mod, "RotaConstraintOptions", None)
|
|
if RotaConstraintOptions is not None:
|
|
opts_model = RotaConstraintOptions()
|
|
else:
|
|
# Last-resort: try to instantiate RotaBuilder and read its
|
|
# constraint_options_model attribute (may have side-effects).
|
|
RotaBuilder = getattr(mod, "RotaBuilder", None)
|
|
if RotaBuilder is not None:
|
|
try:
|
|
rb = RotaBuilder()
|
|
opts_model = getattr(rb, "constraint_options_model", None)
|
|
except Exception:
|
|
opts_model = None
|
|
except Exception:
|
|
opts_model = None
|
|
|
|
self._constraint_defaults = {}
|
|
self._constraint_meta = {}
|
|
|
|
if opts_model is not None:
|
|
try:
|
|
defaults = opts_model.model_dump()
|
|
except Exception:
|
|
defaults = {}
|
|
self._constraint_defaults = defaults
|
|
# try to read field descriptions
|
|
try:
|
|
mf = opts_model.__class__.model_fields
|
|
except Exception:
|
|
mf = {}
|
|
|
|
for k, v in defaults.items():
|
|
field_name = f"opt__{k}"
|
|
help_text = None
|
|
if isinstance(mf, dict) and k in mf:
|
|
info = mf[k]
|
|
try:
|
|
if hasattr(info, "description") and info.description:
|
|
help_text = info.description
|
|
else:
|
|
help_text = info.get("description")
|
|
except Exception:
|
|
help_text = None
|
|
|
|
initial = initial_options.get(k, v)
|
|
|
|
# Map Python types to Django form fields
|
|
if isinstance(v, bool):
|
|
self.fields[field_name] = forms.BooleanField(required=False, initial=bool(initial), label=k.replace("_", " "), help_text=help_text)
|
|
elif isinstance(v, int):
|
|
# Allow None for Optional[int]
|
|
self.fields[field_name] = forms.IntegerField(required=False, initial=initial, label=k.replace("_", " "), help_text=help_text)
|
|
elif isinstance(v, float):
|
|
self.fields[field_name] = forms.FloatField(required=False, initial=initial, label=k.replace("_", " "), help_text=help_text)
|
|
elif isinstance(v, (list, dict)):
|
|
# JSON textarea for structured values
|
|
try:
|
|
init_val = json.dumps(initial) if initial is not None else json.dumps(v)
|
|
except Exception:
|
|
init_val = ""
|
|
self.fields[field_name] = forms.CharField(required=False, initial=init_val, widget=forms.Textarea, label=k.replace("_", " "), help_text=(help_text or "Enter JSON"))
|
|
else:
|
|
self.fields[field_name] = forms.CharField(required=False, initial=initial if initial is not None else v, label=k.replace("_", " "), help_text=help_text)
|
|
|
|
def save_for_rota(self, rota):
|
|
"""Persist cleaned constraint fields into `rota.options`.
|
|
|
|
Only updates keys that are present in the typed constraint defaults.
|
|
For structured fields (JSON), attempt to parse into Python objects.
|
|
"""
|
|
opts = rota.options or {}
|
|
for key in self._constraint_defaults.keys():
|
|
field_name = f"opt__{key}"
|
|
if field_name in self.cleaned_data:
|
|
val = self.cleaned_data[field_name]
|
|
default = self._constraint_defaults.get(key)
|
|
if isinstance(default, (list, dict)):
|
|
try:
|
|
parsed = json.loads(val) if val is not None and val != "" else []
|
|
except Exception:
|
|
parsed = val
|
|
opts[key] = parsed
|
|
else:
|
|
# For optional int/float fields allow empty -> None
|
|
if val == "" or val is None:
|
|
opts[key] = None
|
|
else:
|
|
opts[key] = val
|
|
|
|
rota.options = opts
|
|
rota.save(update_fields=["options"])
|
|
|
|
|
|
class ShiftForm(forms.Form):
|
|
name = forms.CharField(max_length=200)
|
|
sites = forms.CharField(
|
|
help_text="Comma separated list of sites (e.g. exeter,plymouth)",
|
|
required=True,
|
|
)
|
|
length = forms.DecimalField(max_digits=6, decimal_places=2, initial=12.5)
|
|
DAYS = [(d, d) for d in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]]
|
|
days = forms.MultipleChoiceField(choices=DAYS, widget=forms.CheckboxSelectMultiple)
|
|
workers_required = forms.IntegerField(min_value=1, initial=1)
|
|
assign_as_block = forms.BooleanField(required=False, initial=False)
|
|
balance_offset = forms.FloatField(
|
|
required=False,
|
|
help_text="Maximum allowed difference in allocated shifts for this shift (leave blank for default)",
|
|
)
|
|
|
|
# Allow entering shift-specific constraints as JSON. This should be a
|
|
# list of constraint dicts compatible with the scheduler's SingleShift
|
|
# `constraints` field. Example: `[{"name": "night", "options": {...}}]`
|
|
constraints = forms.CharField(
|
|
required=False,
|
|
widget=forms.Textarea(attrs={"rows": 4, "class": "textarea"}),
|
|
help_text="Optional JSON list of constraint objects for this shift (e.g. [{'name':'night','options':{}}])",
|
|
)
|
|
include_groups = forms.CharField(
|
|
required=False,
|
|
help_text="Comma separated list of groups that can work this shift (leave empty for all)",
|
|
label="Include groups",
|
|
)
|
|
exclude_groups = forms.CharField(
|
|
required=False,
|
|
help_text="Comma separated list of groups that cannot work this shift",
|
|
label="Exclude groups",
|
|
)
|
|
|
|
def clean_sites(self):
|
|
val = self.cleaned_data["sites"]
|
|
sites = [s.strip() for s in val.split(",") if s.strip()]
|
|
if not sites:
|
|
raise forms.ValidationError("At least one site is required")
|
|
return sites
|
|
|
|
def clean_constraints(self):
|
|
val = self.cleaned_data.get("constraints")
|
|
if not val:
|
|
return []
|
|
try:
|
|
parsed = json.loads(val)
|
|
if not isinstance(parsed, list):
|
|
raise forms.ValidationError("Constraints must be a JSON list of constraint objects")
|
|
# Validate each constraint object for basic correctness
|
|
errors = []
|
|
for i, c in enumerate(parsed):
|
|
if not isinstance(c, dict):
|
|
errors.append(f"Constraint #{i+1} must be an object")
|
|
continue
|
|
name = c.get('name') or c.get('type')
|
|
if not name:
|
|
errors.append(f"Constraint #{i+1} missing 'name'")
|
|
continue
|
|
if name in ('pre', 'post'):
|
|
opts = c.get('options', {}) or {}
|
|
# days is now required: check top-level or under options
|
|
days_val = c.get('days') if c.get('days') is not None else opts.get('days')
|
|
if days_val is None:
|
|
errors.append(
|
|
f"Constraint #{i+1} ('{name}') requires a 'days' duration (number of days the constraint applies for)."
|
|
)
|
|
if errors:
|
|
raise forms.ValidationError(errors)
|
|
return parsed
|
|
except forms.ValidationError:
|
|
raise
|
|
except Exception as exc:
|
|
raise forms.ValidationError(f"Invalid JSON for constraints: {exc}")
|
|
|
|
def clean_include_groups(self):
|
|
val = self.cleaned_data.get("include_groups") or ""
|
|
return [g.strip() for g in val.split(",") if g.strip()]
|
|
|
|
def clean_exclude_groups(self):
|
|
val = self.cleaned_data.get("exclude_groups") or ""
|
|
return [g.strip() for g in val.split(",") if g.strip()]
|
|
|
|
def clean(self):
|
|
cleaned_data = super().clean()
|
|
include = cleaned_data.get("include_groups") or []
|
|
exclude = cleaned_data.get("exclude_groups") or []
|
|
overlap = set(include).intersection(set(exclude))
|
|
if overlap:
|
|
raise forms.ValidationError(
|
|
f"Include groups and Exclude groups cannot overlap. Overlapping groups: {', '.join(overlap)}"
|
|
)
|
|
return cleaned_data
|