How to use the pexpect.fdpexpect function in pexpect

def spawn_port(port_name, baudrate=115200):
    global ser
    ser = serial.serial_for_url(port_name, baudrate=baudrate)
    return fdpexpect.fdspawn(ser, 'wb', timeout=0, encoding='cp437')

def test_fd_isalive (self):
        fd = os.open ('TESTDATA.txt', os.O_RDONLY)
        s = fdpexpect.fdspawn(fd)
        assert s.isalive()
        os.close(fd)
        assert not s.isalive(), "Should not be alive after close()"

import ttbl

import pexpect
try:
    # FIXME: we don't prolly need this anymore
    from pexpect.exceptions import TIMEOUT as pexpect_TIMEOUT
    from pexpect.exceptions import EOF as pexpect_EOF
except ImportError:
    from pexpect import TIMEOUT as pexpect_TIMEOUT
    from pexpect import EOF as pexpect_EOF
try:
    import pexpect.fdpexpect
except ImportError:
    # RHEL 7 -> fdpexpect is a separate module, not a submod of pexpectg    import fdpexpect
    import fdpexpect
    pexpect.fdpexpect = fdpexpect

class impl_c(ttbl.tt_interface_impl_c):
    """
    Implementation interface for a console driver

    The target will list the available consoles in the targets'
    *consoles* tag

    :param list command_sequence: (optional) when the console is
      enabled (from :meth:`target.console.enable
      ` or when powering up
      a target that also enables the console at the same time via
      :meth:`target.power.on `),
      run a sequence of send/expect commands.

      This is commonly used when the serial line is part of a server

def get_connection(timeout, init_dtr=None, logcls=SerialLogger,
                   logfile=None, *args, **kwargs):
    if init_dtr is not None:
        kwargs['dsrdtr'] = True
    try:
        conn = serial.Serial(*args, **kwargs)
    except serial.SerialException as e:
        raise HostError(str(e))
    if init_dtr is not None:
        conn.setDTR(init_dtr)
    conn.nonblocking()
    conn.flushOutput()
    target = fdpexpect.fdspawn(conn.fileno(), timeout=timeout, logfile=logfile)
    target.logfile_read = logcls('read')
    target.logfile_send = logcls('send')

    # Monkey-patching sendline to introduce a short delay after
    # chacters are sent to the serial. If two sendline s are issued
    # one after another the second one might start putting characters
    # into the serial device before the first one has finished, causing
    # corruption. The delay prevents that.
    tsln = target.sendline

    def sendline(x):
        tsln(x)
        time.sleep(0.1)

    target.sendline = sendline
    return target, conn

def _open(self):
    """Connect to serial device and create pexpect interface."""
    assert self._fd is None
    self._fd = os.open(self._pty_path, os.O_RDWR | os.O_NONBLOCK)
    # Don't allow forked processes to access.
    fcntl.fcntl(self._fd, fcntl.F_SETFD,
                fcntl.fcntl(self._fd, fcntl.F_GETFD) | fcntl.FD_CLOEXEC)
    self._child = fdpexpect.fdspawn(self._fd)
    # pexpect defaults to a 100ms delay before sending characters, to
    # work around race conditions in ssh. We don't need this feature
    # so we'll change delaybeforesend from 0.1 to 0.001 to speed things up.
    if self._fast:
      self._child.delaybeforesend = 0.001

# overwrite an existing log file, since it's still open.
        log_counter = 0
        rc = 1
        while (rc) and (log_counter < 15):
            log_path = config.get('cache_dir') / 'log{}.log'.format(log_counter)
            cmd = 'log using `"{}"\', replace text name(stata_kernel_log)'.format(
                log_path)
            rc = self.automate('DoCommand', cmd)
            log_counter += 1
            sleep(0.1)

        if rc:
            return rc

        self.fd = Path(log_path).open()
        self.log_fd = pexpect.fdpexpect.fdspawn(
            self.fd, encoding='utf-8', maxread=9999999, codec_errors='replace')
        self.log_fd.logfile = (
            config.get('cache_dir') / 'console_debug.log').open(
                'w', encoding='utf-8')

        return 0

def ExecuteCmd(os_cmd, use_shell=False):
  """Execute os command and log results."""
  print "Executing: {}".format(os_cmd if use_shell else ' '.join(os_cmd))
  process = None
  try:
    process = subprocess.Popen(os_cmd, stdin=None, stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE, bufsize=0, shell=use_shell)
    stdout_stream = fdpexpect.fdspawn(process.stdout)
    stderr_stream = fdpexpect.fdspawn(process.stderr)
    # process.returncode is None until the subprocess completes. Then, it gets
    # filled in with the subprocess exit code.
    while process.returncode is None:
      PipePexpectStreamNonBlocking(stdout_stream, sys.stdout)
      PipePexpectStreamNonBlocking(stderr_stream, sys.stderr)
      process.poll()
    PipePexpectStreamNonBlocking(stdout_stream, sys.stdout)
    PipePexpectStreamNonBlocking(stderr_stream, sys.stderr)
    if process.returncode: # Assume a non-zero exit code means error:
      return "Unable to execute %s" % os_cmd
    return process.returncode
  except Exception, e:
    print "FAILED: %s" % e.__str__()
    raise OsCommandError()
  finally:
    # Terminate sub-process on keyboard interrupt or other exception:

def _command_sequence_run(self, target, component):
        write_file_name = os.path.join(target.state_dir,
                                       "console-%s.write" % component)
        read_file_name = os.path.join(target.state_dir,
                                      "console-%s.read" % component)
        log_file_name = os.path.join(
            target.state_dir, "console-%s.command.log" % component)
        with codecs.open(read_file_name, "r", encoding = 'utf-8') as rf, \
             open(write_file_name, "w") as wf, \
             open(log_file_name, "w+") as logf:
            timeout = self.command_timeout
            rfd = rf.fileno()
            flag = fcntl.fcntl(rfd, fcntl.F_GETFL)
            fcntl.fcntl(rfd, fcntl.F_SETFL, flag | os.O_NONBLOCK)
            expect = pexpect.fdpexpect.fdspawn(rf, logfile = logf,
                                               timeout = timeout)
            for command, response in self.command_sequence:
                if command:
                    target.log.debug(
                        "%s: writing command: %s"
                        % (component, command.encode('unicode-escape',
                                                     errors = 'replace')))
                    wf.write(command)
                if response:
                    if hasattr(response, "pattern"):
                        response_str = "(regex) " + response.pattern
                    else:
                        response_str = response
                    target.log.debug("%s: expecting response: %s"
                                     % (component, response_str))
                    self._response(target, component,

import errno
import logging
import os
import re
import socket
import time
import traceback
import types

import pexpect
try:
    import pexpect.fdpexpect
except ImportError:
    # RHEL 7 -> fdpexpect is a separate module, not a submod of pexpectg    import fdpexpect
    import fdpexpect
    pexpect.fdpexpect = fdpexpect

# FIXME: remove -> cleanup
try:
    from pexpect.exceptions import TIMEOUT as pexpect_TIMEOUT
    from pexpect.exceptions import EOF as pexpect_EOF
except ImportError:
    from pexpect import TIMEOUT as pexpect_TIMEOUT
    from pexpect import EOF as pexpect_EOF

import commonl
import ttbl
import ttbl.debug
import ttbl.images
import ttbl.power

# FIXME: rename to address_maps