Spaces:
Paused
Paused
File size: 5,141 Bytes
2004c79 3b9a6b5 2004c79 e466dd5 e3a1efe e466dd5 3b9a6b5 e3a1efe 3b9a6b5 e3a1efe 3b9a6b5 e3a1efe 3b9a6b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
from datetime import datetime, timedelta, date
import pandas as pd
from factory.data.generators import earliest_monday_on_or_after
from constraint_solvers.timetable.working_hours import (
SLOTS_PER_WORKING_DAY,
MORNING_SLOTS,
slot_to_datetime,
)
def schedule_to_dataframe(schedule) -> pd.DataFrame:
"""
Convert an EmployeeSchedule to a pandas DataFrame.
Args:
schedule (EmployeeSchedule): The schedule to convert.
Returns:
pd.DataFrame: The converted DataFrame.
"""
data: list[dict[str, str]] = []
# Get base date from schedule info if available
base_date = None
if hasattr(schedule, "schedule_info"):
if hasattr(schedule.schedule_info, "base_date"):
base_date = schedule.schedule_info.base_date
# Process each task in the schedule
for task in schedule.tasks:
# Get employee name or "Unassigned" if no employee assigned
employee: str = task.employee.name if task.employee else "Unassigned"
# Calculate start and end times (naive local time)
start_time: datetime = slot_to_datetime(task.start_slot, base_date)
end_time: datetime = slot_to_datetime(
task.start_slot + task.duration_slots, base_date
)
# Add task data to list with availability flags
data.append(
{
"Project": getattr(task, "project_id", ""),
"Sequence": getattr(task, "sequence_number", 0),
"Employee": employee,
"Task": task.description,
"Start": start_time,
"End": end_time,
"Duration (hours)": task.duration_slots / 2, # Convert slots to hours
"Required Skill": task.required_skill,
"Pinned": getattr(task, "pinned", False), # Include pinned status
# Check if task falls on employee's unavailable date
"Unavailable": employee != "Unassigned"
and hasattr(task.employee, "unavailable_dates")
and start_time.date() in task.employee.unavailable_dates,
# Check if task falls on employee's undesired date
"Undesired": employee != "Unassigned"
and hasattr(task.employee, "undesired_dates")
and start_time.date() in task.employee.undesired_dates,
# Check if task falls on employee's desired date
"Desired": employee != "Unassigned"
and hasattr(task.employee, "desired_dates")
and start_time.date() in task.employee.desired_dates,
}
)
return pd.DataFrame(data)
def employees_to_dataframe(schedule) -> pd.DataFrame:
"""
Convert an EmployeeSchedule to a pandas DataFrame.
Args:
schedule (EmployeeSchedule): The schedule to convert.
"""
def format_dates(dates_list, max_display=3):
"""Helper function to format dates for display"""
if not dates_list:
return "None"
try:
sorted_dates = sorted(dates_list)
if len(sorted_dates) <= max_display:
return ", ".join(d.strftime("%m/%d") for d in sorted_dates)
else:
displayed = ", ".join(
d.strftime("%m/%d") for d in sorted_dates[:max_display]
)
return f"{displayed} (+{len(sorted_dates) - max_display} more)"
except Exception:
return f"{len(dates_list)} dates"
data: list[dict[str, str]] = []
for emp in schedule.employees:
try:
first, last = emp.name.split(" ", 1) if " " in emp.name else (emp.name, "")
# Safely get preference dates with fallback to empty sets
unavailable_dates = getattr(emp, "unavailable_dates", set())
undesired_dates = getattr(emp, "undesired_dates", set())
desired_dates = getattr(emp, "desired_dates", set())
data.append(
{
"First Name": first,
"Last Name": last,
"Skills": ", ".join(sorted(emp.skills)),
"Unavailable Dates": format_dates(unavailable_dates),
"Undesired Dates": format_dates(undesired_dates),
"Desired Dates": format_dates(desired_dates),
"Total Preferences": f"{len(unavailable_dates)} unavailable, {len(undesired_dates)} undesired, {len(desired_dates)} desired",
}
)
except Exception as e:
# Fallback for any employee that causes issues
data.append(
{
"First Name": str(emp.name),
"Last Name": "",
"Skills": ", ".join(sorted(getattr(emp, "skills", []))),
"Unavailable Dates": "Error loading",
"Undesired Dates": "Error loading",
"Desired Dates": "Error loading",
"Total Preferences": "Error loading preferences",
}
)
return pd.DataFrame(data)
|