pygpibtoolkit/gpibcontroller.py

#
"""
A general purpose GPIB controller based on prologix GPIB device
"""
import sys
import threading
import time
from inspect import isclass

from serial.serialutil import SerialException
from prologix import GPIB, GPIB_CONTROLLER, GPIB_DEVICE
from pygpibtoolkit.pygpib import AbstractCommand

class AbstractGPIBDeviceMetaclass(type):
    def __new__(mcs, name, bases, classdict):
        # Automatically add commands to Device classes.
        # Commands which name starts with '_' are ignored.
        # Only commands defined in the same module as the device
        # are added to the device.
        # Commands are descriptors (see pygpib.py)
        new_cls = type.__new__(mcs, name, bases, classdict)
        modname = new_cls.__module__
        module = sys.modules[modname]
        for pname, param in module.__dict__.items():
            if not pname.startswith('_') and isclass(param) and \
                    issubclass(param, AbstractCommand) and \
                    param.__module__ == modname:
                setattr(new_cls, pname, param())
        return new_cls

class AbstractGPIBDevice(object):
    __metaclass__ = AbstractGPIBDeviceMetaclass
    _accepts = []

    @classmethod
    def accepts(cls, idn):
        return idn in cls._accepts
    
    def __init__(self, idn, address, controller):
        self._idn = idn
        self._address = address
        self._controller = controller
        self._use_cache = True
        
        self._cache = {}
        for pname, param in self.__class__.__dict__.items():
            if isinstance(param, AbstractCommand) and not param._readonly:
                self._cache[pname] = param._init_value
    def use_cache(self, use=None):
        if use is None:
            return self._use_cache
        self._use_cache = bool(use)
        
    def _get(self, name):
        if self._use_cache and name in self._cache and self._cache[name] is not None:
            return self._cache[name]
        
        param = getattr(self.__class__, name) 
        cmd = param.build_get_cmd()
        value = self._controller.send_command(self._address, cmd)
        value = param.convert_from(value)
        if name in self._cache:
            self._cache[name] = value
        return value
            
    def _set(self, name, value):
        param = getattr(self.__class__, name) 
        cmd = param.build_set_cmd(value)
        res = self._controller.send_command(self._address, cmd)
        if name in self._cache:
            self._cache[name] = value
        return res
    
    def manage_srq(self, statusbyte):
        pass

    def send_command(self, cmd):
        return self._controller.send_command(self._address, cmd).strip()
    
class GPIBDeviceRegister(object):
    def __init__(self):
        self._registry = []
    def register_manager(self, mgr):
        self._registry.append(mgr)
    def get_manager(self, idn):
        for mgr in self._registry:
            if mgr.accepts(idn):
                return mgr
        return None
    def get_idn_cmds(self):
        return [mgr._idn for mgr in self._registry]
    def __str__(self):
        msg = "<GPIBDeviceRegister: %s managers\n"%len(self._registry)
        for mgr in self._registry:
            msg += "  %s: %s\n"%(mgr.__name__, ", ".join(mgr._accepts))
        msg += ">"
        return msg
    
deviceRegister = GPIBDeviceRegister()

class GenericGPIBDevice(AbstractGPIBDevice):
    _idn = "*IDN?"
    
deviceRegister.register_manager(GenericGPIBDevice)

class GPIBController(object):
    """
    The main GPIB Controller In Charge.

    This is responsible for any communication with the registered
    devices.

    It hold a thread reponsible for communication with devices,
    performing spolls regularly. It can react and call callbacks on
    events detected while spolling.

    It also keeps a command queue which devices/user program can fill
    with GPIB and device commands, which are executed ASAP (ie. as
    soon as the GPIB bus if free and the device states itself as
    ready.)
    
    """
    def __init__(self, device="/dev/ttyUSB0", controller_address=21):
        self._address = controller_address
        try:
            self._cnx = GPIB(device)
        except SerialException:
            self._cnx = None
        self._devices = {}
        self._setup_threading_system()

    def __del__(self):
        print "deleting controller"
        self._stop.set()
        
    def _setup_threading_system(self):
        self._n_cmds = 0
        self._n_cmds_lock = threading.RLock()
        self._cmd_queue = {}
        self._cmd_condition = threading.Condition()
        self._results_queue = {}
        self._results_condition = threading.Condition()
        self._loop_interrupter = threading.Event()
        self._stop = threading.Event()
        self._cnx_thread = threading.Thread(name="GPIBConnectionThread",
                                            target=self._cnx_loop)
        self._cnx_thread.start()
        self._loop_interrupter.set()
        
    def _check_srq(self):
        if self._cnx and self._cnx.check_srq():
            addrs = sorted(self._devices.keys())
            polled = self._cnx.poll(addrs)
            for add in addrs:
                if polled[add] & 0x40:
                    print "device %s (#%s) requested attention"%(self._devices[add]._idn, add)
                    # TODO: check what to do...
                    self._devices[add].manage_srq(polled[add])
        
    def _cnx_loop(self):
        while 1:
            #sys.stderr.write('.')
            if self._stop.isSet():
                print "_stop set, exiting"
                return
            while not self._loop_interrupter.isSet():
                #sys.stderr.write('|')
                self._loop_interrupter.wait()
            self._loop_interrupter.clear()
            # first we check for SRQ and dispatch its management
            self._check_srq()
            # then we check if we have any pending commands.
            self._run_one_cmd()
            self._loop_interrupter.set()
            time.sleep(0.1)

    def _run_one_cmd(self):
        cond = self._cmd_condition
        cond.acquire()
        if len(self._cmd_queue) == 0:
            cond.release()
            return
        n_cmd = min(self._cmd_queue.keys())
        addr, cmd, cb = self._cmd_queue.pop(n_cmd)
        cond.release()

        # TODO: check device is RDY etc.
        resu = self._cnx.send_command(cmd, addr)
        # TODO: check any error etc.
        if isinstance(cb, threading._Event):
            cond = self._results_condition
            cond.acquire()
            self._results_queue[n_cmd] = resu
            cond.notifyAll()
            cond.release()
            cb.set()
        else:
            if callable(cb):
                cb(resu)

    def stop(self, *args):
        self._stop.set()
    
    def send_command(self, addr, cmd, sync=True, cb=None):
        if cb is not None and callable(cb):
            sync = False
        self._n_cmds_lock.acquire()
        self._n_cmds += 1
        n_cmd = self._n_cmds
        self._n_cmds_lock.release()

        cond = self._cmd_condition
        cond.acquire()
        if sync:
            cb = threading.Event()            
        self._cmd_queue[n_cmd] = (addr, cmd, cb)
        cond.notify()
        cond.release()
        if sync:
            cb.wait()
            cond = self._results_condition
            cond.acquire()
            while n_cmd not in self._results_queue:
                cond.wait()
            resu = self._results_queue.pop(n_cmd)
            cond.release()
            return resu
        
    def detect_devices(self):
        """
        Perform a Serial Poll on all addresses to detect alive devices
        connected on the GPIB bus.
        """
        while not self._loop_interrupter.isSet():
            self._loop_interrupter.wait()
        self._loop_interrupter.clear()
        try:
            spoll = self._cnx.poll()
            for address in spoll:
                if address not in self._devices:
                    # found a new device
                    for id_str in deviceRegister.get_idn_cmds():
                        idn = self._cnx.send_command(id_str, address).strip()
                        if idn:
                            self.register_device(address, idn)
                            break
                        else:
                            # redo a spoll, it should have set the err bit
                            poll = self._cnx.poll(address)
                            if not poll & 0x20: # 0x20 == ERR bit
                                print "Strange, device %d did not answer to a %s command without setting its ERR bit"%(address, id_str)
                    else:
                        print "Can't retrieve IDN of device at address ", address
        finally:
            self._loop_interrupter.set()
                    
    def register_device(self, address, idn):
        """
        Register a device manager for device at given GPIB
        address. The manager is retrived from device registry.
        """
        devicemgr = deviceRegister.get_manager(idn)
        if devicemgr is not None:
            devicemgr = devicemgr(idn, address, self)
            self._devices[address] = devicemgr
            return devicemgr
        else:
            print "can't find a manager for", repr(idn)
            print  deviceRegister._registry

    def idn(self, address):
        """
        Query identity of the addressed device.
        """
        return self.send_command(address, self._devices[address].__class__._idn).strip()
    
    def status(self, address):
        """
        Query status byte for device at given address
        """
        while not self._loop_interrupter.isSet():
            self._loop_interrupter.wait()
        self._loop_interrupter.clear()
        try:
            return self._cnx.poll(address)
        finally:
            self._loop_interrupter.set()