Coverage for python/lsst/pipe/tasks/consolidateSsTables.py: 76%

1# This file is part of pipe_tasks. 

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 

22__all__ = ["ConsolidateSsTablesConfig", "ConsolidateSsTablesTask", "ConsolidateSsTablesConnections"] 

23 

24import astropy.table as tb 

25import numpy as np 

26import warnings 

27 

28import lsst.pipe.base as pipeBase 

29import lsst.pipe.base.connectionTypes as cT 

30from lsst.obs.base.utils import TableVStack 

31from lsst.utils.timer import timeMethod 

32 

33warnings.filterwarnings('ignore') 

34 

35 

36class ConsolidateSsTablesConnections(pipeBase.PipelineTaskConnections, 

37 dimensions=("skymap",), 

38 defaultTemplates={"coaddName": "goodSeeing", 

39 "fakesType": ""}): 

40 inputCatalogs = cT.Input( 

41 doc="associated ssSources from all tract-patches.", 

42 name="{fakesType}{coaddName}Diff_assocSsSrcTable", 

43 storageClass="ArrowAstropy", 

44 dimensions=("tract", "patch"), 

45 multiple=True, 

46 deferLoad=True, 

47 ) 

48 ssSourceTable = cT.Output( 

49 doc="", 

50 name="{fakesType}{coaddName}Diff_ssSrcTable", 

51 storageClass="ArrowAstropy", 

52 dimensions=(), 

53 ) 

54 ssObjectTable = cT.Output( 

55 doc="", 

56 name="{fakesType}{coaddName}Diff_ssObjTable", 

57 storageClass="ArrowAstropy", 

58 dimensions=(), 

59 ) 

60 

61 

62class ConsolidateSsTablesConfig(pipeBase.PipelineTaskConfig, 

63 pipelineConnections=ConsolidateSsTablesConnections): 

64 

65 """Config for ConsolidateSsTablesTask""" 

66 

67 

68class ConsolidateSsTablesTask(pipeBase.PipelineTask): 

69 """ Consolidate per-patch ssSource tables into a single table. 

70 Create ssObject table 

71 TODO (DM-49451): Fit per-object parameters 

72 TODO (DM-49453): Generate MPCORB table. 

73 """ 

74 

75 ConfigClass = ConsolidateSsTablesConfig 

76 _DefaultName = "consolidateSsTables" 

77 

78 @timeMethod 

79 def run(self, inputCatalogs): 

80 """Concatenate per-patch ssSource tables. 

81 Generate ssObject table. 

82 

83 Parameters 

84 ---------- 

85 inputCatalogs `list` of `lsst.daf.butler.DeferredDatasetHandle`: 

86 All per-patch ssSource Tables 

87 

88 Returns 

89 ------- 

90 output : `lsst.pipe.base.Struct` 

91 Results struct with attributes: 

92 

93 ``sSourceTable`` 

94 Table of ssSources. 

95 (`astropy.table.Table`) 

96 ``ssObjectTable`` 

97 Table of ssObjects 

98 (`astropy.table.Table`). 

99 """ 

100 self.log.info("Concatenating %s per-patch ssSource Tables", 

101 len(inputCatalogs)) 

102 ssSourceTable = TableVStack.vstack_handles(inputCatalogs) 

103 ssObjectTable = tb.Table() 

104 ssObjectTable['ssObjectId'], ssObjectTable['numObs'] = np.unique(ssSourceTable['ssObjectId'], 

105 return_counts=True) 

106 ssObjectTable['discoverySubmissionDate'] = np.nan 

107 

108 return pipeBase.Struct( 

109 ssSourceTable=ssSourceTable, 

110 ssObjectTable=ssObjectTable, 

111 )