NCPA

Initialize a NCPA generator.

Parameters:

Name	Type	Description	Default
telescope	Telescope	Telescope object to which the WFS is attached.	required

Source code in SAOS/NCPA.py

def __init__(self,
             telescope,
             logger = None):
    """
    Initialize a NCPA generator.

    Parameters
    ----------
    telescope : Telescope
        Telescope object to which the WFS is attached.
    """        
    # Setup the logger to handle the queue of info, warning and errors msgs in the simulator
    if logger is None:
        self.queue_listerner = self.setup_logging()
        self.logger = logging.getLogger()
        self.external_logger_flag = False
    else:
        self.external_logger_flag = True
        self.logger = logger

    # Define class attributes
    self.tag = 'ncpa'

    # Generate empty OPD

    self.ncpa_opd = np.zeros_like(telescope.pupil, dtype=np.float32)

queue_listerner instance-attribute

queue_listerner = setup_logging()

logger instance-attribute

logger = getLogger()

external_logger_flag instance-attribute

external_logger_flag = False

tag instance-attribute

tag = 'ncpa'

ncpa_opd instance-attribute

ncpa_opd = zeros_like(pupil, dtype=float32)

load

load(filename)

Load the NCPA OPD from a H5 file.

Parameters:

Name	Type	Description	Default
filename	str	Path and base filename (with extension) of the H5 file to load.	required

Returns:

Type	Description
bool	True if loaded successfully.

Source code in SAOS/NCPA.py

def load(self, filename): 
    """
    Load the NCPA OPD from a H5 file.

    Parameters
    ----------
    filename : str
        Path and base filename (with extension) of the H5 file to load.

    Returns
    -------
    bool
        True if loaded successfully.
    """
    self.logger.debug('NCPA::load')


    if not filename.endswith(".h5"):
        filename += ".h5"        

    with h5py.File(filename, 'r') as f:
        temp_opd = f['opd']['data'][()]

        self.logger.info(f"NCPA::load - Loaded OPD.")     

    if (temp_opd.shape[0] != self.ncpa_opd.shape[0]) or (temp_opd.shape[1] != self.ncpa_opd.shape[1]):
        self.logger.warning('NCPA::Load - The dimensions of the NCPA do not match the telescope\'s pupil. Interpolating, may introduce artifacs.')
        self.ncpa_opd = cv2.resize(temp_opd, (self.ncpa_opd.shape[0], self.ncpa_opd.shape[1]), interpolation=cv2.INTER_LINEAR)

    return True

setOPD

setOPD(input_opd)

set the NCPA OPD directly from a variable

Parameters:

Name	Type	Description	Default
input_opd	ndarray	Input OPD in [m]	required

Returns:

Type	Description
bool	True if loaded successfully,.

Source code in SAOS/NCPA.py

def setOPD(self, input_opd):
    """
    set the NCPA OPD directly from a variable

    Parameters
    ----------
    input_opd : np.ndarray
        Input OPD in [m]

    Returns
    -------
    bool
        True if loaded successfully,.
    """
    self.logger.debug('NCPA::load')

    if (input_opd.shape[0] != self.ncpa_opd.shape[0]) or (input_opd.shape[1] != self.ncpa_opd.shape[1]):
        self.logger.error('NCPA::setOPD - The dimensions of the NCPA do not match the telescope\'s pupil. Rejecting.')
        raise ValueError('The dimensions of the NCPA do not match the telescope\'s pupil.')

    self.ncpa_opd = input_opd.copy()

    return True

getPhase

getPhase()

Return the current NCPA OPD.

Returns:

Type	Description
ndarray	NCPA OPD in meters [m].

Source code in SAOS/NCPA.py

def getPhase(self):
    """
    Return the current NCPA OPD.

    Returns
    -------
    np.ndarray
        NCPA OPD in meters [m].
    """
    self.logger.debug('NCPA::getPhase')

    return self.ncpa_opd

setup_logging

setup_logging(logging_level=logging.WARNING)

Source code in SAOS/NCPA.py

def setup_logging(self, logging_level=logging.WARNING):
    #
    #  Setup of logging at the main process using QueueHandler
    log_queue = Queue()
    queue_handler = logging.handlers.QueueHandler(log_queue)
    root_logger = logging.getLogger()
    root_logger.setLevel(logging_level)  # Minimum log level

    # Setup of the formatting
    formatter = logging.Formatter(
        "%(asctime)s - %(levelname)s - %(message)s"
    )

    # Output to terminal
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)

    # Qeue handler captures the messages from the different logs and serialize them
    queue_listener = logging.handlers.QueueListener(log_queue, console_handler)
    root_logger.addHandler(queue_handler)
    queue_listener.start()

    return queue_listener

__del__

__del__()

Source code in SAOS/NCPA.py

def __del__(self):
    if not self.external_logger_flag:
        self.queue_listerner.stop()