OcmCtl.py

Go to the documentation of this file.

import datetime
 
from robot.api import logger
from robot.utils import DotDict
from robot.utils import timestr_to_secs
 
from elt import pymal
from ModDaqif.Daqif import (
    DaqReply,
    DaqStatus,
)
from ModDaqif.Daqif.OcmDaqControl import OcmDaqControlSync
from ModStdif.Stdif.StdCmds import StdCmdsSync
 
 
def get_rr_uri(arg):
    if "@" in arg:
        # Split name@rr-uri,ps-uri
        uris = arg.split("@")[1]
        return uris.split(",")[0]
    return arg
 
 
def dict_from_daqreply(r: DaqReply):
    return DotDict(id=r.getId(), error=r.getError())
 
 
def dict_from_daqstatus(r: DaqStatus):
    # Note: To keep compatibility of C++, return only the stem from enum attr name
    return DotDict(
        id=r.getId(),
        state=str(r.getState()).split(".")[1],
        subState=str(r.getSubState()).split(".")[1],
        error=r.getError(),
        result=r.getResult(),
        message=r.getMessage(),
        timestamp=r.getTimestamp(),
    )
 
 
class Clients:
    daq = None
    std = None
 
 
class OcmCtl:
    """Robot library providing keywords for the daqif.OcmDaqControl and stdif interface"""
 
    # Global scope is necessary since MAL will shutdown protobuf library randomly
    # otherwise, which eventually lead to segfaults. See ECII-304.
    ROBOT_LIBRARY_SCOPE = "GLOBAL"
 
    def __init__(self):
        logger.debug("Loading MAL and creating factory")
        self.malmal = pymal.loadMal("zpb", {})
        self.factoryfactory = pymal.CiiFactory.getInstance()
        self.factoryfactory.registerMal("zpb", self.malmal)
        logger.debug("Loading MAL and creating factory done")
 
    def ocmctl_create(self, root_uri, timeout=None):
        """Creates RR clientss with a given ReplyTime timeout."""
        timeout = timestr_to_secs(timeout) if timeout is not None else 5
        root_uri = get_rr_uri(root_uri)
        logger.info("Create client for uri %s with timeout %d" % (root_uri, timeout))
        clients = Clients()
        clients.daq = self.factoryfactory.getClient(
            "%s/daq" % root_uri,
            OcmDaqControlSync,
            pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout)),
        )
        clients.std = self.factoryfactory.getClient(
            "%s/std" % root_uri,
            StdCmdsSync,
            pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout)),
        )
        return clients
 
    def ocmctl_daq_get_status(self, clients: Clients, daq_id: str):
        logger.info("OcmDaqControl.GetStatus(%s): Sending request", daq_id)
        r = clients.daq.GetStatus(daq_id)
        logger.info("OcmDaqControl.GetStatus(%s): Got reply", daq_id)
        return dict_from_daqstatus(r)
 
    def ocmctl_daq_forceabort(self, clients: Clients, daq_id: str):
        logger.info("OcmDaqControl.ForceAbort(%s): Sending request", daq_id)
        r = clients.daq.ForceAbortDaq(daq_id)
        logger.info("OcmDaqControl.ForceAbort(%s): Got reply", daq_id)
        return dict_from_daqreply(r)
 
    def ocmctl_std_get_state(self, clients: Clients):
        logger.info("StdCmds.GetState(): Sending request")
        r = clients.std.GetState()
        logger.info("StdCmds.GetState(): Got reply")
        return r
 
    def ocmctl_daq_get_active_list(self, clients: Clients):
        logger.info("OcmDaqControl.GetActive(): Sending request")
        daqs = clients.daq.GetActiveList()
        logger.info("OcmDaqControl.GetActive(): Got reply")
        return [dict_from_daqstatus(daq) for daq in daqs]