Coverage for python/lsst/pipe/tasks/rewrite_images/_rewrite_difference_image.py: 0%

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 

22from __future__ import annotations 

23 

24__all__ = ("RewriteDifferenceImageConnections", "RewriteDifferenceImageTask", "RewriteDifferenceImageConfig") 

25 

26from typing import ClassVar 

27 

28import astropy.units 

29 

30import lsst.pipe.base.connectionTypes as cT 

31from lsst.afw.image import PhotoCalib 

32from lsst.images import DifferenceImage 

33from lsst.images.fields import field_from_legacy_photo_calib 

34from lsst.pex.config import Field 

35from lsst.pipe.base import ( 

36 InputQuantizedConnection, 

37 OutputQuantizedConnection, 

38 PipelineTask, 

39 PipelineTaskConfig, 

40 PipelineTaskConnections, 

41 QuantumContext, 

42 Struct, 

43) 

44 

45 

46class RewriteDifferenceImageConnections( 

47 PipelineTaskConnections, 

48 dimensions={"visit", "detector"}, 

49 defaultTemplates={"legacy_prefix": "legacy_", "future_prefix": ""}, 

50): 

51 legacy_exposure = cT.Input( 

52 "{legacy_prefix}difference_image", 

53 # We expect the repository storage class to be ExposureF, but we set 

54 # the storage class we want to DifferenceImage so the butler can do 

55 # most of the conversion on read, and in doing so preserve the 

56 # quantization so we don't have doubly-lossless compression. 

57 storageClass="DifferenceImage", 

58 dimensions={"visit", "detector"}, 

59 deferLoad=True, # So we can pass preserve_quantization=True as a parameter. 

60 doc="The input image to convert.", 

61 ) 

62 visit_summary = cT.Input( 

63 "visit_summary", 

64 storageClass="ExposureCatalog", 

65 dimensions={"visit"}, 

66 doc="A visit summary catalog with the PhotoCalib that was already applied to the image's pixels.", 

67 ) 

68 future_difference_image = cT.Output( 

69 "{future_prefix}difference_image", 

70 storageClass="DifferenceImage", 

71 dimensions={"visit", "detector"}, 

72 doc="The output difference image.", 

73 ) 

74 

75 config: RewriteDifferenceImageConfig 

76 

77 

78class RewriteDifferenceImageConfig( 

79 PipelineTaskConfig, 

80 pipelineConnections=RewriteDifferenceImageConnections, 

81): 

82 instrumental_unit = Field( 

83 "Unit for instrumental flux pixels (i.e. uncalibrated side of a PhotoCalib).", 

84 dtype=str, 

85 default="electron", 

86 ) 

87 

88 

89class RewriteDifferenceImageTask(PipelineTask): 

90 ConfigClass: ClassVar[type[RewriteDifferenceImageConfig]] = RewriteDifferenceImageConfig 

91 config: RewriteDifferenceImageConfig 

92 _DefaultName = "rewriteDifferenceImage" 

93 

94 def runQuantum( 

95 self, 

96 butlerQC: QuantumContext, 

97 inputRefs: InputQuantizedConnection, 

98 outputRefs: OutputQuantizedConnection, 

99 ) -> None: 

100 inputs = butlerQC.get(inputRefs) 

101 difference_image = inputs.pop("legacy_exposure").get(parameters={"preserve_quantization": True}) 

102 visit_summary = inputs.pop("visit_summary") 

103 photo_calib: PhotoCalib = visit_summary.find(butlerQC.quantum.dataId["detector"]).getPhotoCalib() 

104 outputs = self.run(difference_image, photo_calib=photo_calib, **inputs) 

105 butlerQC.put(outputs, outputRefs) 

106 

107 def run(self, difference_image: DifferenceImage, *, photo_calib: PhotoCalib) -> Struct: 

108 instrumental_unit = astropy.units.Unit(self.config.instrumental_unit) 

109 difference_image.photometric_scaling = field_from_legacy_photo_calib( 

110 photo_calib, bounds=difference_image.bbox, instrumental_unit=instrumental_unit 

111 ) 

112 return Struct(future_difference_image=difference_image)