"""
Module to group images based on the target coordinates into planned stack groups
"""
# pylint: disable=duplicate-code
from astropy import coordinates as coords
from astropy import units as u
from astropy.coordinates import Angle
from mirar.data import ImageBatch
from mirar.paths import OBSCLASS_KEY
from mirar.pipelines.lmi.config.constants import LMI_WIDTH_DEG
from mirar.processors.utils.image_selector import split_images_into_batches
[docs]
def label_stack_id(batch: ImageBatch) -> ImageBatch:
"""
Label the stack id of the images in the batch
:param batch: Original batch of images
:return: Labeled batch of images
"""
ras = []
decs = []
for image in batch:
if image[OBSCLASS_KEY] != "science":
image["targnum"] = -1
continue
target_ra = Angle(image["CRVAL1"], unit="hourangle").degree
target_dec = Angle(image["CRVAL2"], unit="degree").degree
if (target_ra == 0.0) & (target_dec == 0.0):
target_ra = image["OBJRA"]
target_dec = image["OBJDEC"]
image["CRVAL1"] = target_ra
image["CRVAL2"] = target_dec
position = coords.SkyCoord(target_ra, target_dec, unit="deg")
match = None
if len(ras) > 0:
idx, d2d, _ = position.match_to_catalog_sky(
coords.SkyCoord(ra=ras, dec=decs, unit="deg")
)
mask = d2d < (LMI_WIDTH_DEG * u.deg * 2.0)
if mask:
match = idx
if match is None:
ras.append(target_ra)
decs.append(target_dec)
image["targnum"] = int(len(ras) - 1)
else:
image["targnum"] = int(match)
new_batches = split_images_into_batches(batch, ["targnum", "filter", "exptime"])
combined_batch = ImageBatch()
for i, split_batch in enumerate(new_batches):
label = f"stack{i}"
for image in split_batch:
image["stackid"] = label
image["OBJRA"] = ras[image["targnum"]]
image["OBJDEC"] = decs[image["targnum"]]
combined_batch.append(image)
return combined_batch