usbtmc_pipe.py

"""USBTMC pipe script

This script is used to pipe commands and queries to and from a USBTMC
device and is intended to be run using the following socat command::

  socat tcp-listen:5025,fork,reuseaddr,crnl,tcpwrap=script\
   EXEC:"python usbtmc_pipe.py",su-d=pi,pty,echo=0

"""

from __future__ import print_function
import os
import sys
import time
import re
import argparse

class USBTMCDevice(object):
    """Object representation of /dev/usbtmcX."""
    def __init__(self, dev_file, terminator="\n", echo=False):
        self.terminator = terminator
        self.echo = echo
        try:
            self.device = os.open(dev_file, os.O_RDWR)
        except:
            print("Error opening device file " + dev_file + \
                  ". Does it exist?")

    def __enter__(self):
        return self

    def __exit__(self, type_, value, traceback):
        self.close()

    def write(self, cmd):
        if self.echo:
            sys.stderr.write(self.get_time() + ">>> " + repr(cmd) + "\n")
        os.write(self.device, cmd + device.terminator)

    def read(self, bufsize=65536):
        result = os.read(self.device, bufsize).strip()
        if self.echo:
            if len(result) < 256:
                sys.stderr.write(self.get_time() + "<<< " + repr(result) + "\n")
            else:
                sys.stderr.write(self.get_time() + "<<< " + \
                                 repr(result)[:256] + \
                                 " ... (long output suppressed)\n")
        sys.stdout.write(result)

    def close(self):
        os.close(self.device)

    def get_time(self):
        return "[" + time.strftime('%Y-%m-%d %H:%M:%S') + "] "

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Pipe commands to/from a USBTMC device.")
    parser.add_argument(
        "-t", "--terminator", dest="term_str", default='\n',
        help="Specify the termination character. Default = '\\n'")
    args = parser.parse_args()
    
    with USBTMCDevice("/dev/usbtmc0",
                      terminator=args.term_str,
                      echo=True) as device:
        while True:
            cmd = raw_input()
            cmd = cmd.strip()
            device.write(cmd)
            if re.compile('\?').search(cmd):
                device.read()

See here for a full description.