-
Notifications
You must be signed in to change notification settings - Fork 1
/
job_allocator.py
124 lines (102 loc) · 4.13 KB
/
job_allocator.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
113
114
115
116
117
118
119
120
121
122
123
124
# Data:
# Tasks queue is a list
# Job represented by a dictionary
# Worker repr by dictionary
from operator import itemgetter
class Type:
def __init__(self, name, skills):
self.name = name
self.skills = skills
class Task:
def __init__(self, jobtype, weighting=1):
self.jobtype = jobtype
self.weighting = weighting
self.done = False
class Worker:
def __init__(self, skills, penalty):
self.skills = skills
self.penalty = penalty
self.task = None
self.fatigue = 1 ## not implemented
self.busy = 0 ## semaphore
def is_penalty(self, task):
return - int(self.penalty in task.jobtype.skills)
def skill_score(self, task):
## list all the matched skills
skill_match = len(set(task.jobtype.skills).intersection(self.skills))
# skill_match = set(task.jobtype.skills) & set(self.skills)
## testing if penalty exists
score = skill_match + self.is_penalty(task)
return score
def compute_duration(self, task):
duration = task.weighting * (100 - (5 * max(0, self.skill_score(task)) ** 1.5)) / (100 * (self.fatigue))
return round(duration)
class Scheduler:
def __init__(self, ):
self.fifo = []
def run_tasks(self):
for task in list(self.fifo):
## finding the best available worker
best_duration = 10000
best_worker = None
for name, worker in workers.items():
contender_duration = worker.compute_duration(task)
if None == worker.task and contender_duration < best_duration:
best_name, best_worker = name, worker
best_duration = contender_duration
if best_worker == None:
print('no worker available')
return
best_worker.task = task
best_worker.busy += best_duration
self.fifo.remove(task)
print(f"task '{task.jobtype.name}' started with \tworker '{best_name}' \tfor duration '{best_duration}'")
def step(self):
for k, v in workers.items():
v.busy = max(0, v.busy -1)
if v.busy < 1 :
v.task = None
v.fatigue = min(1, v.fatigue + 0.1)
else:
v.fatigue *= .9
self.run_tasks()
## job types
job_types = {
'music': Type('Playing music', ['creativity', 'coordination', 'drinking']),
'kicking_ball': Type('Kicking a ball', ['energy', 'coordination', 'determination']),
'software': Type('Writing Software', ['logic','determination','vision']),
'space_travel': Type('Going to Space', ['determination', 'logic', 'curiosity']),
'dig_tunnel': Type('Dig a tunnel', ['energy', 'determination', 'curiosity']),
'daydreaming': Type('Daydreaming', ['creativity', 'drinking', 'communication']),
'cold': Type('Catching a Cold', ['energy', 'drinking', 'communication']),
}
## workers
workers = {
'student': Worker(['drinking', 'curiosity', 'energy'], 'communication'),
'engineer': Worker(['logic', 'vision', 'creativity'], 'coordination'),
'lawyer': Worker(['communication', 'determination', 'vision'], 'drinking'),
'mathematician': Worker(['logic', 'vision', 'determination'], 'coordination'),
'artist': Worker(['creativity', 'coordination', 'drinking'], 'logic'),
'barista': Worker(['coordination', 'communication', 'drinking'], 'energy'),
}
## if main
if __name__ == "__main__":
from random import choice, randint
from time import sleep
task1 = Task(job_types['software'], 2)
task2 = Task(job_types['music'], 3)
scheduler = Scheduler()
N = 5
while(True):
## generate N random tasks
for _ in range(randint(0, N)):
rand_task = choice(list(job_types.keys()))
next_task = Task(job_types[rand_task], randint(1, 3))
scheduler.fifo.append(next_task)
scheduler.step()
## sleep 1 sec
sleep(1)
##################################
# from job_allocator import *
# task1 = Task(job_types['software'], 2)
# task2 = Task(job_types['music'], 3)