hugvey/hugvey/story.py

import json
import time
import threading
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

    @classmethod
    def initFromJson(message, data, story):
        msg = message(data['@id'], data['text'])
        msg.isStart = data['start'] if 'start' in data else False
        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 Interruption(object):
    """
    An Interruption. 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(interruptionClass, data, story):
        interrupt = interruptionClass(data['@id'])
        if 'conditions' in data:
            for conditionId in data['conditions']:
                c = story.get(conditionId)
                interrupt.addCondition(c)
        return interrupt


storyClasses = {
    'Msg': Message,
    'Direction': Direction,
    'Condition': Condition,
    'Interruption': Interruption,
}


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()
    
    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)]),
            'interruptions': len([e for e in self.log if isinstance(e, Interruption)])
        }

    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.interruptions = []
        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) == Interruption:
            self.interruptions.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.
        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 start(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'):
            return 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()