source: moxy/trunk/src/moxy/model/pv.py @ 835

#!/usr/bin/env python

########### SVN repository information ###################
# $Date: 2012-04-26 15:54:14 +0000 (Thu, 26 Apr 2012) $
# $Author: jemian $
# $Revision: 835 $
# $URL$
# $Id: pv.py 835 2012-04-26 15:54:14Z jemian $
########### SVN repository information ###################

'''
details about a model single PV connection

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


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

import epics
from threading import Thread
from time import sleep


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


__svnid__ = "$Id: pv.py 835 2012-04-26 15:54:14Z jemian $"


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

class PvConnection(Thread):
    '''
    A single EPICS Process Variable connection
    
    :param str pvname: full EPICS Process Variable name
    :param channel: PV connection objection
    :type channel: epics.PV() object
    :param int cb_index: callback index number from PyEpics :meth:`add_callback()`
    :param bool connected: Is PV connected with EPICS now?
    :param str connectedText: describes connection state for humans
    '''
    
    def __init__(self, pv = ''):
        Thread.__init__(self)
        self.pvname = pv
        self.channel = None
        self.cb_index = 0
        self.connected = False
        self.connectedText = 'not connected'
        self.ext_callback = None
    
    def connect(self, callback = None, suppliedname = None, **kw):
        '''
        connect to the EPICS PV
        
        The name given in ``self.pvname`` will override a name supplied to this method.
        If ``self.pvname`` is not given, then use ``suppliedname`` if it is given.
        '''
        pvname = self.pvname.strip()
        if len(pvname) == 0 and suppliedname is not None:
            pvname = suppliedname

        if callback is None:
            def callback(pvname=None, value=None, char_value=None, **kwds):
                "generic monitor callback"
                if char_value is None:
                    char_value = repr(value)
                print("%.32s %s %s" % (pvname, epics.pv.fmt_time(), char_value))

        if len(pvname) > 0:    # PV name must not be an empty string
            self.channel = epics.PV(pvname, connection_callback=self.connection_cb)
            if callback is not None:
                self.ext_callback = callback
                self.cb_index = self.channel.add_callback(callback, **kw)
    
    def disconnect(self):
        '''stop monitoring the PV for CA callbacks and drop the channel'''
        if self.channel is not None:
            if self.cb_index != 0:
                self.channel.remove_callback(self.cb_index)
                self.cb_index = 0
            self.channel.disconnect()
            self.connected = False
            self.channel = None
            self.ext_callback = None

    def connection_cb(self, **metadata):
        "PV connection callback"
        self.connected = metadata['conn']       # connection state
        self.connectedText = {True: 'connected', 
                              False: 'not connected'}[self.connected]
        if self.ext_callback is not None:
            self.ext_callback(**metadata)


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


def pvIsAvailable( pv, timeout = None ):
    '''
    Is this PV available?
    
    :param str pv:  EPICS process variable name
    :param float timeout:  seconds to wait before giving up
    :return:  Is this PV available?
    :rtype: bool
    '''
    ch = epics.PV( pv )
    ch.connect( timeout )
    return ch.wait_for_connection( timeout )



def GetBasePv( pv ):
    '''
    Returns the record type of "pv"
    
    :param str pv:  EPICS process variable name
    :return: EPICS record type or None if cannot connect
    :rtype: str
    '''
    if len(pv) == 0:
        return None
    base = pv.split('.')[0]
    return base


def GetRTYP( pv ):
    '''
    Returns the record type of "pv"
    
    :param str pv:  EPICS process variable name
    :return: EPICS record type or None if cannot connect
    :rtype: str
    '''
    base = GetBasePv(pv)
    if pvIsAvailable(base):
        return epics.caget(base + '.RTYP')
    else:
        return None


def pvIsMotorRec( pv ):
    '''
    test if the given PV is a motor record
    
    :param pv: [string] EPICS Process Variable name
    '''
    return GetRTYP(pv) == 'motor'


def __demo_waitmove(pv, position, seconds):
    '''
    Move this motor pv and wait for a fixed time.
    
    .. note: Only use this method for demonstration purposes.
    '''
    print "moving", pv, " to ", position
    epics.caput(pv, position)
    for i in range(seconds, 0, -1):
        print i, " seconds remain"
        sleep(1)
    print "done waiting"


def _demo_():
    '''demonstrate use of this module'''
    pv = "prj:m1"
    if not pvIsAvailable(pv):
        print "Cannot connect to PV = ", pv
        exit(0)
    
    print pv, " is a motor record: ", str( pvIsMotorRec(pv) )
    conn = PvConnection(pv+".RBV")
    conn.connect()

    # now try a couple moves
    __demo_waitmove(pv, position=5,   seconds=5)
    sleep(2.5)
    __demo_waitmove(pv, position=1.1, seconds=5)

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


if __name__ == '__main__':
    _demo_()