-
Notifications
You must be signed in to change notification settings - Fork 3
/
Copy pathtask_manager.py
112 lines (90 loc) · 4.87 KB
/
task_manager.py
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
from utils.task_locations import Task
from utils.logger import logger
import random
from villager import Villager,Werewolf,Player
from utils.logger import logger
class TaskManager:
def __init__(self) -> None:
self.tasks = self.initialize_task_locations()
def initialize_task_locations(self):
return [
Task(1150, 625, "Gather food",5),
Task(1200, 400, "Build a house",10),
Task(880, 625, "Collect wood",8),
Task(60, 400, "Fetch water",10),
Task(880, 400, "Guard the village",15),
Task(360, 625, "Cook food",10),
Task(80, 200, "Hunt for animals",15),
Task(600, 400, "Scout the area",10),
Task(360, 400, "Heal the injured",12),
Task(720, 200, "Teach children",10)
]
def completed_tasks(self):
return [task for task in self.tasks if task.completed]
def incomplete_tasks(self):
return [task for task in self.tasks if not task.completed]
def all_tasks_completed(self):
return all(task.completed for task in self.tasks)
def update_tasks(self, player):
"""
Update tasks based on player proximity.
Parameters:
player (Player): The player object.
Returns:
None
"""
for task in self.tasks:
if player.distance_to_task(task.x, task.y) < 20:
if not player.is_werewolf:
task.complete()
else:
task.sabotage()
def assign_first_task(villagers, task_locations,complete,incomplete):
for i in range(len(villagers)):
villager = villagers[i]
if isinstance(villager,Werewolf):
if complete:
task = random.choice(complete)
task_name = task.task
else:
task= random.choice(incomplete)
task_name = task.task
task_location = [loc for loc in task_locations if loc.task == task_name][0]
task_time = task_location.task_period # Time required for the task
task_sabotage_function = task_location.sabotage
villager.assign_task(task_name, task_location, task_time, task_sabotage_function)
logger.debug(f"Sabotage task '{task_name}' to {villager.agent_id} at location ({task_location.x}, {task_location.y})")
else:
task = random.choice(incomplete)
task_name = task.task
task_location = [loc for loc in task_locations if loc.task == task_name][0]
task_time = task_location.task_period
task_complete_function = task_location.complete
villager.assign_task(task_name, task_location, task_time, task_complete_function)
logger.debug(f"Assigned task '{task_name}' to {villager.agent_id} at location ({task_location.x}, {task_location.y})")
def assign_next_task(villager, task_locations,previous_task):
# task_location = random.choice(task_locations)
# task_name = task_location.task
# villager.assign_task(task_name, task_location, task_location.task_period, task_location.complete)
# logger.debug(f"Assigned task '{task_name}' to {villager.agent_id} at location ({task_location.x}, {task_location.y})")
# return task_name, task_location
if len(task_locations) == 0:
tm = TaskManager()
task_locations = tm.initialize_task_locations()
try:
_,response = villager.agent.generate_reaction(observation=f"only assign one task from the following:{[loc.task for loc in task_locations]} other than {previous_task}",call_to_action_template="What should be the next task for "+villager.agent_id+f"? Expecting the response to be in the format Task: <task_name>. Do not assign other than from the given list. only assign one task from the following:{[loc.task for loc in task_locations]} ")
print(response)
task_name = response.strip().split(':')[1].strip()
task_location = [loc for loc in task_locations if loc.task == task_name][0]
task_time = task_location.task_period # Time required for the task
task_complete_function = task_location.complete
# villager.assign_task(task_name, task_location,task_time, task_complete_function)
logger.debug(f"Assigned task '{task_name}' to {villager.agent_id} at location ({task_location.x}, {task_location.y})")
except Exception as e:
logger.error(f"Error assigning task to {villager.agent_id}: {e}")
default_task_location = random.choice(task_locations)
# default_task_time = default_task_location.task_period
# task_complete_function = default_task_location.complete
# villager.assign_task(default_task_location.task, default_task_location,default_task_time,task_complete_function )
return default_task_location.task, default_task_location
return task_name, task_location