SimCtl.py

Go to the documentation of this file.

import json
import datetime
 
from robot.api import logger
from robot.utils import DotDict
from robot.utils import timestr_to_secs
 
from elt import pymal
from ModDaqsimif.Daqsimif.SimCtl import SimCtlSync
 
 
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 to_list(arg):
    if isinstance(arg, list):
        return arg
    return arg.split(" ")
 
 
class SimCtl:
    """Robot library providing keywords for the SimCtl 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.mal = pymal.loadMal("zpb", {})
        self.factory = pymal.CiiFactory.getInstance()
        self.factory.registerMal("zpb", self.mal)
        logger.debug("Loading MAL and creating factory done")
 
    def simctl_create_clients(self, uris, timeout=None, replace_from=None, replace_to=None):
        """Creates a list of clients with a given ReplyTime timeout."""
        logger.info("Create clients raw arg: %r, type=%s" % (uris, type(uris)))
        timeout = timestr_to_secs(timeout) if timeout is not None else 3
        clients = []
        # uris may be a single string in the OCM format name@rr://URI,ps://URI. This is taken care
        # of automatically
        for uri in to_list(uris):
            uri = get_rr_uri(uri)
            if replace_from and replace_to:
                uri = uri.replace(replace_from, replace_to)
            logger.info("Create client for uri %s with timeout %d" % (uri, timeout))
            client = self.factory.getClient(
                uri, SimCtlSync, pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout))
            )
            clients.append(client)
        return clients
 
    def simctl_create_client(self, uri, timeout=None):
        """Creates a client with a given ReplyTime timeout."""
        timeout = timestr_to_secs(timeout) if timeout is not None else 3
        uri = get_rr_uri(uri)
        logger.info("Create client for uri %s with timeout %d" % (uri, timeout))
        client = self.factory.getClient(
            uri, SimCtlSync, pymal.rr.qos.ReplyTime(datetime.timedelta(seconds=timeout))
        )
        return client
 
    def __setup(self, clients, arg):
        logger.info("Sending setup: %s to %d clients" % (arg, len(clients)))
        replies = []
        for client in clients:
            reply = client.Setup(arg)
            logger.info("Received reply: %s" % reply)
            replies.append(json.loads(reply, object_pairs_hook=DotDict))
        return replies
 
    def simctl_setup(self, clients, arg):
        if not isinstance(clients, list):
            return self.__setup([clients], arg)[0]
        return self.__setup(clients, arg)
 
    def simctl_reset(self, clients):
        logger.info("Sending reset")
        if not isinstance(clients, list):
            return self.__reset([clients])[0]
        return self.__reset(clients)
 
    def __reset(self, clients):
        logger.debug("Sending reset")
        replies = []
        for client in clients:
            replies.append(client.Reset())
        logger.debug("Received replies")
        return replies
 
    def simctl_force_exit(self, clients):
        if not isinstance(clients, list):
            self.__force_exit([clients])[0]
        self.__force_exit(clients)
 
    def __force_exit(self, clients):
        logger.debug("Sending ForceExit. A reply is not expected.")
        for client in clients:
            client.ForceExit()