Coverage for python/lsst/meas/extensions/multiprofit/catalog_actions.py: 17%

1# This file is part of meas_extensions_multiprofit. 

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

23 "CatalogAction", 

24 "MergeMultibandFluxes", 

25) 

26 

27from collections import defaultdict 

28 

29import astropy.table 

30import numpy as np 

31 

32import lsst.pex.config as pexConfig 

33from lsst.multiprofit.fitting.fit_catalog import CatalogFitterConfig 

34from lsst.pex.config.configurableActions import ConfigurableAction 

35 

36 

37class CatalogAction(ConfigurableAction): 

38 """Configurable action to return a catalog.""" 

39 

40 def __call__(self, data, **kwargs): 

41 """Return a catalog, potentially modified in-place. 

42 

43 Parameters 

44 ---------- 

45 data 

46 A dict-like catalog. 

47 **kwargs 

48 Additional keyword arguments. 

49 

50 Returns 

51 ------- 

52 data 

53 The original data, modified in-place. 

54 """ 

55 return data 

56 

57 

58class MergeMultibandFluxes(CatalogAction): 

59 """Configurable action to merge single-band flux tables into one.""" 

60 

61 name_model = pexConfig.Field[str](doc="The name of the model that fluxes are measured from", default="") 

62 

63 def __call__(self, data: astropy.table.Table, **kwargs): 

64 datasetType = kwargs.get("datasetType") 

65 prefix_model = self.name_model + ("_" if self.name_model else "") 

66 

67 # Check if the table metadata has relevant config settings 

68 if ( 

69 self.name_model 

70 and hasattr(data, "meta") 

71 and datasetType 

72 and (config := data.meta.get(datasetType)) 

73 ): 

74 config_dict = config.get("config", {}) 

75 prefix = config_dict.get("prefix_column", CatalogFitterConfig.prefix_column.default) 

76 suffix_error = config_dict.get("suffix_error", CatalogFitterConfig.suffix_error.default) 

77 column_id = config_dict.get("column_id") 

78 else: 

79 prefix = CatalogFitterConfig.prefix_column.default 

80 suffix_error = CatalogFitterConfig.suffix_error.default 

81 column_id = "id" if "id" in data.colnames else None 

82 

83 columns_rest = [] if prefix else ([column_id] if column_id else []) 

84 columns_flux_band = defaultdict(list) 

85 for column in data.columns: 

86 if not prefix or column.startswith(prefix): 

87 if column.endswith("_flux"): 

88 band = column.split("_")[-2] 

89 columns_flux_band[band].append(column) 

90 else: 

91 columns_rest.append(column) 

92 

93 columns_exclude_prefix = set(columns_rest) if prefix_model else set() 

94 

95 for band, columns_band in columns_flux_band.items(): 

96 column_flux = f"{band}_{prefix_model}flux" 

97 column_flux_err = f"{column_flux}{suffix_error}" 

98 if len(columns_band) > 1: 

99 # Sum up component fluxes and make a total flux column 

100 flux = np.nansum([data[column] for column in columns_band], axis=0) 

101 data[column_flux] = flux 

102 

103 columns_band_err = [f"{column}{suffix_error}" for column in columns_band] 

104 errors = [data[column] ** 2 for column in columns_band_err if column in data.columns] 

105 if errors: 

106 flux_err = np.sqrt(np.nansum(errors, axis=0)) 

107 flux_err[flux_err == 0] = np.nan 

108 data[column_flux_err] = flux_err 

109 columns_exclude_prefix.add(column_flux_err) 

110 else: 

111 data.rename_columns( 

112 (columns_band[0], f"{columns_band[0]}{suffix_error}"), (column_flux, column_flux_err) 

113 ) 

114 

115 columns_exclude_prefix.add(column_flux) 

116 columns_exclude_prefix.add(f"{column_flux}{suffix_error}") 

117 

118 if prefix_model: 

119 # Add prefixes to the column names, if needed 

120 colnames = [ 

121 ( 

122 col 

123 if (col in columns_exclude_prefix) 

124 else ( 

125 f"{prefix}{prefix_model if (prefix_model != prefix) else ''}" 

126 f"{col.split(prefix, 1)[1] if prefix else col}" 

127 ) 

128 ) 

129 for col in data.columns 

130 ] 

131 if hasattr(data, "rename_columns"): 

132 data.rename_columns([x for x in data.columns], colnames) 

133 else: 

134 data.columns = colnames 

135 

136 return data