3769342838

Build Type

push

github

Committed by Robert Stein

Commit Message

lintify cal hunter

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

4635 of 6126 relevant lines covered (75.66%)

0.76 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	1✔
5
6	from winterdrp.data import ImageBatch	1✔
7	from winterdrp.errors import ErrorStack	1✔
8	from winterdrp.paths import PROC_FAIL_KEY	1✔
9	from winterdrp.processors.base_processor import BaseImageProcessor	1✔
10
11	logger = logging.getLogger(__name__)	1✔
12
13
14	class ErrorStackAnnotator(BaseImageProcessor):	1✔
15	"""
16	Processor to update image headers with processing failurs
17	"""
18
19	base_key = "errorannotate"	1✔
20
21	def __init__(self, errorstack: ErrorStack, processed_images: list[str]):	1✔
22	super().__init__()	1✔
23	self.errorstack = errorstack	1✔
24	self.image_dict = self.unpack_errorstack()	1✔
25	self.processed_images = processed_images	1✔
26
27	def unpack_errorstack(self) -> dict:	1✔
28	"""
29	Convert an errorstack to an image-indexed dictionary
30
31	:return: dictionary of errors
32	"""
33	image_dict = {}	1✔
34
35	all_reports = self.errorstack.get_all_reports()	1✔
36
37	for error_report in all_reports:	1✔
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	1✔
48
49	def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:	1✔
50
51	for image in batch:	1✔
52
53	base_name = image.get_name()	1✔
54
55	if base_name in self.image_dict:	1✔
56	image[PROC_FAIL_KEY] += ",".join(self.image_dict[base_name])	×
57	elif self.processed_images is not None:	1✔
58	if base_name not in self.processed_images:	1✔
59	image[PROC_FAIL_KEY] += "Not Processed"	1✔
60
61	return batch	1✔

winter-telescope / winterdrp / 3769342838

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