Coverage for python/lsst/analysis/tools/tasks/wholeSkyAnalysis.py: 62%

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__ = ( 

24 "WholeSkyAnalysisConfig", 

25 "WholeSkyAnalysisTask", 

26) 

27 

28from lsst.pex.config import ListField 

29from lsst.pipe.base import connectionTypes as ct 

30from lsst.skymap import BaseSkyMap 

31 

32from ..interfaces import AnalysisBaseConfig, AnalysisBaseConnections, AnalysisPipelineTask 

33 

34 

35class WholeSkyAnalysisConnections( 

36 AnalysisBaseConnections, 

37 dimensions=("skymap",), 

38 defaultTemplates={"outputName": "objectTableCore_wholeSky", "inputName": "objectTableCore_metricsTable"}, 

39): 

40 data = ct.Input( 

41 doc="Tract based table to load from the butler", 

42 name="{inputName}", 

43 storageClass="ArrowAstropy", 

44 deferLoad=True, 

45 dimensions=("skymap",), 

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 def __init__(self, *, config=None): 

56 """Customize the dimensions of the inputs/outputs for a specific 

57 instance. This enables it to be dynamically set at runtime, 

58 allowing the task to work with different datasets. 

59 

60 Parameters 

61 ---------- 

62 config : `WholeSkyAnalysisConfig` 

63 A config for `WholeSkyAnalysisConfig`. 

64 """ 

65 super().__init__(config=config) 

66 if config.inputDimensions: 

67 self.data = ct.Input( 

68 name=self.data.name, 

69 doc=self.data.doc, 

70 storageClass=self.data.storageClass, 

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

72 deferLoad=self.data.deferLoad, 

73 multiple=self.data.multiple, 

74 ) 

75 

76 if config.outputDimensions: 

77 self.dimensions.clear() 

78 self.dimensions.update(frozenset(sorted(config.outputDimensions))) 

79 

80 

81class WholeSkyAnalysisConfig(AnalysisBaseConfig, pipelineConnections=WholeSkyAnalysisConnections): 

82 inputDimensions = ListField[str]( 

83 default=(), 

84 doc=("Override the dimensions of the input data. "), 

85 ) 

86 outputDimensions = ListField[str]( 

87 default=(), doc=("Override the dimensions of the output data." "Also overrides the task dimensions.") 

88 ) 

89 

90 

91class WholeSkyAnalysisTask(AnalysisPipelineTask): 

92 ConfigClass = WholeSkyAnalysisConfig 

93 _DefaultName = "wholeSkyAnalysis"