DpmCtl.py

Go to the documentation of this file.

import datetime
import time
from typing import List
 
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,
    StorageStatus,
    Exception,
    DaqException,
)
 
from ModDaqif.Daqif.DpmControl import DpmControlSync
from ModDaqif.Daqif.DpmDaqControl import DpmDaqControlSync
 
 
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(),
        message=r.getMessage(),
        timestamp=r.getTimestamp(),
    )
 
 
def dict_from_storagestatus(r: StorageStatus):
    return DotDict(
        free=r.getFree(),
        available=r.getAvailable(),
        capacity=r.getCapacity(),
    )
 
 
class Clients:
    daq = None
    dpm = None
 
 
class DpmCtl:
    """Robot library providing keywords for the daqif.DpmDaqControl and daqif.DpmControl
    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 dpmctl_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,
            DpmDaqControlSync,
            pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout)),
        )
        clients.dpm = self.factoryfactory.getClient(
            "%s/dpm" % root_uri,
            DpmControlSync,
            pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout)),
        )
        return clients
 
    def dpmctl_exit(self, clients: Clients) -> str:
        logger.info("Sending DpmControl.Exit(%s) request")
        clients.dpm.Exit()
 
    def dpmctl_query_storage_status(self, clients: Clients) -> StorageStatus:
        logger.info("Sending DpmDaqControl.QueryStorageStatus() request")
        r = clients.dpm.QueryStorageStatus()
        return dict_from_storagestatus(r)
 
    def dpmctl_queue_daq(self, clients: Clients, dp_spec: str) -> DaqReply:
        logger.info(
            "Sending DpmDaqControl.QueueDaq(%s) request",
            dp_spec[0:10].replace("\n", " "),
        )
        try:
            r = clients.daq.QueueDaq(dp_spec)
            return dict_from_daqreply(r)
        except DaqException as e:
            logger.info("QueueDaq() failed: %s" % e.getMessage())
            raise
 
    def dpmctl_abort_daq(self, clients: Clients, daq_id: str) -> DaqReply:
        logger.info("Sending DpmDaqControl.AbortDaq(%s) request", daq_id)
        r = clients.daq.AbortDaq(daq_id)
        return dict_from_daqreply(r)
 
    def dpmctl_get_daq_status(self, clients: Clients, daq_id: str) -> DaqStatus:
        logger.info("Sending DpmDaqControl.GetStatus(%s) request", daq_id)
        r = clients.daq.GetDaqStatus(daq_id)
        return dict_from_daqstatus(r)
 
    def dpmctl_get_active_daqs(self, clients: Clients) -> List[DaqStatus]:
        logger.info("Sending DpmDaqControl.GetActive() request")
        daqs = clients.daq.GetActiveDaqs()
        return [dict_from_daqstatus(daq) for daq in daqs]
 
    def dpmctl_await_error(self, clients: Clients, daq_id: str, timeout: str = None) -> DaqStatus:
        logger.info("Sending DpmDaqControl.GetDaqStatus() until timeout or erro")
        timeout = timestr_to_secs(timeout) if timeout is not None else 3
        until = time.monotonic() + timeout
        while True:
            if time.monotonic() > until:
                raise TimeoutError("Timeout waiting for DAQ error")
            r = clients.daq.GetDaqStatus(daq_id)
            if r.getError():
                return dict_from_daqstatus(r)
            time.sleep(0.25)