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
 
from rich.progress import track
 
try:
    from PyQt5 import QtWidgets
except ModuleNotFoundError as e:
    try:    
        from PySide2 import QtWidgets
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError("No module named 'PyQt5' or PySide2', please install one of them\nException raised: "+e.msg)
 
from shesha.supervisor.canapassSupervisor import CanapassSupervisor
from typing import Any
from docopt import docopt
 
from shesha.widgets.widget_base import WidgetBase
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.CBCB = {}
        self.wpyrwpyr = None
        self.current_buffercurrent_buffer = 1
        
        self.uiAOuiAO.actionShow_Pyramid_Tools.toggled.connect(self.show_pyr_toolsshow_pyr_tools)
        self.wpyrNbBufferwpyrNbBuffer = 1
        
    def init_config(self) -> None:
        self.supervisorsupervisorsupervisor = CanapassSupervisor(self.configconfig)
        WidgetBase.init_config(self)
 
    def init_configFinished(self) -> None:
        widgetAOWindow.init_configFinished(self)
        global server
        server = self.start_pyro_serverstart_pyro_server()
 
    def loop_once(self) -> None:
        widgetAOWindow.loop_once(self)
        if (self.uiAOuiAO.actionShow_Pyramid_Tools.isChecked()):  # PYR only
            self.wpyrwpyr.Fe = 1 / self.configconfig.p_loop.ittime  # needs Fe for PSD...
            if (self.wpyrwpyr.CBNumber == 1):
                self.aiai = self.compute_modal_residuals()
                self.set_pyr_tools_paramsset_pyr_tools_params(self.aiai)
            else:
                if (self.current_buffercurrent_buffer == 1):  # First iter of the CB
                    aiVect = self.compute_modal_residuals()
                    self.aiai = aiVect[np.newaxis, :]
                    self.current_buffercurrent_buffer += 1  # Keep going
 
                else:  # Keep filling the CB
                    aiVect = self.compute_modal_residuals()
                    self.aiai = np.concatenate((self.aiai, aiVect[np.newaxis, :]))
                    if (self.current_buffercurrent_buffer < self.wpyrwpyr.CBNumber):
                        self.current_buffercurrent_buffer += 1  # Keep going
                    else:
                        self.current_buffercurrent_buffer = 1  # reset buffer
                        self.set_pyr_tools_paramsset_pyr_tools_params(self.aiai)  # display
 
    def next(self, nbIters):
        ''' Move atmos -> get_slopes -> applyControl ; One integrator step '''
        for i in track(range(nbIters)):
            self.supervisorsupervisorsupervisor.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.wpyrwpyr = widget_pyrStats()
        self.wpyrNbBufferwpyrNbBuffer = self.wpyrwpyr.CBNumber
        self.wpyrwpyr.show()
 
    def set_pyr_tools_params(self, ai):
        self.wpyrwpyr.pup = self.supervisorsupervisorsupervisor.config.p_geom._spupil
        self.wpyrwpyr.phase = self.supervisorsupervisorsupervisor.target.get_tar_phase(0, pupil=True)
        self.wpyrwpyr.updateResiduals(ai)
        if (self.phase_to_modesphase_to_modes is None):
            print('computing phase 2 Modes basis')
            self.phase_to_modesphase_to_modes = self.supervisorsupervisorsupervisor.basis.compute_phase_to_modes(self.modal_basis)
        self.wpyrwpyr.phase_to_modes = self.phase_to_modesphase_to_modes
 
    def show_pyr_tools(self):
        if (self.wpyrwpyr is None):
            try:
                print("Lauching pyramid widget...")
                self.initPyrToolsinitPyrTools()
                print("Done")
            except:
                raise ValueError("ERROR: ADOPT  not found. Cannot launch Pyramid tools")
        else:
            if (self.uiAOuiAO.actionShow_Pyramid_Tools.isChecked()):
                self.wpyrwpyr.show()
            else:
                self.wpyrwpyr.hide()
 
    def getAi(self):
        return self.wpyrwpyr.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)
 
            if(self.supervisorsupervisorsupervisor.corono == None):
                from shesha.util.pyroEmptyClass import PyroEmptyClass
                coro2pyro = PyroEmptyClass()
            else:
                coro2pyro = self.supervisorsupervisorsupervisor.corono
            devices = [self.supervisorsupervisorsupervisor, self.supervisorsupervisorsupervisor.rtc, self.supervisorsupervisorsupervisor.wfs, 
            self.supervisorsupervisorsupervisor.target, self.supervisorsupervisorsupervisor.tel,self.supervisorsupervisorsupervisor.basis, self.supervisorsupervisorsupervisor.calibration,
            self.supervisorsupervisorsupervisor.atmos, self.supervisorsupervisorsupervisor.dms,  self.supervisorsupervisorsupervisor.config, self.supervisorsupervisorsupervisor.modalgains, coro2pyro, wao_loop]
            
            names = ["supervisor", "supervisor_rtc", "supervisor_wfs", 
            "supervisor_target", "supervisor_tel", "supervisor_basis", "supervisor_calibration", 
            "supervisor_atmos", "supervisor_dms", "supervisor_config", "supervisor_modalgains","supervisor_corono",  "wao_loop"]
            nname = [];  
            for name in names: 
                nname.append(name+"_"+user) 
            server = PyroServer(listDevices=devices, listNames=nname)
            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.waowao = wao
 
    def start(self):
        self.waowao.aoLoopClicked(True)
        self.waowao.uiAO.wao_run.setChecked(True)
 
    def stop(self):
        self.waowao.aoLoopClicked(False)
        self.waowao.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())