gSerial.py

Go to the documentation of this file.

#!/usr/local/bin/python
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "GLAST LAT serial port wrapper classes"
__author__   = "R. Claus <Claus@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "12/27/2005" # Created
__version__  = "$Revision: 1.2 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC


import getopt
import sys
import os
import time
from   ConfigParser import ConfigParser
from   serial       import Serial


class SerialCommon(Serial):
  """!\brief A helper class to add some protocol to the serial byte stream.
  """
  STARTCHARACTER = '\x01'   #ctrl+A = SOH = Start of Heading
  FILECHARACTER  = '\x02'   #ctrl+B = STX = Start of Text
  ENDCHARACTER   = '\x04'   #ctrl+D = EOT = End of Transmission

  def __init__(self, config, echo = None, timeout = None):
    """!\brief  Initialize the class with defaults from a configuration file
    and/or the command line.

    \param config - A ConfigParser configuration object
    \param echo   - Option to override the configuration file's echo value
    """
    self.__config = config

    # Initialize the defaults
    self.__port       = 0
    self.__baudrate   = 9600
    self.__rtscts     = 0
    self.__xonxoff    = 0
    try:
      self._echo      = int(echo)
    except:
      self._echo      = echo
    self._debug       = 0

    self.__serialOptions(config)

    # Open the serial port
    try:
      Serial.__init__(self, self.__port, self.__baudrate, timeout=timeout,
                      rtscts=self.__rtscts, xonxoff=self.__xonxoff)
    except:
      sys.stderr.write("Could not open port\n")
      raise


  def __serialOptions(self, config):
    """
    Serial line options the [serial] block of the configuration file can be:
    port=PORT:             Port, a number, default = 0 or a device name
    baud=BAUD:             Baud rate, default 9600
    rtscts=0/1:            En/dis-able RTS/CTS flow control (default: disabled)
    xonxoff=0/1:           En/dis-able software flow control (default: disabled)
    echo=0/1:              En/dis-able local echo (default: disabled)
    debug=0/1:             En/dis-able
    convert_nl=CR/LF/CRLF: Convert newline character(s) to one of CR, LF, CRLF
                           (default: OS dependent)
    """
    cSect = config.sections()
    if "serial" not in cSect:  return   # Take the defaults

    opts = config.options("serial")

    if "port" in opts:
      port = config.get("serial", "port")
      try:
        self.__port = int(port)
      except ValueError:
        self.__port = port

    if "baudrate" in opts:
      baudrate = config.get("serial", "baudrate")
      try:
        self.__baudrate = int(baudrate)
      except ValueError:
        raise ValueError, "Baudrate must be a integer number, not %r" % baudrate

    if "rtscts" in opts:
      self.__rtscts  = config.get("serial", "rtscts") != "0"

    if "xonxoff" in opts:
      self.__xonxoff = config.get("serial", "xonxoff") != "0"

    if self._echo is None:     # Allow command line option to override
      if "echo" in opts:
        self._echo   = config.get("serial", "echo") != "0"
      else:
        self._echo   = False

    if "debug" in opts:
      self._debug    = config.get("serial", "debug") != "0"

  def getConfig(self):
    """!\brief Return the configuration ConfigParser object
    """
    return self.__config


class SerialSender(SerialCommon):
  """!\brief A helper class to add some protocol to the serial byte stream.
  """
  def __init__(self, config, echo = False):
    """!\brief  Initialize the class with defaults from a configuration file
    and/or the command line.

    \param config - A ConfigParser configuration object
    \param echo   - Option to override the configuration file's echo value
    """
    SerialCommon.__init__(self, config, echo)

  def begin(self, filename = None):
    """!\brief Announce that a block of data is about to be sent.
    Optionally request that the other side tags it with a specific file name.

    \param filename Optional filename
    """
    self.write(self.STARTCHARACTER)
    if filename is not None:
      self.write(self.FILECHARACTER + fn + self.FILECHARACTER)

  def send(self, buf):
    """!\brief Send a buffer of data.
    Optionally print the buffer for debugging

    \param buf Buffer of data to send
    """
    if self._echo:
      if self._debug:  print "%r" % buf
      else:            print buf,
    self.write(buf)

  def end(self):
    """!\brief Announce the end of the block of data.
    """
    self.write(self.ENDCHARACTER)


class SerialReceiver(SerialCommon):
  """!\brief A helper class to handle some protocol on the serial byte stream.
  """
  CONVERT_CRLF   = 2
  CONVERT_CR     = 1
  CONVERT_LF     = 0

  def __init__(self, config, echo = False):
    """!\brief  Initialize the class with defaults from a configuration file.

    \param config - A ConfigParser configuration object
    \param echo   - Option to override the configuration file's echo value
    """
    SerialCommon.__init__(self, config, echo, timeout=1)

    self.__additionalOptions(config)

    self.__quit   = False

  def __additionalOptions(self, config):
    """
    Serial line options the [serial] block of the configuration file can be:
    convert_nl=CR/LF/CRLF: Convert newline character(s) to one of CR, LF, CRLF
                           (default: OS dependent)
    """
    cSect = config.sections()
    if "serial" not in cSect:  return    # Take the defaults

    opts = config.options("serial")

    self.__convert_nl = None
    if "convert_nl" in opts:
      convert_nl     = config.get("serial", "convert_nl")
      if   convert_nl.upper() == "CRLF": # Windows mode
        self.__convert_nl = self.CONVERT_CRLF
      elif convert_nl.upper() == "CR":   # Mac mode
        self.__convert_nl = self.CONVERT_CR
      elif convert_nl.upper() == "LF":   # Unix mode
        self.__convert_nl = self.CONVERT_LF

  def quit(self):
    """!\brief Call this method from another thread to force the receive loop
    to exit.
    """
    self.__quit = True

  def receive(self, filename = None):
    """!\brief Receive a block of data from the serial port and write
    it to a file.

    Receive a block of data from the serial port and write it to a
    file.  In order of decreasing precedence, the filename is given by:
    - an optional argument to this method
    - a name embedded in the serial byte stream
    - the name of the program executing this method followed by an
      underscore and a timestamp with a '.txt' extension, e.g.,
      'serialReceiver_12345678.txt'.  The timestamp used is what is
      returned by Python's time() method, but with the decimal point
      separator removed.

    To avoid conflicts while the file is being written, the data is
    initially written to a temporary file.  When the temporary file is
    complete (as provided by a signal in the serial byte stream), it
    is renamed with the filename as described above.

    \param filename - Optional filename to contain the data
    """
    # Use a temporary file to write the byte stream to
    s = '/'
    if os.name is "nt":  s = '\\'
    fn = sys.argv[0].split('.')[-2].split(s)[-1] + ".txt"
    f = open(fn, "w")                   # In case STARTCHARACTER is never seen

    fnRemote    = ''
    fnStart     = False
    line        = ''
    while not self.__quit:
      data = self.read()              # Get one character
      if   data == self.STARTCHARACTER:
        f.close()
        f = open(fn, "w")
      elif data == self.ENDCHARACTER:
        break                           # Exit if exit character seen
      elif data == self.FILECHARACTER:
        fnStart = not fnStart
      elif data == '\n':                # Convert newlines
        if   self.__convert_nl == self.CONVERT_CRLF:
          line += '\r\n'         #make it a CR+LF
        elif self.__convert_nl == self.CONVERT_CR:
          line += '\r'           #make it a CR
        elif self.__convert_nl == self.CONVERT_LF:
          line += '\n'           #make it a LF
        else:
          line += data
        if self._echo:
          if self._debug:  print "%r" % line
          else:            print line,
        f.write(line)
        line = ''
      else:
        if fnStart:
          fnRemote += data
        else:
          line += data

    f.close()                           # In case Exit

    # Indicate we're done with the file by renaming it to the target name
    if filename is None:
      filename = fnRemote
    if len(filename) == 0:
      fnTuple  = fn.split('.')
      tijd     = str(time.time()).split('.')
      filename = fnTuple[0] + "_" + tijd[0] + tijd[1] + "." + fnTuple[1]
    # Todo: Deal path issues here
    os.rename(fn, filename)

    # ToDo: Probably should delete zero length files here

    return self.__quit


if __name__ == "__main__":
  def senderUsage():
    """!\brief Return sender program usage information
    """
    s = '/'
    if os.name is 'nt':  s = '\\'
    name = sys.argv[0].split('.')[-2].split(s)[-1]
    return "\n\n%s is used to copy a file over a serial port connection.\n" % \
           (name)

  def receiverUsage():
    """!\brief Return receiver program usage information
    """
    s = '/'
    if os.name is 'nt':  s = '\\'
    name = sys.argv[0].split('.')[-2].split(s)[-1]
    return """\n
    %s is used to receive data from a serial port
    connection and store it in a file.  The file's name will be:

    - command line value (always take precedence)
    - a filename transmitted by the other side
    - %s_<timestamp>.txt (default if neither side specifies a name)

    """ % (name, name)

  def senderTest(cfgFile, filename):
    """!\brief Test code that copies a file over a serial port connection.

    \param cfgFile  - The filename of a .cfg file
    \param filename - The name of the file to be copied
    """
    config = ConfigParser()
    config.read(cfgFile)

    sender = SerialSender(config)
    sender.begin(filename.split('/')[-1])   # Just the filename, not the path

    f = open(filename, "r")
    for line in f.xreadlines():
      sender.send(line)

    sender.end()
    f.close()

  def receiverTest(cfgFile, filename):
    config = ConfigParser()
    config.read(cfgFile)

    receiver = SerialReceiver(config)

    # Keep reading files until somebody aborts us
    while True:
      receiver.receive(filename)

    receiver.close()

  def handleOptions(mandatory, optional=[], switches=[], usage=None):
    from LICOS.util.gOptions import Options

    if usage is None:
      def usage():  pass

    options = Options(mandatory, optional, switches)
    try:
      options.parse()
    except Exception, msg:
      options.usage(usage())
      raise Exception, msg

    return options

  #--
  # Options, first list mandatory, second list optional
  if False: #True:
    options = handleOptions(["config", "filename"], usage=senderUsage)

    senderTest(options.config, options.filename)
  else:
    options = handleOptions(["config"], ["filename"], usage=receiverUsage)

    receiverTest(options.config, options.filename)

---