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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
|
# SPDX-License-Identifier: Apache-2.0
#
# Copyright (c) 2023 Malik Talha
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from agl_service_voiceagent.utils.config import get_config_value
from agl_service_voiceagent.utils.common import load_json_file, words_to_number
class Intent2VSSMapper:
"""
Intent2VSSMapper is a class that facilitates the mapping of natural language intent to
corresponding vehicle signal specifications (VSS) for automated vehicle control systems.
"""
def __init__(self):
"""
Initializes the Intent2VSSMapper class by loading Intent-to-VSS signal mappings
and VSS signal specifications from external configuration files.
"""
intents_vss_map_file = get_config_value("intents_vss_map", "Mapper")
vss_signals_spec_file = get_config_value("vss_signals_spec", "Mapper")
self.intents_vss_map = load_json_file(intents_vss_map_file).get("intents", {})
self.vss_signals_spec = load_json_file(vss_signals_spec_file).get("signals", {})
if not self.validate_signal_spec_structure():
raise ValueError("[-] Invalid VSS signal specification structure.")
def validate_signal_spec_structure(self):
"""
Validates the structure of the VSS signal specification data.
"""
signals = self.vss_signals_spec
# Iterate over each signal in the 'signals' dictionary
for signal_name, signal_data in signals.items():
# Check if the required keys are present in the signal data
if not all(key in signal_data for key in ['default_value', 'default_change_factor', 'actions', 'values', 'default_fallback', 'value_set_intents']):
print(f"[-] {signal_name}: Missing required keys in signal data.")
return False
actions = signal_data['actions']
# Check if 'actions' is a dictionary with at least one action
if not isinstance(actions, dict) or not actions:
print(f"[-] {signal_name}: Invalid 'actions' key in signal data. Must be an object with at least one action.")
return False
# Check if the actions match the allowed actions ["set", "increase", "decrease"]
for action in actions.keys():
if action not in ["set", "increase", "decrease"]:
print(f"[-] {signal_name}: Invalid action in signal data. Allowed actions: ['set', 'increase', 'decrease']")
return False
# Check if the 'synonyms' list is present for each action and is either a list or None
for action_data in actions.values():
synonyms = action_data.get('synonyms')
if synonyms is not None and (not isinstance(synonyms, list) or not all(isinstance(synonym, str) for synonym in synonyms)):
print(f"[-] {signal_name}: Invalid 'synonyms' value in signal data. Must be a list of strings.")
return False
values = signal_data['values']
# Check if 'values' is a dictionary with the required keys
if not isinstance(values, dict) or not all(key in values for key in ['ranged', 'start', 'end', 'ignore', 'additional']):
print(f"[-] {signal_name}: Invalid 'values' key in signal data. Required keys: ['ranged', 'start', 'end', 'ignore', 'additional']")
return False
# Check if 'ranged' is a boolean
if not isinstance(values['ranged'], bool):
print(f"[-] {signal_name}: Invalid 'ranged' value in signal data. Allowed values: [true, false]")
return False
default_fallback = signal_data['default_fallback']
# Check if 'default_fallback' is a boolean
if not isinstance(default_fallback, bool):
print(f"[-] {signal_name}: Invalid 'default_fallback' value in signal data. Allowed values: [true, false]")
return False
# If all checks pass, the self.vss_signals_spec structure is valid
return True
def map_intent_to_signal(self, intent_name):
"""
Maps an intent name to the corresponding VSS signals and their specifications.
Args:
intent_name (str): The name of the intent to be mapped.
Returns:
dict: A dictionary containing VSS signals as keys and their specifications as values.
"""
intent_data = self.intents_vss_map.get(intent_name, None)
result = {}
if intent_data:
signals = intent_data.get("signals", [])
for signal in signals:
signal_info = self.vss_signals_spec.get(signal, {})
if signal_info:
result.update({signal: signal_info})
return result
def parse_intent(self, intent_name, intent_slots = []):
"""
Parses an intent, extracting relevant VSS signals, actions, modifiers, and values
based on the intent and its associated slots.
Args:
intent_name (str): The name of the intent to be parsed.
intent_slots (list): A list of dictionaries representing intent slots.
Returns:
list: A list of dictionaries describing actions and signal-related details for execution.
Note:
- If no relevant VSS signals are found for the intent, an empty list is returned.
- If no specific action or modifier is determined, default values are used.
"""
vss_signal_data = self.map_intent_to_signal(intent_name)
execution_list = []
for signal_name, signal_data in vss_signal_data.items():
action = self.determine_action(signal_data, intent_slots)
modifier = self.determine_modifier(signal_data, intent_slots)
value = self.determine_value(signal_data, intent_slots)
if value != None and not self.verify_value(signal_data, value):
value = None
change_factor = signal_data["default_change_factor"]
if action in ["increase", "decrease"]:
if value and modifier == "to":
execution_list.append({"action": action, "signal": signal_name, "value": str(value)})
elif value and modifier == "by":
execution_list.append({"action": action, "signal": signal_name, "factor": str(value)})
elif value:
execution_list.append({"action": action, "signal": signal_name, "value": str(value)})
elif signal_data["default_fallback"]:
execution_list.append({"action": action, "signal": signal_name, "factor": str(change_factor)})
# if no value found set the default value
if value == None and signal_data["default_fallback"]:
value = signal_data["default_value"]
if action == "set" and value != None:
execution_list.append({"action": action, "signal": signal_name, "value": str(value)})
return execution_list
def determine_action(self, signal_data, intent_slots):
"""
Determines the action (e.g., set, increase, decrease) based on the intent slots
and VSS signal data.
Args:
signal_data (dict): The specification data for a VSS signal.
intent_slots (list): A list of dictionaries representing intent slots.
Returns:
str: The determined action or None if no action can be determined.
"""
action_res = None
for intent_slot in intent_slots:
for action, action_data in signal_data["actions"].items():
if intent_slot["name"] in action_data["intents"] and intent_slot["value"] in action_data["synonyms"]:
action_res = action
break
return action_res
def determine_modifier(self, signal_data, intent_slots):
"""
Determines the modifier (e.g., 'to' or 'by') based on the intent slots
and VSS signal data.
Args:
signal_data (dict): The specification data for a VSS signal.
intent_slots (list): A list of dictionaries representing intent slots.
Returns:
str: The determined modifier or None if no modifier can be determined.
"""
modifier_res = None
for intent_slot in intent_slots:
for _, action_data in signal_data["actions"].items():
intent_val = intent_slot["value"]
if "modifier_intents" in action_data and intent_slot["name"] in action_data["modifier_intents"] and ("to" in intent_val or "by" in intent_val):
modifier_res = "to" if "to" in intent_val else "by" if "by" in intent_val else None
break
return modifier_res
def determine_value(self, signal_data, intent_slots):
"""
Determines the value associated with the intent slot, considering the data type
and converting it to a numeric string representation if necessary.
Args:
signal_data (dict): The specification data for a VSS signal.
intent_slots (list): A list of dictionaries representing intent slots.
Returns:
str: The determined value or None if no value can be determined.
"""
result = None
for intent_slot in intent_slots:
for value, value_data in signal_data["value_set_intents"].items():
if intent_slot["name"] == value:
result = intent_slot["value"]
if value_data["datatype"] == "number":
result = words_to_number(result) # we assume our model will always return a number in words
# the value should always returned as str because Kuksa expects str values
return str(result) if result != None else None
def verify_value(self, signal_data, value):
"""
Verifies that the value is valid based on the VSS signal data.
Args:
signal_data (dict): The specification data for a VSS signal.
value (str): The value to be verified.
Returns:
bool: True if the value is valid, False otherwise.
"""
if value in signal_data["values"]["ignore"]:
return False
elif signal_data["values"]["ranged"] and isinstance(value, (int, float)):
return value >= signal_data["values"]["start"] and value <= signal_data["values"]["end"]
else:
return value in signal_data["values"]["additional"]
|
