Coverage for python / lsst / multiprofit / sourceconfig.py: 16%
142 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-17 08:58 +0000
1# This file is part of 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/>.
22__all__ = [
23 "ComponentConfigs",
24 "ComponentGroupConfig",
25 "SourceConfig",
26]
28import string
30import lsst.gauss2d.fit as g2f
31import lsst.pex.config as pexConfig
33from .componentconfig import (
34 CentroidConfig,
35 EllipticalComponentConfig,
36 Fluxes,
37 GaussianComponentConfig,
38 SersicComponentConfig,
39)
41ComponentConfigs = dict[str, EllipticalComponentConfig]
44class ComponentGroupConfig(pexConfig.Config):
45 """Configuration for a group of lsst.gauss2d.fit Components.
47 ComponentGroups may have linked CentroidParameters
48 and IntegralModels, e.g. if is_fractional is True.
50 Notes
51 -----
52 Gaussian components are generated first, then Sersic.
54 This config class has no equivalent in gauss2d_fit, because gauss2d_fit
55 model parameter dependencies implicitly. This class implements only a
56 subset of typical use cases, i.e. PSFs sharing a fractional integral
57 model with fixed unit flux, and galaxies/PSF components sharing a single
58 common centroid.
59 If greater flexibility in linking parameter values is needed,
60 users must assemble their own gauss2d_fit models directly.
61 """
63 centroids = pexConfig.ConfigDictField[str, CentroidConfig](
64 doc="Centroids by key, which can be a component name or 'default'."
65 "The 'default' key-value pair must be specified if it is needed.",
66 default={"default": CentroidConfig},
67 )
68 # TODO: Change this to just one EllipticalComponentConfig field
69 # when pex_config supports derived types in ConfigDictField
70 # (possibly DM-41049)
71 components_gauss = pexConfig.ConfigDictField[str, GaussianComponentConfig](
72 doc="Gaussian Components in the source",
73 optional=False,
74 default={},
75 )
76 components_sersic = pexConfig.ConfigDictField[str, SersicComponentConfig](
77 doc="Sersic Components in the component mixture",
78 optional=False,
79 default={},
80 )
81 is_fractional = pexConfig.Field[bool](doc="Whether the integral_model is fractional", default=False)
82 transform_fluxfrac_name = pexConfig.Field[str](
83 doc="The name of the reference transform for flux parameters",
84 default="logit_fluxfrac",
85 optional=True,
86 )
87 transform_flux_name = pexConfig.Field[str](
88 doc="The name of the reference transform for flux parameters",
89 default="log10",
90 optional=True,
91 )
93 @staticmethod
94 def format_label(label: str, name_component: str) -> str:
95 return string.Template(label).safe_substitute(name_component=name_component)
97 @staticmethod
98 def get_integral_label_default() -> str:
99 return "comp: ${name_component} " + EllipticalComponentConfig.get_integral_label_default()
101 def get_component_configs(self) -> ComponentConfigs:
102 component_configs: ComponentConfigs = dict(self.components_gauss)
103 for name, component in self.components_sersic.items():
104 component_configs[name] = component
105 return component_configs
107 @staticmethod
108 def get_fluxes_default(
109 channels: tuple[g2f.Channel],
110 component_configs: ComponentConfigs,
111 is_fractional: bool = False,
112 ) -> list[Fluxes]:
113 """Get default flux values for a ComponentConfigs instance.
115 Parameters
116 ----------
117 channels
118 A tuple of channels to populate with flux values.
119 component_configs
120 A dict of named EllipticalComponentConfigs to provide initial flux
121 values for.
122 is_fractional
123 Whether to return values for a fractional model. If True, all
124 components must have a fluxfrac config set and the first must
125 also have a valid flux config.
127 Returns
128 -------
129 fluxes
130 A dict of flux values by channel for each component.
131 """
132 if len(component_configs) == 0:
133 raise ValueError("Must provide at least one ComponentConfig")
134 fluxes = []
135 component_configs_iter = tuple(component_configs.values())[: len(component_configs) - is_fractional]
136 for idx, component_config in enumerate(component_configs_iter):
137 if is_fractional and (idx == 0):
138 value = component_config.flux.value_initial
139 fluxes.append({channel: value for channel in channels})
140 config_flux = component_config.fluxfrac if is_fractional else component_config.flux
141 value = config_flux.value_initial
142 fluxes.append({channel: value for channel in channels})
143 return fluxes
145 def make_components(
146 self,
147 component_fluxes: list[Fluxes],
148 label_integral: str | None = None,
149 ) -> tuple[list[g2f.Component], list[g2f.Prior]]:
150 """Make a list of lsst.gauss2d.fit.Component from this configuration.
152 Parameters
153 ----------
154 component_fluxes
155 A list of Fluxes to populate an appropriate
156 `lsst.gauss2d.fit.IntegralModel` with.
157 If self.is_fractional, the first item in the list must be
158 total fluxes while the remainder are fractions (the final
159 fraction is always fixed at 1.0 and must not be provided).
160 label_integral
161 A label to apply to integral parameters. Can reference the
162 relevant component name with ${name_component}}.
164 Returns
165 -------
166 componentdata
167 An appropriate ComponentData including the initialized component.
168 """
169 component_configs = self.get_component_configs()
170 fluxes_first = component_fluxes[0]
171 channels = fluxes_first.keys()
172 fluxes_all = (component_fluxes[1:] + [None]) if self.is_fractional else component_fluxes
173 if len(fluxes_all) != len(component_configs):
174 raise ValueError(f"{len(fluxes_all)=} != {len(component_configs)=}")
175 priors = []
176 idx_final = len(component_configs) - 1
177 components = []
178 last = None
180 centroid_default = None
181 for idx, (fluxes_component, (name_component, config_comp)) in enumerate(
182 zip(fluxes_all, component_configs.items())
183 ):
184 label_integral_comp = self.format_label(
185 label_integral if label_integral is not None else (config_comp.get_integral_label_default()),
186 name_component=name_component,
187 )
189 if self.is_fractional:
190 if idx == 0:
191 last = config_comp.make_linear_integral_model(
192 fluxes=fluxes_first,
193 label_integral=label_integral_comp,
194 )
196 is_final = idx == idx_final
197 if is_final:
198 params_frac = [
199 (channel, g2f.ProperFractionParameterD(1.0, fixed=True)) for channel in channels
200 ]
201 else:
202 if fluxes_component.keys() != channels:
203 raise ValueError(f"{name_component=} {fluxes_component=}")
204 params_frac = [
205 (
206 channel,
207 config_comp.make_fluxfrac_parameter(value=fluxfrac),
208 )
209 for channel, fluxfrac in fluxes_component.items()
210 ]
212 integral_model = g2f.FractionalIntegralModel(
213 params_frac,
214 model=last,
215 is_final=is_final,
216 )
217 # TODO: Omitting this crucial step should raise but doesn't
218 # There shouldn't be two IntegralModels with the same last
219 # especially not one is_final and one not
220 last = integral_model
221 else:
222 integral_model = config_comp.make_linear_integral_model(
223 fluxes_component,
224 label_integral=label_integral_comp,
225 )
227 centroid = self.centroids.get(name_component)
228 if not centroid:
229 if centroid_default is None:
230 centroid_default = self.centroids["default"].make_centroid()
231 centroid = centroid_default
232 componentdata = config_comp.make_component(
233 centroid=centroid,
234 integral_model=integral_model,
235 )
236 components.append(componentdata.component)
237 priors.extend(componentdata.priors)
238 return components, priors
240 def validate(self) -> None:
241 super().validate()
242 errors = []
243 components: ComponentConfigs = dict(self.components_gauss)
245 for name, component in self.components_sersic.items():
246 if name in components:
247 errors.append(
248 f"key={name} cannot be used in both self.components_gauss and self.components_sersic"
249 )
250 components[name] = component
252 keys = set(self.centroids.keys())
253 has_default = "default" in keys
254 for name in components.keys():
255 if name in keys:
256 keys.remove(name)
257 elif not has_default:
258 errors.append(f"component {name=} has no entry in self.centroids and default not specified")
259 if errors:
260 newline = "\n"
261 raise ValueError(f"ComponentMixtureConfig has validation errors:\n{newline.join(errors)}")
264class SourceConfig(pexConfig.Config):
265 """Configuration for an lsst.gauss2d.fit Source.
267 Sources may contain components with distinct centroids that may be linked
268 by a prior (e.g. a galaxy + AGN + star clusters),
269 although such priors are not yet implemented.
270 """
272 component_groups = pexConfig.ConfigDictField[str, ComponentGroupConfig](
273 doc="Components in the source",
274 optional=False,
275 )
277 def _make_components_priors(
278 self,
279 component_group_fluxes: list[list[Fluxes]],
280 label_integral: str,
281 validate_psf: bool = False,
282 ) -> tuple[list[g2f.Component], list[g2f.Prior]]:
283 """Make components and priors for this source.
285 Parameters
286 ----------
287 component_group_fluxes
288 A list of inputs to pass to make_components for each group in
289 self.component_groups.
290 label_integral
291 An integral parameter label to pass to self.format_label.
292 validate_psf
293 Whether to validate that component_group_fluxes only uses the NONE
294 channel, as required for a PSF model.
296 Returns
297 -------
298 components
299 The list of components for this source.
300 priors
301 The list of priors for this source.
303 Notes
304 -----
305 Components and priors are concatenated by iterating through
306 self.component_groups.
307 """
308 if len(component_group_fluxes) != len(self.component_groups):
309 raise ValueError(f"{len(component_group_fluxes)=} != {len(self.component_groups)=}")
310 components = []
311 priors = []
312 if validate_psf:
313 # PSFs must use only the NONE channel
314 keys_expected = (g2f.Channel.NONE,)
315 for component_fluxes, (name_group, component_group) in zip(
316 component_group_fluxes, self.component_groups.items()
317 ):
318 if validate_psf:
319 for idx, fluxes_comp in enumerate(component_fluxes):
320 keys = tuple(fluxes_comp.keys())
321 if keys != keys_expected:
322 raise ValueError(
323 f"{name_group=} comp[{idx}] {keys=} != {keys_expected=} with {validate_psf=}"
324 )
326 components_i, priors_i = component_group.make_components(
327 component_fluxes=component_fluxes,
328 label_integral=self.format_label(label=label_integral, name_group=name_group),
329 )
330 components.extend(components_i)
331 priors.extend(priors_i)
333 return components, priors
335 @staticmethod
336 def format_label(label: str, name_group: str) -> str:
337 return string.Template(label).safe_substitute(name_group=name_group)
339 def get_component_configs(self) -> ComponentConfigs:
340 has_prefix_group = self.has_prefix_group()
341 component_configs = {}
342 for name_group, config_group in self.component_groups.items():
343 prefix_group = f"{name_group}_" if has_prefix_group else ""
344 for name_comp, component_config in config_group.get_component_configs().items():
345 component_configs[f"{prefix_group}{name_comp}"] = component_config
346 return component_configs
348 def get_integral_label_default(self) -> str:
349 prefix = "mix: ${name_group} " if self.has_prefix_group() else ""
350 return f"{prefix}{ComponentGroupConfig.get_integral_label_default()}"
352 def has_prefix_group(self) -> bool:
353 return (len(self.component_groups) > 1) or bool(next(iter(self.component_groups.keys())))
355 def make_source(
356 self,
357 component_group_fluxes: list[list[Fluxes]],
358 label_integral: str | None = None,
359 ) -> tuple[g2f.Source, list[g2f.Prior]]:
360 """Make an lsst.gauss2d.fit.Source from this configuration.
362 Parameters
363 ----------
364 component_group_fluxes
365 A list of Fluxes for each of the self.component_groups to use
366 when calling make_components.
367 label_integral
368 A label to apply to integral parameters. Can reference the
369 relevant component mixture name with ${name_group}.
371 Returns
372 -------
373 source
374 An appropriate lsst.gauss2d.fit.Source.
375 priors
376 A list of priors from all constituent components.
377 """
378 if label_integral is None:
379 label_integral = self.get_integral_label_default()
380 components, priors = self._make_components_priors(
381 component_group_fluxes=component_group_fluxes,
382 label_integral=label_integral,
383 )
384 source = g2f.Source(components)
385 return source, priors
387 def make_psf_model(
388 self,
389 component_group_fluxes: list[list[Fluxes]],
390 label_integral: str | None = None,
391 ) -> tuple[g2f.PsfModel, list[g2f.Prior]]:
392 """Make an lsst.gauss2d.fit.PsfModel from this configuration.
394 This method will validate that the arguments make a valid PSF model,
395 i.e. with a unity total flux, and only one config for the none band.
397 Parameters
398 ----------
399 component_group_fluxes
400 A list of CentroidFluxes for each of the self.component_groups
401 when calling make_components.
402 label_integral
403 A label to apply to integral parameters. Can reference the
404 relevant component mixture name with ${name_group}.
406 Returns
407 -------
408 psf_model
409 An appropriate lsst.gauss2d.fit.PsfModel.
410 priors
411 A list of priors from all constituent components.
412 """
413 if label_integral is None:
414 label_integral = f"PSF {self.get_integral_label_default()}"
415 components, priors = self._make_components_priors(
416 component_group_fluxes=component_group_fluxes,
417 label_integral=label_integral,
418 validate_psf=True,
419 )
420 model = g2f.PsfModel(components=components)
422 return model, priors
424 def validate(self) -> None:
425 super().validate()
426 if not self.component_groups:
427 raise ValueError("Must have at least one componentgroup")