3699777599

Build Type

push

github

Committed by GitHub

Commit Message

Add initial mypy integration (#241)

Run Details

223 of 223 new or added lines in 45 files covered. (100.0%)

4575 of 6107 relevant lines covered (74.91%)

0.75 hits per line

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

80.56

/winterdrp/processors/utils/error_annotator.py

import logging
from pathlib import Path

import astropy.io.fits
import numpy as np

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

logger = logging.getLogger(__name__)


class ErrorStackAnnotator(BaseImageProcessor):

    base_key = "errorannotate"

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

    def unpack_errorstack(self) -> dict:
        image_dict = 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.keys():
                    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 i, image in enumerate(batch):

            base_name = str(Path(image[RAW_IMG_KEY]).name)

            if base_name in self.image_dict.keys():
                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	import logging	1✔
2	from pathlib import Path	1✔
3
4	import astropy.io.fits	1✔
5	import numpy as np	1✔
6
7	from winterdrp.data import ImageBatch	1✔
8	from winterdrp.errors import ErrorStack	1✔
9	from winterdrp.paths import PROC_FAIL_KEY, RAW_IMG_KEY	1✔
10	from winterdrp.processors.base_processor import BaseImageProcessor	1✔
11
12	logger = logging.getLogger(__name__)	1✔
13
14
15	class ErrorStackAnnotator(BaseImageProcessor):	1✔
16
17	base_key = "errorannotate"	1✔
18
19	def __init__(	1✔
20	self, errorstack: ErrorStack, processed_images: list[str], args, *kwargs
21	):
22	super().__init__(args, *kwargs)	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	image_dict = dict()	1✔
29
30	all_reports = self.errorstack.get_all_reports()	1✔
31
32	for error_report in all_reports:	1✔
33	image_names = error_report.contents	×
34	error_name = error_report.get_error_name()	×
35
36	for image_name in image_names:	×
37	if image_name not in image_dict.keys():	×
38	image_dict[image_name] = [error_name]	×
39	else:
40	image_dict[image_name].append(error_name)	×
41
42	return image_dict	1✔
43
44	def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:	1✔
45
46	for i, image in enumerate(batch):	1✔
47
48	base_name = str(Path(image[RAW_IMG_KEY]).name)	1✔
49
50	if base_name in self.image_dict.keys():	1✔
51	image[PROC_FAIL_KEY] += ",".join(self.image_dict[base_name])	×
52	elif self.processed_images is not None:	1✔
53	if base_name not in self.processed_images:	1✔
54	image[PROC_FAIL_KEY] += "Not Processed"	1✔
55
56	return batch	1✔

winter-telescope / winterdrp / 3699777599

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