Coverage for python / lsst / multiprofit / modelconfig.py: 41%

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/>. 

21 

22__all__ = ["ModelConfig"] 

23 

24import string 

25from typing import Iterable 

26 

27import lsst.gauss2d.fit as g2f 

28import lsst.pex.config as pexConfig 

29 

30from .componentconfig import Fluxes 

31from .sourceconfig import SourceConfig 

32 

33 

34class ModelConfig(pexConfig.Config): 

35 """Configuration for an lsst.gauss2d.fit Model.""" 

36 

37 sources = pexConfig.ConfigDictField[str, SourceConfig](doc="The configuration for sources", default={}) 

38 

39 @staticmethod 

40 def format_label(label: str, name_source: str) -> str: 

41 return string.Template(label).safe_substitute(name_source=name_source) 

42 

43 def get_integral_label_default(self, sourceconfig: SourceConfig) -> str: 

44 prefix = "src: {name_source} " if self.has_prefix_source() else "" 

45 return f"{prefix}{sourceconfig.get_integral_label_default()}" 

46 

47 def has_prefix_source(self) -> bool: 

48 return (len(self.sources) > 1) or next(iter(self.sources.keys())) 

49 

50 def make_sources( 

51 self, 

52 component_group_fluxes_srcs: Iterable[list[list[Fluxes]]], 

53 label_integral: str | None = None, 

54 ) -> tuple[list[g2f.Source], list[g2f.Prior]]: 

55 n_src = len(self.sources) 

56 if component_group_fluxes_srcs is None or len(component_group_fluxes_srcs) != n_src: 

57 raise ValueError(f"{len(component_group_fluxes_srcs)=} != {n_src=}") 

58 

59 sources = [] 

60 priors = [] 

61 for component_group_fluxes, (name_src, config_src) in zip( 

62 component_group_fluxes_srcs, self.sources.items() 

63 ): 

64 label_integral_src = ( 

65 label_integral 

66 if label_integral is not None 

67 else (self.get_integral_label_default(config_src)) 

68 ) 

69 

70 source, priors_src = config_src.make_source( 

71 component_group_fluxes=component_group_fluxes, 

72 label_integral=self.format_label(label=label_integral_src, name_source=name_src), 

73 ) 

74 sources.append(source) 

75 priors.extend(priors_src) 

76 

77 return sources, priors 

78 

79 def make_model( 

80 self, 

81 component_group_fluxes_srcs: Iterable[list[list[Fluxes]]], 

82 data: g2f.DataD, 

83 psf_models: list[g2f.PsfModel], 

84 label_integral: str | None = None, 

85 ) -> g2f.ModelD: 

86 sources, priors = self.make_sources( 

87 component_group_fluxes_srcs=component_group_fluxes_srcs, 

88 label_integral=label_integral, 

89 ) 

90 

91 model = g2f.ModelD(data=data, psfmodels=psf_models, sources=sources, priors=priors) 

92 

93 return model