mirrorDeviceSet.py

Go to the documentation of this file.

from __future__ import division, absolute_import

from twistedActor import DeviceSet, LinkCommands, UserCmd

__all__ = ["MirrorDeviceSet"]

class MirrorDeviceSet(DeviceSet):
    SlotList = ("prim", "sec", "tert")
    """!Mirror devices
    """
    DefaultTimeLim = 2
    def __init__(self, actor, devList):
        """!Construct an MirrorDeviceSet

        @param[in] actor  actor (instance of twistedActor.BaseActor);
            used for writeToUsers in this class, and subclasses may make additonal use of it
        @param[in] devList  prim, sec, and tert device; any can be None if it does not exist

        @throw RuntimeError if len(devList) != 3
        """
        DeviceSet.__init__(self,
            actor = actor,
            slotList = self.SlotList,
            devList = devList,
            connStateKeyword = "mirrorConnState",
        )
        # start keyword forwarding
        for dev in devList:
            if dev != None:
                dev.startKWForwarding()

    def connectOrInit(self, slotList=None, userCmd=None, timeLim=DefaultTimeLim):
        """!Connect and stop axis controllers

        Specified axis controllers that are already connected are simply stopped.

        @param[in] slotList  collection of slot names, or None for all filled slots
        @param[in] userCmd  user command whose set state is set to Done or Failed when all device commands are done;
            if None a new UserCmd is created and returned
        @param[in] timeLim  time limit for command (sec)
        @return userCmd: the supplied userCmd or a newly created UserCmd
        """
        # print "connectOrInit(slotList=%r, userCmd=%r)" % (slotList, userCmd)
        expSlotList = self.expandSlotList(slotList)
        connSlotList = [slot for slot in expSlotList if self.get(slot).isConnected]
        notConnSlotList = list(set(expSlotList) - set(connSlotList))
        initUserCmd = self.init(connSlotList, timeLim=timeLim)
        connUserCmd = self.connect(notConnSlotList, timeLim=timeLim)
        if userCmd is None:
            userCmd = UserCmd()
        LinkCommands(userCmd, (initUserCmd, connUserCmd))
        return userCmd

    def status(self, slotList=None, userCmd=None, timeLim=DefaultTimeLim):
        """!Get status from the mirror controllers

        @param[in] slotList  collection of slot names, or None for all filled slots
        @param[in] userCmd  user command whose set state is set to Done or Failed when all device commands are done;
            if None a new UserCmd is created and returned
        @param[in] timeLim  time limit for command (sec)
        @return userCmd: the supplied userCmd or a newly created UserCmd

        @throw RuntimeError if slotList has empty or non-existent slots
        """
        self.showConnState(userCmd=userCmd)
        return self.startCmd(cmdStrOrList="status", slotList=slotList, userCmd=userCmd, timeLim=timeLim)

    def init(self, slotList=None, userCmd=None, timeLim=DefaultTimeLim):
        """!Initialize the mirror controllers

        @param[in] slotList  collection of slot names, or None for all filled slots
        @param[in] userCmd  user command whose set state is set to Done or Failed when all device commands are done;
            if None a new UserCmd is created and returned
        @param[in] timeLim  time limit for command (sec)
        @return userCmd: the supplied userCmd or a newly created UserCmd

        @throw RuntimeError if slotList has empty or non-existent slots
        """
        return self.startCmd(cmdStrOrList="stop", slotList = slotList, userCmd=userCmd, timeLim=timeLim)