""" Automated Scheduling System for Staff and Contractors """ import json import math import datetime import dataclasses import hashlib from typing import Dict, List, Optional, Any, Tuple @dataclasses.dataclass class Shift: shift_id: str staff_id: str start_time: datetime.datetime end_time: datetime.datetime location: str role: str status: str # "scheduled", "in_progress", "completed", "cancelled" hourly_rate: float overtime_eligible: bool @dataclasses.dataclass class ScheduleRequest: request_id: str requester: str required_role: str start_time: datetime.datetime end_time: datetime.datetime location: str priority: str # "low", "medium", "high", "critical" skills_required: List[str] max_hourly_rate: Optional[float] notes: str @dataclasses.dataclass class TimeOff: timeoff_id: str staff_id: str start_date: datetime.date end_date: datetime.date reason: str status: str # "pending", "approved", "rejected" approved_by: Optional[str] class SchedulingSystem: """Automated scheduling system for staff and contractors""" def __init__(self, platform): self.platform = platform self.shifts: Dict[str, Shift] = {} self.requests: Dict[str, ScheduleRequest] = {} self.time_off_requests: Dict[str, TimeOff] = {} self.schedule_patterns: Dict[str, Dict] = {} def add_staff_member(self, staff_id: str, name: str, role: str, skills: List[str], max_hours_per_week: int = 40, hourly_rate: float = 15.0) -> Dict: """Add new staff member to system""" try: # Initialize availability (7 days a week, 24 hours initially empty) availability = {} for day in range(7): day_name = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"][day] availability[day_name] = [] staff = StaffMember( staff_id=staff_id, name=name, role=role, skills=skills, availability=availability, current_assignment=None, hourly_rate=hourly_rate ) self.platform.staff[staff_id] = staff return { "success": True, "staff_id": staff_id, "name": name, "role": role, "skills": skills, "hourly_rate": hourly_rate } except Exception as e: return {"success": False, "error": str(e)} def set_staff_availability(self, staff_id: str, day: str, start_time: str, end_time: str) -> Dict: """Set staff availability for specific day""" if staff_id not in self.platform.staff: return {"error": "Staff member not found"} if day not in ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]: return {"error": "Invalid day"} # Parse times try: start = datetime.datetime.strptime(start_time, "%H:%M").time() end = datetime.datetime.strptime(end_time, "%H:%M").time() except ValueError: return {"error": "Invalid time format. Use HH:MM"} staff = self.platform.staff[staff_id] # Check for conflicts with existing shifts conflicts = self._check_availability_conflicts(staff_id, day, start, end) if conflicts: return { "error": "Availability conflicts with existing shifts", "conflicts": conflicts } # Add availability staff.availability[day].append((start, end)) return { "success": True, "staff_id": staff_id, "day": day, "availability": f"{start_time}-{end_time}" } def create_schedule_request(self, requester: str, required_role: str, start_time: datetime.datetime, end_time: datetime.datetime, location: str, priority: str = "medium", skills_required: List[str] = None, max_hourly_rate: Optional[float] = None, notes: str = "") -> Dict: """Create new schedule request""" try: request_id = self._generate_request_id() request = ScheduleRequest( request_id=request_id, requester=requester, required_role=required_role, start_time=start_time, end_time=end_time, location=location, priority=priority, skills_required=skills_required or [], max_hourly_rate=max_hourly_rate, notes=notes ) self.requests[request_id] = request return { "success": True, "request_id": request_id, "start_time": start_time.isoformat(), "end_time": end_time.isoformat(), "required_role": required_role } except Exception as e: return {"success": False, "error": str(e)} def auto_assign_shift(self, request_id: str) -> Dict: """Automatically assign best available staff to schedule request""" if request_id not in self.requests: return {"error": "Request not found"} request = self.requests[request_id] # Find available and qualified staff candidates = self._find_best_candidates(request) if not candidates: return { "error": "No qualified staff available", "request_id": request_id } # Select best candidate best_candidate = candidates[0] # Create shift shift_id = self._generate_shift_id() shift = Shift( shift_id=shift_id, staff_id=best_candidate["staff_id"], start_time=request.start_time, end_time=request.end_time, location=request.location, role=request.required_role, status="scheduled", hourly_rate=best_candidate["hourly_rate"], overtime_eligible=best_candidate["overtime_eligible"] ) self.shifts[shift_id] = shift # Update staff assignment staff = self.platform.staff[best_candidate["staff_id"]] staff.current_assignment = shift_id # Update request status request.status = "assigned" return { "success": True, "shift_id": shift_id, "staff_id": best_candidate["staff_id"], "staff_name": best_candidate["name"], "start_time": request.start_time.isoformat(), "end_time": request.end_time.isoformat(), "location": request.location, "hourly_rate": best_candidate["hourly_rate"], "match_score": best_candidate["score"] } def get_weekly_schedule(self, week_start: datetime.date) -> Dict: """Get complete schedule for a week""" week_end = week_start + datetime.timedelta(days=6) # Get all shifts for the week week_shifts = [] for shift in self.shifts.values(): if (week_start <= shift.start_time.date() <= week_end): week_shifts.append({ "shift_id": shift.shift_id, "staff_id": shift.staff_id, "staff_name": self.platform.staff[shift.staff_id].name, "role": shift.role, "start_time": shift.start_time.isoformat(), "end_time": shift.end_time.isoformat(), "location": shift.location, "status": shift.status, "duration_hours": (shift.end_time - shift.start_time).total_seconds() / 3600, "cost": shift.hourly_rate * (shift.end_time - shift.start_time).total_seconds() / 3600 }) # Group by day daily_schedule = {} for i in range(7): day = week_start + datetime.timedelta(days=i) day_shifts = [s for s in week_shifts if datetime.datetime.fromisoformat(s["start_time"]).date() == day] daily_schedule[day.isoformat()] = day_shifts # Calculate statistics total_shifts = len(week_shifts) total_hours = sum(s["duration_hours"] for s in week_shifts) total_cost = sum(s["cost"] for s in week_shifts) coverage_gaps = self._identify_coverage_gaps(week_shifts, week_start, week_end) return { "week_start": week_start.isoformat(), "week_end": week_end.isoformat(), "daily_schedule": daily_schedule, "summary": { "total_shifts": total_shifts, "total_hours": round(total_hours, 2), "total_cost": round(total_cost, 2), "average_cost_per_hour": round(total_cost / total_hours, 2) if total_hours > 0 else 0, "coverage_gaps": coverage_gaps, "staff_utilization": self._calculate_staff_utilization(week_shifts) } } def request_time_off(self, staff_id: str, start_date: datetime.date, end_date: datetime.date, reason: str) -> Dict: """Submit time off request""" if staff_id not in self.platform.staff: return {"error": "Staff member not found"} if start_date > end_date: return {"error": "End date must be after start date"} # Check for conflicts with scheduled shifts conflicts = self._check_timeoff_conflicts(staff_id, start_date, end_date) if conflicts: return { "error": "Time off conflicts with scheduled shifts", "conflicts": conflicts } timeoff_id = self._generate_timeoff_id() timeoff = TimeOff( timeoff_id=timeoff_id, staff_id=staff_id, start_date=start_date, end_date=end_date, reason=reason, status="pending", approved_by=None ) self.time_off_requests[timeoff_id] = timeoff return { "success": True, "timeoff_id": timeoff_id, "staff_id": staff_id, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "days_requested": (end_date - start_date).days + 1, "status": "pending" } def approve_time_off(self, timeoff_id: str, approved_by: str) -> Dict: """Approve time off request""" if timeoff_id not in self.time_off_requests: return {"error": "Time off request not found"} timeoff = self.time_off_requests[timeoff_id] timeoff.status = "approved" timeoff.approved_by = approved_by return { "success": True, "timeoff_id": timeoff_id, "staff_id": timeoff.staff_id, "approved_by": approved_by, "approved_at": datetime.datetime.now().isoformat() } def _find_best_candidates(self, request: ScheduleRequest) -> List[Dict]: """Find best candidates for a shift""" candidates = [] day_name = request.start_time.strftime("%A").lower() start_time = request.start_time.time() end_time = request.end_time.time() for staff_id, staff in self.platform.staff.items(): # Check role match if staff.role != request.required_role: continue # Check skills if request.skills_required: if not all(skill in staff.skills for skill in request.skills_required): continue # Check rate constraints if request.max_hourly_rate and staff.hourly_rate > request.max_hourly_rate: continue # Check availability if not self._is_staff_available(staff_id, day_name, start_time, end_time): continue # Calculate match score score = self._calculate_match_score(staff, request) candidates.append({ "staff_id": staff_id, "name": staff.name, "hourly_rate": staff.hourly_rate, "skills": staff.skills, "score": score, "overtime_eligible": True # Simplified assumption }) return sorted(candidates, key=lambda x: x["score"], reverse=True) def _is_staff_available(self, staff_id: str, day_name: str, start_time: datetime.time, end_time: datetime.time) -> bool: """Check if staff is available during the requested time""" staff = self.platform.staff[staff_id] # Check availability windows available_periods = staff.availability.get(day_name, []) for period_start, period_end in available_periods: if period_start <= start_time and end_time <= period_end: # Check for conflicts with existing shifts conflicts = self._check_shift_conflicts(staff_id, start_time, end_time) return not conflicts return False def _calculate_match_score(self, staff: StaffMember, request: ScheduleRequest) -> float: """Calculate match score for staff to request""" score = 0.0 # Base score for role match (already filtered) score += 50.0 # Skills match bonus if request.skills_required: skill_matches = sum(1 for skill in request.skills_required if skill in staff.skills) skill_score = (skill_matches / len(request.skills_required)) * 30.0 score += skill_score # Rate competitiveness (lower rate gets higher score if under max) if request.max_hourly_rate: rate_diff = request.max_hourly_rate - staff.hourly_rate rate_score = min(20.0, (rate_diff / request.max_hourly_rate) * 20.0) score += rate_score return score def _check_shift_conflicts(self, staff_id: str, start_time: datetime.time, end_time: datetime.time) -> bool: """Check for conflicts with existing shifts""" for shift in self.shifts.values(): if shift.staff_id == staff_id and shift.status in ["scheduled", "in_progress"]: # Simple time overlap check if (start_time < shift.end_time.time() and end_time > shift.start_time.time()): return True return False def _check_availability_conflicts(self, staff_id: str, day: str, start_time: datetime.time, end_time: datetime.time) -> List[str]: """Check for availability conflicts with existing shifts""" conflicts = [] for shift in self.shifts.values(): if (shift.staff_id == staff_id and shift.start_time.strftime("%A").lower() == day): if (start_time < shift.end_time.time() and end_time > shift.start_time.time()): conflicts.append(shift.shift_id) return conflicts def _check_timeoff_conflicts(self, staff_id: str, start_date: datetime.date, end_date: datetime.date) -> List[Dict]: """Check for time off conflicts with scheduled shifts""" conflicts = [] for shift in self.shifts.values(): if shift.staff_id == staff_id: shift_date = shift.start_time.date() if start_date <= shift_date <= end_date: conflicts.append({ "shift_id": shift.shift_id, "date": shift_date.isoformat(), "start_time": shift.start_time.time().isoformat(), "end_time": shift.end_time.time().isoformat() }) return conflicts def _identify_coverage_gaps(self, shifts: List[Dict], week_start: datetime.date, week_end: datetime.date) -> List[Dict]: """Identify coverage gaps in schedule""" gaps = [] required_hours = 8 # Assuming 8-hour coverage needed per day locations = set(s["location"] for s in shifts) for location in locations: for i in range(7): day = week_start + datetime.timedelta(days=i) day_shifts = [s for s in shifts if datetime.datetime.fromisoformat(s["start_time"]).date() == day and s["location"] == location] day_hours = sum(s["duration_hours"] for s in day_shifts) if day_hours < required_hours: gaps.append({ "date": day.isoformat(), "location": location, "required_hours": required_hours, "scheduled_hours": day_hours, "shortage": required_hours - day_hours }) return gaps def _calculate_staff_utilization(self, shifts: List[Dict]) -> Dict: """Calculate staff utilization metrics""" staff_hours = {} for shift in shifts: staff_id = shift["staff_id"] if staff_id not in staff_hours: staff_hours[staff_id] = 0 staff_hours[staff_id] += shift["duration_hours"] total_staff = len(staff_hours) if total_staff == 0: return {"average_hours": 0, "max_hours": 0, "min_hours": 0} hours_list = list(staff_hours.values()) return { "average_hours": sum(hours_list) / total_staff, "max_hours": max(hours_list), "min_hours": min(hours_list), "staff_count": total_staff } def _generate_request_id(self) -> str: """Generate unique request ID""" import hashlib return hashlib.md5( f"req_{datetime.datetime.now().isoformat()}_{len(self.requests)}".encode() ).hexdigest()[:10] def _generate_shift_id(self) -> str: """Generate unique shift ID""" import hashlib return hashlib.md5( f"shift_{datetime.datetime.now().isoformat()}_{len(self.shifts)}".encode() ).hexdigest()[:10] def _generate_timeoff_id(self) -> str: """Generate unique time off ID""" import hashlib return hashlib.md5( f"to_{datetime.datetime.now().isoformat()}_{len(self.time_off_requests)}".encode() ).hexdigest()[:10] # Import required classes import hashlib from supply_chain_platform import StaffMember