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

94.64

/winterdrp/errors/error_stack.py

"""
Module for ErrorStack objects.

A :class:`~winterdrp.errors.error_stack.ErrorStack` object will contain a list of
:class:`~winterdrp.errors.error_report.ErrorReport` objects, and can correspond to
 multiple errors raised by the code.
"""

import logging
from typing import Optional

import numpy as np

from winterdrp.errors.error_report import ErrorReport
from winterdrp.paths import __version__, package_name

logger = logging.getLogger(__name__)


class ErrorStack:
    """
    Container class to hold multiple
    :class:`~winterdrp.errors.error_report.ErrorReport` objects
    """

    def __init__(self, reports: Optional[list[ErrorReport]] = None):
        self.reports = []
        self.noncritical_reports = []
        self.failed_images = []

        if reports is not None:
            for report in reports:
                self.add_report(report)

    def add_report(self, report: ErrorReport):
        """
        Adds a new ErrorReport

        :param report: ErrorReport to add
        :return: None
        """
        if report.non_critical_bool:
            self.noncritical_reports.append(report)
        else:
            self.reports.append(report)
        all_failed_images = self.failed_images + report.contents
        self.failed_images = sorted(list(set(all_failed_images)))

    def __add__(self, other):
        for report in other.reports:
            self.add_report(report)
        return self

    def get_all_reports(self) -> list[ErrorReport]:
        """
        Returns the full list of error reports (both critical and non-critical).

        :return: list of ErrorReports
        """
        return self.reports + self.noncritical_reports

    def summarise_error_stack(self, output_path=None, verbose: bool = True) -> str:
        """
        Returns a string summary of all ErrorReports.

        :param output_path: Path to write summary in .txt format (optional)
        :param verbose: boolean whether to provide a verbose summary
        :return: String summary of errors
        """

        is_known_error = [x.known_error_bool for x in self.reports]

        summary = (
            f"Error report summarising {len(self.reports)} errors. \n"
            f"Code version: {package_name}=={__version__} \n \n"
            f"{int(len(is_known_error) - np.sum(is_known_error))}/{len(is_known_error)}"
            f" errors were errors not raised by {package_name}. \n"
            f"The remaining {int(np.sum(is_known_error))}/{len(is_known_error)} "
            f"errors were known errors raised by {package_name}. \n"
            f"An additional {len(self.noncritical_reports)} non-critical "
            f"errors were raised. \n"
        )
        all_reports = self.get_all_reports()

        if len(all_reports) > 0:

            summary += (
                f"The following {len(self.failed_images)} images were affected "
                f"by at least one error during processing: \n "
            )

            if verbose:
                summary += f"{self.failed_images} \n \n"

            summary += "Summarising each error: \n\n"

            logger.error(f"Found {len(self.reports)} errors caught by code.")

            error_lines = [err.get_error_message() for err in all_reports]

            for error_type in list(set(error_lines)):

                matching_errors = [
                    x for x in all_reports if x.get_error_message() == error_type
                ]

                img_paths = []
                error_name = None
                for err in matching_errors:
                    img_paths += err.contents
                    if error_name is None:
                        error_name = err.get_error_name()
                img_paths = list(set(img_paths))

                line = error_type.split("\n")[0]
                summary += (
                    f"Found {error_lines.count(error_type)} counts of error "
                    f"{error_name}, "
                    f"affecting {len(img_paths)} images: \n{line}.\n \n"
                )

            summary += " \n"

            if verbose:
                for report in all_reports:
                    summary += str(report.generate_full_traceback())

        else:
            msg = "No raised errors found in processing"
            logger.info(msg)
            summary += f"\n {msg}"

        if output_path is not None:
            logger.error(f"Saving tracebacks of caught errors to {output_path}")
            with open(output_path, "w", encoding="utf-8") as err_file:
                err_file.write(summary)

        return summary

1	"""
2	Module for ErrorStack objects.
3
4	A :class:`~winterdrp.errors.error_stack.ErrorStack` object will contain a list of
5	:class:`~winterdrp.errors.error_report.ErrorReport` objects, and can correspond to
6	multiple errors raised by the code.
7	"""
8
9	import logging	1✔
10	from typing import Optional	1✔
11
12	import numpy as np	1✔
13
14	from winterdrp.errors.error_report import ErrorReport	1✔
15	from winterdrp.paths import __version__, package_name	1✔
16
17	logger = logging.getLogger(__name__)	1✔
18
19
20	class ErrorStack:	1✔
21	"""
22	Container class to hold multiple
23	:class:`~winterdrp.errors.error_report.ErrorReport` objects
24	"""
25
26	def __init__(self, reports: Optional[list[ErrorReport]] = None):	1✔
27	self.reports = []	1✔
28	self.noncritical_reports = []	1✔
29	self.failed_images = []	1✔
30
31	if reports is not None:	1✔
32	for report in reports:	×
33	self.add_report(report)	×
34
35	def add_report(self, report: ErrorReport):	1✔
36	"""
37	Adds a new ErrorReport
38
39	:param report: ErrorReport to add
40	:return: None
41	"""
42	if report.non_critical_bool:	1✔
43	self.noncritical_reports.append(report)	×
44	else:
45	self.reports.append(report)	1✔
46	all_failed_images = self.failed_images + report.contents	1✔
47	self.failed_images = sorted(list(set(all_failed_images)))	1✔
48
49	def __add__(self, other):	1✔
50	for report in other.reports:	1✔
51	self.add_report(report)	1✔
52	return self	1✔
53
54	def get_all_reports(self) -> list[ErrorReport]:	1✔
55	"""
56	Returns the full list of error reports (both critical and non-critical).
57
58	:return: list of ErrorReports
59	"""
60	return self.reports + self.noncritical_reports	1✔
61
62	def summarise_error_stack(self, output_path=None, verbose: bool = True) -> str:	1✔
63	"""
64	Returns a string summary of all ErrorReports.
65
66	:param output_path: Path to write summary in .txt format (optional)
67	:param verbose: boolean whether to provide a verbose summary
68	:return: String summary of errors
69	"""
70
71	is_known_error = [x.known_error_bool for x in self.reports]	1✔
72
73	summary = (
74	f"Error report summarising {len(self.reports)} errors. \n"
75	f"Code version: {package_name}=={__version__} \n \n"
76	f"{int(len(is_known_error) - np.sum(is_known_error))}/{len(is_known_error)}"
77	f" errors were errors not raised by {package_name}. \n"
78	f"The remaining {int(np.sum(is_known_error))}/{len(is_known_error)} "
79	f"errors were known errors raised by {package_name}. \n"
80	f"An additional {len(self.noncritical_reports)} non-critical "
81	f"errors were raised. \n"
82	)
83	all_reports = self.get_all_reports()	1✔
84
85	if len(all_reports) > 0:	1✔
86
87	summary += (	1✔
88	f"The following {len(self.failed_images)} images were affected "
89	f"by at least one error during processing: \n "
90	)
91
92	if verbose:	1✔
93	summary += f"{self.failed_images} \n \n"	1✔
94
95	summary += "Summarising each error: \n\n"	1✔
96
97	logger.error(f"Found {len(self.reports)} errors caught by code.")
98
99	error_lines = [err.get_error_message() for err in all_reports]	1✔
100
101	for error_type in list(set(error_lines)):	1✔
102
103	matching_errors = [	1✔
104	x for x in all_reports if x.get_error_message() == error_type
105	]
106
107	img_paths = []	1✔
108	error_name = None	1✔
109	for err in matching_errors:	1✔
110	img_paths += err.contents	1✔
111	if error_name is None:	1✔
112	error_name = err.get_error_name()	1✔
113	img_paths = list(set(img_paths))	1✔
114
115	line = error_type.split("\n")[0]	1✔
116	summary += (	1✔
117	f"Found {error_lines.count(error_type)} counts of error "
118	f"{error_name}, "
119	f"affecting {len(img_paths)} images: \n{line}.\n \n"
120	)
121
122	summary += " \n"	1✔
123
124	if verbose:	1✔
125	for report in all_reports:	1✔
126	summary += str(report.generate_full_traceback())	1✔
127
128	else:
129	msg = "No raised errors found in processing"
130	logger.info(msg)	1✔
131	summary += f"\n {msg}"	1✔
132
133	if output_path is not None:	1✔
134	logger.error(f"Saving tracebacks of caught errors to {output_path}")
135	with open(output_path, "w", encoding="utf-8") as err_file:	1✔
136	err_file.write(summary)	1✔
137
138	return summary	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