Coverage for python / lsst / cell_coadds / _stitched_exposure_catalog.py: 36%

1# This file is part of cell_coadds. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <https://www.gnu.org/licenses/>. 

21 

22from __future__ import annotations 

23 

24import logging 

25from collections.abc import Generator, Iterable, Mapping 

26 

27import lsst.geom as geom 

28from lsst.skymap import Index2D 

29 

30from ._identifiers import ObservationIdentifiers 

31from ._uniform_grid import UniformGrid 

32 

33__all__ = ("StitchedExposureCatalog",) 

34 

35logger = logging.getLogger(__name__) 

36 

37 

38class StitchedExposureCatalog: 

39 """A class that quacks like an `lsst.afw.table.ExposureCatalog`. 

40 

41 Parameters 

42 ---------- 

43 ugrid : `~lsst.cell_coadds.UniformGrid` 

44 The uniform grid that defines the bounding boxes for each piece. 

45 gc : `Mapping[Index2D, tuple[ObservationIdentifiers, ...]` 

46 The grid container that holds the values for each cell. 

47 """ 

48 

49 def __init__(self, ugrid: UniformGrid, gc: Mapping[Index2D, tuple[ObservationIdentifiers, ...]]): 

50 self.ugrid = ugrid 

51 self.gc = gc 

52 self._inputs: list[ObservationIdentifiers] = [] 

53 

54 def __iter__(self) -> Generator[ObservationIdentifiers]: 

55 """Iterate over the unique set of `ObservationIdentifiers` 

56 corresponding to the visits that overlap the full bounding box. 

57 """ 

58 if not self._inputs: 

59 inputs: list = [] 

60 for idx in self.gc: 

61 inputs += self.gc[idx] 

62 self._inputs = sorted(set(inputs)) 

63 yield from self._inputs 

64 

65 def __len__(self) -> int: 

66 if not self._inputs: 

67 return len([self.__iter__()]) 

68 return len(self._inputs) 

69 

70 def subsetContaining(self, point: geom.Point2D) -> Iterable[ObservationIdentifiers]: 

71 """Obtain the set of `ObservationIdentifiers` corresponding to the 

72 visits at a given point on the image. 

73 

74 Parameters 

75 ---------- 

76 point : `~lsst.geom.Point2D` or `~lsst.geom.Point2I` 

77 The point at which to evaluate the BoundedField. 

78 

79 Returns 

80 ------- 

81 value: `~collections.abc.Iterable[ObservationIdentifiers]` 

82 The set of visits overlapping a given point on the image. 

83 """ 

84 eval_point = geom.Point2I(point) 

85 idx = self.ugrid.index(eval_point) 

86 try: 

87 return self.gc[idx] 

88 except KeyError: 

89 logger.info("No inputs found for %s", idx) 

90 return ()