simulator.py

Go to the documentation of this file.

"""
@file
@ingroup daq_recifsim
@copyright 2022 ESO - European Southern Observatory
 
@brief Implements simulation of recdaqif
"""
# pylint: skip-file
import inspect
import json
import logging
import os
import time
 
import numpy as np
from astropy.io import fits
from ModDaqsimif.Daqsimif import Error as SimCtlError
from ModRecif.Recif import (ExceptionErr, RecProperties, RecStatus,
                            RecStatusNames, RecWaitSpec, RecWaitStatus,
                            RecWaitStatusNames)
from ModRecif.Recif.RecCmds import RecCmdsSyncService
 
 
class Recording:
    """Holds simulated status for recording"""
 
    def __init__(self, id):
        self.idid = id
        self.dp_filesdp_files = []
        self.files_generatedfiles_generated = 0
        self.start_timestart_time = 1.0
        self.end_timeend_time = 1.0
        self.time_elapsedtime_elapsed = 0.0
        self.remaining_timeremaining_time = 1.0
        self.frames_processedframes_processed = 0
        self.frames_remainingframes_remaining = 1
        self.size_recordedsize_recorded = 0
        self.infoinfo = ""
        self.statusstatus = RecStatusNames.Undefined
 
 
class Simulator:
    """Simulator that implements RecCmdsSyncService 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, dataroot):
        self.namename = name
        self.malmal = mal
        self.datarootdataroot = dataroot
        self.current_reccurrent_rec = None
        self.recordingsrecordings = {}  # id: status
        # Oneshot commands
        self.sim_hookssim_hooks = {}
 
    def RecStart(self, properties: RecProperties) -> RecStatus:
        """@implements RecCmds.StartDaq"""
        rec_id = properties.getId()
        logging.info("Request: StartDaq(%s)", rec_id)
        if rec_id in self.recordingsrecordings:
            raise ExceptionErr("Recording with id already exist: %s" % rec_id, 1234)
        self.current_reccurrent_rec = rec = Recording(rec_id)
        logging.info("Request: RecStart(%s) ref=%s", rec_id, id(rec))
        rec.status = RecStatusNames.Active
        self.recordingsrecordings[rec_id] = rec
        result = self.make_rec_statusmake_rec_status(rec_id)
        return self.run_sim_hookrun_sim_hook(result, rec_id)
 
    def RecStop(self) -> RecStatus:
        """@implements RecCmds.RecStop"""
        try:
            if not self.current_reccurrent_rec:
                raise ExceptionErr("No current recording active", 1234)
            rec_id = self.current_reccurrent_rec.id
            logging.info("Request: RecStop(%s)", rec_id)
 
            rec = self.get_recordingget_recording(rec_id)
 
            if rec.status in (RecStatusNames.Completed, RecStatusNames.Stopped):
                logging.info(
                    "Request: RecStop(%s): Recording already complete. Returning error",
                    rec_id,
                )
                raise ExceptionErr("Recording already stopped '%s'" % rec_id, 1234)
 
            path = os.path.join(self.datarootdataroot, "%s_%s.fits" % (rec_id, self.namename))
            start = time.monotonic()
            hdul = fits.HDUList([fits.PrimaryHDU()])
            hdul.append(fits.ImageHDU(np.ones((100, 100)), name=self.namename))
            hdul.writeto(path)
            stop = time.monotonic()
            logging.info(
                "Request: RecStop(%s): FITS written to %s (took %.1fs)",
                rec_id,
                path,
                stop - start,
            )
 
            rec.dp_files = [path]
            rec.status = RecStatusNames.Stopped
 
            result = self.make_rec_statusmake_rec_status(rec_id)
            # Due to MAL bug we have to continue keeping the current recording alive...
            # self.current_rec = None
            return self.run_sim_hookrun_sim_hook(result, rec_id)
        except Exception as e:
            logging.exception("RecStop failed")
            raise ExceptionErr("RecStop failed: {str(e)}", 1234)
 
    def RecAbort(self) -> str:
        """@implements RecCmds.RecAbort"""
        if not self.current_reccurrent_rec:
            raise ExceptionErr("No current recording active", 1234)
 
        try:
            rec_id = self.current_reccurrent_rec.id
            logging.info("Request: RecAbort(%s)", rec_id)
            rec = self.get_recordingget_recording(rec_id)
            rec.status = RecStatusNames.Aborted
            self.recordingsrecordings[rec_id] = rec
            result = ""
            self.current_reccurrent_rec = None
            return self.run_sim_hookrun_sim_hook(result, rec_id)
        except Exception as e:
            logging.exception("RecAbort faied")
            raise ExceptionErr("RecAbort failed: {str(e)}", 1234)
        except:
            logging.exception("RecAbort faied with unknown error")
            raise ExceptionErr("RecAbort failed with unknown error", 1234)
 
    def RecWait(
        self, spec: RecWaitSpec
    ) -> RecStatus:
        """@implements RecCmds.RecWait
 
        For the moment we simply simulate that the recording was completed
        as soon as this request was received.
        """
        rec_id = self.current_reccurrent_rec.id
        logging.info("Request: RecWait(%s)", rec_id)
        result = self.make_rec_wait_statusmake_rec_wait_status(rec_id)
        return self.run_sim_hookrun_sim_hook(result, rec_id)
 
    def RecStatus(self, rec_id: str) -> RecStatus:
        """@implements RecCmds.RecStatus"""
        logging.info("Request: RecStatus(%s)", rec_id)
        result = self.make_rec_statusmake_rec_status(rec_id)
        return self.run_sim_hookrun_sim_hook(result, rec_id)
 
    def RecPause(self) -> RecStatus:
        """@implements RecCmds.RecPause"""
        logging.info("Request: RecPause(): Throwing not implemented error")
        raise ExceptionErr("Not implemented", 1234)
 
    def get_recording(self, rec_id) -> Recording:
        """Get recording or raise ExceptionErr."""
        if rec_id in self.recordingsrecordings:
            return self.recordingsrecordings[rec_id]
        raise ExceptionErr("Unknown recording id '%s'" % rec_id, 1234)
 
    def make_rec_status(self, rec_id) -> RecStatus:
        """
        <struct name="RecStatus">
            <member name="dpFiles" type="string" arrayDimensions="(32)"/>
            <member name="endTime" type="double"/>
            <member name="filesGenerated" type="int32_t"/>
            <member name="framesProcessed" type="int32_t"/>
            <member name="framesRemaining" type="int32_t"/>
            <member name="id" type="string"/>
            <member name="info" type="string"/>
            <member name="remainingTime" type="double"/>
            <member name="sizeRecorded" type="int32_t"/>
            <member name="startTime" type="double"/>
            <member name="status" type="nonBasic" nonBasicTypeName="RecStatusNames"/>
            <member name="timeElapsed" type="double"/>
        </struct>
        """
        logging.debug("Creating return type: RecStatus for id %s", rec_id)
        rec = self.get_recordingget_recording(rec_id)
        status = self.malmal.createDataEntity(RecStatus)
        self.populate_rec_statuspopulate_rec_status(rec, status)
        return status
 
    def populate_rec_status(self, rec: Recording, status: RecStatus) -> RecStatus:
        """Populates RecStatus from status"""
        logging.debug(
            "Populating return type: RecStatus: id=%s, status=%r (ref=%s)",
            rec.id,
            rec.status,
            id(rec),
        )
        status.setDpFiles(rec.dp_files)
        status.setEndTime(rec.end_time)
        status.setFilesGenerated(rec.files_generated)
        status.setFramesProcessed(rec.frames_processed)
        status.setFramesRemaining(rec.frames_remaining)
        status.setId(rec.id)
        status.setInfo(rec.info)
        status.setRemainingTime(rec.remaining_time)
        status.setSizeRecorded(rec.size_recorded)
        status.setStartTime(rec.start_time)
        status.setStatus(rec.status)
        status.setTimeElapsed(rec.time_elapsed)
 
    def make_rec_wait_status(self, rec_id) -> RecWaitStatus:
        """Creates RecWaitStatus and populates attributes"""
        logging.debug("Creating return type: RecWaitStatus")
        rec = self.get_recordingget_recording(rec_id)
        status = self.malmal.createDataEntity(RecWaitStatus)
        self.populate_rec_statuspopulate_rec_status(rec, status.getRecStatus())
        if rec.status in (RecStatusNames.Completed, RecStatusNames.Stopped):
            logging.debug(
                "Recording is Completed or Stopped, RecWaitStatus set to success"
            )
            status.setStatus(RecWaitStatusNames.Success)
        else:
            status.setStatus(RecWaitStatusNames.Timeout)
        return status
 
    def add_sim_hook(self, cmd_name: str, hook):
        """Add the simulation hook `hook` for the specified command `cmd_name` (e.g.
        "RecStart").
 
        Any previous hooks will be replaced.
        """
        self.sim_hookssim_hooks[cmd_name] = hook
 
    def reset(self):
        """Reset simulator to initial default state."""
        self.clear_sim_hooksclear_sim_hooks()
        self.recordingsrecordings = {}
        self.current_reccurrent_rec = None
 
    def clear_sim_hooks(self):
        """Remove all simulation hooks."""
        self.sim_hookssim_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_hookssim_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_hookssim_hooks[cmd_name]
            return hook(result, *args)
        else:
            logging.debug(
                "No hook for %s, returning %s (%s)", cmd_name, result, type(result)
            )
            return result
 
 
class SimulatorCtl:
    """Simulator controller that implements ModDaqsimif and sets up
    simulation behaviour for Simulator.
    """
 
    def __init__(self, name, mal, server, dataroot):
        self.namename = name
        self.malmal = mal
        self.serverserver = server
        self.datarootdataroot = dataroot
 
        self.simsim = Simulator(name, mal, dataroot)
        logging.info("Registering service 'rec'")
        self.serverserver.registerService("rec", RecCmdsSyncService, self.simsim)
 
    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 == "complete-recording":
                # finish current recording
                rec = self.simsim.current_rec
                logging.info(
                    "SimCtl.Setup: Marking %s as Completed ref=%s", rec.id, id(rec)
                )
                rec.status = RecStatusNames.Completed
                path = os.path.join(
                    self.datarootdataroot, "auto_%s_%s.fits" % (rec.id, self.namename)
                )
 
                hdul = fits.HDUList([fits.PrimaryHDU()])
                hdul.append(fits.ImageHDU(np.ones((100, 100)), name=self.namename))
                hdul.writeto(path)
                logging.info("SimCtl.Setup: %s FITS written to %s", rec.id, path)
                rec.dp_files = [path]
                assert id(self.simsim.get_recording(rec.id)) == id(rec)
 
            elif action == "throw":
 
                def throw(res, *args):
                    logging.debug("Throwing exception")
                    raise ExceptionErr(
                        "Simulated error (current id: %s)"
                        % getattr(self.simsim.current_rec, "id", "n/a"),
                        1234,
                    )
 
                self.simsim.add_sim_hook(cmd, throw)
            else:
                raise SimCtlError("Unknown action '%s' provided" % action)
            logging.info("SimCtl.Setup: Returning empty string")
            return json.dumps("")
        except SimCtlError:
            logging.exception("SimCtl.Setup: Failed")
            raise
        except Exception as e:
            logging.exception("SimCtl.Setup: Failed")
            raise SimCtlError('"%s"' % str(e))
 
    def Reset(self):
        """@implements SimCtl.Reset"""
        try:
            self.simsim.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()