keyboard.py 文件源码

python
阅读 34 收藏 0 点赞 0 评论 0

项目:HPCinstall 作者: NCAR 项目源码 文件源码
def _read_until(term, pattern, timeout):
    """
    Convenience read-until-pattern function, supporting :meth:`~.get_location`.

    :arg blessed.Terminal term: :class:`~.Terminal` instance.
    :arg float timeout: timeout period, may be set to None to indicate no
        timeout (where 0 is always returned).
    :arg str pattern: target regular expression pattern to seek.
    :rtype: tuple
    :returns: tuple in form of ``(match, str)``, *match*
        may be :class:`re.MatchObject` if pattern is discovered
        in input stream before timeout has elapsed, otherwise
        None. ``str`` is any remaining text received exclusive
        of the matching pattern).

    The reason a tuple containing non-matching data is returned, is that the
    consumer should push such data back into the input buffer by
    :meth:`~.Terminal.ungetch` if any was received.

    For example, when a user is performing rapid input keystrokes while its
    terminal emulator surreptitiously responds to this in-band sequence, we
    must ensure any such keyboard data is well-received by the next call to
    term.inkey() without delay.
    """
    stime = time.time()
    match, buf = None, u''

    # first, buffer all pending data. pexpect library provides a
    # 'searchwindowsize' attribute that limits this memory region.  We're not
    # concerned about OOM conditions: only (human) keyboard input and terminal
    # response sequences are expected.

    while True:
        # block as long as necessary to ensure at least one character is
        # received on input or remaining timeout has elapsed.
        ucs = term.inkey(timeout=_time_left(stime, timeout))
        if ucs:
            buf += ucs
            # while the keyboard buffer is "hot" (has input), we continue to
            # aggregate all awaiting data.  We do this to ensure slow I/O
            # calls do not unnecessarily give up within the first 'while' loop
            # for short timeout periods.
            while True:
                ucs = term.inkey(timeout=0)
                if not ucs:
                    break
                buf += ucs

        match = re.search(pattern=pattern, string=buf)
        if match is not None:
            # match
            break

        if timeout is not None:
            if not _time_left(stime, timeout):
                # timeout
                break

    return match, buf
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号