4623937781

Build Type

github

Committed by GitHub

Commit Message

Merge 2c994e02b into c82ce9712

Pull Request Pull Request #319: Multi-extension and many files processing

Run Details

116 of 116 new or added lines in 8 files covered. (100.0%)

5471 of 6661 relevant lines covered (82.13%)

1.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/winterdrp/processors/utils/multi_ext_parser.py

"""
Module for loading and creating multi-extension FITS (MEF) images
"""
import logging
import os
from collections.abc import Callable
from glob import glob
from pathlib import Path

import astropy.io.fits
import numpy as np

from winterdrp.data import Image, ImageBatch
from winterdrp.errors import ImageNotFoundError
from winterdrp.io import open_fits
from winterdrp.paths import RAW_IMG_SUB_DIR, base_raw_dir
from winterdrp.processors.base_processor import BaseImageProcessor

logger = logging.getLogger(__name__)


class MultiExtParser(BaseImageProcessor):
    """Processor to split multi-extension FITS (MEF) images into single-extension files.
    Should be run before ImageLoader, especially for SEDMv2
    """

    base_key = "load"  # should this be changed?

    def __init__(
        self,
        input_sub_dir: str = RAW_IMG_SUB_DIR,
        input_img_dir: str = base_raw_dir,
        load_image: Callable[[str], [np.ndarray, astropy.io.fits.Header]] = open_fits,
    ):
        super().__init__()
        self.input_sub_dir = input_sub_dir
        self.input_img_dir = input_img_dir
        self.load_image = load_image

    def __str__(self):
        return (
            f"Processor to parse MEF images from the {self.input_sub_dir} subdirectory "
        )

    def parse(self, path: str) -> list:
        """
        Function to open a raw MEF image, write each extension to a new file

        :param path: path of raw MEF image
        :return: new paths of single-extension files (moved out of raw/mef/, into raw/)

        *** need to manually place MEF science images in a /mef/ subdirectory ***
            ex: /[instrument]/[night]/raw/mef/
        """

        new_paths = []
        with astropy.io.fits.open(path) as hdu:
            num_ext = len(hdu)
            logger.info(f"This file has {num_ext} extensions.")

            hdr0 = hdu[0].header  # pylint: disable=no-member
            # zip hdr0's values and comments
            zipped = list(zip(hdr0.values(), hdr0.comments))
            # combining main header (hdr0) with extension header
            for ext in range(1, num_ext):
                data = hdu[ext].data
                hdrext = hdu[ext].header

                # append hdr0 to hdrext
                for count, key in enumerate(list(hdr0.keys())):
                    hdrext.append((key, zipped[count][0], zipped[count][1]))

                # save to new file with 1 extension
                notmefpath = path.split("/mef/")[0] + path.split("/mef")[1]
                newpath = notmefpath.split(".fits")[0] + "_" + str(ext) + ".fits"
                astropy.io.fits.writeto(
                    newpath, data, hdrext, overwrite=True
                )  # pylint: disable=no-member
                new_paths.append(newpath)

        return new_paths

    def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:
        input_dir = os.path.join(
            self.input_img_dir, os.path.join(self.night_sub_dir, self.input_sub_dir)
        )
        return load_from_dir(input_dir, parse_f=self.parse)


def load_from_dir(
    input_dir: str | Path, parse_f: Callable[[str | Path], Image]
) -> ImageBatch:
    """
    Function to parse all MEF images in a directory

    :param input_dir: directory path
    :param parse_f: function to parse MEF images
    :return: nothing...
    """

    img_list = sorted(glob(f"{input_dir}/*.fits"))

    logger.info(f"Loading from {input_dir}, with {len(img_list)} images")

    if len(img_list) < 1:
        err = f"No images found in {input_dir}. Please check path is correct!"
        logger.error(err)
        raise ImageNotFoundError(err)

    for path in img_list:
        parse_f(path)

    empty_batch = ImageBatch()
    return empty_batch

1	"""
2	Module for loading and creating multi-extension FITS (MEF) images
3	"""
4	import logging	2✔
5	import os	2✔
6	from collections.abc import Callable	2✔
7	from glob import glob	2✔
8	from pathlib import Path	2✔
9
10	import astropy.io.fits	2✔
11	import numpy as np	2✔
12
13	from winterdrp.data import Image, ImageBatch	2✔
14	from winterdrp.errors import ImageNotFoundError	2✔
15	from winterdrp.io import open_fits	2✔
16	from winterdrp.paths import RAW_IMG_SUB_DIR, base_raw_dir	2✔
17	from winterdrp.processors.base_processor import BaseImageProcessor	2✔
18
19	logger = logging.getLogger(__name__)	2✔
20
21
22	class MultiExtParser(BaseImageProcessor):	2✔
23	"""Processor to split multi-extension FITS (MEF) images into single-extension files.
24	Should be run before ImageLoader, especially for SEDMv2
25	"""
26
27	base_key = "load" # should this be changed?	2✔
28
29	def __init__(	2✔
30	self,
31	input_sub_dir: str = RAW_IMG_SUB_DIR,
32	input_img_dir: str = base_raw_dir,
33	load_image: Callable[[str], [np.ndarray, astropy.io.fits.Header]] = open_fits,
34	):
35	super().__init__()	2✔
36	self.input_sub_dir = input_sub_dir	2✔
37	self.input_img_dir = input_img_dir	2✔
38	self.load_image = load_image	2✔
39
40	def __str__(self):	2✔
41	return (	2✔
42	f"Processor to parse MEF images from the {self.input_sub_dir} subdirectory "
43	)
44
45	def parse(self, path: str) -> list:	2✔
46	"""
47	Function to open a raw MEF image, write each extension to a new file
48
49	:param path: path of raw MEF image
50	:return: new paths of single-extension files (moved out of raw/mef/, into raw/)
51
52	* need to manually place MEF science images in a /mef/ subdirectory *
53	ex: /[instrument]/[night]/raw/mef/
54	"""
55
56	new_paths = []	×
57	with astropy.io.fits.open(path) as hdu:	×
58	num_ext = len(hdu)	×
59	logger.info(f"This file has {num_ext} extensions.")	×
60
61	hdr0 = hdu[0].header # pylint: disable=no-member	×
62	# zip hdr0's values and comments
63	zipped = list(zip(hdr0.values(), hdr0.comments))	×
64	# combining main header (hdr0) with extension header
65	for ext in range(1, num_ext):	×
66	data = hdu[ext].data	×
67	hdrext = hdu[ext].header	×
68
69	# append hdr0 to hdrext
70	for count, key in enumerate(list(hdr0.keys())):	×
71	hdrext.append((key, zipped[count][0], zipped[count][1]))	×
72
73	# save to new file with 1 extension
74	notmefpath = path.split("/mef/")[0] + path.split("/mef")[1]	×
75	newpath = notmefpath.split(".fits")[0] + "_" + str(ext) + ".fits"	×
76	astropy.io.fits.writeto(	×
77	newpath, data, hdrext, overwrite=True
78	) # pylint: disable=no-member
79	new_paths.append(newpath)	×
80
81	return new_paths	×
82
83	def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:	2✔
84	input_dir = os.path.join(	×
85	self.input_img_dir, os.path.join(self.night_sub_dir, self.input_sub_dir)
86	)
87	return load_from_dir(input_dir, parse_f=self.parse)	×
88
89
90	def load_from_dir(	2✔
91	input_dir: str \| Path, parse_f: Callable[[str \| Path], Image]
92	) -> ImageBatch:
93	"""
94	Function to parse all MEF images in a directory
95
96	:param input_dir: directory path
97	:param parse_f: function to parse MEF images
98	:return: nothing...
99	"""
100
101	img_list = sorted(glob(f"{input_dir}/*.fits"))	×
102
103	logger.info(f"Loading from {input_dir}, with {len(img_list)} images")	×
104
105	if len(img_list) < 1:	×
106	err = f"No images found in {input_dir}. Please check path is correct!"
107	logger.error(err)
108	raise ImageNotFoundError(err)
109
110	for path in img_list:	×
111	parse_f(path)	×
112
113	empty_batch = ImageBatch()	×
114	return empty_batch	×

winter-telescope / winterdrp / 4623937781

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous