""" Scheduling Engine — Deterministic logic for availability checking, overlap prevention, and provider ranking. This module contains NO LLM calls. All logic is pure Python. """ from datetime import datetime, timedelta from zoneinfo import ZoneInfo from config import DEFAULT_WORK_HOURS, DEFAULT_LUNCH_BREAK from core import settings_manager as settings # --- Time utility helpers --- def parse_time(time_str: str) -> datetime: """Parse a time string to a datetime (date portion ignored). Accepts 'HH:MM', 'H:MM', or 'HHMM' (e.g. '0700'). """ s = str(time_str).strip() for fmt in ("%H:%M", "%H%M"): try: return datetime.strptime(s, fmt) except ValueError: continue raise ValueError(f"Unrecognized time format: {time_str!r}") def normalize_time_str(time_str: str) -> str | None: """Normalize a user-entered time string to 'HH:MM'. Returns None if invalid.""" if time_str is None: return None try: return parse_time(time_str).strftime("%H:%M") except ValueError: return None def times_overlap(start1: str, end1: str, start2: str, end2: str) -> bool: """Check if two time ranges overlap. All args in 'HH:MM' format.""" s1, e1 = parse_time(start1), parse_time(end1) s2, e2 = parse_time(start2), parse_time(end2) return s1 < e2 and s2 < e1 def add_minutes(time_str: str, minutes: int) -> str: """Add minutes to a time string. Returns 'HH:MM'.""" result = parse_time(time_str) + timedelta(minutes=minutes) return result.strftime("%H:%M") # --- Visit duration helpers --- def get_visit_duration(visit_level: int | str) -> int: """Return the duration in minutes for a given visit level.""" return settings.get_visit_levels()[visit_level]["duration_minutes"] def get_total_block_duration(visit_level: int | str) -> int: """Return visit duration + buffer in minutes.""" return get_visit_duration(visit_level) + settings.get_buffer_minutes() # --- Availability checks --- def is_within_work_hours(provider: dict, start_time: str, end_time: str) -> bool: """Check if a proposed time slot falls within the provider's work hours. Args: provider: Provider dict with 'work_hours' field start_time: "HH:MM" format end_time: "HH:MM" format """ work = provider.get("work_hours", DEFAULT_WORK_HOURS) return ( parse_time(start_time) >= parse_time(work["start"]) and parse_time(end_time) <= parse_time(work["end"]) ) def conflicts_with_lunch(provider: dict, start_time: str, end_time: str) -> bool: """Check if a proposed time slot overlaps with the provider's lunch break. An empty or missing lunch break (e.g. {"start": "", "end": ""}) is treated as "no lunch break" — returns False. """ lunch = provider.get("lunch_break") or DEFAULT_LUNCH_BREAK lunch_start = normalize_time_str(lunch.get("start")) lunch_end = normalize_time_str(lunch.get("end")) if not lunch_start or not lunch_end: return False return times_overlap(start_time, end_time, lunch_start, lunch_end) def has_overlap(existing_appointments: list[dict], proposed_start: str, proposed_end: str, travel_time_to: int = 0, travel_time_from: int = 0) -> bool: """Check if a proposed slot overlaps with any existing appointment, accounting for buffer and travel times. For each existing appointment, the blocked window is: [appt_start - travel_time_to, appt_end + BUFFER_MINUTES + travel_time_from] Args: existing_appointments: List of appointment dicts for the provider on that date proposed_start: "HH:MM" format proposed_end: "HH:MM" format travel_time_to: Minutes of travel to the existing appointment travel_time_from: Minutes of travel from the existing appointment """ buffer = settings.get_buffer_minutes() for appt in existing_appointments: blocked_start = add_minutes(appt["start_time"], -travel_time_to) blocked_end = add_minutes(appt["end_time"], buffer + travel_time_from) if times_overlap(proposed_start, proposed_end, blocked_start, blocked_end): return True return False def is_oncall_for_slot(provider: dict, date: str, start_time: str, end_time: str) -> bool: """Return True if the slot [date+start_time, date+end_time] falls within one of the provider's 24-hour on-call windows (07:00 of an on-call date → 07:00 the next day).""" on_call_dates = provider.get("on_call", []) or [] if not on_call_dates: return False try: slot_start = datetime.strptime(f"{date} {start_time}", "%Y-%m-%d %H:%M") slot_end = datetime.strptime(f"{date} {end_time}", "%Y-%m-%d %H:%M") except ValueError: return False on_call_start_hour = settings.get_on_call_start_hour() for d in on_call_dates: try: window_start = datetime.strptime(f"{d} {on_call_start_hour}", "%Y-%m-%d %H:%M") except ValueError: continue window_end = window_start + timedelta(hours=24) if slot_start >= window_start and slot_end <= window_end: return True return False def is_provider_available(provider: dict, date: str, start_time: str, visit_level: int | str, existing_appointments: list[dict]) -> bool: """Master check: is this provider available for a visit at the given date/time? Checks: 1. Provider is active 2. Date is not in provider's days_off 3. Date's weekday is not in provider's recurring_days_off 4. Provider supports the visit_level 5. Slot is within work hours 6. Slot doesn't conflict with lunch 7. No overlap with existing appointments (including buffer) 8. Meets 24-hour lead time requirement Returns True if all checks pass. """ # 1. Active if not provider.get("active", True): return False duration = get_visit_duration(visit_level) end_time = add_minutes(start_time, duration) # On-call window supersedes days off, recurring days off, work hours, # and lunch break — but NOT the lead-time rule, which always applies. on_call = is_oncall_for_slot(provider, date, start_time, end_time) if not on_call: # 2. Specific date/time off for off in provider.get("days_off", []): if isinstance(off, str): # Legacy format: plain date string = all day off if date == off: return False elif isinstance(off, dict) and off.get("date") == date: if off.get("all_day", True): return False # Specific time block off — check overlap off_start = off.get("start_time") off_end = off.get("end_time") if off_start and off_end and times_overlap(start_time, end_time, off_start, off_end): return False # 3. Recurring day off (e.g., every Friday) weekday = datetime.strptime(date, "%Y-%m-%d").strftime("%A") if weekday in provider.get("recurring_days_off", []): return False # 4. Supports the requested visit level if visit_level not in provider.get("service_levels", []): return False if not on_call: # 5. Within work hours if not is_within_work_hours(provider, start_time, end_time): return False # 6. No lunch conflict if conflicts_with_lunch(provider, start_time, end_time): return False # 7. No overlap with existing appointments if has_overlap(existing_appointments, start_time, end_time): return False # 8. Lead-time rule — always applies, even for on-call providers. # Compare in the configured timezone so a UTC server doesn't mis-evaluate local times. proposed_dt = datetime.strptime(f"{date} {start_time}", "%Y-%m-%d %H:%M") now_local = datetime.now(ZoneInfo(settings.get_timezone())).replace(tzinfo=None) if proposed_dt < now_local + timedelta(hours=settings.get_min_lead_time_hours()): return False return True def get_available_providers(all_providers: list[dict], date: str, start_time: str, visit_level: int | str, appointments_by_provider: dict[str, list[dict]]) -> list[dict]: """Filter all providers down to those available for the proposed visit. Args: all_providers: List of all provider dicts date: "YYYY-MM-DD" start_time: "HH:MM" visit_level: Visit level (1, 2, 3, or "Nurse") appointments_by_provider: Dict mapping provider_id -> list of their appointments on that date Returns: List of available provider dicts """ available = [] for provider in all_providers: existing = appointments_by_provider.get(provider["id"], []) if is_provider_available(provider, date, start_time, visit_level, existing): available.append(provider) return available def rank_providers_by_travel(providers_with_travel: list[dict]) -> list[dict]: """Sort providers by travel time (ascending). Boost providers who already have visits in the same area. Each dict in the input list should have: - All standard provider fields - 'travel_time_minutes': int (travel time to the patient) - 'has_nearby_visits': bool (True if provider has other visits nearby that day) Ranking logic: - Primary sort: is_oncall (True first — travel time is ignored) - Secondary sort: has_nearby_visits (True first) - Tertiary sort: travel_time_minutes (ascending) Returns the sorted list. """ return sorted( providers_with_travel, key=lambda p: ( not p.get("is_oncall", False), not p.get("has_nearby_visits", False), p.get("travel_time_minutes", 0), ), )