Systemy_dialogowe/system_mockup/monitor_stanu_dialogowego.py

69 lines
2.1 KiB
Python

dialogue_state = {
# current context should be set to an act that is currently taking place
'current_context' : None,
# topics will be an array of dicts structured similar to this
# {'act': 'notifications', 'slots': [('time_when', None), ('liczba', ''), ('timeunit', None)]}
'topics': [],
'counter': 0
}
def monitor_stanu_dialogowego(frame):
global dialogue_state
# Some frames can reset the dialogue state, like saying hello.
incr_counter(dialogue_state)
# Discarding acts that were not identified properly, but lets keep increasing the counter
if frame['act'] == 'null':
return dialogue_state
reset_state_if_needed(frame)
dialogue_state['current_context'] = frame['act']
append_or_merge_frame(frame, dialogue_state)
return dialogue_state
def append_or_merge_frame(frame, dialogue_state):
act = frame['act']
act_from_state = list(filter(lambda x: x['act'] == act, dialogue_state['topics']))
if len(act_from_state) > 0:
append_or_merge_slots(frame['slots'], act_from_state[0]['slots'])
else:
dialogue_state['topics'].append(frame)
def append_or_merge_slots(frame_slots, state_slots):
"""
This function expects both parameters to be an array
of tuples (slot_name, slot_value), and will merge
frame slots into state_slots (on unique slot names).
"""
for frame_slot in frame_slots:
merged = False
for i in range(len(state_slots)):
if state_slots[i][0] == frame_slot[0]:
merged = True
# Do not merge empty incomming slots
if frame_slot[1] is not None and not len(frame_slot[1]) == 0:
state_slots[i] = frame_slot
break
if not merged:
state_slots.append(frame_slot)
def incr_counter(dialogue_state):
current_val = dialogue_state['counter']
dialogue_state['counter'] = current_val + 1
def reset_state_if_needed(frame):
global iterator
if frame['act'] == "hello" or frame['act'] == "bye":
dialogue_state.clear()
iterator = 1