Coverage for python/lsst/analysis/tools/tasks/exposureCatalogAnalysis.py: 48%

1# This file is part of analysis_tools. 

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/>. 

21from __future__ import annotations 

22 

23__all__ = ("ExposureCatalogAnalysisConfig", "ExposureCatalogAnalysisTask") 

24 

25import numpy as np 

26from astropy.table import Column 

27 

28from lsst.pex.config import Field, ListField 

29from lsst.pipe.base import connectionTypes as cT 

30from lsst.skymap import BaseSkyMap 

31 

32from ..interfaces import AnalysisBaseConfig, AnalysisBaseConnections, AnalysisPipelineTask 

33 

34 

35class ExposureCatalogAnalysisConnections( 

36 AnalysisBaseConnections, 

37 dimensions=(), 

38 defaultTemplates={"inputName": ""}, 

39): 

40 data = cT.Input( 

41 doc="Exposure Catalog to analyze.", 

42 name="{inputName}", 

43 storageClass="ExposureCatalog", 

44 dimensions=(), 

45 deferLoad=True, 

46 ) 

47 

48 skymap = cT.Input( 

49 doc="The skymap that covers the tract that the data is from.", 

50 name=BaseSkyMap.SKYMAP_DATASET_TYPE_NAME, 

51 storageClass="SkyMap", 

52 dimensions=("skymap",), 

53 ) 

54 

55 camera = cT.PrerequisiteInput( 

56 doc="Input camera to use for focal plane geometry.", 

57 name="camera", 

58 storageClass="Camera", 

59 dimensions=("instrument",), 

60 isCalibration=True, 

61 ) 

62 

63 def __init__(self, *, config=None): 

64 super().__init__(config=config) 

65 

66 self.data = cT.Input( 

67 doc=self.data.doc, 

68 name=self.data.name, 

69 storageClass=self.data.storageClass, 

70 deferLoad=self.data.deferLoad, 

71 dimensions=frozenset(sorted(config.inputTableDimensions)), 

72 multiple=self.data.multiple, 

73 ) 

74 

75 self.dimensions.update(frozenset(sorted(config.taskDimensions))) 

76 

77 if not config.loadSkymap: 

78 del self.skymap 

79 if not config.loadCamera: 

80 del self.camera 

81 

82 

83class ExposureCatalogAnalysisConfig( 

84 AnalysisBaseConfig, pipelineConnections=ExposureCatalogAnalysisConnections 

85): 

86 

87 inputTableDimensions = ListField[str](doc="Dimensions of the input table.", optional=False) 

88 

89 taskDimensions = ListField[str](doc="Task and output dimensions.", optional=False) 

90 

91 loadSkymap = Field[bool](doc="Whether to load the skymap.", default=False) 

92 

93 loadCamera = Field[bool](doc="Whether to load the camera.", default=False) 

94 

95 

96class ExposureCatalogAnalysisTask(AnalysisPipelineTask): 

97 """An ``AnalysisPipelineTask`` that loads an ExposureCatalog passes it 

98 to the atools for analysis. 

99 

100 It will also pass the ``camera`` and ``skyMap`` inputs to the parent run 

101 methods if these are requested by the config parameters. 

102 """ 

103 

104 ConfigClass = ExposureCatalogAnalysisConfig 

105 _DefaultName = "exposureCatalogAnalysisTask" 

106 

107 def loadData(self, handle): 

108 # The parent class loadData does not work for ExposureCatalog. 

109 

110 data = handle.get().asAstropy() 

111 for colname in list(data.columns.keys()): 

112 # An ExposureCatalog may contain elements that are themselves 

113 # arrays. In such cases, the arrays are expanded into multiple 

114 # columns named: <colname>_0, <colname>_1, etc. 

115 if isinstance(data[colname][0], np.ndarray): 

116 for index in np.arange(len(data[colname][0])): 

117 values = [row[index] for row in data[colname]] 

118 data.add_column(Column(name=f"{colname}_{index}", data=values)) 

119 

120 return data