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

93.51

/winterdrp/data/base_data.py

"""
This contains the base data classes for the :module:`wintedrp.processors`.

The smallest unit is a :class:`~winterdrp.data.base_data.DataBlock` object,
corresponding to a single image.
These :class:`~winterdrp.data.base_data.DataBlock` objects are grouped into
:class:`~winterdrp.data.base_data.DataBatch` objects.
Each :class:`~wintedrp.processors.BaseProcessor` will operate on a individual
:class:`~winterdrp.data.base_data.DataBatch` object.

The :class:`~winterdrp.data.base_data.DataBatch` objects are stored within a larger
:class:`~winterdrp.data.base_data.DataSet` object.
A :class:`~wintedrp.processors.BaseProcessor` will iterate over each
:class:`~winterdrp.data.base_data.DataBatch` in a
:class:`~winterdrp.data.base_data.Dataset`.
"""
import logging
from pathlib import Path
from typing import Optional, Type

from winterdrp.paths import BASE_NAME_KEY, RAW_IMG_KEY

logger = logging.getLogger(__name__)


class DataBlock:
    """Base unit for processing, corresponding to a single image."""

    def __init__(self):
        self.raw_img_list = [Path(x) for x in self[RAW_IMG_KEY].split(",")]
        self.base_name = self[BASE_NAME_KEY]

    def __getitem__(self, item):
        raise NotImplementedError

    def __setitem__(self, key, value):
        raise NotImplementedError

    def get_name(self) -> str:
        """Function to retrieve the :variable:`winterdrp.paths.BASE_NAME_KEY`
        of the parent image

        :return: Base name of parent image
        """
        return self.base_name

    def get_raw_img_list(self) -> list[Path]:
        """Function to retrieve the paths of all raw images from
        which this object is derived.
        Because of stacking, this list may include multiple entries.

        :return: List of path strings
        """
        return self.raw_img_list


class PseudoList:
    """
    Base Class for a list-like object which contains a list of data.
    Other classes inherit from this object.

    The basic idea is that this class holds all the functions
    for safely creating an object with a specified data type.

    This class also contains the relevant magic functions so that `len(x)`, `x[i] = N`,
    and `for y in x` work as intended.
    """

    @property
    def data_type(self):
        """
        Each list should take one specific data type.
        This is where that type is defined.
        """
        raise NotImplementedError()

    def __init__(self, data_list=None):
        self._datalist = []

        if data_list is None:
            data_list = []
        elif isinstance(data_list, self.data_type):
            data_list = [data_list]

        if not isinstance(data_list, list):
            err = f"Found {data_list} of type {type(data_list)}. Expected a list."
            logger.error(err)
            raise ValueError(err)

        for item in data_list:
            self.append(item)

    def get_data_list(self):
        """
        Retrieve the data list

        :return: The saved list of objects
        """
        return self._datalist

    def append(self, item):
        """
        Function to append, list-style, new objects.

        :param item: Object to be added
        :return: None
        """
        self._append(item)

    def _append(self, item):
        """
        Protected method to append, list-style, new objects.
        This function also checks the data type to ensure they are correct.

        :param item: Object to be added
        :return: None
        """

        if not isinstance(item, self.data_type):
            err = (
                f"Error appending item {item} of type {type(item)}. "
                f"Expected a {self.data_type} item"
            )
            logger.error(err)
            raise ValueError(err)

        if len(self._datalist) > 0:
            if not isinstance(item, type(self._datalist[0])):
                err = (
                    f"Error appending item {item} of type {type(item)}. "
                    f"This {self.__class__.__name__} object already contains "
                    f"data of type {type(self._datalist[0])}. "
                    f"Please ensure all data is of the same type."
                )
                logger.error(err)
                raise ValueError(err)
        self._datalist.append(item)

    def __getitem__(self, item):
        return self._datalist.__getitem__(item)

    def __setitem__(self, key, value):
        self._datalist.__setitem__(key, value)

    def __add__(self, other):
        new = self.__class__()
        for item in self.get_data_list():
            new.append(item)
        for item in other.get_data_list():
            new.append(item)
        return new

    def __iadd__(self, other):
        for item in other.get_data_list():
            self._datalist.append(item)
        return self

    def __len__(self):
        return self._datalist.__len__()

    def __iter__(self):
        return self._datalist.__iter__()


class DataBatch(PseudoList):
    """
    Base class for a collection of individual
    :class:`~winterdrp.data.base_data.DataBlock` objects.
    Each :class:`~winterdrp.data.base_data.DataBatch` will be operated on
    by a :class:`~wintedrp.processors.BaseProcessor`
    """

    @property
    def data_type(self) -> Type[DataBlock]:
        raise NotImplementedError()

    def __init__(self, batch: Optional[list[DataBlock] | DataBlock] = None):
        super().__init__(data_list=batch)

    def get_batch(self) -> list[DataBlock]:
        """Returns the :class:`~winterdrp.data.base_data.DataBlock`
        items within the batch

        :return: list of :class:`~winterdrp.data.base_data.DataBlock` objects
        """
        return self.get_data_list()

    def get_raw_image_names(self) -> list[Path]:
        """Returns the name of each parent raw image

        :return: list of raw image names
        """
        img_list = []
        for data_block in self.get_batch():
            img_list += [Path(x).name for x in data_block.get_raw_img_list()]
        return img_list

    def __str__(self):
        return (
            f"<An {self.__class__.__name__} object, "
            f"containing {[x.get_name() for x in self.get_batch()]}>"
        )


class Dataset(PseudoList):
    """
    Base class for a collection of individual
    :class:`~winterdrp.data.base_data.DataBatch` objects.
    A :class:`~wintedrp.processors.BaseProcessor` will iterate over these.
    """

    data_type = DataBatch

    def get_batches(self):
        """Returns the :class:`~winterdrp.data.base_data.DataBatch`
        items within the batch

        :return: list of :class:`~winterdrp.data.base_data.DataBatch` objects
        """
        return self.get_data_list()

    def __init__(self, batches: Optional[list[DataBatch] | DataBatch] = None):
        super().__init__(data_list=batches)

    def append(self, item: DataBatch):
        """
        Function to append, list-style, new objects.

        :param item: Object to be added
        :return: None
        """
        super()._append(item)

1	"""
2	This contains the base data classes for the :module:`wintedrp.processors`.
3
4	The smallest unit is a :class:`~winterdrp.data.base_data.DataBlock` object,
5	corresponding to a single image.
6	These :class:`~winterdrp.data.base_data.DataBlock` objects are grouped into
7	:class:`~winterdrp.data.base_data.DataBatch` objects.
8	Each :class:`~wintedrp.processors.BaseProcessor` will operate on a individual
9	:class:`~winterdrp.data.base_data.DataBatch` object.
10
11	The :class:`~winterdrp.data.base_data.DataBatch` objects are stored within a larger
12	:class:`~winterdrp.data.base_data.DataSet` object.
13	A :class:`~wintedrp.processors.BaseProcessor` will iterate over each
14	:class:`~winterdrp.data.base_data.DataBatch` in a
15	:class:`~winterdrp.data.base_data.Dataset`.
16	"""
17	import logging	2✔
18	from pathlib import Path	2✔
19	from typing import Optional, Type	2✔
20
21	from winterdrp.paths import BASE_NAME_KEY, RAW_IMG_KEY	2✔
22
23	logger = logging.getLogger(__name__)	2✔
24
25
26	class DataBlock:	2✔
27	"""Base unit for processing, corresponding to a single image."""
28
29	def __init__(self):	2✔
30	self.raw_img_list = [Path(x) for x in self[RAW_IMG_KEY].split(",")]	2✔
31	self.base_name = self[BASE_NAME_KEY]	2✔
32
33	def __getitem__(self, item):	2✔
34	raise NotImplementedError
35
36	def __setitem__(self, key, value):	2✔
37	raise NotImplementedError
38
39	def get_name(self) -> str:	2✔
40	"""Function to retrieve the :variable:`winterdrp.paths.BASE_NAME_KEY`
41	of the parent image
42
43	:return: Base name of parent image
44	"""
45	return self.base_name	2✔
46
47	def get_raw_img_list(self) -> list[Path]:	2✔
48	"""Function to retrieve the paths of all raw images from
49	which this object is derived.
50	Because of stacking, this list may include multiple entries.
51
52	:return: List of path strings
53	"""
54	return self.raw_img_list	×
55
56
57	class PseudoList:	2✔
58	"""
59	Base Class for a list-like object which contains a list of data.
60	Other classes inherit from this object.
61
62	The basic idea is that this class holds all the functions
63	for safely creating an object with a specified data type.
64
65	This class also contains the relevant magic functions so that `len(x)`, `x[i] = N`,
66	and `for y in x` work as intended.
67	"""
68
69	@property	2✔
70	def data_type(self):	2✔
71	"""
72	Each list should take one specific data type.
73	This is where that type is defined.
74	"""
75	raise NotImplementedError()
76
77	def __init__(self, data_list=None):	2✔
78	self._datalist = []	2✔
79
80	if data_list is None:	2✔
81	data_list = []	2✔
82	elif isinstance(data_list, self.data_type):	2✔
83	data_list = [data_list]	2✔
84
85	if not isinstance(data_list, list):	2✔
86	err = f"Found {data_list} of type {type(data_list)}. Expected a list."
87	logger.error(err)
88	raise ValueError(err)
89
90	for item in data_list:	2✔
91	self.append(item)	2✔
92
93	def get_data_list(self):	2✔
94	"""
95	Retrieve the data list
96
97	:return: The saved list of objects
98	"""
99	return self._datalist	2✔
100
101	def append(self, item):	2✔
102	"""
103	Function to append, list-style, new objects.
104
105	:param item: Object to be added
106	:return: None
107	"""
108	self._append(item)	×
109
110	def _append(self, item):	2✔
111	"""
112	Protected method to append, list-style, new objects.
113	This function also checks the data type to ensure they are correct.
114
115	:param item: Object to be added
116	:return: None
117	"""
118
119	if not isinstance(item, self.data_type):	2✔
120	err = (
121	f"Error appending item {item} of type {type(item)}. "
122	f"Expected a {self.data_type} item"
123	)
124	logger.error(err)
125	raise ValueError(err)
126
127	if len(self._datalist) > 0:	2✔
128	if not isinstance(item, type(self._datalist[0])):	2✔
129	err = (
130	f"Error appending item {item} of type {type(item)}. "
131	f"This {self.__class__.__name__} object already contains "
132	f"data of type {type(self._datalist[0])}. "
133	f"Please ensure all data is of the same type."
134	)
135	logger.error(err)
136	raise ValueError(err)
137	self._datalist.append(item)	2✔
138
139	def __getitem__(self, item):	2✔
140	return self._datalist.__getitem__(item)	2✔
141
142	def __setitem__(self, key, value):	2✔
143	self._datalist.__setitem__(key, value)	2✔
144
145	def __add__(self, other):	2✔
146	new = self.__class__()	2✔
147	for item in self.get_data_list():	2✔
148	new.append(item)	2✔
149	for item in other.get_data_list():	2✔
150	new.append(item)	2✔
151	return new	2✔
152
153	def __iadd__(self, other):	2✔
154	for item in other.get_data_list():	2✔
155	self._datalist.append(item)	2✔
156	return self	2✔
157
158	def __len__(self):	2✔
159	return self._datalist.__len__()	2✔
160
161	def __iter__(self):	2✔
162	return self._datalist.__iter__()	2✔
163
164
165	class DataBatch(PseudoList):	2✔
166	"""
167	Base class for a collection of individual
168	:class:`~winterdrp.data.base_data.DataBlock` objects.
169	Each :class:`~winterdrp.data.base_data.DataBatch` will be operated on
170	by a :class:`~wintedrp.processors.BaseProcessor`
171	"""
172
173	@property	2✔
174	def data_type(self) -> Type[DataBlock]:	2✔
175	raise NotImplementedError()
176
177	def __init__(self, batch: Optional[list[DataBlock] \| DataBlock] = None):	2✔
178	super().__init__(data_list=batch)	2✔
179
180	def get_batch(self) -> list[DataBlock]:	2✔
181	"""Returns the :class:`~winterdrp.data.base_data.DataBlock`
182	items within the batch
183
184	:return: list of :class:`~winterdrp.data.base_data.DataBlock` objects
185	"""
186	return self.get_data_list()	×
187
188	def get_raw_image_names(self) -> list[Path]:	2✔
189	"""Returns the name of each parent raw image
190
191	:return: list of raw image names
192	"""
193	img_list = []	2✔
194	for data_block in self.get_batch():	2✔
195	img_list += [Path(x).name for x in data_block.get_raw_img_list()]	×
196	return img_list	2✔
197
198	def __str__(self):	2✔
199	return (	×
200	f"<An {self.__class__.__name__} object, "
201	f"containing {[x.get_name() for x in self.get_batch()]}>"
202	)
203
204
205	class Dataset(PseudoList):	2✔
206	"""
207	Base class for a collection of individual
208	:class:`~winterdrp.data.base_data.DataBatch` objects.
209	A :class:`~wintedrp.processors.BaseProcessor` will iterate over these.
210	"""
211
212	data_type = DataBatch	2✔
213
214	def get_batches(self):	2✔
215	"""Returns the :class:`~winterdrp.data.base_data.DataBatch`
216	items within the batch
217
218	:return: list of :class:`~winterdrp.data.base_data.DataBatch` objects
219	"""
220	return self.get_data_list()	2✔
221
222	def __init__(self, batches: Optional[list[DataBatch] \| DataBatch] = None):	2✔
223	super().__init__(data_list=batches)	2✔
224
225	def append(self, item: DataBatch):	2✔
226	"""
227	Function to append, list-style, new objects.
228
229	:param item: Object to be added
230	:return: None
231	"""
232	super()._append(item)	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