scheduler / api /calendar.py
umangchaudhry's picture
Upload 31 files
0d04b76 verified
Raw
History Blame Contribute Delete
8.32 kB
"""Calendar route — produces FullCalendar-compatible events from appointments + provider blocks."""
from datetime import datetime, timedelta
from fastapi import APIRouter, Depends
from core import data_manager
from core import settings_manager as settings
from core.scheduling_engine import add_minutes
from api.deps import get_current_session
router = APIRouter(prefix="/api/calendar", tags=["calendar"])
# Deterministic color palette assigned to providers in load order.
_PROVIDER_COLORS = [
"#2563eb", "#16a34a", "#dc2626", "#9333ea", "#ea580c",
"#0891b2", "#ca8a04", "#db2777", "#4f46e5", "#65a30d",
]
def _provider_color_map(providers: list[dict]) -> dict[str, str]:
return {p["id"]: _PROVIDER_COLORS[i % len(_PROVIDER_COLORS)] for i, p in enumerate(providers)}
@router.get("/events")
def get_events(
start: str | None = None,
end: str | None = None,
provider_ids: str | None = None,
show_unavailable: bool = True,
show_travel: bool = True,
show_oncall: bool = True,
session: dict = Depends(get_current_session),
):
providers = data_manager.get_all_providers(active_only=False)
color_map = _provider_color_map(providers)
selected_ids: set[str] | None = None
if session.get("user_role") == "provider":
selected_ids = {session.get("provider_id")}
elif provider_ids:
selected_ids = set(provider_ids.split(",")) if provider_ids else None
visible_providers = [p for p in providers if (selected_ids is None or p["id"] in selected_ids)]
events = []
is_provider_view = session.get("user_role") == "provider"
# Appointments — include both scheduled and completed; render completed in grey.
appointments = [a for a in data_manager.get_all_appointments() if a.get("status") in ("scheduled", "completed")]
for a in appointments:
pid = a.get("provider_id")
if selected_ids is not None and pid not in selected_ids:
continue
status = a.get("status", "scheduled")
is_completed = status == "completed"
color = "#94a3b8" if is_completed else color_map.get(pid, "#2563eb")
if is_provider_view:
title = f"Visit — {a.get('visit_level', '')}"
else:
title = f"{a.get('patient_name', '')}{a.get('provider_name', '')}"
if is_completed:
title = f"✓ {title}"
events.append({
"id": f"appt_{a['id']}",
"title": title,
"start": f"{a['date']}T{a['start_time']}:00",
"end": f"{a['date']}T{a['end_time']}:00",
"backgroundColor": color,
"borderColor": color,
"classNames": ["event-completed"] if is_completed else [],
"extendedProps": {
"type": "appointment",
"status": status,
"appointment_id": a["id"],
"patient_name": a.get("patient_name", ""),
"provider_name": a.get("provider_name", ""),
"provider_id": pid,
"visit_level": str(a.get("visit_level", "")),
"visit_address": a.get("visit_address", ""),
"recommended_departure_time": a.get("recommended_departure_time", ""),
"start_time": a.get("start_time", ""),
"end_time": a.get("end_time", ""),
"date": a.get("date", ""),
"completed_at": a.get("completed_at", ""),
},
})
# Travel block (30 min before, if departure time exists) — skip for completed visits.
if show_travel and not is_completed:
departure = a.get("recommended_departure_time")
if departure:
events.append({
"id": f"travel_{a['id']}",
"title": f"Travel to {a.get('patient_name', 'visit')}",
"start": f"{a['date']}T{departure}:00",
"end": f"{a['date']}T{a['start_time']}:00",
"backgroundColor": color,
"borderColor": color,
"classNames": ["event-travel"],
"extendedProps": {
"type": "travel",
"provider_id": pid,
"provider_name": a.get("provider_name", ""),
},
})
# Days-off blocks
if show_unavailable:
for p in visible_providers:
for off in p.get("days_off", []) or []:
if isinstance(off, str):
off = {"date": off, "all_day": True, "title": "Day Off"}
d = off.get("date")
if not d:
continue
title = off.get("title", "Unavailable")
if off.get("all_day", True):
start_iso = f"{d}T00:00:00"
end_iso = f"{d}T23:59:59"
else:
start_iso = f"{d}T{off.get('start_time','00:00')}:00"
end_iso = f"{d}T{off.get('end_time','23:59')}:00"
events.append({
"id": f"off_{p['id']}_{d}_{title}",
"title": f"{p['name']}{title}",
"start": start_iso,
"end": end_iso,
"backgroundColor": "#fecaca",
"borderColor": "#dc2626",
"textColor": "#7f1d1d",
"extendedProps": {
"type": "days_off",
"provider_id": p["id"],
"provider_name": p["name"],
"title": title,
},
})
# Recurring weekdays — render as background events for whole date range
recurring = p.get("recurring_days_off", []) or []
if recurring and start and end:
weekday_to_num = {"Monday":0,"Tuesday":1,"Wednesday":2,"Thursday":3,"Friday":4,"Saturday":5,"Sunday":6}
try:
s_dt = datetime.strptime(start[:10], "%Y-%m-%d")
e_dt = datetime.strptime(end[:10], "%Y-%m-%d")
except ValueError:
s_dt = e_dt = None
if s_dt and e_dt:
cur = s_dt
while cur < e_dt:
if cur.strftime("%A") in recurring:
d_str = cur.strftime("%Y-%m-%d")
events.append({
"id": f"rec_{p['id']}_{d_str}",
"title": f"{p['name']} — Standing Day Off",
"start": f"{d_str}T00:00:00",
"end": f"{d_str}T23:59:59",
"display": "background",
"backgroundColor": "#e5e7eb",
"extendedProps": {"type": "recurring_off", "provider_id": p["id"]},
})
cur += timedelta(days=1)
# On-call dates highlight — 24h window from on_call_start_hour on date D
# to the same hour on D+1 (matches scheduling_engine.is_oncall_for_slot).
start_hour = settings.get_on_call_start_hour() # "HH:MM"
for d in (p.get("on_call", []) or []) if show_oncall else []:
try:
window_start = datetime.strptime(f"{d} {start_hour}", "%Y-%m-%d %H:%M")
except ValueError:
continue
window_end = window_start + timedelta(hours=24)
events.append({
"id": f"oncall_{p['id']}_{d}",
"title": f"{p['name']} — ON CALL",
"start": window_start.strftime("%Y-%m-%dT%H:%M:%S"),
"end": window_end.strftime("%Y-%m-%dT%H:%M:%S"),
"backgroundColor": "#10b981",
"borderColor": "#047857",
"textColor": "#ffffff",
"extendedProps": {"type": "on_call", "provider_id": p["id"]},
})
return {
"events": events,
"providers": [
{"id": p["id"], "name": p["name"], "color": color_map[p["id"]]}
for p in providers
],
}