widget_canapass.py

 
"""
Widget to simulate a closed loop using CANAPASS
 
Usage:
  widget_canapass.py [<parameters_filename>]
 
with 'parameters_filename' the path to the parameters file
 
Options:
  -h --help          Show this help message and exit
  -i, --interactive  keep the script interactive
"""
 
import os, sys
import numpy as np
import time
 
import pyqtgraph as pg
from shesha.util.tools import plsh, plpyr
from tqdm import trange
import astropy.io.fits as pfits
from PyQt5 import QtWidgets
from shesha.supervisor.canapassSupervisor import CanapassSupervisor
 
from typing import Any, Dict, Tuple, Callable, List
from docopt import docopt
 
from shesha.widgets.widget_ao import widgetAOWindow, widgetAOWindow
 
global server
server = None
 
 
class widgetCanapassWindowPyro(widgetAOWindow):
 
    def __init__(self, config_file: Any = None, cacao: bool = False,
                 expert: bool = False) -> None:
        widgetAOWindow.__init__(self, config_file, cacao, hide_histograms=True)
        #Pyro.core.ObjBase.__init__(self)
 
        self.CB = {}
        self.wpyr = None
        self.current_buffer = 1
        
        self.uiAO.actionShow_Pyramid_Tools.toggled.connect(self.show_pyr_tools)
        self.wpyrNbBuffer = 1
        
    def init_configThread(self) -> None:
        self.uiAO.wao_deviceNumber.setDisabled(True)
        self.supervisor = CanapassSupervisor(self.config)
 
    def init_config(self) -> None:
        widgetAOWindow.init_config(self)
        global server
        server = self.start_pyro_server()
 
    def load_config(self, *args, config_file=None, supervisor=None) -> None:
        '''
            Callback when 'LOAD' button is hit
        '''
        if supervisor is None:
            supervisor = CanapassSupervisor(config_file)
 
        widgetAOWindow.load_config(self, args, config_file=config_file,
                                  supervisor=supervisor)
 
    def loop_once(self) -> None:
        widgetAOWindow.loop_once(self)
        if (self.uiAO.actionShow_Pyramid_Tools.isChecked()):  # PYR only
            self.wpyr.Fe = 1 / self.config.p_loop.ittime  # needs Fe for PSD...
            if (self.wpyr.CBNumber == 1):
                self.ai = self.compute_modal_residuals()
                self.set_pyr_tools_params(self.ai)
            else:
                if (self.current_buffer == 1):  # First iter of the CB
                    aiVect = self.compute_modal_residuals()
                    self.ai = aiVect[np.newaxis, :]
                    self.current_buffer += 1  # Keep going
 
                else:  # Keep filling the CB
                    aiVect = self.compute_modal_residuals()
                    self.ai = np.concatenate((self.ai, aiVect[np.newaxis, :]))
                    if (self.current_buffer < self.wpyr.CBNumber):
                        self.current_buffer += 1  # Keep going
                    else:
                        self.current_buffer = 1  # reset buffer
                        self.set_pyr_tools_params(self.ai)  # display
 
    def next(self, nbIters):
        ''' Move atmos -> get_slopes -> applyControl ; One integrator step '''
        for i in trange(nbIters):
            self.supervisor.next()
 
    def initPyrTools(self):
        ADOPTPATH = os.getenv("ADOPTPATH")
        sys.path.append(ADOPTPATH + "/widgets")
        from pyrStats import widget_pyrStats
        print("OK Pyramid Tools Widget initialized")
        self.wpyr = widget_pyrStats()
        self.wpyrNbBuffer = self.wpyr.CBNumber
        self.wpyr.show()
 
    def set_pyr_tools_params(self, ai):
        self.wpyr.pup = self.supervisor.config.p_geom._spupil
        self.wpyr.phase = self.supervisor.target.get_tar_phase(0, pupil=True)
        self.wpyr.updateResiduals(ai)
        if (self.phase_to_modes is None):
            print('computing phase 2 Modes basis')
            self.phase_to_modes = self.supervisor.basis.compute_phase_to_modes(self.modal_basis)
        self.wpyr.phase_to_modes = self.phase_to_modes
 
    def show_pyr_tools(self):
        if (self.wpyr is None):
            try:
                print("Lauching pyramid widget...")
                self.initPyrTools()
                print("Done")
            except:
                raise ValueError("ERROR: ADOPT  not found. Cannot launch Pyramid tools")
        else:
            if (self.uiAO.actionShow_Pyramid_Tools.isChecked()):
                self.wpyr.show()
            else:
                self.wpyr.hide()
 
    def getAi(self):
        return self.wpyr.ai
 
    def start_pyro_server(self):
        try:
            from subprocess import Popen, PIPE
            from hraa.server.pyroServer import PyroServer
 
            # Init looper
            wao_loop = loopHandler(self)
 
            # Find username
            p = Popen("whoami", shell=True, stdout=PIPE, stderr=PIPE)
            out, err = p.communicate()
            if (err != b''):
                print(err)
                raise Exception("ERROR CANNOT RECOGNIZE USER")
            else:
                user = out.split(b"\n")[0].decode("utf-8")
                print("User is " + user)
 
            server = PyroServer()
            server.add_device(self.supervisor, "waoconfig_" + user)
            server.add_device(wao_loop, "waoloop_" + user)
            server.start()
        except:
            raise Exception("Error could not connect to Pyro server.\n It can  be:\n - Missing dependencies? (check if Pyro4 is installed)\n - pyro server not running")
 
        return server
 
 
class loopHandler:
 
    def __init__(self, wao):
        self.wao = wao
 
    def start(self):
        self.wao.aoLoopClicked(True)
        self.wao.uiAO.wao_run.setChecked(True)
 
    def stop(self):
        self.wao.aoLoopClicked(False)
        self.wao.uiAO.wao_run.setChecked(False)
 
    def alive(self):
        return "alive"
 
 
if __name__ == '__main__':
    arguments = docopt(__doc__)
    app = QtWidgets.QApplication(sys.argv)
    app.setStyle('cleanlooks')
    wao = widgetCanapassWindowPyro(arguments["<parameters_filename>"], cacao=True)
    wao.show()
    # if arguments["--interactive"]:
    #     from shesha.util.ipython_embed import embed
    #     from os.path import basename
    #     embed(basename(__file__), locals())