-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
230 lines (196 loc) · 8.54 KB
/
run.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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
""" This file contains the code for the argumentation model. """
import random
import pandas as pd
from mesa import Model
from mesa.time import RandomActivation
from communication.agent.CommunicatingAgent import CommunicatingAgent
from communication.message.MessageService import MessageService
from communication.message.Message import Message
from communication.message.MessagePerformative import MessagePerformative
from communication.preferences.Preferences import Preferences
from communication.preferences.CriterionValue import CriterionValue
from communication.preferences.Item import Item
from communication.preferences.CriterionName import criterion_find
from arguments.couplevalue import CoupleValue
from arguments.argument import Argument
DATASET_PATH = "weapons_dataset.csv"
class ArgumentAgent(CommunicatingAgent):
"""ArgumentAgent which inherit from CommunicatingAgent ."""
def __init__(self, unique_id, model, name, preferences):
super().__init__(unique_id, model, name)
self.preferences = preferences
class ArgumentModel(Model):
"""ArgumentModel which inherit from Model ."""
def __init__(self):
super().__init__()
self.schedule = RandomActivation(self)
self.__messages_service = MessageService(self.schedule)
self.preferences = Preferences()
self.commits = 0
# Read the dataset
dataset = pd.read_csv(DATASET_PATH, sep=";")
self.preferences.set_criterion_name_list([0, 1, 2, 3, 4, 5])
# Add values
self.items = []
for _, row in dataset.iterrows():
new_item = Item(row["WEAPON"], "")
self.items.append(new_item)
for column in dataset.columns:
if column != "WEAPON":
self.preferences.add_criterion_value(
CriterionValue(
new_item,
criterion_find(column),
row[column],
)
)
# Create agents
agent1 = ArgumentAgent(1, self, "A", self.preferences)
self.schedule.add(agent1)
agent2 = ArgumentAgent(2, self, "B", self.preferences)
self.schedule.add(agent2)
self.running = True
self.__messages_service.send_message(
Message("A", "B", MessagePerformative.PROPOSE, random.choice(self.items))
)
def support_proposal(self, item):
"""
Used when the agent receives " ASK_WHY " after having proposed an item
: param item : str - name of the item which was proposed
: return : string - the strongest supportive argument
"""
argument = Argument(True, item)
premisses = argument.list_supporting_proposal(item, self.preferences)
return ["Because", item, random.choice(premisses)]
def counter_proposal(self, proposed_item: Item, couple_value: CoupleValue):
"""
Find a counter proposal to the given item
- Try to find an item with a better value for the given criterion
- If not, consider a better criterion
- If not, propose a random item
"""
criterion = couple_value.criterion_name
value = couple_value.value
# Find an item with a better value for the given criterion
for item in self.items:
item_value = self.preferences.get_value(item, criterion)
if item_value > value:
return [
"Found better item for this criterion",
item,
CoupleValue(criterion, item_value),
]
# Consider a better criterion
best_criterion = None
for criterion in self.preferences.get_criterion_name_list():
if criterion != couple_value.criterion_name:
best_criterion = criterion
else:
break
# Find an item with a better value for the best criterion
if best_criterion is not None:
best_criterion_value = self.preferences.get_value(
proposed_item, best_criterion
)
for item in self.items:
item_value = self.preferences.get_value(item, best_criterion)
if item_value > best_criterion_value:
return [
"The criterion is not the most important one",
item,
CoupleValue(best_criterion, item_value),
]
# Else random item
item = random.choice(self.items)
return [
"The item is not satisfying, how about...",
item,
self.support_proposal(item)[2],
]
def step(self):
self.__messages_service.dispatch_messages()
self.schedule.step()
for agent in self.schedule.agents:
unread_messages = agent.get_new_messages()
for message in unread_messages:
# Parse the message
sender = message.get_exp()
receiver = message.get_dest()
performative = message.get_performative()
content = message.get_content()
# Print the message
print(f"Message from {sender}: {performative}")
if isinstance(content, list):
for element in content:
print(element)
else:
print(content)
print("")
# If PROPOSE, send ACCEPT or ASK_WHY
if performative == MessagePerformative.PROPOSE:
if self.preferences.is_item_among_top_10_percent(
content, self.items
):
self.__messages_service.send_message(
Message(
receiver, sender, MessagePerformative.ACCEPT, content
)
)
else:
self.__messages_service.send_message(
Message(
receiver, sender, MessagePerformative.ASK_WHY, content
)
)
# If ACCEPT, send COMMIT
elif performative == MessagePerformative.ACCEPT:
# Send COMMIT message
self.__messages_service.send_message(
Message(receiver, sender, MessagePerformative.COMMIT, content)
)
self.commits += 1
# If COMMIT, send COMMIT and stop the simulation
elif performative == MessagePerformative.COMMIT:
if self.commits == 1:
self.__messages_service.send_message(
Message(
receiver, sender, MessagePerformative.COMMIT, content
)
)
self.commits += 1
elif self.commits == 2:
print(f"Commitment reached for {content} !\n")
self.running = False
# If ASK_WHY, send ARGUE
elif performative == MessagePerformative.ASK_WHY:
self.__messages_service.send_message(
Message(
receiver,
sender,
MessagePerformative.ARGUE,
self.support_proposal(content),
)
)
# If ARGUE, send ACCEPT or ARGUE
elif performative == MessagePerformative.ARGUE:
[comment, item, couple_value] = content
if self.preferences.is_item_among_top_10_percent(item, self.items):
self.__messages_service.send_message(
Message(receiver, sender, MessagePerformative.ACCEPT, item)
)
else:
self.__messages_service.send_message(
Message(
receiver,
sender,
MessagePerformative.ARGUE,
self.counter_proposal(item, couple_value),
)
)
# Run the simulation
argument_model = ArgumentModel()
for i in range(20):
if not argument_model.running:
break
print(f"Step {i}:")
argument_model.step()