source: moxy/trunk/src/moxy/m_axis.py @ 833

#!/usr/bin/env python

########### SVN repository information ###################
# $Date: 2012-04-26 15:30:26 +0000 (Thu, 26 Apr 2012) $
# $Author: jemian $
# $Revision: 833 $
# $URL$
# $Id: m_axis.py 833 2012-04-26 15:30:26Z jemian $
########### SVN repository information ###################

'''
details about a model single axis

Copyright (c) 2009 - 2011, UChicago Argonne, LLC.
See LICENSE file for details.
'''


# - - - - - - - - - - - - - - - - - - Imports


import m_pv
import paxis


# - - - - - - - - - - - - - - - - - - Global


__svnid__ = "$Id: m_axis.py 833 2012-04-26 15:30:26Z jemian $"


# - - - - - - - - - - - - - - - - - - class

class SingleAxis:
    '''
    A single axis connects to several EPICS PVs.
    Since the axis may not be an EPICS motor record,
    we allow them to be completely separate.
    
    :param str name: short name for this axis
    :param obj axis: Either a Motor object or a PAxis object
    '''
    
    def __init__(self, name, val, isMotorRecord = True,
                 desc = '', dmov = '', egu = '', rbv = '', stop = ''):
        '''
        Constructor
    
        :param str name: short name for this axis
        :param str val: EPICS PV name of the target position of this axis
        :param bool isMotorRecord: True if val refers to an EPICS motor record
        :param str rbv: EPICS PV name of the readback position of this axis
        :param str stop: EPICS PV name of the stop field of this axis, writing 1 to this PV will stop the axis if it is moving
        :param str dmov: EPICS PV name of the done-moving field of this axis, 1 if axis is not moving, 0 if axis is moving
        :param str desc: EPICS PV name of the description field of this axis
        :param str egu: EPICS PV name of the engineering units field of this axis
        '''
        self.name = name
        self.isMotorRecord = isMotorRecord
        self.axis = None
        if isMotorRecord:
            pv = m_pv.GetBasePv(val)
            if not m_pv.pvIsAvailable(pv, timeout = 0.5):
                raise Exception, "Could not connect with PV: " + pv
            if not m_pv.pvIsMotorRec(pv):
                raise Exception, "%s.RTYP = %s, must be 'motor'" % (pv, m_pv.GetRTYP(pv))
            self.axis = paxis.Motor(val)
        else:
            # elegant, but hard to read, need better feedback if PV not found
            #status = map(m_pv.pvIsAvailable, (val, rbv, desc, egu, dmov, stop))
            #ready = reduce(lambda x, y: x and y, status)
            for pv in (val, rbv, desc, egu, dmov, stop):
                if not m_pv.pvIsAvailable(pv):
                    raise Exception, "Could not connect with PV: " + pv
            self.axis = paxis.PAxis(val, rbv, desc, egu, dmov, stop)
        if self.axis is None:
            raise Exception, "Did not connect axis: " + name
        self.callback_ids = {}
    
    def move(self, position):
        '''tell EPICS to move this axis'''
        self.axis.VAL = position
    
    def stopMotion(self):
        '''tell EPICS to stop moving this axis'''
        self.axis.stop()
    
    def connect(self, handler = None):
        '''connect the callbacks'''
        if len(self.callback_ids) > 0:
            self.disconnect()
        for field in ('VAL', 'RBV', 'DESC', 'EGU', 'DMOV', 'STOP'):
            # remember the callback id
            cb = self.axis.start_callback(field, handler, {'field': field})
            self.callback_ids[field] = cb
        self._connected = True
    
    def disconnect(self):
        '''disconnect all the PVs with EPICS'''
        for field in self.callback_ids.keys():
            self.axis.stop_callback(field, self.callback_ids[field])
        self.callback_ids = {}
        self._connected = False


# - - - - - - - - - - - - - - - - - - methods


def __demo_waitmove(axis, position, seconds):
    '''move this axis and wait for a fixed time (showing EPICS CA monitors)'''
    from time import sleep
    print "moving", axis._pvs['VAL'], " to ", position
    axis.move(position)
    for i in range(seconds, 0, -1):
        print i, " seconds remain"
        sleep(1)
    print "done waiting"


def handler(**kw):
    '''trivial example callback handler'''
    print kw['pvname'], kw['field'], kw['value']


def _demo_():
    '''demonstrate use of this module'''
    from time import sleep
    pv = "prj:m1"
    
    motor = SingleAxis( name = "test axis", val = pv, isMotorRecord = True, )
    motor.connect(handler=handler)
    name = motor.axis.DESC
    motor.axis.DESC = "new name"
    sleep(1)
    motor.axis.DESC = name
    sleep(1)

    # now try a couple moves
    __demo_waitmove(motor.axis, position=5.5, seconds=5)
    __demo_waitmove(motor.axis, position=1.1, seconds=5)
    motor.disconnect()
    del motor

    # - - - -
    
    non_motor = SingleAxis( name = "second axis", isMotorRecord = False, 
                            val = pv + '.VAL', 
                            rbv = pv + '.RBV', 
                            desc = pv + '.DESC', 
                            egu = pv + '.EGU', 
                            dmov = pv + '.DMOV', 
                            stop = pv + '.STOP', 
                            )
    non_motor.connect(handler=handler)
    __demo_waitmove(non_motor.axis, position=5.5, seconds=5)
    __demo_waitmove(non_motor.axis, position=1.1, seconds=5)
    non_motor.disconnect()

# - - - - - - - - - - - - - - - - - - main


if __name__ == '__main__':
    _demo_()