from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum
import heapq
import copy
class Location(Enum):
HOME = "Home (Boston)"
AIRPORT = "Airport"
GRANDMA_HOUSE = "Grandma's House"
class TaskType(Enum):
COOKING = "Cooking"
TRAVEL = "Travel"
PICKUP = "Pickup"
PREP = "Preparation"
@dataclass
class Task:
id: str
name: str
duration: int # minutes
task_type: TaskType
location: Location
earliest_start: datetime
latest_end: datetime
required_person: Optional[str] = None
possible_people: List[str] = field(default_factory=list)
prerequisites: List[str] = field(default_factory=list)
continuous: bool = False
@dataclass
class Person:
name: str
location: Location
available_time: datetime
can_drive: bool = True
@dataclass
class ScheduledTask:
task: Task
person: str
start_time: datetime
end_time: datetime
@dataclass
class State:
scheduled: List[ScheduledTask] = field(default_factory=list)
people: Dict[str, Person] = field(default_factory=dict)
remaining_tasks: List[str] = field(default_factory=list)
def __hash__(self):
# Simple hash based on scheduled tasks
return hash(tuple(sorted([(t.task.id, t.person, t.start_time.hour, t.start_time.minute)
for t in self.scheduled])))
def __eq__(self, other):
return self.__hash__() == other.__hash__()
@dataclass
class SearchNode:
state: State
cost: float
heuristic: float
total: float
def __lt__(self, other):
return self.total < other.total
class ThanksgivingAStar:
def __init__(self):
self.dinner_time = datetime(2023, 11, 23, 18, 0) # 6 PM
self.base_date = datetime(2023, 11, 23)
# Initialize people
self.initial_people = {
"Sarah": Person("Sarah", Location.HOME, self.base_date.replace(hour=10)),
"James": Person("James", Location.AIRPORT, self.base_date.replace(hour=13)),
"Emily": Person("Emily", Location.AIRPORT, self.base_date.replace(hour=14, minute=30), False),
"Michael": Person("Michael", Location.HOME, self.base_date.replace(hour=15)),
"Grandma": Person("Grandma", Location.GRANDMA_HOUSE, self.base_date.replace(hour=12), False)
}
# Travel times
self.travel_times = {
(Location.HOME, Location.AIRPORT): 60,
(Location.AIRPORT, Location.HOME): 60,
(Location.HOME, Location.GRANDMA_HOUSE): 30,
(Location.GRANDMA_HOUSE, Location.HOME): 30,
(Location.AIRPORT, Location.GRANDMA_HOUSE): 60,
(Location.GRANDMA_HOUSE, Location.AIRPORT): 60
}
self.tasks = self._create_tasks()
def _create_tasks(self) -> List[Task]:
"""Create all tasks that need to be scheduled"""
tasks = []
# Turkey - must start 4 hours before dinner
turkey_start = self.dinner_time - timedelta(hours=4)
tasks.append(Task(
id="turkey",
name="Cook turkey (4 hours, continuous supervision)",
duration=240, # 4 hours
task_type=TaskType.COOKING,
location=Location.HOME,
earliest_start=turkey_start,
latest_end=self.dinner_time,
required_person="Sarah",
continuous=True
))
# Side dishes - 2 hours, can overlap with turkey
tasks.append(Task(
id="sides",
name="Prepare side dishes",
duration=120, # 2 hours
task_type=TaskType.COOKING,
location=Location.HOME,
earliest_start=self.base_date.replace(hour=12),
latest_end=self.dinner_time,
possible_people=["Sarah", "Grandma"] # Grandma can help, but only after pickup
))
# James gets car and comes home
tasks.append(Task(
id="james_car",
name="James gets rental car",
duration=60, # luggage + rental
task_type=TaskType.PREP,
location=Location.AIRPORT,
earliest_start=self.base_date.replace(hour=13),
latest_end=self.base_date.replace(hour=17),
required_person="James"
))
tasks.append(Task(
id="james_home",
name="James drives home",
duration=60,
task_type=TaskType.TRAVEL,
location=Location.HOME,
earliest_start=self.base_date.replace(hour=13),
latest_end=self.base_date.replace(hour=17),
required_person="James",
prerequisites=["james_car"]
))
# Emily pickup - someone must get her
tasks.append(Task(
id="emily_pickup",
name="Pick up Emily from airport",
duration=120, # to airport + back
task_type=TaskType.PICKUP,
location=Location.HOME,
earliest_start=self.base_date.replace(hour=15), # after her landing + luggage
latest_end=self.dinner_time,
possible_people=["James", "Michael"]
))
# Grandma pickup
tasks.append(Task(
id="grandma_pickup",
name="Pick up Grandma",
duration=60, # to grandma + back
task_type=TaskType.PICKUP,
location=Location.HOME,
earliest_start=self.base_date.replace(hour=14),
latest_end=self.dinner_time,
possible_people=["James", "Michael"]
))
return tasks
def get_initial_state(self) -> State:
"""Create initial state"""
return State(
scheduled=[],
people=copy.deepcopy(self.initial_people),
remaining_tasks=[t.id for t in self.tasks]
)
def is_goal(self, state: State) -> bool:
"""Check if all tasks are scheduled"""
return len(state.remaining_tasks) == 0
def get_task_by_id(self, task_id: str) -> Task:
"""Get task by ID"""
return next(t for t in self.tasks if t.id == task_id)
def can_schedule_task(self, state: State, task_id: str, person: str, start_time: datetime) -> bool:
"""Check if a task can be scheduled"""
task = self.get_task_by_id(task_id)
# Check if person can do this task
if task.required_person and task.required_person != person:
return False
if task.possible_people and person not in task.possible_people:
return False
# Check if person can drive (for pickup tasks)
if task.task_type == TaskType.PICKUP and not state.people[person].can_drive:
return False
# Check time constraints
end_time = start_time + timedelta(minutes=task.duration)
if start_time < task.earliest_start or end_time > task.latest_end:
return False
# Check if person is available
if start_time < state.people[person].available_time:
return False
# Check prerequisites
for prereq in task.prerequisites:
if prereq in state.remaining_tasks:
return False
# Check conflicts with existing schedule
for scheduled in state.scheduled:
if scheduled.person == person:
if not (end_time <= scheduled.start_time or start_time >= scheduled.end_time):
return False
# CRITICAL FIX: Check location requirements
person_location = state.people[person].location
# For cooking tasks, person must be at the cooking location
if task.task_type == TaskType.COOKING and person_location != task.location:
return False
# Special case: If Grandma is doing a task at HOME, she must be picked up first
if person == "Grandma" and task.location == Location.HOME:
grandma_pickup_done = "grandma_pickup" not in state.remaining_tasks
if not grandma_pickup_done:
return False
# For pickup tasks, calculate if person can reach pickup location and return
if task.task_type == TaskType.PICKUP:
# Check if we have travel time to destination
if task_id == "emily_pickup":
# Person must be able to go to airport and back
if person_location != Location.HOME and person_location != Location.AIRPORT:
return False
elif task_id == "grandma_pickup":
# Person must be able to go to grandma's house and back
if person_location != Location.HOME and person_location != Location.GRANDMA_HOUSE:
return False
return True
def apply_action(self, state: State, task_id: str, person: str, start_time: datetime) -> State:
"""Apply scheduling action to create new state"""
task = self.get_task_by_id(task_id)
end_time = start_time + timedelta(minutes=task.duration)
# Create new state
new_state = State(
scheduled=state.scheduled.copy(),
people=copy.deepcopy(state.people),
remaining_tasks=[t for t in state.remaining_tasks if t != task_id]
)
# Add scheduled task
new_state.scheduled.append(ScheduledTask(task, person, start_time, end_time))
# Update person availability and location
new_state.people[person].available_time = end_time
new_state.people[person].location = task.location
# Special handling for pickup tasks - update the picked up person's location too
if task.task_type == TaskType.PICKUP:
if task_id == "emily_pickup":
new_state.people["Emily"].location = Location.HOME
new_state.people["Emily"].available_time = end_time
elif task_id == "grandma_pickup":
new_state.people["Grandma"].location = Location.HOME
new_state.people["Grandma"].available_time = end_time
return new_state
def get_successors(self, state: State) -> List[Tuple[State, float]]:
"""Get all possible successor states"""
successors = []
if not state.remaining_tasks:
return successors
# Prioritize tasks - try most constrained tasks first
remaining_task_ids = list(state.remaining_tasks)
# Sort by priority: turkey first (most constrained), then others
def task_priority(task_id):
if task_id == "turkey":
return 0 # Highest priority
elif "pickup" in task_id:
return 1 # Second priority
else:
return 2 # Lower priority
remaining_task_ids.sort(key=task_priority)
# Try scheduling the highest priority available task
for task_id in remaining_task_ids:
task = self.get_task_by_id(task_id)
# Check if prerequisites are satisfied
prereqs_satisfied = all(prereq not in state.remaining_tasks for prereq in task.prerequisites)
if not prereqs_satisfied:
continue
# Get possible people
possible_people = []
if task.required_person:
possible_people = [task.required_person]
elif task.possible_people:
possible_people = task.possible_people
else:
possible_people = list(state.people.keys())
# Try different start times (every 30 minutes)
current_time = max(task.earliest_start,
min(p.available_time for p in state.people.values()))
latest_start = task.latest_end - timedelta(minutes=task.duration)
while current_time <= latest_start:
for person in possible_people:
if self.can_schedule_task(state, task_id, person, current_time):
new_state = self.apply_action(state, task_id, person, current_time)
cost = 1.0
# Add penalties for optimization
end_time = current_time + timedelta(minutes=task.duration)
if end_time > self.dinner_time:
cost += 1000 # Heavy penalty for going over dinner time
# CRITICAL: Penalize waiting time heavily
if task_id == "emily_pickup":
emily_lands = self.base_date.replace(hour=14, minute=30)
emily_ready = emily_lands + timedelta(minutes=30) # luggage time
wait_time = max(0, (current_time - emily_ready).total_seconds() / 60)
cost += wait_time * 0.5 # Heavy penalty for making Emily wait
# Preference penalty (prefer Michael for Grandma pickup)
if task_id == "grandma_pickup" and person != "Michael":
cost += 5
# Encourage earlier completion of critical path items
if task_id in ["turkey", "emily_pickup", "grandma_pickup"]:
# Small penalty for later start times
hours_from_earliest = (current_time - task.earliest_start).total_seconds() / 3600
cost += hours_from_earliest * 0.2
# Reward efficient resource utilization
# If James is available and can do pickups, prefer him for efficiency
if task_id == "emily_pickup" and person == "James":
james_home_tasks = [t for t in state.scheduled if t.person == "James" and t.task.id == "james_home"]
if james_home_tasks:
james_available = james_home_tasks[0].end_time
if current_time >= james_available:
cost -= 1 # Small reward for using James efficiently
successors.append((new_state, cost))
current_time += timedelta(minutes=30)
# Only try one task type at a time to prevent explosion
if successors:
break
return successors
def heuristic(self, state: State) -> float:
"""Improved heuristic - estimate cost to complete remaining tasks"""
if not state.remaining_tasks:
return 0.0
h_cost = len(state.remaining_tasks) * 1.0 # Base cost per remaining task
# Heavy penalty if critical tasks aren't scheduled yet
if "turkey" in state.remaining_tasks:
h_cost += 20.0 # Turkey is critical path
if "emily_pickup" in state.remaining_tasks:
# Check how long Emily has been waiting
emily_lands = self.base_date.replace(hour=14, minute=30)
emily_ready = emily_lands + timedelta(minutes=30)
if state.scheduled:
current_time = max(task.end_time for task in state.scheduled)
wait_time = max(0, (current_time - emily_ready).total_seconds() / 60)
h_cost += wait_time * 0.3 # Penalty for Emily waiting
h_cost += 10.0 # Emily pickup is critical
# Check time pressure
if state.scheduled:
latest_end = max(task.end_time for task in state.scheduled)
time_left = (self.dinner_time - latest_end).total_seconds() / 60
avg_task_time = 90 # Average task duration
if time_left < len(state.remaining_tasks) * avg_task_time:
h_cost += 100.0 # Running out of time
# Bonus for having key logistics tasks done
logistics_done = sum(1 for task_id in ["james_home", "emily_pickup", "grandma_pickup"]
if task_id not in state.remaining_tasks)
h_cost -= logistics_done * 2.0
return h_cost
def search(self) -> Optional[State]:
"""A* search for optimal schedule"""
print("π Starting A* search...")
initial_state = self.get_initial_state()
initial_node = SearchNode(
state=initial_state,
cost=0.0,
heuristic=self.heuristic(initial_state),
total=self.heuristic(initial_state)
)
open_list = [initial_node]
closed_set = set()
best_costs = {initial_state: 0.0}
nodes_explored = 0
max_nodes = 5000 # Limit to prevent infinite search
while open_list and nodes_explored < max_nodes:
# Get node with lowest total cost
current_node = heapq.heappop(open_list)
current_state = current_node.state
if current_state in closed_set:
continue
closed_set.add(current_state)
nodes_explored += 1
# Check if goal
if self.is_goal(current_state):
print(f"β
Found solution after exploring {nodes_explored} nodes!")
return current_state
# Progress update
if nodes_explored % 100 == 0:
remaining = len(current_state.remaining_tasks)
print(f" Explored {nodes_explored} nodes, {remaining} tasks remaining...")
# Generate successors
for next_state, action_cost in self.get_successors(current_state):
if next_state in closed_set:
continue
new_cost = current_node.cost + action_cost
if next_state not in best_costs or new_cost < best_costs[next_state]:
best_costs[next_state] = new_cost
h_cost = self.heuristic(next_state)
total_cost = new_cost + h_cost
next_node = SearchNode(
state=next_state,
cost=new_cost,
heuristic=h_cost,
total=total_cost
)
heapq.heappush(open_list, next_node)
print(f"β Search completed after {nodes_explored} nodes. No complete solution found.")
# Return best partial solution
if open_list:
best_node = min(open_list, key=lambda n: len(n.state.remaining_tasks))
print(f"π Returning best partial solution with {len(best_node.state.remaining_tasks)} tasks remaining.")
return best_node.state
return None
def print_schedule(self, state: State):
"""Print the final schedule"""
if not state:
print("β No schedule generated!")
return
print("\nπ¦ THANKSGIVING SCHEDULE (A* Search Result) π¦")
print("=" * 60)
print(f"Target dinner time: {self.dinner_time.strftime('%I:%M %p')}")
if state.remaining_tasks:
print(f"β οΈ Unscheduled tasks: {', '.join(state.remaining_tasks)}")
else:
print("β
All tasks successfully scheduled!")
print("\nπ
SCHEDULE:")
print("-" * 60)
# Sort tasks by start time
sorted_tasks = sorted(state.scheduled, key=lambda x: x.start_time)
for task in sorted_tasks:
start_str = task.start_time.strftime("%I:%M %p")
end_str = task.end_time.strftime("%I:%M %p")
duration = task.task.duration
continuous_note = " [CONTINUOUS]" if task.task.continuous else ""
print(f"β° {start_str} - {end_str} ({duration} min)")
print(f" π€ {task.person}: {task.task.name}{continuous_note}")
print(f" π {task.task.location.value}")
print()
print(f"π½οΈ DINNER TIME: {self.dinner_time.strftime('%I:%M %p')}")
# Check constraints
violations = self._check_constraints(state)
if violations:
print(f"\nβ οΈ CONSTRAINT VIOLATIONS:")
for violation in violations:
print(f" β {violation}")
else:
print(f"\nβ
ALL CONSTRAINTS SATISFIED!")
def _check_constraints(self, state: State) -> List[str]:
"""Check for constraint violations"""
violations = []
# Check if all tasks are scheduled
if state.remaining_tasks:
violations.append(f"Unscheduled tasks: {', '.join(state.remaining_tasks)}")
# Check if any tasks end after dinner
for task in state.scheduled:
if task.end_time > self.dinner_time:
violations.append(f"{task.task.name} ends after dinner time")
# Check if Emily gets picked up (she can't drive)
emily_pickup = any(t.task.id == "emily_pickup" for t in state.scheduled)
if not emily_pickup and "emily_pickup" not in state.remaining_tasks:
violations.append("Emily has no pickup scheduled (she cannot drive)")
# Check turkey supervision
turkey_task = next((t for t in state.scheduled if t.task.id == "turkey"), None)
if turkey_task:
# Simplified check: Sarah must be available during turkey cooking
sarah_tasks = [t for t in state.scheduled if t.person == "Sarah"]
if not any(t.start_time <= turkey_task.start_time and t.end_time >= turkey_task.end_time
for t in sarah_tasks):
violations.append("Turkey lacks continuous supervision")
# Check location logic - Grandma can't be at HOME without being picked up first
for task in state.scheduled:
if task.person == "Grandma" and task.task.location == Location.HOME:
# Check if grandma_pickup happened before this task
grandma_pickups = [t for t in state.scheduled if t.task.id == "grandma_pickup"]
if not grandma_pickups:
violations.append(f"Grandma assigned to task at home but no pickup scheduled")
elif grandma_pickups and task.start_time <= grandma_pickups[0].end_time:
violations.append(f"Grandma task at home starts before pickup completion")
return violations
def main():
"""Run the Thanksgiving A* scheduler"""
print("π¦ Thanksgiving Dinner A* Scheduler π¦")
print("Using A* search algorithm to find optimal schedule")
print("=" * 50)
scheduler = ThanksgivingAStar()
# Run search
result = scheduler.search()
# Print results
scheduler.print_schedule(result)
if result and not result.remaining_tasks:
print("\nπ SUCCESS! Complete optimal schedule found!")
elif result:
print(f"\nβ οΈ Partial solution found. {len(result.remaining_tasks)} tasks could not be scheduled.")
print("This may indicate conflicting constraints or insufficient time.")
else:
print("\nβ No solution found. Check constraints and time requirements.")
# Run the scheduler
if __name__ == "__main__":
main()