449 lines
14 KiB
Python
449 lines
14 KiB
Python
import json
|
|
import time
|
|
import logging
|
|
import asyncio
|
|
|
|
|
|
logger = logging.getLogger("narrative")
|
|
|
|
|
|
class Message(object):
|
|
def __init__(self, id, text):
|
|
self.id = id
|
|
self.text = text
|
|
self.isStart = False
|
|
self.reply = None
|
|
self.audioFile= None
|
|
|
|
@classmethod
|
|
def initFromJson(message, data, story):
|
|
msg = message(data['@id'], data['text'])
|
|
msg.isStart = data['start'] if 'start' in data else False
|
|
if 'audio' in data:
|
|
msg.audioFile = data['audio']['file']
|
|
return msg
|
|
|
|
def setReply(self, text):
|
|
self.reply = text
|
|
|
|
def hasReply(self):
|
|
return self.reply is not None
|
|
|
|
def getReply(self):
|
|
if self.reply is None:
|
|
raise Exception(
|
|
"Getting reply while there is none! {0}".format(self.id))
|
|
|
|
return self.reply
|
|
|
|
|
|
class Condition(object):
|
|
"""
|
|
A condition, basic conditions are built in, custom condition can be given by
|
|
providing a custom method.
|
|
"""
|
|
|
|
def __init__(self, id):
|
|
self.id = id
|
|
self.method = None
|
|
self.vars = {}
|
|
|
|
@classmethod
|
|
def initFromJson(conditionClass, data, story):
|
|
condition = conditionClass(data['@id'])
|
|
# TODO: should Condition be subclassed?
|
|
if data['type'] == "replyContains":
|
|
condition.method = condition._hasMetReplyContains
|
|
if data['type'] == "timeout":
|
|
condition.method = condition._hasMetTimeout
|
|
|
|
if 'vars' in data:
|
|
condition.vars = data['vars']
|
|
|
|
return condition
|
|
|
|
def isMet(self, story):
|
|
"""
|
|
Validate if condition is met for the current story state
|
|
"""
|
|
return self.method(story)
|
|
|
|
def _hasMetTimeout(self, story):
|
|
now = time.time()
|
|
# check if the message already finished playing
|
|
if not story.lastMsgFinishTime:
|
|
return False
|
|
|
|
return now - story.lastMsgFinishTime >= self.vars['seconds']
|
|
|
|
def _hasMetReplyContains(self, story):
|
|
if not story.currentMessage.hasReply():
|
|
return False
|
|
|
|
if 'regex' in self.vars:
|
|
if 'regexCompiled' not in self.vars:
|
|
# Compile once, as we probably run it more than once
|
|
self.vars['regexCompiled'] = re.compile(self.vars['regex'])
|
|
result = re.match(self.vars['regexCompiled'])
|
|
if result is None:
|
|
return False
|
|
results = result.groupdict()
|
|
for captureGroup in results:
|
|
story.variables[captureGroup] = results[captureGroup]
|
|
logger.critical("Regex not implemented yet")
|
|
return False
|
|
|
|
if 'contains' in self.vars:
|
|
if self.vars['contains'] == '*':
|
|
return True
|
|
return self.vars['contains'] in story.currentMessage.getReply()
|
|
|
|
|
|
class Direction(object):
|
|
"""
|
|
A condition based edge in the story graph
|
|
"""
|
|
|
|
def __init__(self, id, msgFrom: Message, msgTo: Message):
|
|
self.id = id
|
|
self.msgFrom = msgFrom
|
|
self.msgTo = msgTo
|
|
self.conditions = []
|
|
|
|
def addCondition(self, condition: Condition):
|
|
self.conditions.append(condition)
|
|
|
|
@classmethod
|
|
def initFromJson(direction, data, story):
|
|
msgFrom = story.get(data['source'])
|
|
msgTo = story.get(data['target'])
|
|
direction = direction(data['@id'], msgFrom, msgTo)
|
|
if 'conditions' in data:
|
|
for conditionId in data['conditions']:
|
|
c = story.get(conditionId)
|
|
direction.addCondition(c)
|
|
return direction
|
|
|
|
|
|
class Diversion(object):
|
|
"""
|
|
An Diversion. Used to catch events outside of story flow.
|
|
"""
|
|
|
|
def __init__(self, id):
|
|
self.id = id
|
|
self.conditions = []
|
|
|
|
def addCondition(self, condition: Condition):
|
|
self.conditions.append(condition)
|
|
|
|
@classmethod
|
|
def initFromJson(diversionClass, data, story):
|
|
diversion = diversionClass(data['@id'])
|
|
if 'conditions' in data:
|
|
for conditionId in data['conditions']:
|
|
c = story.get(conditionId)
|
|
diversion.addCondition(c)
|
|
return diversion
|
|
|
|
|
|
storyClasses = {
|
|
'Msg': Message,
|
|
'Direction': Direction,
|
|
'Condition': Condition,
|
|
'Diversion': Diversion,
|
|
}
|
|
|
|
|
|
class Stopwatch(object):
|
|
"""
|
|
Keep track of elapsed time. Use multiple markers, but a single pause/resume button
|
|
"""
|
|
def __init__(self):
|
|
self.isRunning = asyncio.Event()
|
|
self.reset()
|
|
|
|
def getElapsed(self, since_mark='start'):
|
|
t = time.time()
|
|
if self.paused_at != 0:
|
|
pause_duration = t - self.paused_at
|
|
else:
|
|
pause_duration = 0
|
|
return t - self.marks[since_mark] - pause_duration
|
|
|
|
def pause(self):
|
|
self.paused_at = time.time()
|
|
self.isRunning.clear()
|
|
|
|
def resume(self):
|
|
if self.paused_at == 0:
|
|
return
|
|
|
|
pause_duration = time.time() - self.paused_at
|
|
for m in self.marks:
|
|
self.marks[m] += pause_duration
|
|
|
|
self.paused_at = 0
|
|
self.isRunning.set()
|
|
|
|
def reset(self):
|
|
self.marks = {}
|
|
self.setMark('start')
|
|
self.paused_at = 0
|
|
self.isRunning.set()
|
|
|
|
def setMark(self, name):
|
|
self.marks[name] = time.time()
|
|
|
|
|
|
def clearMark(self, name):
|
|
if name in self.marks:
|
|
self.marks.pop(name)
|
|
|
|
class Story(object):
|
|
"""Story represents and manages a story/narrative flow"""
|
|
# TODO should we separate 'narrative' (the graph) from the story (the
|
|
# current user flow)
|
|
|
|
def __init__(self, hugvey_state):
|
|
super(Story, self).__init__()
|
|
self.hugvey = hugvey_state
|
|
|
|
self.events = [] # queue of received events
|
|
self.commands = [] # queue of commands to send
|
|
self.log = [] # all nodes/elements that are triggered
|
|
self.currentMessage = None
|
|
self.timer = Stopwatch()
|
|
self.isRunning = False
|
|
|
|
def pause(self):
|
|
logger.debug('pause hugvey')
|
|
self.timer.pause()
|
|
|
|
def resume(self):
|
|
logger.debug('resume hugvey')
|
|
self.timer.resume()
|
|
|
|
def getStoryCounts(self):
|
|
# counts = {}
|
|
# for item in self.log:
|
|
# n =item.__class__.__name__
|
|
# if n not in counts:
|
|
# counts[n] = 0
|
|
# counts[n] += 1
|
|
# return counts
|
|
return {
|
|
'messages': len([e for e in self.log if isinstance(e, Message)]),
|
|
'diversions': len([e for e in self.log if isinstance(e, Diversion)])
|
|
}
|
|
|
|
def setStoryData(self, story_data):
|
|
"""
|
|
Parse self.data into a working story engine
|
|
"""
|
|
self.data = story_data
|
|
|
|
# keep to be able to reset it in the end
|
|
currentId = self.currentMessage.id if self.currentMessage else None
|
|
|
|
self.elements = {}
|
|
self.diversions = []
|
|
self.directionsPerMsg = {}
|
|
self.startMessage = None # The entrypoint to the graph
|
|
self.reset()
|
|
|
|
for el in self.data:
|
|
className = storyClasses[el['@type']]
|
|
obj = className.initFromJson(el, self)
|
|
self.add(obj)
|
|
|
|
logger.debug(self.elements)
|
|
logger.debug(self.directionsPerMsg)
|
|
|
|
if currentId:
|
|
self.currentMessage = self.get(currentId)
|
|
if self.currentMessage:
|
|
logger.info(
|
|
f"Reinstantiated current message: {self.currentMessage.id}")
|
|
else:
|
|
logger.warn(
|
|
"Could not reinstatiate current message. Starting over")
|
|
|
|
def reset(self):
|
|
self.timer.reset()
|
|
# self.startTime = time.time()
|
|
# currently active message, determines active listeners etc.
|
|
self.currentMessage = None
|
|
self.lastMsgTime = None
|
|
self.lastSpeechStartTime = None
|
|
self.lastSpeechEndTime = None
|
|
self.variables = {} # captured variables from replies
|
|
self.finish_time = False
|
|
|
|
self.events = [] # queue of received events
|
|
self.commands = [] # queue of commands to send
|
|
self.log = [] # all nodes/elements that are triggered
|
|
|
|
def add(self, obj):
|
|
if obj.id in self.elements:
|
|
# print(obj)
|
|
raise Exception("Duplicate id for ''".format(obj.id))
|
|
|
|
if type(obj) == Message and obj.isStart:
|
|
self.startMessage = obj
|
|
|
|
self.elements[obj.id] = obj
|
|
|
|
if type(obj) == Diversion:
|
|
self.diversions.append(obj)
|
|
|
|
if type(obj) == Direction:
|
|
if obj.msgFrom.id not in self.directionsPerMsg:
|
|
self.directionsPerMsg[obj.msgFrom.id] = []
|
|
self.directionsPerMsg[obj.msgFrom.id].append(obj)
|
|
|
|
def get(self, id):
|
|
"""
|
|
Get a story element by its id
|
|
"""
|
|
if id in self.elements:
|
|
return self.elements[id]
|
|
return None
|
|
|
|
def stop(self):
|
|
logger.info("Stop Story")
|
|
if self.isRunning:
|
|
self.isRunning = False
|
|
|
|
def _processPendingEvents(self):
|
|
# Gather events:
|
|
nr = len(self.events)
|
|
for i in range(nr):
|
|
e = self.events.pop(0)
|
|
logger.info("handle '{}'".format(e))
|
|
if e['event'] == "exit":
|
|
self.stop()
|
|
if e['event'] == 'connect':
|
|
# a client connected. Shold only happen in the beginning or in case of error
|
|
# that is, until we have a 'reset' or 'start' event.
|
|
# reinitiate current message
|
|
self.setCurrentMessage(self.currentMessage)
|
|
|
|
if e['event'] == "playbackFinish":
|
|
if e['msgId'] == self.currentMessage.id:
|
|
self.lastMsgFinishTime = time.time()
|
|
|
|
if self.currentMessage.id not in self.directionsPerMsg:
|
|
logger.info("THE END!")
|
|
self.stop()
|
|
return
|
|
|
|
if e['event'] == 'speech':
|
|
# log if somebody starts speaking
|
|
# TODO: use pausing timer
|
|
if self.lastSpeechStartTime is None or self.lastSpeechStartTime < self.lastMsgTime:
|
|
self.lastSpeechStartTime = e['time']
|
|
|
|
if e['is_final']:
|
|
# final result
|
|
self.lastSpeechEndTime = e['time']
|
|
self.currentMessage.setReply(e['transcript'])
|
|
|
|
def _processDirections(self, directions):
|
|
for direction in directions:
|
|
for condition in direction.conditions:
|
|
if condition.isMet(self):
|
|
logger.info("Condition is met: {0}, going to {1}".format(
|
|
condition.id, direction.msgTo.id))
|
|
self.log.append(condition)
|
|
self.log.append(direction)
|
|
self.setCurrentMessage(direction.msgTo)
|
|
return direction
|
|
|
|
async def _renderer(self):
|
|
"""
|
|
every 1/10 sec. determine what needs to be done based on the current story state
|
|
"""
|
|
loopDuration = 0.1 # Configure fps
|
|
lastTime = time.time()
|
|
logger.info("Start renderer")
|
|
while self.isRunning:
|
|
if self.isRunning is False:
|
|
break
|
|
|
|
# pause on timer paused
|
|
await self.timer.isRunning.wait() # wait for un-pause
|
|
|
|
for i in range(len(self.events)):
|
|
self._processPendingEvents()
|
|
|
|
if self.currentMessage.id not in self.directionsPerMsg:
|
|
self.finish()
|
|
|
|
directions = self.getCurrentDirections()
|
|
self._processDirections(directions)
|
|
|
|
# TODO create timer event
|
|
# self.commands.append({'msg':'TEST!'})
|
|
|
|
# wait for next iteration to avoid too high CPU
|
|
t = time.time()
|
|
await asyncio.sleep(max(0, loopDuration - (t - lastTime)))
|
|
lastTime = t
|
|
|
|
logger.info("Stop renderer")
|
|
|
|
def setCurrentMessage(self, message):
|
|
self.currentMessage = message
|
|
self.lastMsgTime = time.time()
|
|
self.lastMsgFinishTime = None # to be filled in by the event
|
|
|
|
logger.info("Current message: ({0}) \"{1}\"".format(
|
|
message.id, message.text))
|
|
self.log.append(message)
|
|
# TODO: prep events & timer etc.
|
|
if message.audioFile:
|
|
self.hugvey.sendCommand({
|
|
'action': 'play',
|
|
'file': message.audioFile,
|
|
'id': message.id,
|
|
})
|
|
else:
|
|
self.hugvey.sendCommand({
|
|
'action': 'play',
|
|
'msg': message.text,
|
|
'id': message.id,
|
|
})
|
|
|
|
logger.debug("Pending directions: ")
|
|
|
|
for direction in self.getCurrentDirections():
|
|
conditions = [c.id for c in direction.conditions]
|
|
logger.debug(
|
|
"- {0} -> {1} (when: {2}) ".format(direction.msgFrom.id, direction.msgTo.id, conditions))
|
|
|
|
def getCurrentDirections(self):
|
|
if self.currentMessage.id not in self.directionsPerMsg:
|
|
return []
|
|
else:
|
|
return self.directionsPerMsg[self.currentMessage.id]
|
|
|
|
async def run(self):
|
|
logger.info("Starting story")
|
|
self.timer.reset()
|
|
self.isRunning = True
|
|
self.setCurrentMessage(self.startMessage)
|
|
await self._renderer()
|
|
|
|
def isFinished(self):
|
|
if hasattr(self, 'finish_time') and self.finish_time:
|
|
return time.time() - self.finish_time
|
|
|
|
return False
|
|
|
|
def finish(self):
|
|
logger.info(f"Finished story for {self.hugvey.id}")
|
|
self.hugvey.pause()
|
|
self.finish_time = time.time()
|
|
self.timer.pause()
|
|
|