I've been helping to develop and conduct a class that introduces kids to
electronics, computer engineering, and software development concepts
using Arduino & ATMega microcontrollers. We chose to use AMForth for a
part of the programming section of the class and were concerned the kids
would have trouble understanding the results of uploading using the
currently bundled amforth-shell.py and amforth-uplaod.py scripts when
something went wrong.
To address those concerns I whipped together the script below which,
after a bit of scope creep, implements the following enhancements to the
existing scripts:
1) Based on the pyserial module, it aggressively reads character echos
which appears to prevent occasional microcontroller resets during
uploads that I think may be related to serial tx overruns. It also
checks for lines that are too long (preventing rx overruns) and
attempts to compress out excess whitespace and comment lines and
avoid sending them to the microcontroller to shorten the upload time.
2) When uploading it prints the contents of each line sent to the
microcontroller as they are sent, with some status information
prefixed so it is easy to understand the progress of the upload.
Lines successfully sent are prefixed with an 'S', whitespace lines
that weren't sent at all are prefixed with a 'W', errors, with 'E',
etc.
3) By default it stops an upload on the first error observed, detected
by matching on the pattern of amforth error messages or receiving a
timeout. This can be overridden for a single line, an entire file,
or an entire group of files (using nested includes).
4) In interactive mode it implements host-side command history, word
completion (including host-side directives), and filename completion
for the directives for including (uploading) and editing files. The
file editing directive has a default mode where it opens an editor at
the location of the last error.
Typical usage is very simple:
./frt-interact.py -p /dev/ttyUSB0
Replace the path to the port device with whatever port is connected to
your microcontroller. For Windows users, I believe you can specify a
COM device and it should work but I haven't tested that. This will get
you an interaction prompt just as if you were interacting with amforth
over a regular terminal but with all the above enhancements. There is a
lot more information about the operation of the script, available
host-side directives, etc. in the comment block at the beginning of the
script below.
We've been using this in the class for a couple of weeks now and it
seems to be working well. I'd be very pleased if this could be included
in the base distribution of amforth.
Regards, Keith
--- Cut here ----
#!/usr/bin/python
#
# pySerial based upload & interpreter interaction module for amforth.
#
# Copyright 2011 Keith Amidon (cama...@picncipark.org)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# =====================================================================
# DOCUMENTATION
# =====================================================================
# This module module may be used as a script or imported for use as a
# part of a larger python program.
#
# Script Usage
# ------------
# When used as a script this module provides two main functions, the
# ability to reliably upload files to the amforth interpreter and an
# interpreter interaction mode with line editing, word completion and
# previous input history. For information on how to access these
# features when invoking the module as a script, execute it with the
# --help option and read the following sections on the interaction
# protocol and local directives.
#
#
# Interaction Protocol
# --------------------
# The amforth interaction protocol used by this module is to send a
# line to amforth character by character, reading the echos as quickly
# as possible. Once the entire line has been sent it then reads the
# response up until the " ok" prompt or a prompt that looks like an
# error response. The character by character handling of echos
# appears to eliminate unexpected device resets when compared to the
# line by line method used by previous tools, possibly by eliminating
# the possibility of serial tx overrun on the device.
#
# To further optimize interaction with the device lines are evaluated
# before sending and redundant whitespace is compressed out. Lines
# which are all whitespace or whitespace and comments are not sent and
# the next line is handled.
#
#
# Local Directives
# ----------------
# A number of special directives are supported which instruct the
# script to do something and are handled locally without being sent to
# amforth. Directives may be specified within comments or outside
# comments as controlled by the "#directive" directive. They must be
# the only contents of a line or they will be ignored. The directives
# include:
#
# #include <file>
# Upload the file named by <file> before proceeding further.
#
# #cd <dir>
# Change the current local directory to the location specified.
# During uploads, this directive affects the current file and
# files it includes. Once the current file is complete the old
# value will be restored.
#
# #directive <config>
# Change how directives are discovered. The valid values for
# <config> are:
# none : Stop looking for any directives
# commented : Only look for directives within comments
# Commented directives must be the first word of the
# comment. The remaining text in the comment is the
# argument provided to the directive. There must
# not be any other non-whitespace text other than
# the comment start and (if required) end characters
# and the directive and any directive argument on a
# commented directive line. If any other text is
# present on the line an error will be generated.
# uncommented : Only look for directives outside comments.
# Uncommented directives must be the first word of a
# line and extend to the end of the line. If a
# directive name exists in a subsequent word of a
# line it will be sent to the interpreter as a word
# like any other.
# all : Allow both commented and uncommented directives.
# This is the default.
# During uploads, this directive affects the current file and
# files it includes. Once the current file is complete the old
# value will be restored.
#
# #timeout <float>
# Change the timeout value to <float> seconds. Fractional
# values are supported. During uploads, this directive affects
# the current file and files it includes. Once the current file
# is complete the old value will be restored.
#
# #timeout-next <float>
# Change the timeout value for the next line sent to the
# interpreter to <float> seconds. Fractional values are
# supported. The timeout returns to its previous value after
# the next line is sent to the interpreter. If this directive
# is encountered as the very last line of an upload file it will
# have no effect.
#
# #error-on-output [<yes-or-no>]
# Controls whether an error is generated if unexpected output
# occurs during an upload. The default is yes. This directive
# can not be used in interactive mode as it would not have any
# effect. During uploads it affects the rest of the current file
# and any files it includes. The argument is optional. If not
# given it is assumed to be "yes".
#
# #ignore-error [<yes-or-no>]
# Ignore any error that occurs later in the current upload file
# or a file it includes. The argument is optional. If given
# the behavior is set as specified. If not given it is assumed
# to be "yes".
#
# #ignore-error-next [<yes-or-no>]
# Ignore any error that occurs on the next line. The argument
# is optional. If given the behavior is set as specified. If
# not given it is assumed to be "yes".
#
# #expect-output-next [<regexp>]
# Expect specific output on the next line. The argument is
# optional. If it is not specified a default regular expression
# of ".*" (match everything) is assumed. This overrides the
# #error-on-output directive. An error is raised if the output
# doesn't match the regular expression. It will be ignored if
# #ignore-error is yes. Use of this directive without an
# argument is the way to prevent an error on output when
# #error-on-output is yes
#
# #start-string-word <word>
# Add a word that starts a string. The string will end when a
# double quote character is read.
#
# #quote-char-word <word>
# Add a word that quotes the immediately next word
#
# #interact
# Start an interactive session before proceeding with file upload.
# This only makes sense during a file upload.
#
# #edit [<filename>]
# Edit a file. The filename is optional. If it is provided the
# named file will be edited. If it is not provided and the last
# upload ended in an error the file that had the error will be
# edited at the location of the error. If there was no previous
# upload or the last upload completed successfully but an #edit
# directive was previously issued with a filename, edit the file
# previously named. Finally, if none of these apply an error is
# printed. The editor used can be specified with the --editor
# option when starting the program or through the EDITOR
# environment variable.
#
# #update-words
# This directive is only available in an interactive session.
# It cause the interaction code to reload the list of words used
# for completion from the amforth interpreter. Typically it is
# not required as words are updated automatically when the
# session starts and any time a file is uploaded as a results of
# a #include directive. The case where it is required is when
# completion is needed for words defined interactively during
# the session.
#
# #exit
# Exit an interactive session or the current upload immediately.
# If encountered during an upload, no further lines from the
# file will be processed.
#
#
# Programmatic Usage
# ------------------
# For programmatic usage, a single class named AMForth is provided.
# It can be instantiated with no arguments but typically a serial port
# device and port speed will be provided as the defaults are unlikely
# to be correct.
#
# Once an instance is obtained, and connected the high-level entry
# points are the "upload_file" and "interact" methods, the former
# uploading a file to the AMForth interperter and the latter providing
# an interative interpreter shell with command history and word
# completion. These methods provide progress information in various
# cases by calling the function stored in the "progress_callback"
# property with three arguments, the type of progress being reported,
# a line number if available (otherwise it is None) and a message with
# further information. The default progress callback prints this
# information to the screen in a terse format. Other programs may
# wish to replace this with their own progress presentation function.
#
# Low-level interaction with the AMForth interpreter would typically
# use the "send_line" and "read_response" methods. Before these can
# be used the serial connection must be established. The
# serial_connected property indicates whether a connection currently
# exists. A good way to obtain a connection and rule out errors in
# serial communication is to call "find_prompt" which ensures the
# existence of a serial connection and sends a newline to the AMForth
# interperter and watches for the echo. This is usually the best way
# of establishing a connection but the "serial_connect" method will
# open a connection without sending anything if that is required.
#
# Elimination of whitespace and discovery of directives (see below) is
# provided through the "preprocess_line" method and directives that
# have common implementations can be handled with the
# "handle_common_directives" method.
# TODO: - Update comments on most functions explaining what they do.
import argparse
import atexit
import copy
import glob
import os
import re
import readline
import serial
import StringIO
import subprocess
import sys
import traceback
class AMForthException(Exception):
pass
class Behaviors(object):
"""Simple class for storing configurable processing behaviors"""
def __init__(self):
self.working_directory = os.getcwd()
self.filename = None
self.timeout = 15.0
self.quote_char_words = ["[char]", "char"]
self.start_string_words = ['s"', '."', 'abort"']
self.error_on_output = True
self.ignore_errors = False
self.directive_uncommented = True
self.directive_commented = True
self.expected_output_regexp = None
@property
def directive_config(self):
"Get the current directive configuration"
if self.directive_uncommented:
if self.directive_commented:
return "all"
else:
return "uncommented"
else:
if self.directive_commented:
return "commented"
else:
return "none"
@directive_config.setter
def directive_config(self, value):
"Set the directive configuration"
if value == "none":
self.directive_uncommented = False
self.directive_commented = False
elif value == "all":
self.directive_uncommented = True
self.directive_commented = True
elif value == "uncommented":
self.directive_uncommented = True
self.directive_commented = False
elif value == "commented":
self.directive_uncommented = False
self.directive_commented = True
else:
raise AMForthException("Unknown directive config: %s" % value)
class BehaviorManager(object):
"""Class for determining currently configured behavior
This class manages the lifetime of behaviors established through
configuration options and directives to minimize the impact of
that support on the AMForth class. """
def __init__(self):
self.default_behavior = Behaviors()
self.clear()
def clear(self):
"Clear out accumulated behavior"
self._next_line_behavior = None
self._current_line_behavior = None
self._file_behaviors = []
@property
def current_behavior(self):
"""The behavior currently in effect"""
if self._current_line_behavior:
return self._current_line_behavior
elif self._file_behaviors:
return self._file_behaviors[0]
else:
return self.default_behavior
def advance_line(self):
"""Call when changing to the next line"""
self._current_line_behavior = self._next_line_behavior
self._next_line_behavior = None
def push_file(self, filename):
"""Call when starting processing a new nested file"""
behavior = copy.deepcopy(self.current_behavior)
behavior.filename = filename
self._file_behaviors.insert(0, behavior)
def pop_file(self):
"""Call when returning from a nested file"""
del(self._file_behaviors[0])
@property
def next_line_behavior(self):
"""The behavior to use for the next line"""
return self._next_line_behavior
@next_line_behavior.setter
def next_line_behavior(self, behavior):
self._next_line_behavior = behavior
@property
def current_file_behavior(self):
"""The behavior for the current file.
Will raise an exception if there is no file currently."""
return self._file_behaviors[0]
@current_file_behavior.setter
def current_file_behavior(self, behavior):
self._file_behaviors[0] = behavior
class AMForth(object):
"Class for interacting with the AMForth interpreter"
amforth_error_cre = re.compile(" \?\? -\d+ \d+ \r\n> $")
upload_directives = [
"#cd", "#include", "#directive", "#ignore-error",
"#ignore-error-next", "#error-on-output", "#expect-output-next",
"#string-start-word", "#quote-char-word",
"#timeout", "#timeout-next", "#interact", "#exit"
]
interact_directives = [
"#cd", "#edit", "#include", "#directive", "#ignore-error",
"#error-on-output", "#string-start-word", "#quote-char-word",
"#timeout", "#timeout-next", "#update-words", "#exit"
]
def __init__(self, serial_port="/dev/amforth", speed=9600):
self.debug = False
self.max_line_length = 80
self.progress_callback = self.print_progress
self.editor = None
self._serial_port = serial_port
self._serial_speed = speed
self._serialconn = None
self._readline_initialized = False
self._amforth_dp = None
self._amforth_words = []
self._last_error = ()
self._last_edited_file = None
self._config = BehaviorManager()
@property
def serial_port(self):
"Serial port device attached to AMForth"
return self._serial_port
@serial_port.setter
def serial_port(self, value):
"""Set the serial port device attached to AMForth
If the value provided is different than the current value any
existing serial connection will be closed and a new connection
opened."""
if self._serial_port != value:
self._serial_port = value
self.serial_reconnect()
@property
def serial_speed(self):
"Speed of the serial connection to AMForth"
return self._serial_speed
@serial_speed.setter
def serial_speed(self, value):
if self._serial_speed != value:
self._serial_speed = value
self.serial_reconnect()
@property
def serial_connected(self):
"Boolean status for whether currently connected to AMForth"
return self._serialconn is not None
def main(self):
"Main function called when module is used as a script"
upload_files, interact = self.parse_arg()
try:
for fn in upload_files:
if fn == "-":
self.interact()
else:
self.upload_file(fn)
if interact:
self.interact()
except AMForthException:
return 1
except KeyboardInterrupt:
print "\nAborted with keyboard interrupt"
except Exception, e:
print "\n---- Unexpected exception ----"
traceback.print_exc()
return 1
finally:
self.serial_disconnect()
return 0
def parse_arg(self):
"Argument parsing used when module is used as a script"
parser = argparse.ArgumentParser(description="Interact with AMForth")
parser.add_argument("--timeout", "-t", action="store",
type=float, default=15.0,
help="Timeout for response in seconds (float value)")
parser.add_argument("--port", "-p", action="store",
default=self.serial_port,
help="Name of serial port on which AMForth is connected")
parser.add_argument("--speed", "-s", action="store",
type=int, default=self.serial_speed,
help="Speed of serial port on which AMForth is connected")
parser.add_argument("--line-length", "-l", action="store",
type=int, default=self.max_line_length,
help="Maximum length of amforth input line")
parser.add_argument("--interact", "-i", action="store_true",
help="Enter interactive prompt after upload")
parser.add_argument("--directive", "-d", action="store",
default="all",
help="Local directive configuration (where found)")
parser.add_argument("--editor", action="store",
default = os.environ.get("EDITOR", None),
help="Editor to use for #edit directive")
parser.add_argument("--no-error-on-output", action="store_true",
help="Indicate an error if upload causes output")
parser.add_argument("--ignore-error", action="store_true",
help="Ignore errors during upload (not recommended)")
parser.add_argument("--debug-serial", action="store_true",
help="Output extra info about serial transfers in stderr")
parser.add_argument("files", nargs="*")
arg = parser.parse_args()
self.debug = arg.debug_serial
self.max_line_length = arg.line_length
self._serial_port = arg.port
self._serial_speed = arg.speed
self.editor = arg.editor
behavior = self._config.current_behavior
behavior.error_on_output = not arg.no_error_on_output
behavior.directive_config = arg.directive
behavior.timeout = arg.timeout
behavior.ignore_errors = arg.ignore_error
return arg.files, (arg.interact or len(arg.files) == 0)
def serial_connect(self, port=None, speed=None):
"""Connect to AMForth on a serial port
The port and speed argument are optional. If not specified
the current values set in the object are used. These will be
the defaults if the have not been changed. If either is
specified corresponding property of the instance will be
updated to the new value.
This is safe to call even if a connection already exists as
existing an existing connection will be closed before the new
connection is made."""
if port != None:
self.serial_port = port
if speed != None:
self.serial_speed = speed
if self._serialconn:
self.serial_disconnect()
try:
timeout = self._config.current_behavior.timeout
self._serialconn = serial.Serial(self.serial_port,
self.serial_speed,
serial.EIGHTBITS,
serial.PARITY_NONE,
serial.STOPBITS_ONE,
timeout, False, False, None, False)
except serial.SerialException, e:
raise AMForthException("Serial port connect failure: %s" % str(e))
def serial_disconnect(self):
"""Disconnect the serial connection to AMForth
This is safe to call even if there is currently no connection."""
if self._serialconn:
self._serialconn.close()
self._serialconn = None
def serial_reconnect(self):
"""Reconnect the serial connection to AMForth
This is the same as calling serial_connect while there is an
existing connection. It is provided to make the clear when
the intent is to re-establish an existing connection (usually
to apply new settings) versus creating a new connectoion."""
self.serial_connect()
def find_prompt(self):
"Attempt to find a prompt by sending a newline and verifying echo"
if not self.serial_connected:
self.serial_connect()
# Use a short timeout to quickly detect if can't communicate
self._serialconn.timeout = 2.0
try:
try:
self.send_line("\n") # Get empty line echo to make sure ready
self.read_response() # Throw away the response.
except serial.SerialException, e:
self.progress_callback("Error", None, str(e))
raise AMForthException("Failed to get prompt: %s" % str(e))
finally:
# Restore the current timeout
self._serialconn.timeout = self._config.current_behavior.timeout
def upload_file(self, filename):
wd = self._config.current_behavior.working_directory
fpath = os.path.normpath(os.path.join(wd, filename))
self._config.push_file(fpath)
try:
try:
self.find_prompt()
except AMForthException, e:
self.progress_callback("Error", None, str(e))
raise
self.progress_callback("File", None, fpath)
try:
with open(fpath, "r") as f:
self._send_file_contents(f)
except (OSError, IOError), e:
self.progress_callback("Error", None, str(e))
raise AMForthException("Unknown file: " + fpath)
self._last_error = ()
finally:
self._config.pop_file()
self._serialconn.timeout = self._config.current_behavior.timeout
try:
os.chdir(self._config.current_behavior.working_directory)
except OSError, e:
errmsg = ("Failed to change to directory '%s': %s"
% (self._config.current_behavior.working_directory,
str(e)))
self.progress_callback("Error", None, errmsg)
raise AMForthException(errmsg)
def _send_file_contents(self, f):
in_comment = False
lineno = 0
for full_line in f:
self._config.advance_line()
self._serialconn.timeout = self._config.current_behavior.timeout
try:
os.chdir(self._config.current_behavior.working_directory)
except OSError, e:
errmsg = ("Failed to change to directory '%s': %s"
% (self._config.current_behavior.working_directory,
str(e)))
self.progress_callback("Error", None, errmsg)
raise AMForthException(errmsg)
lineno += 1
if full_line and full_line[-1] == "\n":
full_line = full_line[:-1]
if full_line and full_line[-1] == "\r":
full_line = full_line[:-1]
line = full_line.strip()
if len(line) == 0:
if in_comment:
self.progress_callback("Comment", lineno, full_line)
else:
self.progress_callback("Whitespace", lineno, full_line)
continue
try:
(line, in_comment,
directive,
directive_arg) = self.preprocess_line(full_line, in_comment,
self.upload_directives)
except AMForthException, e:
self._record_error(lineno)
self.progress_callback("Error", lineno, full_line)
self.progress_callback("Error", None, str(e))
raise
if directive:
self.progress_callback("Directive", lineno, full_line)
if directive == "#exit":
break
elif directive == "#interact":
self.interact()
continue
self.handle_common_directives(directive, directive_arg)
continue
if len(line) == 0:
self.progress_callback("Comment", lineno, full_line)
continue
try:
self.send_line(line)
except AMForthException, e:
self._record_error(lineno)
self.progress_callback("Error", lineno, full_line)
self.progress_callback("Error", None, str(e))
raise
response = self.read_response()
self.progress_callback("Sent", lineno, full_line)
if response[-3:] == " ok":
if len(response) > 3:
for l in StringIO.StringIO(response[:-3]):
self.progress_callback("Output", lineno, l.rstrip())
r = self._config.current_behavior.expected_output_regexp
if r:
m = re.match(r, response[:-3], re.MULTILINE)
response_ok = m is not None
else:
response_ok = False
if not response_ok:
if self._config.current_behavior.error_on_output:
errmsg = "Unexpected output after line."
errmsg += " To allow, specify --no-error-on-output."
self.progress_callback("Error", lineno, errmsg)
if not self._config.current_behavior.ignore_errors:
self._record_error(lineno)
raise AMForthException(errmsg)
else:
self.progress_callback("Error", None, response)
if not self._config.current_behavior.ignore_errors:
self._record_error(lineno)
raise AMForthException("Error in line sent")
def preprocess_line(self, line, in_delim_comment=False, directives=[]):
# Compresses whitespace, including comments so send minimum
# data to atmega
result = []
comment_words = []
char_quote = False
in_string = False
in_line_comment = False
directive = None
directive_arg = []
words = self._split_space_or_tab(line)
for w in words:
if in_string:
try:
i = w.index('"')
except ValueError:
result[-1] += " " + w
continue
in_string = False
result[-1] += " " + w[:i+1]
result[-1] = result[-1][1:] # remove extra initial space
w = w[i+1:]
if not w:
continue
if char_quote:
result.append(w)
char_quote = False
continue
if w == "(":
if not in_delim_comment:
in_delim_comment = True
else:
raise AMForthException("Illegal nested comment")
continue
if w == ")":
if in_delim_comment:
in_delim_comment = False
else:
raise AMForthException("Comment end without begin")
continue
if not in_delim_comment and not in_line_comment:
if w == "\\":
in_line_comment = True
continue
elif w in self._config.current_behavior.start_string_words:
in_string = True
result.append(w)
result.append('')
continue
if w in self._config.current_behavior.quote_char_words:
char_quote = True # no continue deliberately
if directive:
directive_arg.append(w)
else:
if (self._config.current_behavior.directive_uncommented
and not result
and w in directives):
directive = w
else:
result.append(w)
else:
if directive:
directive_arg.append(w)
else:
if (self._config.current_behavior.directive_commented
and not result
and not comment_words
and w in directives):
directive = w
else:
comment_words.append(w)
if directive and len(result):
raise AMForthError("Directive must not have other content: %s",
" ".join(result))
return (" ".join(result), in_delim_comment,
directive, " ".join(directive_arg))
def _record_error(self, lineno):
fn = self._config.current_behavior.filename
if fn:
self._last_error = (fn, lineno)
def _split_space_or_tab(self, line):
result = [""]
for c in line:
if c == " " or c == "\t":
result.append("")
else:
result[-1] += c
return result
def handle_common_directives(self, directive, directive_arg):
if directive == "#include":
fn = directive_arg.strip()
self.upload_file(fn)
resume_fn = self._config.current_behavior.filename
if resume_fn:
self.progress_callback("File", None, resume_fn + " (resumed)")
elif directive == "#cd":
dirname = directive_arg.strip()
if os.path.isabs(dirname):
dirpath = os.path.normpath(dirname)
else:
oldpath = self._config.current_behavior.working_directory
dirpath = os.path.normpath(os.path.join(oldpath, dirname))
self._config.current_behavior.working_directory = dirpath
elif directive == "#timeout":
try:
timeout = float(directive_arg)
except ValueError, e:
self.progress_callback("Error", None, "Invalid timeout")
return
self._config.current_file_behavior.timeout = timeout
elif directive == "#timeout-next":
try:
timeout = float(directive_arg)
except ValueError, e:
self.progress_callback("Error", None, "Invalid timeout")
return
behavior = copy.deepcopy(self._config.current_behavior)
behavior.timeout = timeout
self._config.next_line_behavior = behavior
elif directive == "#ignore-error":
v = self._yes_or_no_arg(directive_arg)
self._config.current_file_behavior.ignore_errors = v
elif directive == "#ignore-error-next":
v = self._yes_or_no_arg(directive_arg)
behavior = copy.deepcopy(self._config.current_behavior)
behavior.ignore_errors = v
self._config.next_line_behavior = behavior
elif directive == "#error-on-output":
v = self._yes_or_no_arg(directive_arg)
behavior = self._config.current_file_behavior
behavior.error_on_output = v
elif directive == "#expect-output-next":
regexp = directive_arg.strip()
if not regexp:
regexp = ".*"
behavior = copy.deepcopy(self._config.current_behavior)
behavior.expected_output_regexp = regexp
self._config.next_line_behavior = behavior
elif directive == "#start-string-word":
behavior = self._config.current_file_behavior
behavior.start_string_words.append(directive_arg.strip().split(" "))
elif directive == "#quote-char-word":
behavior = self._config.current_file_behavior
behavior.quote_char_words.append(directive_arg.strip().split(" "))
elif directive == "#directive":
behavior = self._config.current_file_behavior
behavior.directive_config = directive_arg.strip()
else:
errmsg = "Unknown directive: %s %s" % (directive, directive_arg)
raise AMForthException(errmsg)
def _yes_or_no_arg(self, directive_arg):
if not directive_arg:
return True
else:
if directive_arg.lower() == "yes":
return True
elif directive_arg.lower() == "no":
return False
else:
errmsg = "Invalid directive argument. Must be yes or no."
raise AMForthExcetion(errmsg)
def send_line(self, line):
if len(line) > self.max_line_length - 1: # For newline
raise AMForthException("Input line > %d char"
% self.max_line_length)
if self.debug:
sys.stderr.write("|a( )" + repr(line)[1:-1] + "\n")
sys.stderr.write("|s( )")
for c in line + "\n":
if self.debug:
sys.stderr.write(repr(c)[1:-1]+"->")
sys.stderr.flush()
self._serialconn.write(c)
self._serialconn.flush()
r = self._serialconn.read(1) # Read echo of character we just sent
while r and (r != c or (c == '\t' and r != ' ')):
if self.debug:
sys.stderr.write(repr(r)[1:-1])
sys.stderr.flush()
r = self._serialconn.read(1)
if not r:
raise AMForthException("Input character not echoed.")
if self.debug:
sys.stderr.write(repr(r)[1:-1] + "|")
sys.stderr.flush()
if self.debug:
sys.stderr.write("\n")
def read_response(self):
if self.debug:
sys.stderr.write("|r( )")
response = ""
r = self._serialconn.read(1)
while r != "":
if self.debug:
sys.stderr.write(repr(r)[1:-1])
sys.stderr.flush()
response = response + r
if response[-3:] == " ok":
# Interactive prompt read and discarded while handling
# echo of next line sent.
break
elif self.amforth_error_cre.search(response) is not None:
response = response[:-3] # Don't return prompt in response
break
r = self._serialconn.read(1)
if not response:
response = "Timed out waiting for ok response"
if self.debug:
sys.stderr.write("\n")
return response
def print_progress(self, type, lineno, info):
if not lineno:
print "|%s=%s" % (type[:1], info)
else:
print "|%s|%5d|%s" % (type[:1], lineno, info)
def interact(self):
self.progress_callback("Interact", None,
"Entering amforth interactive interpreter")
# Use null filename "file" to capture interactive config
self._config.push_file(None)
try:
self.find_prompt()
except AMForthException, e:
self.progress_callback("Error", None, str(e))
self._config.pop_file()
raise
self._init_readline()
in_comment = False
while True:
try:
full_line = raw_input("> ")
except EOFError, e:
print ""
break
self._config.advance_line()
self._serialconn.timeout = self._config.current_behavior.timeout
try:
os.chdir(self._config.current_behavior.working_directory)
except OSError, e:
errmsg = ("Failed to change to directory '%s': %s"
% (self._config.current_behavior.working_directory,
str(e)))
self.progress_callback("Error", None, errmsg)
raise AMForthException(errmsg)
(line, in_comment,
directive,
directive_arg) = self.preprocess_line(full_line, in_comment,
self.interact_directives)
try:
if directive:
self.progress_callback("Directive", None, full_line)
if directive == "#exit":
break
elif directive == "#update-words":
self._update_words()
continue
elif directive == "#edit":
if directive_arg:
self.edit_file(directive_arg.strip())
elif self._last_error:
self.edit_file(*self._last_error)
elif self._last_edited_file:
self.edit_file(self._last_edited_file)
else:
print "No file to edit"
continue
self.handle_common_directives(directive, directive_arg)
if directive == "#include":
self._update_words()
continue
if in_comment or not line:
continue
else:
self.send_line(line)
print self.read_response()
except AMForthException, e:
print "Error: " + str(e)
self._config.pop_file()
self._serialconn.timeout = self._config.current_behavior.timeout
try:
os.chdir(self._config.current_behavior.working_directory)
except OSError, e:
errmsg = ("Failed to change to directory '%s': %s"
% (self._config.current_behavior.working_directory,
str(e)))
self.progress_callback("Error", None, errmsg)
raise AMForthException(errmsg)
self.progress_callback("Interact", None,
"Leaving interactive interpreter")
def _init_readline(self):
if not self._readline_initialized:
readline.set_completer_delims(" ")
readline.set_completer(self._rlcompleter)
readline.parse_and_bind("tab: complete")
histfn = os.path.join(os.path.expanduser("~"),
".frt-interact.history")
try:
readline.read_history_file(histfn)
except IOError, e:
pass
self._update_words()
atexit.register(readline.write_history_file, histfn)
def _update_words(self):
# TODO: - handle multiple wordlists in _update_words
self.send_line("dp .")
dp = self.read_response()
if dp[-3:] != " ok":
return # Something went wrong, just silently ignore
dp = int(dp[:-3])
if self._amforth_dp != dp:
self._amforth_dp = dp
self.send_line("words")
words = self.read_response()
if words[-3:] != " ok":
return # Something went wrong, just silently ignore
self._amforth_words = words.split(" ") + self.interact_directives
def _rlcompleter(self, text, state):
if state == 0:
line_words = readline.get_line_buffer().split(" ")
if line_words and line_words[-1] == text:
line_words = line_words[:-1]
while line_words and line_words[-1] == "":
line_words = line_words[:-1]
if line_words:
if line_words[-1] in ["#include", "#edit"]:
self._rl_matches = glob.glob(text + "*")
elif line_words[-1] == "#cd":
fnames = glob.glob(text + '*')
self._rl_matches = [f + "/" for f in fnames
if os.path.isdir(f)]
elif line_words[-1] == "#directive":
self._rl_matches = [w for w in ("all ", "uncommented ",
"commented ", "none ")
if w.startswith(text)]
elif line_words[-1] in ["#error-on-output",
"#ignore-error", "#ignore-error-next"]:
self._rl_matches = [w for w in ["yes", "no"]
if w.startswith(text)]
elif line_words[-1] in ["#exit", "#update-words",
"#timeout", "#timeout-next"]:
self._rl_matches = []
else:
self._rl_matches = [w + " " for w in self._amforth_words
if not text or w.startswith(text)]
else:
self._rl_matches = [w + " " for w in self._amforth_words
if not text or w.startswith(text)]
if self._rl_matches:
return self._rl_matches[0]
else:
return None
else:
if state < len(self._rl_matches):
return self._rl_matches[state]
else:
return None
def edit_file(self, filename, lineno=0):
if self.editor:
# Have to construct command line differently for different
# editors to be able to move to specific line...
exename = os.path.basename(self.editor)
if exename in ["emacs", "emacsclient", "nano"]:
cmd = [self.editor, "+" + str(lineno), filename]
elif exename in ["vi", "vim"]:
cmd = [self.editor, filename, "+" + str(lineno)]
elif exename == "gedit":
cmd = [self.editor, "-b", filename, "+" + str(lineno)]
else:
cmd = [self.editor, filename]
try:
subprocess.call(cmd)
self._last_edited_file = filename
except OSError, e:
raise AMForthException("Could not start editor: "+self.editor)
else:
raise AMForthException("No editor specified. Use --editor or
EDITOR environment variable")
if __name__ == "__main__":
sys.exit(AMForth().main())
------------------------------------------------------------------------------
Live Security Virtual Conference
Exclusive live event will cover all the ways today's security and
threat landscape has changed and how IT managers can respond. Discussions
will include endpoint security, mobile security and the latest in malware
threats. http://www.accelacomm.com/jaw/sfrnl04242012/114/50122263/
_______________________________________________
Amforth-devel mailing list for http://amforth.sf.net/
Amforth-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/amforth-devel
