view skaapsteker/dialogue.py @ 286:0dbb50d07764

Poke the current state before checking to see if there is text in case the world has changed and it is time for the state machine to move on.
author Simon Cross <hodgestar@gmail.com>
date Fri, 08 Apr 2011 20:58:12 +0200
parents 71f15f6e9274
children 04be4219742b
line wrap: on
line source

import json

from . import data


class DSM(object):
    """Dialogue State Machine!

       Parameters
       ----------
       json_filename : str
           Path to file under data/ that contains JSON description
           of state machine.
       world : object
           Something to allow states to introspect the game state with.
       """

    def __init__(self, name, world, json_filename, state):
        self.state = state
        self.world = world
        self.states = AttrDict()
        src = json.loads(data.load(json_filename).read())
        for state, state_src in src.iteritems():
            pseudo_path = [json_filename, state]
            self.states[state] = DsmState(state, state_src, pseudo_path)
        assert self.state in self.states, "DSM must have start state %r" % (self.state,)

    def get_state(self):
        return self.states[self.state]

    def has_text(self):
        self.poke()
        return bool(self.states[self.state].text)

    def event(self, ev):
        my_locals = {
            "state": self.states,
            "world": self.world,
            "npcs": self.world.npcs,
        }
        my_locals.update(ev.items)
        state = self.states[self.state]
        next_state = state.event(my_locals)
        if next_state is not None and next_state.name in self.states:
            self.states[self.state].leave(my_locals)
            self.state = next_state.name
            # TODO: update self.world to reflect new state?
            self.states[self.state].enter(my_locals)

    def choice(self, i):
        self.event(DsmEvent(choice=i))

    def auto_next(self):
        self.event(DsmEvent(auto_next=True))

    def poke(self):
        # poke the current state to see if it feels like making
        # a transition.
        self.event(DsmEvent(poke=True))


class AttrDict(dict):

    def __getattr__(self, name):
        if name not in self:
            raise AttributeError("No attribute %r" % (name,))
        return self[name]


class DsmEvent(object):

    def __init__(self, choice=None, auto_next=False, poke=False):
        self.items = {
            "choice": choice,
            "auto_next": auto_next,
            "poke": poke,
        }


class DsmState(object):
    """State within a DSM.
       """

    def __init__(self, name, state_src, base_path):
        self.name = name
        self.text = state_src.get("text", None)
        self.choices = []
        self.triggers = []

        choices = state_src.get("choices", [])
        for i, choice in enumerate(choices):
            pseudo_path = base_path + ["choice-%d" % i]
            self.choices.append((i, choice["text"]))
            next_state_code = choice.get("next", None)
            if next_state_code is not None:
                self.triggers.append(
                    Trigger("choice == %d" % i, next_state_code, pseudo_path))

        events = state_src.get("events", [])
        for i, event in enumerate(events):
            pseudo_path = base_path + ["event-%d" % i]
            self.triggers.append(Trigger(event["matches"], event["next"],
                                         pseudo_path))

        auto_next = state_src.get("auto_next", None)
        self.auto_next = False
        if auto_next is not None:
            self.auto_next = True
            assert not self.choices, "%s: auto_next and choices are not compatible" % ":".join(base_path)
            pseudo_path = base_path + ["auto_next"]
            self.triggers.append(Trigger("""auto_next""", auto_next, pseudo_path))

        on_entry = state_src.get("on_entry", None)
        if on_entry is not None:
            self.on_entry = compile(on_entry,
                                    "<%s>" % ":".join(base_path + ["on_entry"]),
                                    "exec")
        else:
            self.on_entry = None

        on_exit = state_src.get("on_exit", None)
        if on_exit is not None:
            self.on_exit = compile(on_exit,
                                   "<%s>" % ":".join(base_path + ["on_exit"]),
                                   "exec")
        else:
            self.on_exit = None

    def __repr__(self):
        return "<%r name=%r>" % (self.__class__.__name__, self.name)

    def event(self, my_locals):
        for trigger in self.triggers:
            next_state = trigger.fire(my_locals)
            if next_state is not None:
                return next_state

    def enter(self, my_locals):
        if self.on_entry is not None:
            exec(self.on_entry, {}, my_locals.copy())

    def leave(self, my_locals):
        if self.on_exit is not None:
            exec(self.on_exit, {}, my_locals.copy())


class Trigger(object):
    """Matches DSM events and triggers state transitions.
       """

    def __init__(self, matches_code, next_state_code, pseudo_path):
        self._matches = compile(matches_code,
                                "<%s>" % ":".join(pseudo_path + ["match"]),
                                "eval")
        self._next_state = compile(next_state_code,
                                   "<%s>" % ":".join(pseudo_path + ["next"]),
                                   "eval")

    def fire(self, my_locals):
        if eval(self._matches, {}, my_locals.copy()):
            return eval(self._next_state, {}, my_locals.copy())
        return None