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 

30from lsst.multiprofit.fitting.fit_catalog import CatalogFitterConfig 

31import lsst.pex.config as pexConfig 

32from lsst.pex.config.configurableActions import ConfigurableAction 

33import numpy as np 

34 

35 

36class CatalogAction(ConfigurableAction): 

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

38 

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

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

41 

42 Parameters 

43 ---------- 

44 data 

45 A dict-like catalog. 

46 **kwargs 

47 Additional keyword arguments. 

48 

49 Returns 

50 ------- 

51 data 

52 The original data, modified in-place. 

53 """ 

54 return data 

55 

56 

57class MergeMultibandFluxes(CatalogAction): 

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

59 

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

61 

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

63 datasetType = kwargs.get("datasetType") 

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

65 

66 # Check if the table metadata has relevant config settings 

67 if ( 

68 self.name_model 

69 and hasattr(data, "meta") 

70 and datasetType 

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

72 ): 

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

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

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

76 column_id = config_dict.get("column_id") 

77 else: 

78 prefix = CatalogFitterConfig.prefix_column.default 

79 suffix_error = CatalogFitterConfig.suffix_error.default 

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

81 

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

83 columns_flux_band = defaultdict(list) 

84 for column in data.columns: 

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

86 if column.endswith("_flux"): 

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

88 columns_flux_band[band].append(column) 

89 else: 

90 columns_rest.append(column) 

91 

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

93 

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

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

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

97 if len(columns_band) > 1: 

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

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

100 data[column_flux] = flux 

101 

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

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

104 if errors: 

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

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

107 data[column_flux_err] = flux_err 

108 columns_exclude_prefix.add(column_flux_err) 

109 else: 

110 data.rename_columns( 

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

112 ) 

113 

114 columns_exclude_prefix.add(column_flux) 

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

116 

117 if prefix_model: 

118 # Add prefixes to the column names, if needed 

119 colnames = [ 

120 ( 

121 col 

122 if (col in columns_exclude_prefix) 

123 else ( 

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

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

126 ) 

127 ) 

128 for col in data.columns 

129 ] 

130 if hasattr(data, "rename_columns"): 

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

132 else: 

133 data.columns = colnames 

134 

135 return data