magic-wormhole/src/wormhole/_input.py

from __future__ import print_function, absolute_import, unicode_literals
from zope.interface import implementer
from attr import attrs, attrib
from attr.validators import provides
from twisted.internet import defer
from automat import MethodicalMachine
from . import _interfaces, errors

def first(outputs):
    return list(outputs)[0]

@attrs
@implementer(_interfaces.IInput)
class Input(object):
    _timing = attrib(validator=provides(_interfaces.ITiming))
    m = MethodicalMachine()
    set_trace = getattr(m, "_setTrace", lambda self, f: None)

    def __attrs_post_init__(self):
        self._all_nameplates = set()
        self._nameplate = None
        self._wordlist = None
        self._wordlist_waiters = []

    def wire(self, code, lister):
        self._C = _interfaces.ICode(code)
        self._L = _interfaces.ILister(lister)

    def when_wordlist_is_available(self):
        if self._wordlist:
            return defer.succeed(None)
        d = defer.Deferred()
        self._wordlist_waiters.append(d)
        return d

    @m.state(initial=True)
    def S0_idle(self): pass # pragma: no cover
    @m.state()
    def S1_typing_nameplate(self): pass # pragma: no cover
    @m.state()
    def S2_typing_code_no_wordlist(self): pass # pragma: no cover
    @m.state()
    def S3_typing_code_yes_wordlist(self): pass # pragma: no cover
    @m.state(terminal=True)
    def S4_done(self): pass # pragma: no cover

    # from Code
    @m.input()
    def start(self): pass

    # from Lister
    @m.input()
    def got_nameplates(self, all_nameplates): pass

    # from Nameplate
    @m.input()
    def got_wordlist(self, wordlist): pass

    # API provided to app as ICodeInputHelper
    @m.input()
    def refresh_nameplates(self): pass
    @m.input()
    def get_nameplate_completions(self, prefix): pass
    @m.input()
    def choose_nameplate(self, nameplate): pass
    @m.input()
    def get_word_completions(self, prefix): pass
    @m.input()
    def choose_words(self, words): pass

    @m.output()
    def do_start(self):
        self._start_timing = self._timing.add("input code", waiting="user")
        self._L.refresh()
        return Helper(self)
    @m.output()
    def do_refresh(self):
        self._L.refresh()
    @m.output()
    def record_nameplates(self, all_nameplates):
        # we get a set of nameplate id strings
        self._all_nameplates = all_nameplates
    @m.output()
    def _get_nameplate_completions(self, prefix):
        completions = set()
        for nameplate in self._all_nameplates:
            if nameplate.startswith(prefix):
                # TODO: it's a little weird that Input is responsible for the
                # hyphen on nameplates, but WordList owns it for words
                completions.add(nameplate+"-")
        return completions
    @m.output()
    def record_all_nameplates(self, nameplate):
        self._nameplate = nameplate
        self._C.got_nameplate(nameplate)
    @m.output()
    def record_wordlist(self, wordlist):
        from ._rlcompleter import debug
        debug("  -record_wordlist")
        self._wordlist = wordlist
    @m.output()
    def notify_wordlist_waiters(self, wordlist):
        while self._wordlist_waiters:
            d = self._wordlist_waiters.pop()
            d.callback(None)

    @m.output()
    def no_word_completions(self, prefix):
        return set()
    @m.output()
    def _get_word_completions(self, prefix):
        assert self._wordlist
        return self._wordlist.get_completions(prefix)

    @m.output()
    def raise_must_choose_nameplate1(self, prefix):
        raise errors.MustChooseNameplateFirstError()
    @m.output()
    def raise_must_choose_nameplate2(self, words):
        raise errors.MustChooseNameplateFirstError()
    @m.output()
    def raise_already_chose_nameplate1(self):
        raise errors.AlreadyChoseNameplateError()
    @m.output()
    def raise_already_chose_nameplate2(self, prefix):
        raise errors.AlreadyChoseNameplateError()
    @m.output()
    def raise_already_chose_nameplate3(self, nameplate):
        raise errors.AlreadyChoseNameplateError()
    @m.output()
    def raise_already_chose_words1(self, prefix):
        raise errors.AlreadyChoseWordsError()
    @m.output()
    def raise_already_chose_words2(self, words):
        raise errors.AlreadyChoseWordsError()

    @m.output()
    def do_words(self, words):
        code = self._nameplate + "-" + words
        self._start_timing.finish()
        self._C.finished_input(code)

    S0_idle.upon(start, enter=S1_typing_nameplate,
                 outputs=[do_start], collector=first)
    # wormholes that don't use input_code (i.e. they use allocate_code or
    # generate_code) will never start() us, but Nameplate will give us a
    # wordlist anyways (as soon as the nameplate is claimed), so handle it.
    S0_idle.upon(got_wordlist, enter=S0_idle, outputs=[record_wordlist,
                                                       notify_wordlist_waiters])
    S1_typing_nameplate.upon(got_nameplates, enter=S1_typing_nameplate,
                             outputs=[record_nameplates])
    # but wormholes that *do* use input_code should not get got_wordlist
    # until after we tell Code that we got_nameplate, which is the earliest
    # it can be claimed
    S1_typing_nameplate.upon(refresh_nameplates, enter=S1_typing_nameplate,
                             outputs=[do_refresh])
    S1_typing_nameplate.upon(get_nameplate_completions,
                             enter=S1_typing_nameplate,
                             outputs=[_get_nameplate_completions],
                             collector=first)
    S1_typing_nameplate.upon(choose_nameplate, enter=S2_typing_code_no_wordlist,
                             outputs=[record_all_nameplates])
    S1_typing_nameplate.upon(get_word_completions,
                             enter=S1_typing_nameplate,
                             outputs=[raise_must_choose_nameplate1])
    S1_typing_nameplate.upon(choose_words, enter=S1_typing_nameplate,
                             outputs=[raise_must_choose_nameplate2])

    S2_typing_code_no_wordlist.upon(got_nameplates,
                                    enter=S2_typing_code_no_wordlist, outputs=[])
    S2_typing_code_no_wordlist.upon(got_wordlist,
                                    enter=S3_typing_code_yes_wordlist,
                                    outputs=[record_wordlist,
                                             notify_wordlist_waiters])
    S2_typing_code_no_wordlist.upon(refresh_nameplates,
                                    enter=S2_typing_code_no_wordlist,
                                    outputs=[raise_already_chose_nameplate1])
    S2_typing_code_no_wordlist.upon(get_nameplate_completions,
                                    enter=S2_typing_code_no_wordlist,
                                    outputs=[raise_already_chose_nameplate2])
    S2_typing_code_no_wordlist.upon(choose_nameplate,
                                    enter=S2_typing_code_no_wordlist,
                                    outputs=[raise_already_chose_nameplate3])
    S2_typing_code_no_wordlist.upon(get_word_completions,
                                    enter=S2_typing_code_no_wordlist,
                                    outputs=[no_word_completions],
                                    collector=first)
    S2_typing_code_no_wordlist.upon(choose_words, enter=S4_done,
                                    outputs=[do_words])

    S3_typing_code_yes_wordlist.upon(got_nameplates,
                                     enter=S3_typing_code_yes_wordlist,
                                     outputs=[])
    # got_wordlist: should never happen
    S3_typing_code_yes_wordlist.upon(refresh_nameplates,
                                     enter=S3_typing_code_yes_wordlist,
                                     outputs=[raise_already_chose_nameplate1])
    S3_typing_code_yes_wordlist.upon(get_nameplate_completions,
                                     enter=S3_typing_code_yes_wordlist,
                                     outputs=[raise_already_chose_nameplate2])
    S3_typing_code_yes_wordlist.upon(choose_nameplate,
                                     enter=S3_typing_code_yes_wordlist,
                                     outputs=[raise_already_chose_nameplate3])
    S3_typing_code_yes_wordlist.upon(get_word_completions,
                                     enter=S3_typing_code_yes_wordlist,
                                     outputs=[_get_word_completions],
                                     collector=first)
    S3_typing_code_yes_wordlist.upon(choose_words, enter=S4_done,
                                     outputs=[do_words])

    S4_done.upon(got_nameplates, enter=S4_done, outputs=[])
    S4_done.upon(got_wordlist, enter=S4_done, outputs=[])
    S4_done.upon(refresh_nameplates,
                 enter=S4_done,
                 outputs=[raise_already_chose_nameplate1])
    S4_done.upon(get_nameplate_completions,
                 enter=S4_done,
                 outputs=[raise_already_chose_nameplate2])
    S4_done.upon(choose_nameplate, enter=S4_done,
                 outputs=[raise_already_chose_nameplate3])
    S4_done.upon(get_word_completions, enter=S4_done,
                 outputs=[raise_already_chose_words1])
    S4_done.upon(choose_words, enter=S4_done,
                 outputs=[raise_already_chose_words2])

# we only expose the Helper to application code, not _Input
@attrs
class Helper(object):
    _input = attrib()

    def refresh_nameplates(self):
        self._input.refresh_nameplates()
    def get_nameplate_completions(self, prefix):
        return self._input.get_nameplate_completions(prefix)
    def choose_nameplate(self, nameplate):
        self._input.choose_nameplate(nameplate)
    def when_wordlist_is_available(self):
        return self._input.when_wordlist_is_available()
    def get_word_completions(self, prefix):
        return self._input.get_word_completions(prefix)
    def choose_words(self, words):
        self._input.choose_words(words)