3597802805

Build Type

push

github

Committed by GitHub

Commit Message

Lint and document base data structure (#226)

Run Details

11 of 11 new or added lines in 3 files covered. (100.0%)

4571 of 6109 relevant lines covered (74.82%)

0.75 hits per line

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

95.89

/winterdrp/data/base_data.py

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

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

The :class:`~winterdrp.data.image_data.DataBatch` objects are stored within a larger
:class:`~winterdrp.data.image_data.DataSet` object.
A :class:`~wintedrp.processors.BaseProcessor` will iterate over each
:class:`~winterdrp.data.image_data.DataBatch` in a
:class:`~winterdrp.data.image_data.DataSet`.
"""
import logging
from pathlib import Path
from typing import 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 = 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[str]:
        """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.image_data.DataBlock` objects.
    Each :class:`~winterdrp.data.image_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: list[DataBlock] | DataBlock = None):
        super().__init__(data_list=batch)

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

        :return: list of :class:`~winterdrp.data.image_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


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

    data_type = DataBatch

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

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

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

1	"""
2	This contains the base data classes for the :module:`wintedrp.processors`.
3
4	The smallest unit is a :class:`~winterdrp.data.image_data.DataBlock` object,
5	corresponding to a single image.
6	These :class:`~winterdrp.data.image_data.DataBlock` objects are grouped into
7	:class:`~winterdrp.data.image_data.DataBatch` objects.
8	Each :class:`~wintedrp.processors.BaseProcessor` will operate on a individual
9	:class:`~winterdrp.data.image_data.DataBatch` object.
10
11	The :class:`~winterdrp.data.image_data.DataBatch` objects are stored within a larger
12	:class:`~winterdrp.data.image_data.DataSet` object.
13	A :class:`~wintedrp.processors.BaseProcessor` will iterate over each
14	:class:`~winterdrp.data.image_data.DataBatch` in a
15	:class:`~winterdrp.data.image_data.DataSet`.
16	"""
17	import logging	1✔
18	from pathlib import Path	1✔
19	from typing import Type	1✔
20
21	from winterdrp.paths import base_name_key, raw_img_key	1✔
22
23	logger = logging.getLogger(__name__)	1✔
24
25
26	class DataBlock:	1✔
27	"""Base unit for processing, corresponding to a single image."""
28
29	def __init__(self):	1✔
30	self.raw_img_list = self[raw_img_key].split(",")	1✔
31	self.base_name = self[base_name_key]	1✔
32
33	def __getitem__(self, item):	1✔
34	raise NotImplementedError
35
36	def __setitem__(self, key, value):	1✔
37	raise NotImplementedError
38
39	def get_name(self) -> str:	1✔
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	1✔
46
47	def get_raw_img_list(self) -> list[str]:	1✔
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:	1✔
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	1✔
70	def data_type(self):	1✔
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):	1✔
78
79	self._datalist = []	1✔
80
81	if data_list is None:	1✔
82	data_list = []	1✔
83	elif isinstance(data_list, self.data_type):	1✔
84	data_list = [data_list]	1✔
85
86	if not isinstance(data_list, list):	1✔
87	err = f"Found {data_list} of type {type(data_list)}. Expected a list."
88	logger.error(err)
89	raise ValueError(err)
90
91	for item in data_list:	1✔
92	self.append(item)	1✔
93
94	def get_data_list(self):	1✔
95	"""
96	Retrieve the data list
97
98	:return: The saved list of objects
99	"""
100	return self._datalist	1✔
101
102	def append(self, item):	1✔
103	"""
104	Function to append, list-style, new objects.
105
106	:param item: Object to be added
107	:return: None
108	"""
109	self._append(item)	1✔
110
111	def _append(self, item):	1✔
112	"""
113	Protected method to append, list-style, new objects.
114	This function also checks the data type to ensure they are correct.
115
116	:param item: Object to be added
117	:return: None
118	"""
119
120	if not isinstance(item, self.data_type):	1✔
121	err = (
122	f"Error appending item {item} of type {type(item)}. "
123	f"Expected a {self.data_type} item"
124	)
125	logger.error(err)
126	raise ValueError(err)
127
128	if len(self._datalist) > 0:	1✔
129	if not isinstance(item, type(self._datalist[0])):	1✔
130	err = (
131	f"Error appending item {item} of type {type(item)}. "
132	f"This {self.__class__.__name__} object already contains "
133	f"data of type {type(self._datalist[0])}. "
134	f"Please ensure all data is of the same type."
135	)
136	logger.error(err)
137	raise ValueError(err)
138	self._datalist.append(item)	1✔
139
140	def __getitem__(self, item):	1✔
141	return self._datalist.__getitem__(item)	1✔
142
143	def __setitem__(self, key, value):	1✔
144	self._datalist.__setitem__(key, value)	1✔
145
146	def __add__(self, other):	1✔
147	new = self.__class__()	1✔
148	for item in self.get_data_list():	1✔
149	new.append(item)	1✔
150	for item in other.get_data_list():	1✔
151	new.append(item)	1✔
152	return new	1✔
153
154	def __iadd__(self, other):	1✔
155	for item in other.get_data_list():	1✔
156	self._datalist.append(item)	1✔
157	return self	1✔
158
159	def __len__(self):	1✔
160	return self._datalist.__len__()	1✔
161
162	def __iter__(self):	1✔
163	return self._datalist.__iter__()	1✔
164
165
166	class DataBatch(PseudoList):	1✔
167	"""
168	Base class for a collection of individual
169	:class:`~winterdrp.data.image_data.DataBlock` objects.
170	Each :class:`~winterdrp.data.image_data.DataBatch` will be operated on
171	by a :class:`~wintedrp.processors.BaseProcessor`
172	"""
173
174	@property	1✔
175	def data_type(self) -> Type[DataBlock]:	1✔
176	raise NotImplementedError()
177
178	def __init__(self, batch: list[DataBlock] \| DataBlock = None):	1✔
179	super().__init__(data_list=batch)	1✔
180
181	def get_batch(self) -> list[DataBlock]:	1✔
182	"""Returns the :class:`~winterdrp.data.image_data.DataBlock`
183	items within the batch
184
185	:return: list of :class:`~winterdrp.data.image_data.DataBlock` objects
186	"""
187	return self.get_data_list()	×
188
189	def get_raw_image_names(self) -> list[Path]:	1✔
190	"""Returns the name of each parent raw image
191
192	:return: list of raw image names
193	"""
194	img_list = []	1✔
195	for data_block in self.get_batch():	1✔
196	img_list += [Path(x).name for x in data_block.get_raw_img_list()]	×
197	return img_list	1✔
198
199
200	class Dataset(PseudoList):	1✔
201	"""
202	Base class for a collection of individual
203	:class:`~winterdrp.data.image_data.DataBatch` objects.
204	A :class:`~wintedrp.processors.BaseProcessor` will iterate over these.
205	"""
206
207	data_type = DataBatch	1✔
208
209	def get_batches(self):	1✔
210	"""Returns the :class:`~winterdrp.data.image_data.DataBatch`
211	items within the batch
212
213	:return: list of :class:`~winterdrp.data.image_data.DataBatch` objects
214	"""
215	return self.get_data_list()	1✔
216
217	def __init__(self, batches: list[DataBatch] \| DataBatch = None):	1✔
218	super().__init__(data_list=batches)	1✔

winter-telescope / winterdrp / 3597802805

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