OpenCellular/util/ec3po/console.py

#!/usr/bin/python2
# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""EC-3PO Interactive Console Interface

console provides the console interface between the user and the interpreter.  It
handles the presentation of the EC console including editing methods as well as
session-persistent command history.
"""

from __future__ import print_function

import argparse
import binascii
# pylint: disable=cros-logging-import
import logging
import multiprocessing
import os
import pty
import re
import select
import stat
import sys

import interpreter


PROMPT = '> '
CONSOLE_INPUT_LINE_SIZE = 80  # Taken from the CONFIG_* with the same name.
CONSOLE_MAX_READ = 100  # Max bytes to read at a time from the user.
LOOK_BUFFER_SIZE = 256  # Size of search window when looking for the enhanced EC
                        # image string.

# In console_init(), the EC will print a string saying that the EC console is
# enabled.  Enhanced images will print a slightly different string.  These
# regular expressions are used to determine at reboot whether the EC image is
# enhanced or not.
ENHANCED_IMAGE_RE = re.compile(r'Enhanced Console is enabled '
                               r'\(v([0-9]+\.[0-9]+\.[0-9]+)\)')
NON_ENHANCED_IMAGE_RE = re.compile(r'Console is enabled; ')

# The timeouts are really only useful for enhanced EC images, but otherwise just
# serve as a delay for non-enhanced EC images.  Therefore, we can keep this
# value small enough so that there's not too much of a delay, but long enough
# that enhanced EC images can respond in time.  Once we've detected an enhanced
# EC image, we can increase the timeout for stability just in case it takes a
# bit longer to receive an ACK for some reason.
NON_ENHANCED_EC_INTERROGATION_TIMEOUT = 0.3  # Maximum number of seconds to wait
                                             # for a response to an
                                             # interrogation of a non-enhanced
                                             # EC image.
ENHANCED_EC_INTERROGATION_TIMEOUT = 1.0  # Maximum number of seconds to wait for
                                         # a response to an interrogation of an
                                         # enhanced EC image.
INTERROGATION_MODES = ['never', 'always', 'auto']  # List of modes which control
                                                   # when interrogations are
                                                   # performed with the EC.


class EscState(object):
  """Class which contains an enumeration for states of ESC sequences."""
  ESC_START = 1
  ESC_BRACKET = 2
  ESC_BRACKET_1 = 3
  ESC_BRACKET_3 = 4
  ESC_BRACKET_8 = 5


class ControlKey(object):
  """Class which contains codes for various control keys."""
  BACKSPACE = 0x08
  CTRL_A = 0x01
  CTRL_B = 0x02
  CTRL_D = 0x04
  CTRL_E = 0x05
  CTRL_F = 0x06
  CTRL_K = 0x0b
  CTRL_N = 0xe
  CTRL_P = 0x10
  CARRIAGE_RETURN = 0x0d
  ESC = 0x1b


class Console(object):
  """Class which provides the console interface between the EC and the user.

  This class essentially represents the console interface between the user and
  the EC.  It handles all of the console editing behaviour

  Attributes:
    logger: A logger for this module.
    master_pty: File descriptor to the master side of the PTY.  Used for driving
      output to the user and receiving user input.
    user_pty: A string representing the PTY name of the served console.
    cmd_pipe: A multiprocessing.Connection object which represents the console
      side of the command pipe.  This must be a bidirectional pipe.  Console
      commands and responses utilize this pipe.
    dbg_pipe: A multiprocessing.Connection object which represents the console's
      read-only side of the debug pipe.  This must be a unidirectional pipe
      attached to the intepreter.  EC debug messages use this pipe.
    oobm_queue: A multiprocessing.Queue which is used for out of band management
      for the interactive console.
    input_buffer: A string representing the current input command.
    input_buffer_pos: An integer representing the current position in the buffer
      to insert a char.
    partial_cmd: A string representing the command entered on a line before
      pressing the up arrow keys.
    esc_state: An integer represeting the current state within an escape
      sequence.
    line_limit: An integer representing the maximum number of characters on a
      line.
    history: A list of strings containing the past entered console commands.
    history_pos: An integer representing the current history buffer position.
      This index is used to show previous commands.
    prompt: A string representing the console prompt displayed to the user.
    enhanced_ec: A boolean indicating if the EC image that we are currently
      communicating with is enhanced or not.  Enhanced EC images will support
      packed commands and host commands over the UART.  This defaults to False
      until we perform some handshaking.
    interrogation_timeout: A float representing the current maximum seconds to
      wait for a response to an interrogation.
    receiving_oobm_cmd: A boolean indicating whether or not the console is in
      the middle of receiving an out of band command.
    pending_oobm_cmd: A string containing the pending OOBM command.
    interrogation_mode: A string containing the current mode of whether
      interrogations are performed with the EC or not and how often.
  """

  def __init__(self, master_pty, user_pty, cmd_pipe, dbg_pipe):
    """Initalises a Console object with the provided arguments.

    Args:
    master_pty: File descriptor to the master side of the PTY.  Used for driving
      output to the user and receiving user input.
    user_pty: A string representing the PTY name of the served console.
    cmd_pipe: A multiprocessing.Connection object which represents the console
      side of the command pipe.  This must be a bidirectional pipe.  Console
      commands and responses utilize this pipe.
    dbg_pipe: A multiprocessing.Connection object which represents the console's
      read-only side of the debug pipe.  This must be a unidirectional pipe
      attached to the intepreter.  EC debug messages use this pipe.
    """
    logger = logging.getLogger('EC3PO.Console')
    self.logger = interpreter.LoggerAdapter(logger, {'pty': user_pty})
    self.master_pty = master_pty
    self.user_pty = user_pty
    self.cmd_pipe = cmd_pipe
    self.dbg_pipe = dbg_pipe
    self.oobm_queue = multiprocessing.Queue()
    self.input_buffer = ''
    self.input_buffer_pos = 0
    self.partial_cmd = ''
    self.esc_state = 0
    self.line_limit = CONSOLE_INPUT_LINE_SIZE
    self.history = []
    self.history_pos = 0
    self.prompt = PROMPT
    self.enhanced_ec = False
    self.interrogation_timeout = NON_ENHANCED_EC_INTERROGATION_TIMEOUT
    self.receiving_oobm_cmd = False
    self.pending_oobm_cmd = ''
    self.interrogation_mode = 'auto'
    self.look_buffer = ''

  def __str__(self):
    """Show internal state of Console object as a string."""
    string = []
    string.append('master_pty: %s' % self.master_pty)
    string.append('user_pty: %s' % self.user_pty)
    string.append('cmd_pipe: %s' % self.cmd_pipe)
    string.append('dbg_pipe: %s' % self.dbg_pipe)
    string.append('oobm_queue: %s' % self.oobm_queue)
    string.append('input_buffer: %s' % self.input_buffer)
    string.append('input_buffer_pos: %d' % self.input_buffer_pos)
    string.append('esc_state: %d' % self.esc_state)
    string.append('line_limit: %d' % self.line_limit)
    string.append('history: [\'' + '\', \''.join(self.history) + '\']')
    string.append('history_pos: %d' % self.history_pos)
    string.append('prompt: \'%s\'' % self.prompt)
    string.append('partial_cmd: \'%s\''% self.partial_cmd)
    string.append('interrogation_mode: \'%s\'' % self.interrogation_mode)
    string.append('look_buffer: \'%s\'' % self.look_buffer)
    return '\n'.join(string)

  def PrintHistory(self):
    """Print the history of entered commands."""
    fd = self.master_pty
    # Make it pretty by figuring out how wide to pad the numbers.
    wide = (len(self.history) / 10) + 1
    for i in range(len(self.history)):
      line = ' %*d %s\r\n' % (wide, i, self.history[i])
      os.write(fd, line)

  def ShowPreviousCommand(self):
    """Shows the previous command from the history list."""
    # There's nothing to do if there's no history at all.
    if not self.history:
      self.logger.debug('No history to print.')
      return

    # Don't do anything if there's no more history to show.
    if self.history_pos == 0:
      self.logger.debug('No more history to show.')
      return

    self.logger.debug('current history position: %d.', self.history_pos)

    # Decrement the history buffer position.
    self.history_pos -= 1
    self.logger.debug('new history position.: %d', self.history_pos)

    # Save the text entered on the console if any.
    if self.history_pos == len(self.history)-1:
      self.logger.debug('saving partial_cmd: \'%s\'', self.input_buffer)
      self.partial_cmd = self.input_buffer

    # Backspace the line.
    for _ in range(self.input_buffer_pos):
      self.SendBackspace()

    # Print the last entry in the history buffer.
    self.logger.debug('printing previous entry %d - %s', self.history_pos,
                      self.history[self.history_pos])
    fd = self.master_pty
    prev_cmd = self.history[self.history_pos]
    os.write(fd, prev_cmd)
    # Update the input buffer.
    self.input_buffer = prev_cmd
    self.input_buffer_pos = len(prev_cmd)

  def ShowNextCommand(self):
    """Shows the next command from the history list."""
    # Don't do anything if there's no history at all.
    if not self.history:
      self.logger.debug('History buffer is empty.')
      return

    fd = self.master_pty

    self.logger.debug('current history position: %d', self.history_pos)
    # Increment the history position.
    self.history_pos += 1

    # Restore the partial cmd.
    if self.history_pos == len(self.history):
      self.logger.debug('Restoring partial command of \'%s\'', self.partial_cmd)
      # Backspace the line.
      for _ in range(self.input_buffer_pos):
        self.SendBackspace()
      # Print the partially entered command if any.
      os.write(fd, self.partial_cmd)
      self.input_buffer = self.partial_cmd
      self.input_buffer_pos = len(self.input_buffer)
      # Now that we've printed it, clear the partial cmd storage.
      self.partial_cmd = ''
      # Reset history position.
      self.history_pos = len(self.history)
      return

    self.logger.debug('new history position: %d', self.history_pos)
    if self.history_pos > len(self.history)-1:
      self.logger.debug('No more history to show.')
      self.history_pos -= 1
      self.logger.debug('Reset history position to %d', self.history_pos)
      return

    # Backspace the line.
    for _ in range(self.input_buffer_pos):
      self.SendBackspace()

    # Print the newer entry from the history buffer.
    self.logger.debug('printing next entry %d - %s', self.history_pos,
                      self.history[self.history_pos])
    next_cmd = self.history[self.history_pos]
    os.write(fd, next_cmd)
    # Update the input buffer.
    self.input_buffer = next_cmd
    self.input_buffer_pos = len(next_cmd)
    self.logger.debug('new history position: %d.', self.history_pos)

  def SliceOutChar(self):
    """Remove a char from the line and shift everything over 1 column."""
    fd = self.master_pty
    # Remove the character at the input_buffer_pos by slicing it out.
    self.input_buffer = self.input_buffer[0:self.input_buffer_pos] + \
                        self.input_buffer[self.input_buffer_pos+1:]
    # Write the rest of the line
    moved_col = os.write(fd, self.input_buffer[self.input_buffer_pos:])
    # Write a space to clear out the last char
    moved_col += os.write(fd, ' ')
    # Update the input buffer position.
    self.input_buffer_pos += moved_col
    # Reset the cursor
    self.MoveCursor('left', moved_col)

  def HandleEsc(self, byte):
    """HandleEsc processes escape sequences.

    Args:
      byte: An integer representing the current byte in the sequence.
    """
    # We shouldn't be handling an escape sequence if we haven't seen one.
    assert self.esc_state != 0

    if self.esc_state is EscState.ESC_START:
      self.logger.debug('ESC_START')
      if byte == ord('['):
        self.esc_state = EscState.ESC_BRACKET
        return

      else:
        self.logger.error('Unexpected sequence. %c', byte)
        self.esc_state = 0

    elif self.esc_state is EscState.ESC_BRACKET:
      self.logger.debug('ESC_BRACKET')
      # Left Arrow key was pressed.
      if byte == ord('D'):
        self.logger.debug('Left arrow key pressed.')
        self.MoveCursor('left', 1)
        self.esc_state = 0 # Reset the state.
        return

      # Right Arrow key.
      elif byte == ord('C'):
        self.logger.debug('Right arrow key pressed.')
        self.MoveCursor('right', 1)
        self.esc_state = 0 # Reset the state.
        return

      # Up Arrow key.
      elif byte == ord('A'):
        self.logger.debug('Up arrow key pressed.')
        self.ShowPreviousCommand()
        # Reset the state.
        self.esc_state = 0 # Reset the state.
        return

      # Down Arrow key.
      elif byte == ord('B'):
        self.logger.debug('Down arrow key pressed.')
        self.ShowNextCommand()
        # Reset the state.
        self.esc_state = 0 # Reset the state.
        return

      # For some reason, minicom sends a 1 instead of 7. /shrug
      # TODO(aaboagye): Figure out why this happens.
      elif byte == ord('1') or byte == ord('7'):
        self.esc_state = EscState.ESC_BRACKET_1

      elif byte == ord('3'):
        self.esc_state = EscState.ESC_BRACKET_3

      elif byte == ord('8'):
        self.esc_state = EscState.ESC_BRACKET_8

      else:
        self.logger.error(r'Bad or unhandled escape sequence. got ^[%c\(%d)',
                          chr(byte), byte)
        self.esc_state = 0
        return

    elif self.esc_state is EscState.ESC_BRACKET_1:
      self.logger.debug('ESC_BRACKET_1')
      # HOME key.
      if byte == ord('~'):
        self.logger.debug('Home key pressed.')
        self.MoveCursor('left', self.input_buffer_pos)
        self.esc_state = 0 # Reset the state.
        self.logger.debug('ESC sequence complete.')
        return

    elif self.esc_state is EscState.ESC_BRACKET_3:
      self.logger.debug('ESC_BRACKET_3')
      # DEL key.
      if byte == ord('~'):
        self.logger.debug('Delete key pressed.')
        if self.input_buffer_pos != len(self.input_buffer):
          self.SliceOutChar()
        self.esc_state = 0 # Reset the state.

    elif self.esc_state is EscState.ESC_BRACKET_8:
      self.logger.debug('ESC_BRACKET_8')
      # END key.
      if byte == ord('~'):
        self.logger.debug('End key pressed.')
        self.MoveCursor('right',
                        len(self.input_buffer) - self.input_buffer_pos)
        self.esc_state = 0 # Reset the state.
        self.logger.debug('ESC sequence complete.')
        return

      else:
        self.logger.error('Unexpected sequence. %c', byte)
        self.esc_state = 0

    else:
      self.logger.error('Unexpected sequence. %c', byte)
      self.esc_state = 0

  def ProcessInput(self):
    """Captures the input determines what actions to take."""
    # There's nothing to do if the input buffer is empty.
    if len(self.input_buffer) == 0:
      return

    # Don't store 2 consecutive identical commands in the history.
    if (self.history and self.history[-1] != self.input_buffer
        or not self.history):
      self.history.append(self.input_buffer)

    # Split the command up by spaces.
    line = self.input_buffer.split(' ')
    self.logger.debug('cmd: %s', self.input_buffer)
    cmd = line[0].lower()

    # The 'history' command is a special case that we handle locally.
    if cmd == 'history':
      self.PrintHistory()
      return

    # Send the command to the interpreter.
    self.logger.debug('Sending command to interpreter.')
    self.cmd_pipe.send(self.input_buffer)

  def CheckForEnhancedECImage(self):
    """Performs an interrogation of the EC image.

    Send a SYN and expect an ACK.  If no ACK or the response is incorrect, then
    assume that the current EC image that we are talking to is not enhanced.

    Returns:
      is_enhanced: A boolean indicating whether the EC responded to the
        interrogation correctly.
    """
    # Send interrogation byte and wait for the response.
    self.logger.debug('Performing interrogation.')
    self.cmd_pipe.send(interpreter.EC_SYN)

    response = ''
    if self.dbg_pipe.poll(self.interrogation_timeout):
      response = self.dbg_pipe.recv()
      self.logger.debug('response: \'%s\'', binascii.hexlify(response))
    else:
      self.logger.debug('Timed out waiting for EC_ACK')

    # Verify the acknowledgment.
    is_enhanced = response == interpreter.EC_ACK

    if is_enhanced:
      # Increase the interrogation timeout for stability purposes.
      self.interrogation_timeout = ENHANCED_EC_INTERROGATION_TIMEOUT
      self.logger.debug('Increasing interrogation timeout to %rs.',
                        self.interrogation_timeout)
    else:
      # Reduce the timeout in order to reduce the perceivable delay.
      self.interrogation_timeout = NON_ENHANCED_EC_INTERROGATION_TIMEOUT
      self.logger.debug('Reducing interrogation timeout to %rs.',
                        self.interrogation_timeout)

    return is_enhanced

  def HandleChar(self, byte):
    """HandleChar does a certain action when it receives a character.

    Args:
      byte: An integer representing the character received from the user.
    """
    fd = self.master_pty

    # Enter the OOBM prompt mode if the user presses '%'.
    if byte == ord('%'):
      self.logger.debug('Begin OOBM command.')
      self.receiving_oobm_cmd = True
      # Print a "prompt".
      os.write(self.master_pty, '\r\n% ')
      return

    # Add chars to the pending OOBM command if we're currently receiving one.
    if self.receiving_oobm_cmd and byte != ControlKey.CARRIAGE_RETURN:
      self.pending_oobm_cmd += chr(byte)
      self.logger.debug('%s', chr(byte))
      os.write(self.master_pty, chr(byte))
      return

    if byte == ControlKey.CARRIAGE_RETURN:
      if self.receiving_oobm_cmd:
        # Terminate the command and place it in the OOBM queue.
        self.logger.debug('End OOBM command.')
        if self.pending_oobm_cmd:
          self.oobm_queue.put(self.pending_oobm_cmd)
          self.logger.debug('Placed \'%s\' into OOBM command queue.',
                            self.pending_oobm_cmd)

        # Reset the state.
        os.write(self.master_pty, '\r\n' + self.prompt)
        self.input_buffer = ''
        self.input_buffer_pos = 0
        self.receiving_oobm_cmd = False
        self.pending_oobm_cmd = ''
        return

      if self.interrogation_mode == 'never':
        self.logger.debug('Skipping interrogation because interrogation mode'
                          ' is set to never.')
      elif self.interrogation_mode == 'always':
        # Only interrogate the EC if the interrogation mode is set to 'always'.
        self.enhanced_ec = self.CheckForEnhancedECImage()
        self.logger.debug('Enhanced EC image? %r', self.enhanced_ec)

    if not self.enhanced_ec:
      # Send everything straight to the EC to handle.
      self.cmd_pipe.send(chr(byte))
      # Reset the input buffer.
      self.input_buffer = ''
      self.input_buffer_pos = 0
      self.logger.debug('Reset input buffer.')
      return

    # Keep handling the ESC sequence if we're in the middle of it.
    if self.esc_state != 0:
      self.HandleEsc(byte)
      return

    # When we're at the end of the line, we should only allow going backwards,
    # backspace, carriage return, up, or down.  The arrow keys are escape
    # sequences, so we let the escape...escape.
    if (self.input_buffer_pos >= self.line_limit and
        byte not in [ControlKey.CTRL_B, ControlKey.ESC, ControlKey.BACKSPACE,
                     ControlKey.CTRL_A, ControlKey.CARRIAGE_RETURN,
                     ControlKey.CTRL_P, ControlKey.CTRL_N]):
      return

    # If the input buffer is full we can't accept new chars.
    buffer_full = len(self.input_buffer) >= self.line_limit


    # Carriage_Return/Enter
    if byte == ControlKey.CARRIAGE_RETURN:
      self.logger.debug('Enter key pressed.')
      # Put a carriage return/newline and the print the prompt.
      os.write(fd, '\r\n')

      # TODO(aaboagye): When we control the printing of all output, print the
      # prompt AFTER printing all the output.  We can't do it yet because we
      # don't know how much is coming from the EC.

      # Print the prompt.
      os.write(fd, self.prompt)
      # Process the input.
      self.ProcessInput()
      # Now, clear the buffer.
      self.input_buffer = ''
      self.input_buffer_pos = 0
      # Reset history buffer pos.
      self.history_pos = len(self.history)
      # Clear partial command.
      self.partial_cmd = ''

    # Backspace
    elif byte == ControlKey.BACKSPACE:
      self.logger.debug('Backspace pressed.')
      if self.input_buffer_pos > 0:
        # Move left 1 column.
        self.MoveCursor('left', 1)
        # Remove the character at the input_buffer_pos by slicing it out.
        self.SliceOutChar()

      self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos)

    # Ctrl+A. Move cursor to beginning of the line
    elif byte == ControlKey.CTRL_A:
      self.logger.debug('Control+A pressed.')
      self.MoveCursor('left', self.input_buffer_pos)

    # Ctrl+B. Move cursor left 1 column.
    elif byte == ControlKey.CTRL_B:
      self.logger.debug('Control+B pressed.')
      self.MoveCursor('left', 1)

    # Ctrl+D. Delete a character.
    elif byte == ControlKey.CTRL_D:
      self.logger.debug('Control+D pressed.')
      if self.input_buffer_pos != len(self.input_buffer):
        # Remove the character by slicing it out.
        self.SliceOutChar()

    # Ctrl+E. Move cursor to end of the line.
    elif byte == ControlKey.CTRL_E:
      self.logger.debug('Control+E pressed.')
      self.MoveCursor('right',
                      len(self.input_buffer) - self.input_buffer_pos)

    # Ctrl+F. Move cursor right 1 column.
    elif byte == ControlKey.CTRL_F:
      self.logger.debug('Control+F pressed.')
      self.MoveCursor('right', 1)

    # Ctrl+K. Kill line.
    elif byte == ControlKey.CTRL_K:
      self.logger.debug('Control+K pressed.')
      self.KillLine()

    # Ctrl+N. Next line.
    elif byte == ControlKey.CTRL_N:
      self.logger.debug('Control+N pressed.')
      self.ShowNextCommand()

    # Ctrl+P. Previous line.
    elif byte == ControlKey.CTRL_P:
      self.logger.debug('Control+P pressed.')
      self.ShowPreviousCommand()

    # ESC sequence
    elif byte == ControlKey.ESC:
      # Starting an ESC sequence
      self.esc_state = EscState.ESC_START

    # Only print printable chars.
    elif IsPrintable(byte):
      # Drop the character if we're full.
      if buffer_full:
        self.logger.debug('Dropped char: %c(%d)', byte, byte)
        return
      # Print the character.
      os.write(fd, chr(byte))
      # Print the rest of the line (if any).
      extra_bytes_written = os.write(fd,
                                     self.input_buffer[self.input_buffer_pos:])

      # Recreate the input buffer.
      self.input_buffer = (self.input_buffer[0:self.input_buffer_pos] +
                           ('%c' % byte) +
                           self.input_buffer[self.input_buffer_pos:])
      # Update the input buffer position.
      self.input_buffer_pos += 1 + extra_bytes_written

      # Reset the cursor if we wrote any extra bytes.
      if extra_bytes_written:
        self.MoveCursor('left', extra_bytes_written)

      self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos)

  def MoveCursor(self, direction, count):
    """MoveCursor moves the cursor left or right by count columns.

    Args:
      direction: A string that should be either 'left' or 'right' representing
        the direction to move the cursor on the console.
      count: An integer representing how many columns the cursor should be
        moved.

    Raises:
      AssertionError: If the direction is not equal to 'left' or 'right'.
    """
    # If there's nothing to move, we're done.
    if not count:
      return
    fd = self.master_pty
    seq = '\033[' + str(count)
    if direction == 'left':
      # Bind the movement.
      if count > self.input_buffer_pos:
        count = self.input_buffer_pos
      seq += 'D'
      self.logger.debug('move cursor left %d', count)
      self.input_buffer_pos -= count

    elif direction == 'right':
      # Bind the movement.
      if (count + self.input_buffer_pos) > len(self.input_buffer):
        count = 0
      seq += 'C'
      self.logger.debug('move cursor right %d', count)
      self.input_buffer_pos += count

    else:
      raise AssertionError(('The only valid directions are \'left\' and '
                            '\'right\''))

    self.logger.debug('input_buffer_pos: %d', self.input_buffer_pos)
    # Move the cursor.
    if count != 0:
      os.write(fd, seq)

  def KillLine(self):
    """Kill the rest of the line based on the input buffer position."""
    # Killing the line is killing all the text to the right.
    diff = len(self.input_buffer) - self.input_buffer_pos
    self.logger.debug('diff: %d', diff)
    # Diff shouldn't be negative, but if it is for some reason, let's try to
    # correct the cursor.
    if diff < 0:
      self.logger.warning('Resetting input buffer position to %d...',
                          len(self.input_buffer))
      self.MoveCursor('left', -diff)
      return
    if diff:
      self.MoveCursor('right', diff)
      for _ in range(diff):
        self.SendBackspace()
      self.input_buffer_pos -= diff
      self.input_buffer = self.input_buffer[0:self.input_buffer_pos]

  def SendBackspace(self):
    """Backspace a character on the console."""
    os.write(self.master_pty, '\033[1D \033[1D')

  def ProcessOOBMQueue(self):
    """Retrieve an item from the OOBM queue and process it."""
    item = self.oobm_queue.get()
    self.logger.debug('OOBM cmd: %s', item)
    cmd = item.split(' ')

    if cmd[0] == 'loglevel':
      # An integer is required in order to set the log level.
      if len(cmd) < 2:
        self.logger.debug('Insufficient args')
        self.PrintOOBMHelp()
        return
      try:
        self.logger.debug('Log level change request.')
        new_log_level = int(cmd[1])
        self.logger.logger.setLevel(new_log_level)
        self.logger.info('Log level changed to %d.', new_log_level)

        # Forward the request to the interpreter as well.
        self.cmd_pipe.send(item)
      except ValueError:
        # Ignoring the request if an integer was not provided.
        self.PrintOOBMHelp()

    elif cmd[0] == 'interrogate' and len(cmd) >= 2:
      enhanced = False
      mode = cmd[1]
      if len(cmd) >= 3 and cmd[2] == 'enhanced':
        enhanced = True

      # Set the mode if correct.
      if mode in INTERROGATION_MODES:
        self.interrogation_mode = mode
        self.logger.debug('Updated interrogation mode to %s.', mode)

        # Update the assumptions of the EC image.
        self.enhanced_ec = enhanced
        self.logger.debug('Enhanced EC image is now %r', self.enhanced_ec)

        # Send command to interpreter as well.
        self.cmd_pipe.send('enhanced ' + str(self.enhanced_ec))
      else:
        self.PrintOOBMHelp()

    else:
      self.PrintOOBMHelp()

  def PrintOOBMHelp(self):
    """Prints out the OOBM help."""
    # Print help syntax.
    os.write(self.master_pty, '\r\n' + 'Known OOBM commands:\r\n')
    os.write(self.master_pty, '  interrogate <never | always | auto> '
             '[enhanced]\r\n')
    os.write(self.master_pty, '  loglevel <int>\r\n')

  def CheckBufferForEnhancedImage(self, data):
    """Adds data to a look buffer and checks to see for enhanced EC image.

    The EC's console task prints a string upon initialization which says that
    "Console is enabled; type HELP for help.".  The enhanced EC images print a
    different string as a part of their init.  This function searches through a
    "look" buffer, scanning for the presence of either of those strings and
    updating the enhanced_ec state accordingly.

    Args:
      data: A string containing the data sent from the interpreter.
    """
    self.look_buffer += data

    # Search the buffer for any of the EC image strings.
    enhanced_match = re.search(ENHANCED_IMAGE_RE, self.look_buffer)
    non_enhanced_match = re.search(NON_ENHANCED_IMAGE_RE, self.look_buffer)

    # Update the state if any matches were found.
    if enhanced_match or non_enhanced_match:
      if enhanced_match:
        self.enhanced_ec = True
      elif non_enhanced_match:
        self.enhanced_ec = False

      # Inform the interpreter of the result.
      self.cmd_pipe.send('enhanced ' + str(self.enhanced_ec))
      self.logger.debug('Enhanced EC image? %r', self.enhanced_ec)

      # Clear look buffer since a match was found.
      self.look_buffer = ''

    # Move the sliding window.
    self.look_buffer = self.look_buffer[-LOOK_BUFFER_SIZE:]


def IsPrintable(byte):
  """Determines if a byte is printable.

  Args:
    byte: An integer potentially representing a printable character.

  Returns:
    A boolean indicating whether the byte is a printable character.
  """
  return byte >= ord(' ') and byte <= ord('~')


def StartLoop(console):
  """Starts the infinite loop of console processing.

  Args:
    console: A Console object that has been properly initialzed.
  """
  console.logger.info('EC Console is being served on %s.', console.user_pty)
  console.logger.debug(console)
  try:
    while True:
      # Check to see if pipes or the console are ready for reading.
      read_list = [console.master_pty, console.cmd_pipe, console.dbg_pipe]
      ready_for_reading = select.select(read_list, [], [])[0]

      for obj in ready_for_reading:
        if obj is console.master_pty:
          console.logger.debug('Input from user')
          # Convert to bytes so we can look for non-printable chars such as
          # Ctrl+A, Ctrl+E, etc.
          line = bytearray(os.read(console.master_pty, CONSOLE_MAX_READ))
          for i in line:
            # Handle each character as it arrives.
            console.HandleChar(i)

        elif obj is console.cmd_pipe:
          data = console.cmd_pipe.recv()
          # Write it to the user console.
          console.logger.debug('|CMD|->\'%s\'', data)
          os.write(console.master_pty, data)

        elif obj is console.dbg_pipe:
          data = console.dbg_pipe.recv()
          if console.interrogation_mode == 'auto':
            # Search look buffer for enhanced EC image string.
            console.CheckBufferForEnhancedImage(data)
          # Write it to the user console.
          console.logger.debug('|DBG|->\'%s\'', data)
          os.write(console.master_pty, data)

      while not console.oobm_queue.empty():
        console.logger.debug('OOBM queue ready for reading.')
        console.ProcessOOBMQueue()

  finally:
    # Close pipes.
    console.dbg_pipe.close()
    console.cmd_pipe.close()
    # Close file descriptor.
    os.close(console.master_pty)
    # Exit.
    sys.exit(0)


def main(argv):
  """Kicks off the EC-3PO interactive console interface and interpreter.

  We create some pipes to communicate with an interpreter, instantiate an
  interpreter, create a PTY pair, and begin serving the console interface.

  Args:
    argv: A list of strings containing the arguments this module was called
      with.
  """
  # Set up argument parser.
  parser = argparse.ArgumentParser(description=('Start interactive EC console '
                                                'and interpreter.'))
  parser.add_argument('ec_uart_pty',
                      help=('The full PTY name that the EC UART'
                            ' is present on. eg: /dev/pts/12'))
  parser.add_argument('--log-level',
                      default='info',
                      help='info, debug, warning, error, or critical')

  # Parse arguments.
  opts = parser.parse_args(argv)

  # Set logging level.
  opts.log_level = opts.log_level.lower()
  if opts.log_level == 'info':
    log_level = logging.INFO
  elif opts.log_level == 'debug':
    log_level = logging.DEBUG
  elif opts.log_level == 'warning':
    log_level = logging.WARNING
  elif opts.log_level == 'error':
    log_level = logging.ERROR
  elif opts.log_level == 'critical':
    log_level = logging.CRITICAL
  else:
    parser.error('Invalid log level. (info, debug, warning, error, critical)')

  # Start logging with a timestamp, module, and log level shown in each log
  # entry.
  logging.basicConfig(level=log_level, format=('%(asctime)s - %(module)s -'
                                               ' %(levelname)s - %(message)s'))

  # Create some pipes to communicate between the interpreter and the console.
  # The command pipe is bidirectional.
  cmd_pipe_interactive, cmd_pipe_interp = multiprocessing.Pipe()
  # The debug pipe is unidirectional from interpreter to console only.
  dbg_pipe_interactive, dbg_pipe_interp = multiprocessing.Pipe(duplex=False)

  # Create an interpreter instance.
  itpr = interpreter.Interpreter(opts.ec_uart_pty, cmd_pipe_interp,
                                 dbg_pipe_interp, log_level)

  # Spawn an interpreter process.
  itpr_process = multiprocessing.Process(target=interpreter.StartLoop,
                                         args=(itpr,))
  # Make sure to kill the interpreter when we terminate.
  itpr_process.daemon = True
  # Start the interpreter.
  itpr_process.start()

  # Open a new pseudo-terminal pair
  (master_pty, user_pty) = pty.openpty()
  # Set the permissions to 660.
  os.chmod(os.ttyname(user_pty), (stat.S_IRGRP | stat.S_IWGRP |
                                  stat.S_IRUSR | stat.S_IWUSR))
  # Create a console.
  console = Console(master_pty, os.ttyname(user_pty), cmd_pipe_interactive,
                    dbg_pipe_interactive)
  # Start serving the console.
  StartLoop(console)


if __name__ == '__main__':
  main(sys.argv[1:])