simulator.py

Go to the documentation of this file.

"""
@file
@ingroup daq_sim_metadaqsim
@copyright 2021 ESO - European Southern Observatory
 
@brief Implements simulation of metadaqif
"""
# pylint: disable=no-name-in-module,import-error,no-self-use
import os
import json
import logging
import calendar
import inspect
 
from ModMetadaqif.Metadaqif import (
        DaqReply,
        DaqStopReply,
        DaqState,
        DaqStatus,
        DaqException,
        )
 
from ModMetadaqif.Metadaqif.MetaDaq import MetaDaqSyncService
from ModDaqsimif.Daqsimif import Error as SimCtlError
 
 
class Daq:
    """Holds simulated status for DAQ
    """
 
    def __init__(self, id):
        self.id = id
        self.state = DaqState.NotStarted
        self.keywords = ""
        self.files = []
 
 
class Simulator:  # pylint: disable=invalid-name
    """Simulator that implements Modmetadaqsimif and sets up
    simulation behaviour for Simulator.
 
    Registered hooks are run once and then removed before executed.
    Each hook is a callable that takes the result from the default implementation
    as well as the arguments given to the original implementation.
    The hook can then choose to modify the existing result or replace it.
    """
 
    def __init__(self, name, mal):
        self.name = name
        self.mal = mal
        self.current_daq = None
        self.daqs = {}  # id: status
        # Oneshot commands
        self.sim_hooks = {}
 
    def StartDaq(self, daq_id: str) -> DaqReply:
        """@implements MetaDaq.StartDaq
        """
        logging.info("Request: StartDaq(%s)", daq_id)
        if daq_id in self.daqs:
            raise DaqException(daq_id, "Data acquisition with id already exist")
        self.current_daq = daq = Daq(daq_id)
        daq.state = DaqState.Acquiring
        self.daqs[daq_id] = daq
        result = self.make_daq_reply(daq_id)
        return self.run_sim_hook(result, daq_id)
 
    def StopDaq(self, daq_id: str) -> DaqStopReply:
        """@implements MetaDaq.StopDaq
        """
        logging.info("Request: StopDaq(%s)", daq_id)
        daq = self.get_daq(daq_id)
        daq.files = ["/tmp/%s.fits" % daq_id]
        result = self.make_daq_stop_reply(daq_id)
        return self.run_sim_hook(result, daq_id)
 
    def AbortDaq(self, daq_id: str) -> DaqReply:
        """@implements MetaDaq.AbortDaq
        """
        logging.info("Request: AbortDaq(%s)", daq_id)
        daq = self.get_daq(daq_id)
        daq.state = DaqState.Failed
        self.daqs[daq_id] = daq
        result = self.make_daq_reply(daq_id)
        return self.run_sim_hook(result, daq_id)
 
    def GetDaqStatus(self, daq_id: str) -> DaqStatus:
        """@implements MetaDaq.GetDaqStatus
        """
        logging.info("Request: GetDaqStatus(%s)", daq_id)
        result = self.make_daq_status(daq_id)
        return self.run_sim_hook(result, daq_id)
 
    def get_daq(self, daq_id) -> Daq:
        """Get daq or raise DaqException.
        """
        if daq_id in self.daqs:
            return self.daqs[daq_id]
        raise DaqException(daq_id, "Unknown daq id '%s'" % daq_id)
 
    def make_daq_reply(self, daq_id) -> DaqReply:
        logging.debug("Creating return type: DaqReply")
        reply = self.mal.createDataEntity(DaqReply)
        reply.setId(daq_id)
        logging.debug("Creating return type: Done")
        return reply
 
    def make_daq_status(self, daq_id) -> DaqStatus:
        """Create DaqStatus instance
        """
        daq = self.get_daq(daq_id)
        status = self.mal.createDataEntity(DaqStatus)
        status.setId(daq.id)
        status.setState(daq.state)
        status.setFiles(daq.files)
        status.setTimestamp(calendar.timegm())
        return status
 
    def make_daq_stop_reply(self, daq_id) -> DaqStopReply:
        """Create DaqStopReply instance
        """
        daq = self.get_daq(daq_id)
        reply = self.mal.createDataEntity(DaqStopReply)
        reply.setId(daq.id)
        reply.setFiles(daq.files)
        reply.setKeywords(daq.keywords)
        return reply
 
    def add_sim_hook(self, cmd_name: str, hook):
        """Add the simulation hook `hook` for the specified command `cmd_name` (e.g.
        "StartDaq").
 
        Any previous hooks will be replaced.
        """
        self.sim_hooks[cmd_name] = hook
 
    def reset(self):
        """Reset simulator to initial default state.
        """
        self.clear_sim_hooks()
        self.daqs = {}
        self.current_daq = None
 
    def clear_sim_hooks(self):
        """Remove all simulation hooks.
        """
        self.sim_hooks = {}
 
    def run_sim_hook(self, result, *args):
        """Runs simulation hook that may modify or replace the default implementation.
 
        The hook to execute is automatically determined from the call stack, so
        this should only ever be executed directly from the service command
        implementation method.
        """
        cmd_name = inspect.stack()[1][3]
        hook = self.sim_hooks.get(cmd_name)
        logging.debug("Checking for hook for command %s", cmd_name)
        if hook:
            logging.debug("Running simulation hook for command %s", cmd_name)
            del self.sim_hooks[cmd_name]
            return hook(result, *args)
        return result
 
 
class SimulatorCtl:  # pylint: skip-file
    """Simulator controller that implements Modmetadaqsimif and sets up
    simulation behaviour for Simulator.
    """
 
    def __init__(self, name, mal, server):
        self.name = name
        self.mal = mal
        self.server = server
 
        self.sim = Simulator(name, mal)
        logging.info("Registering service 'daq'")
        self.server.registerService("daq", MetaDaqSyncService, self.sim)
 
    def Setup(self, spec):
        """@implements SimCtl.Setup.
        """
        logging.info('SimCtl.Setup: spec="%s"', spec)
        s = json.loads(spec)
        try:
            cmd = s.get("command")
            action = s.get("action")
            if action == "throw":
                def throw(res, *args):
                    raise DaqException(self.sim.current_daq.id, "Simulated error")
                self.sim.add_sim_hook(cmd, throw)
            else:
                raise SimCtlError("Unknown action '%s' provided" % action)
            logging.info('CimCtl.Setup: Returning empty string')
            return json.dumps("")
        except SimCtlError:
            raise
        except Exception as e:
            raise SimCtlError("\"%s\"" % str(e))
 
    def Reset(self):
        """@implements SimCtl.Reset
        """
        try:
            self.sim.reset()
        except Exception as e:
            raise SimCtlError(str(e))
 
    def ForceExit(self):
        """@implements SimCtl.ForceExit.
        """
        # Notify main thread that it should exit
        logging.warning("Force Exiting without cleanup!")
        os._exit()  # pylint: disable=protected-access