Source code for pynpoint.processing.extract
"""
Pipeline modules for locating and extracting the position of a star.
"""
import math
import warnings
from typing import Optional, Tuple, Union
import numpy as np
from typeguard import typechecked
from pynpoint.core.processing import ProcessingModule
from pynpoint.util.apply_func import crop_around_star, crop_rotating_star
from pynpoint.util.image import rotate_coordinates
[docs]class StarExtractionModule(ProcessingModule):
"""
Pipeline module to locate the position of the star in each image and to crop all the images
around this position.
"""
__author__ = 'Markus Bonse, Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
image_in_tag: str,
image_out_tag: str,
index_out_tag: Optional[str] = None,
image_size: float = 2.,
fwhm_star: float = 0.2,
position: Optional[Union[Tuple[int, int, float],
Tuple[None, None, float]]] = None) -> None:
"""
Parameters
----------
name_in : str
Unique name of the module instance.
image_in_tag : str
Tag of the dataset with the input images.
image_out_tag : str
Tag of the dataset that is stored as output, containing the extracted images.
index_out_tag : str, None
List with image indices for which the image size is too large to be cropped around the
brightest pixel. No data is written if set to None. This tag name can be provided to
the ``frames``` parameter in
:class:`~pynpoint.processing.frameselection.RemoveFramesModule`. This argument is
ignored if ``CPU`` is set to a value larger than 1.
image_size : float
Cropped image size (arcsec).
fwhm_star : float
Full width at half maximum (arcsec) of the Gaussian kernel that is used to smooth the
images to lower contributions of bad pixels.
position : tuple(int, int, float), None
Subframe that is selected to search for the star. The tuple should contain a position
(pix) and size (arcsec) as (pos_x, pos_y, size). The full image is used if set to None.
The center of the image will be used with ``position=(None, None, size)``.
Returns
-------
NoneType
None
"""
super().__init__(name_in)
self.m_image_in_port = self.add_input_port(image_in_tag)
self.m_image_out_port = self.add_output_port(image_out_tag)
if index_out_tag is None:
self.m_index_out_port = None
else:
self.m_index_out_port = self.add_output_port(index_out_tag)
self.m_image_size = image_size
self.m_fwhm_star = fwhm_star
self.m_position = position
[docs] @typechecked
def run(self) -> None:
"""
Run method of the module. Locates the position of the star (only pixel precision) by
selecting the highest pixel value. A Gaussian kernel with a FWHM similar to the PSF is
used to lower the contribution of bad pixels which may have higher values than the peak
of the PSF. Images are cropped and written to an output port.
Returns
-------
NoneType
None
"""
cpu = self._m_config_port.get_attribute('CPU')
if cpu > 1 and self.m_index_out_port is not None:
warnings.warn('The \'index_out_port\' can only be used if CPU = 1. No data will '
'be stored to this output port.')
del self._m_output_ports[self.m_index_out_port.tag]
self.m_index_out_port = None
pixscale = self.m_image_in_port.get_attribute('PIXSCALE')
self.m_image_size = int(math.ceil(self.m_image_size/pixscale))
self.m_fwhm_star = int(math.ceil(self.m_fwhm_star/pixscale))
self.apply_function_to_images(crop_around_star,
self.m_image_in_port,
self.m_image_out_port,
'Extracting stellar position',
func_args=(self.m_position,
self.m_image_size,
self.m_fwhm_star,
pixscale,
self.m_index_out_port,
self.m_image_out_port))
history = f'fwhm_star (pix) = {self.m_fwhm_star}'
if self.m_index_out_port is not None:
self.m_index_out_port.copy_attributes(self.m_image_in_port)
self.m_index_out_port.add_history('StarExtractionModule', history)
self.m_image_out_port.copy_attributes(self.m_image_in_port)
self.m_image_out_port.add_history('StarExtractionModule', history)
self.m_image_out_port.close_port()
[docs]class ExtractBinaryModule(ProcessingModule):
"""
Pipeline module to extract a binary star (or another point source) which is rotating across the
image stack.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
image_in_tag: str,
image_out_tag: str,
pos_center: Tuple[float, float],
pos_binary: Tuple[float, float],
image_size: float = 2.,
search_size: float = 0.1,
filter_size: Optional[float] = None) -> None:
"""
Parameters
----------
name_in : str
Unique name of the module instance.
image_in_tag : str
Tag of the dataset with the input images.
image_out_tag : str
Tag of the dataset that is stored as output, containing the extracted images.
pos_center : tuple(float, float)
Approximate position (x, y) of the center of rotation (pix).
pos_binary : tuple(float, float)
Approximate position (x, y) of the binary star in the first image (pix).
image_size : float
Cropped image size (arcsec).
search_size : float
Window size (arcsec) in which the brightest pixel is selected as position of the binary
star. The search window is centered on the position that for each image is calculated
from the ``pos_center``, ``pos_binary``, and parallactic angle (``PARANG``) of the
image.
filter_size : float, None
Full width at half maximum (arcsec) of the Gaussian kernel that is used to smooth the
images to lower contributions of bad pixels.
Returns
-------
NoneType
None
"""
super().__init__(name_in)
self.m_image_in_port = self.add_input_port(image_in_tag)
self.m_image_out_port = self.add_output_port(image_out_tag)
self.m_pos_center = (pos_center[1], pos_center[0]) # (y, x)
self.m_pos_binary = (pos_binary[1], pos_binary[0]) # (y, x)
self.m_image_size = image_size
self.m_search_size = search_size
self.m_filter_size = filter_size
[docs] @typechecked
def run(self) -> None:
"""
Run method of the module. Locates the position of a binary star (or some other point
source) which rotates across the stack of images due to parallactic rotation. The
approximate position of the binary star is calculated by taking into account the
parallactic angle of each image separately. The brightest pixel is then selected as
center around which the image is cropped.
Returns
-------
NoneType
None
"""
pixscale = self.m_image_in_port.get_attribute('PIXSCALE')
parang = self.m_image_in_port.get_attribute('PARANG')
positions = np.zeros((parang.shape[0], 2), dtype=int)
for i, item in enumerate(parang):
# rotates in counterclockwise direction, hence the minus sign in angle
positions[i, :] = rotate_coordinates(center=self.m_pos_center,
position=self.m_pos_binary,
angle=item-parang[0])
self.m_image_size = int(math.ceil(self.m_image_size/pixscale))
self.m_search_size = int(math.ceil(self.m_search_size/pixscale))
if self.m_filter_size is not None:
self.m_filter_size = int(math.ceil(self.m_filter_size/pixscale))
self.apply_function_to_images(crop_rotating_star,
self.m_image_in_port,
self.m_image_out_port,
'Extracting binary position',
func_args=(positions,
self.m_image_size,
self.m_filter_size,
self.m_search_size))
history = f'filter (pix) = {self.m_filter_size}'
self.m_image_out_port.copy_attributes(self.m_image_in_port)
self.m_image_out_port.add_history('ExtractBinaryModule', history)
self.m_image_out_port.close_port()