Coverage for python / lsst / afw / display / _write_fits.py: 25%

1# This file is part of afw. 

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__ = ["writeFitsImage"] 

25 

26import io 

27import logging 

28import os 

29import subprocess 

30 

31import lsst.afw.fits 

32import lsst.afw.geom 

33import lsst.afw.image 

34import lsst.pex.exceptions 

35from lsst.daf.base import PropertyList, PropertySet 

36 

37_LOG = logging.getLogger(__name__) 

38 

39 

40def _add_wcs(wcs_name: str, ps: PropertyList, x0: int = 0, y0: int = 0) -> None: 

41 ps.setInt(f"CRVAL1{wcs_name}", x0, "(output) Column pixel of Reference Pixel") 

42 ps.setInt(f"CRVAL2{wcs_name}", y0, "(output) Row pixel of Reference Pixel") 

43 ps.setDouble(f"CRPIX1{wcs_name}", 1.0, "Column Pixel Coordinate of Reference") 

44 ps.setDouble(f"CRPIX2{wcs_name}", 1.0, "Row Pixel Coordinate of Reference") 

45 ps.setString(f"CTYPE1{wcs_name}", "LINEAR", "Type of projection") 

46 ps.setString(f"CTYPE1{wcs_name}", "LINEAR", "Type of projection") 

47 ps.setString(f"CUNIT1{wcs_name}", "PIXEL", "Column unit") 

48 ps.setString(f"CUNIT2{wcs_name}", "PIXEL", "Row unit") 

49 

50 

51def writeFitsImage( 

52 file: str | int | io.BytesIO, 

53 data: lsst.afw.image.Image | lsst.afw.image.Mask, 

54 wcs: lsst.afw.geom.SkyWcs | None = None, 

55 title: str = "", 

56 metadata: PropertySet | None = None, 

57) -> None: 

58 """Write a simple FITS file with no extensions. 

59 

60 Parameters 

61 ---------- 

62 file : `str` or `int` 

63 Path to a file or a file descriptor. 

64 data : `lsst.afw.Image` or `lsst.afw.Mask` 

65 Data to be displayed. 

66 wcs : `lsst.afw.geom.SkyWcs` or `None`, optional 

67 WCS to be written to header to FITS file. 

68 title : `str`, optional 

69 If defined, the value to be stored in the ``OBJECT`` header. 

70 Overrides any value found in ``metadata``. 

71 metadata : `lsst.daf.base.PropertySet` or `None`, optional 

72 Additional information to be written to FITS header. 

73 """ 

74 ps = PropertyList() 

75 

76 # Seed with the external metadata, stripping wcs keywords. 

77 if metadata: 

78 lsst.afw.geom.stripWcsMetadata(metadata) 

79 ps.update(metadata) 

80 

81 # Write WcsB, so that pixel (0,0) is correctly labelled (but ignoring XY0) 

82 _add_wcs("B", ps) 

83 

84 if not wcs: 

85 _add_wcs("", ps) # Works around a ds9 bug that WCSA/B is ignored if no WCS is present. 

86 else: 

87 try: 

88 ps.update(wcs.getFitsMetadata(bbox=data.getBBox())) 

89 except lsst.pex.exceptions.RuntimeError: 

90 _LOG.warning("WCS is not FITS-compatible and has no FITS approximation; displaying without WCS.") 

91 

92 if title: 

93 ps.set("OBJECT", title, "Image being displayed") 

94 

95 if isinstance(file, str): 

96 data.writeFits(file, metadata=ps) 

97 else: 

98 mem = lsst.afw.fits.MemFileManager() 

99 data.writeFits(manager=mem, metadata=ps) 

100 if isinstance(file, int): 

101 # Duplicate to prevent a double close, assuming the caller 

102 # will close the file descriptor they passed in. 

103 with os.fdopen(os.dup(file), "wb") as fh: 

104 fh.write(mem.getData()) 

105 elif isinstance(file, subprocess.Popen): 

106 file.communicate(input=mem.getData()) 

107 else: 

108 # Try the write() method directly. 

109 file.write(mem.getData())