4119519683

Build # 4119519683

Build Type

push

github

Committed by GitHub

Commit Message

Lintify (#287)

Run Details

36 of 36 new or added lines in 9 files covered. (100.0%)

5321 of 6332 relevant lines covered (84.03%)

1.68 hits per line

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

78.79

/winterdrp/processors/utils/error_annotator.py

"""
Module for adding saved errors into image header metadata
"""
import logging

from winterdrp.data import ImageBatch
from winterdrp.errors import ErrorStack
from winterdrp.paths import PROC_FAIL_KEY
from winterdrp.processors.base_processor import BaseImageProcessor

logger = logging.getLogger(__name__)


class ErrorStackAnnotator(BaseImageProcessor):
    """
    Processor to update image headers with processing failurs
    """

    base_key = "errorannotate"

    def __init__(self, errorstack: ErrorStack, processed_images: list[str]):
        super().__init__()
        self.errorstack = errorstack
        self.image_dict = self.unpack_errorstack()
        self.processed_images = processed_images

    def unpack_errorstack(self) -> dict:
        """
        Convert an errorstack to an image-indexed dictionary

        :return: dictionary of errors
        """
        image_dict = {}

        all_reports = self.errorstack.get_all_reports()

        for error_report in all_reports:
            image_names = error_report.contents
            error_name = error_report.get_error_name()

            for image_name in image_names:
                if image_name not in image_dict:
                    image_dict[image_name] = [error_name]
                else:
                    image_dict[image_name].append(error_name)

        return image_dict

    def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:
        for image in batch:
            base_name = image.get_name()

            if base_name in self.image_dict:
                image[PROC_FAIL_KEY] += ",".join(self.image_dict[base_name])
            elif self.processed_images is not None:
                if base_name not in self.processed_images:
                    image[PROC_FAIL_KEY] += "Not Processed"

        return batch

1	"""
2	Module for adding saved errors into image header metadata
3	"""
4	import logging	2✔
5
6	from winterdrp.data import ImageBatch	2✔
7	from winterdrp.errors import ErrorStack	2✔
8	from winterdrp.paths import PROC_FAIL_KEY	2✔
9	from winterdrp.processors.base_processor import BaseImageProcessor	2✔
10
11	logger = logging.getLogger(__name__)	2✔
12
13
14	class ErrorStackAnnotator(BaseImageProcessor):	2✔
15	"""
16	Processor to update image headers with processing failurs
17	"""
18
19	base_key = "errorannotate"	2✔
20
21	def __init__(self, errorstack: ErrorStack, processed_images: list[str]):	2✔
22	super().__init__()	2✔
23	self.errorstack = errorstack	2✔
24	self.image_dict = self.unpack_errorstack()	2✔
25	self.processed_images = processed_images	2✔
26
27	def unpack_errorstack(self) -> dict:	2✔
28	"""
29	Convert an errorstack to an image-indexed dictionary
30
31	:return: dictionary of errors
32	"""
33	image_dict = {}	2✔
34
35	all_reports = self.errorstack.get_all_reports()	2✔
36
37	for error_report in all_reports:	2✔
38	image_names = error_report.contents	×
39	error_name = error_report.get_error_name()	×
40
41	for image_name in image_names:	×
42	if image_name not in image_dict:	×
43	image_dict[image_name] = [error_name]	×
44	else:
45	image_dict[image_name].append(error_name)	×
46
47	return image_dict	2✔
48
49	def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:	2✔
50	for image in batch:	2✔
51	base_name = image.get_name()	2✔
52
53	if base_name in self.image_dict:	2✔
54	image[PROC_FAIL_KEY] += ",".join(self.image_dict[base_name])	×
55	elif self.processed_images is not None:	2✔
56	if base_name not in self.processed_images:	2✔
57	image[PROC_FAIL_KEY] += "Not Processed"	2✔
58
59	return batch	2✔

winter-telescope / winterdrp / 4119519683

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