Coverage for tests / test_ndf_layout.py: 14%

1# This file is part of lsst-images. 

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# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12"""Layout sanity tests for NdfOutputArchive. 

13 

14Opens files written by NdfOutputArchive with raw h5py and verifies 

15the on-disk layout matches the HDS-on-HDF5 / NDF spec. 

16 

17Notes on mask routing 

18--------------------- 

19NDF serialization stores ``Mask`` arrays as a 3-D ``uint8`` DATA primitive 

20whose HDS axes are ``(x, y, mask-byte)``. The HDF5 dataset shape is reversed 

21from that, following hds-v5 convention. It also writes a 2-D ``QUALITY`` 

22view: single-byte masks are copied directly, while wider masks collapse to 

230/1 values. 

24""" 

25 

26from __future__ import annotations 

27 

28import unittest 

29 

30import numpy as np 

31 

32from lsst.images import Box, Image, MaskedImage, MaskPlane, MaskSchema 

33from lsst.images.tests import RoundtripNdf 

34 

35try: 

36 import h5py 

37 

38 from lsst.images.ndf import _hds 

39 

40 HAVE_H5PY = True 

41except ImportError: 

42 HAVE_H5PY = False 

43 

44 

45def _cls(node: h5py.Group) -> str: 

46 """Return the HDS type (CLASS attribute) of an h5py group as a Python 

47 str. 

48 """ 

49 val = node.attrs.get(_hds.ATTR_CLASS) 

50 if val is None: 

51 # Legacy fallback used by older HDS variants. 

52 val = node.attrs.get("HDSTYPE") 

53 if isinstance(val, bytes): 

54 return val.decode("ascii") 

55 return str(val) 

56 

57 

58def _hds_type(dataset: h5py.Dataset) -> str: 

59 """Return the HDS primitive type string inferred from a dataset's numpy 

60 dtype or low-level HDF5 type class. 

61 """ 

62 dataset_type = dataset.id.get_type() 

63 if dataset_type.get_class() == h5py.h5t.BITFIELD: 

64 return "_LOGICAL" 

65 return _hds.hds_type_for_dtype(dataset.dtype) 

66 

67 

68def _hds_shape(dataset: h5py.Dataset) -> tuple[int, ...]: 

69 """Return the dataset shape in HDS/Fortran axis order.""" 

70 return tuple(reversed(dataset.shape)) 

71 

72 

73@unittest.skipUnless(HAVE_H5PY, "h5py is not installed") 

74class NdfImageLayoutTestCase(unittest.TestCase): 

75 """Verify the on-disk layout produced by ``ndf.write()`` for a plain 

76 ``Image``. 

77 """ 

78 

79 def test_image_layout(self) -> None: 

80 """Write an Image and verify root CLASS, DATA_ARRAY, ORIGIN, and LSST 

81 ext. 

82 """ 

83 image = Image( 

84 np.arange(20, dtype=np.float32).reshape(4, 5), 

85 bbox=Box.factory[10:14, 20:25], 

86 ) 

87 with RoundtripNdf(self, image) as roundtrip: 

88 f = roundtrip.inspect() 

89 # Root group carries CLASS="NDF". 

90 self.assertEqual(_cls(f["/"]), "NDF") 

91 

92 # DATA_ARRAY is an ARRAY structure. 

93 self.assertIn("DATA_ARRAY", f) 

94 self.assertEqual(_cls(f["/DATA_ARRAY"]), "ARRAY") 

95 

96 # DATA is a 2-D _REAL primitive whose shape matches the image. 

97 self.assertIn("DATA", f["/DATA_ARRAY"]) 

98 ds = f["/DATA_ARRAY/DATA"] 

99 self.assertEqual(_hds_type(ds), "_REAL") 

100 self.assertEqual(ds.ndim, 2) 

101 self.assertEqual(ds.shape, image.array.shape) 

102 

103 # ORIGIN stores bbox lower bounds as int64 in (x_min, y_min) 

104 # order. 

105 self.assertIn("ORIGIN", f["/DATA_ARRAY"]) 

106 origin = f["/DATA_ARRAY/ORIGIN"][()] 

107 self.assertEqual(origin.dtype, np.int64) 

108 self.assertEqual(int(origin[0]), 20) # x_min from Box.factory[10:14, 20:25] 

109 self.assertEqual(int(origin[1]), 10) # y_min 

110 

111 # /MORE is the standard NDF extension container (EXT) and 

112 # /MORE/LSST carries the type "LSST" matching its name. 

113 self.assertIn("MORE", f) 

114 self.assertEqual(_cls(f["/MORE"]), "EXT") 

115 self.assertIn("LSST", f["/MORE"]) 

116 self.assertEqual(_cls(f["/MORE/LSST"]), "LSST") 

117 

118 # Main JSON serialisation tree is present. 

119 self.assertIn("JSON", f["/MORE/LSST"]) 

120 

121 

122@unittest.skipUnless(HAVE_H5PY, "h5py is not installed") 

123class NdfCompatibleMaskLayoutTestCase(unittest.TestCase): 

124 """Layout test for a MaskedImage whose mask fits in a single uint8 byte. 

125 

126 Even though the mask schema has only 2 planes (which would fit in a single 

127 NDF QUALITY byte), MaskedImage writes the native 3-D uint8 backing array 

128 in ``/MORE/LSST/MASK`` and a direct 2-D copy in ``/QUALITY``. 

129 """ 

130 

131 def test_masked_image_compatible_mask_layout(self) -> None: 

132 """Write a MaskedImage with a ≤8-plane mask; verify QUALITY, 

133 LSST/MASK, and VARIANCE. 

134 """ 

135 planes = [MaskPlane("BAD", "Bad pixel"), MaskPlane("SAT", "Saturated")] 

136 schema = MaskSchema(planes) # default dtype=uint8, mask_size=1 

137 image = Image( 

138 np.arange(20, dtype=np.float32).reshape(4, 5), 

139 bbox=Box.factory[10:14, 20:25], 

140 ) 

141 # Pass an explicit float64 Image as variance so we can verify _DOUBLE 

142 # on disk (the default variance is float32, matching the image dtype). 

143 variance = Image(np.ones((4, 5), dtype=np.float64), bbox=image.bbox) 

144 masked = MaskedImage(image, mask_schema=schema, variance=variance) 

145 masked.mask.set("BAD", image.array % 2 == 0) 

146 masked.mask.set("SAT", image.array > 10) 

147 

148 with RoundtripNdf(self, masked) as roundtrip: 

149 f = roundtrip.inspect() 

150 self.assertIn("QUALITY", f) 

151 self.assertEqual(_cls(f["/QUALITY"]), "QUALITY") 

152 self.assertEqual(_cls(f["/QUALITY/QUALITY"]), "ARRAY") 

153 quality_ds = f["/QUALITY/QUALITY/DATA"] 

154 self.assertEqual(_hds_type(quality_ds), "_UBYTE") 

155 self.assertEqual(quality_ds.shape, image.array.shape) 

156 self.assertEqual(_hds_shape(quality_ds), (image.array.shape[1], image.array.shape[0])) 

157 np.testing.assert_array_equal(quality_ds[()], masked.mask.array[:, :, 0]) 

158 quality_origin = f["/QUALITY/QUALITY/ORIGIN"] 

159 self.assertEqual(_hds_type(quality_origin), "_INTEGER") 

160 self.assertEqual(list(quality_origin[()]), [20, 10]) 

161 bad_pixel = f["/QUALITY/QUALITY/BAD_PIXEL"] 

162 self.assertEqual(_hds_type(bad_pixel), "_LOGICAL") 

163 self.assertFalse(bad_pixel[()]) 

164 self.assertEqual(f["/QUALITY/BADBITS"][()], 255) 

165 

166 # /MORE/LSST/MASK is a sub-NDF (CLASS="NDF") with a 

167 # canonical DATA_ARRAY structure containing DATA + ORIGIN. 

168 self.assertIn("MORE", f) 

169 self.assertIn("LSST", f["/MORE"]) 

170 self.assertIn("MASK", f["/MORE/LSST"]) 

171 self.assertEqual(_cls(f["/MORE/LSST/MASK"]), "NDF") 

172 self.assertEqual(_cls(f["/MORE/LSST/MASK/DATA_ARRAY"]), "ARRAY") 

173 mask_ds = f["/MORE/LSST/MASK/DATA_ARRAY/DATA"] 

174 self.assertEqual(_hds_type(mask_ds), "_UBYTE") 

175 self.assertEqual(mask_ds.ndim, 3) 

176 self.assertEqual(mask_ds.shape, (1, 4, 5)) 

177 self.assertEqual(_hds_shape(mask_ds), (5, 4, 1)) 

178 origin = f["/MORE/LSST/MASK/DATA_ARRAY/ORIGIN"] 

179 self.assertEqual(origin.dtype, np.int64) 

180 # The mask shares the parent image's bbox; the trailing mask 

181 # byte axis keeps a zero origin. 

182 self.assertEqual(list(origin[()]), [20, 10, 0]) 

183 bad_pixel = f["/MORE/LSST/MASK/DATA_ARRAY/BAD_PIXEL"] 

184 self.assertEqual(_hds_type(bad_pixel), "_LOGICAL") 

185 self.assertFalse(bad_pixel[()]) 

186 

187 # VARIANCE is an ARRAY structure whose DATA is _DOUBLE 

188 # (float64). 

189 self.assertIn("VARIANCE", f) 

190 self.assertEqual(_cls(f["/VARIANCE"]), "ARRAY") 

191 self.assertIn("DATA", f["/VARIANCE"]) 

192 self.assertEqual(_hds_type(f["/VARIANCE/DATA"]), "_DOUBLE") 

193 

194 

195@unittest.skipUnless(HAVE_H5PY, "h5py is not installed") 

196class NdfIncompatibleMaskLayoutTestCase(unittest.TestCase): 

197 """Layout test for a MaskedImage with more than 8 mask planes. 

198 

199 A 12-plane uint8 mask has ``mask_size=2`` (two bytes per pixel), and the 

200 on-disk HDS axes are ``(x, y, mask-byte)``. 

201 """ 

202 

203 def test_masked_image_incompatible_mask_layout(self) -> None: 

204 """Write a MaskedImage with 12 planes; verify LSST/MASK and absent 

205 QUALITY. 

206 """ 

207 planes = [MaskPlane(f"P{i}", f"Plane {i}") for i in range(12)] 

208 schema = MaskSchema(planes) # default uint8; mask_size = ceil(12/8) = 2 

209 image = Image( 

210 np.arange(20, dtype=np.float32).reshape(4, 5), 

211 bbox=Box.factory[10:14, 20:25], 

212 ) 

213 masked = MaskedImage(image, mask_schema=schema) 

214 masked.mask.set("P0", image.array % 2 == 0) 

215 masked.mask.set("P11", image.array > 10) 

216 expected_quality = np.any(masked.mask.array != 0, axis=2).astype(np.uint8) 

217 

218 with RoundtripNdf(self, masked) as roundtrip: 

219 f = roundtrip.inspect() 

220 self.assertIn("QUALITY", f) 

221 self.assertEqual(_cls(f["/QUALITY/QUALITY"]), "ARRAY") 

222 quality_ds = f["/QUALITY/QUALITY/DATA"] 

223 self.assertEqual(_hds_type(quality_ds), "_UBYTE") 

224 self.assertEqual(quality_ds.shape, image.array.shape) 

225 np.testing.assert_array_equal(quality_ds[()], expected_quality) 

226 self.assertEqual(f["/QUALITY/BADBITS"][()], 255) 

227 

228 # /MORE/LSST/MASK is a sub-NDF. 

229 self.assertIn("MORE", f) 

230 self.assertIn("LSST", f["/MORE"]) 

231 self.assertIn("MASK", f["/MORE/LSST"]) 

232 self.assertEqual(_cls(f["/MORE/LSST/MASK"]), "NDF") 

233 self.assertEqual(_cls(f["/MORE/LSST/MASK/DATA_ARRAY"]), "ARRAY") 

234 

235 ds = f["/MORE/LSST/MASK/DATA_ARRAY/DATA"] 

236 self.assertEqual(_hds_type(ds), "_UBYTE") 

237 self.assertEqual(ds.ndim, 3) 

238 rows, cols = image.array.shape 

239 self.assertEqual(ds.shape, (2, rows, cols)) 

240 self.assertEqual(_hds_shape(ds), (cols, rows, 2)) 

241 bad_pixel = f["/MORE/LSST/MASK/DATA_ARRAY/BAD_PIXEL"] 

242 self.assertEqual(_hds_type(bad_pixel), "_LOGICAL") 

243 self.assertFalse(bad_pixel[()]) 

244 

245 def test_masked_image_many_plane_mask_layout(self) -> None: 

246 """Write a MaskedImage with more than 31 planes as one native mask.""" 

247 planes = [MaskPlane(f"P{i}", f"Plane {i}") for i in range(40)] 

248 schema = MaskSchema(planes) 

249 image = Image( 

250 np.arange(20, dtype=np.float32).reshape(4, 5), 

251 bbox=Box.factory[10:14, 20:25], 

252 ) 

253 masked = MaskedImage(image, mask_schema=schema) 

254 masked.mask.set("P0", image.array % 2 == 0) 

255 masked.mask.set("P17", image.array > 10) 

256 masked.mask.set("P39", image.array == 19) 

257 expected_quality = np.any(masked.mask.array != 0, axis=2).astype(np.uint8) 

258 

259 with RoundtripNdf(self, masked) as roundtrip: 

260 f = roundtrip.inspect() 

261 self.assertIn("QUALITY", f) 

262 self.assertEqual(_cls(f["/QUALITY/QUALITY"]), "ARRAY") 

263 quality_ds = f["/QUALITY/QUALITY/DATA"] 

264 self.assertEqual(_hds_type(quality_ds), "_UBYTE") 

265 self.assertEqual(quality_ds.shape, image.array.shape) 

266 np.testing.assert_array_equal(quality_ds[()], expected_quality) 

267 self.assertEqual(f["/QUALITY/BADBITS"][()], 255) 

268 ds = f["/MORE/LSST/MASK/DATA_ARRAY/DATA"] 

269 self.assertEqual(_hds_type(ds), "_UBYTE") 

270 self.assertEqual(ds.ndim, 3) 

271 rows, cols = image.array.shape 

272 self.assertEqual(ds.shape, (5, rows, cols)) 

273 self.assertEqual(_hds_shape(ds), (cols, rows, 5)) 

274 

275 

276if __name__ == "__main__": 

277 unittest.main()