Coverage for tests/test_ndf_hds.py: 97%

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 

12from __future__ import annotations 

13 

14import os 

15from pathlib import Path 

16 

17import numpy as np 

18import pytest 

19 

20try: 

21 import h5py 

22 

23 from lsst.images.ndf import _hds 

24 

25 HAVE_H5PY = True 

26except ImportError: 

27 HAVE_H5PY = False 

28 

29skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed") 

30 

31EXAMPLE = os.path.join(os.path.dirname(__file__), "data", "example-ndf.sdf") 

32 

33 

34def _attr_str(value: object) -> str | None: 

35 """Return an h5py attribute decoded to str, or None if not string-like.""" 

36 if isinstance(value, bytes): 36 ↛ 38line 36 didn't jump to line 38 because the condition on line 36 was always true

37 return value.decode("ascii") 

38 if isinstance(value, str): 

39 return value 

40 return None 

41 

42 

43@skip_no_h5py 

44def test_real_array_round_trip(tmp_path: Path) -> None: 

45 """Verify float32 arrays round-trip through _hds write_array / 

46 read_array. 

47 """ 

48 path = tmp_path / "test.sdf" 

49 data = np.arange(12, dtype=np.float32).reshape(3, 4) 

50 with h5py.File(path, "w") as f: 

51 _hds.write_array(f, "DATA", data) 

52 with h5py.File(path, "r") as f: 

53 ds = f["DATA"] 

54 assert ds.dtype == np.float32 

55 assert ds.shape == (3, 4) 

56 assert dict(ds.attrs) == {} 

57 np.testing.assert_array_equal(_hds.read_array(ds), data) 

58 

59 

60@skip_no_h5py 

61def test_double_array_round_trip(tmp_path: Path) -> None: 

62 """Verify float64 arrays round-trip through _hds write_array / 

63 read_array. 

64 """ 

65 path = tmp_path / "test.sdf" 

66 data = np.linspace(0, 1, 6, dtype=np.float64).reshape(2, 3) 

67 with h5py.File(path, "w") as f: 

68 _hds.write_array(f, "DATA", data) 

69 with h5py.File(path, "r") as f: 

70 assert f["DATA"].dtype == np.float64 

71 np.testing.assert_array_equal(_hds.read_array(f["DATA"]), data) 

72 

73 

74@skip_no_h5py 

75def test_ubyte_and_integer(tmp_path: Path) -> None: 

76 """Verify uint8 and int32 arrays preserve dtype through _hds round-trip.""" 

77 path = tmp_path / "test.sdf" 

78 data_u = np.array([[1, 2, 3], [4, 5, 6]], dtype=np.uint8) 

79 data_i = np.array([10, 20, 30], dtype=np.int32) 

80 with h5py.File(path, "w") as f: 

81 _hds.write_array(f, "Q", data_u) 

82 _hds.write_array(f, "I", data_i) 

83 with h5py.File(path, "r") as f: 

84 assert f["Q"].dtype == np.uint8 

85 assert f["I"].dtype == np.int32 

86 np.testing.assert_array_equal(_hds.read_array(f["Q"]), data_u) 

87 np.testing.assert_array_equal(_hds.read_array(f["I"]), data_i) 

88 

89 

90@skip_no_h5py 

91def test_logical_uses_hdf5_bitfield(tmp_path: Path) -> None: 

92 """Verify boolean arrays are stored as HDF5 BITFIELD, not integer.""" 

93 path = tmp_path / "test.sdf" 

94 with h5py.File(path, "w") as f: 

95 _hds.write_array(f, "SCALAR", np.array(False, dtype=np.bool_)) 

96 _hds.write_array(f, "ARRAY", np.array([True, False], dtype=np.bool_)) 

97 with h5py.File(path, "r") as f: 

98 assert f["SCALAR"].id.get_type().get_class() == h5py.h5t.BITFIELD 

99 assert f["SCALAR"].id.get_type().get_size() == 1 

100 assert not _hds.read_array(f["SCALAR"]) 

101 assert f["ARRAY"].id.get_type().get_class() == h5py.h5t.BITFIELD 

102 assert f["ARRAY"].id.get_type().get_size() == 1 

103 np.testing.assert_array_equal(_hds.read_array(f["ARRAY"]), np.array([True, False])) 

104 

105 

106@skip_no_h5py 

107def test_unsupported_dtype_raises_on_write(tmp_path: Path) -> None: 

108 """Verify write_array raises NotImplementedError for complex128.""" 

109 path = tmp_path / "test.sdf" 

110 data = np.array([1.0], dtype=np.complex128) 

111 with h5py.File(path, "w") as f: 

112 with pytest.raises(NotImplementedError): 

113 _hds.write_array(f, "X", data) 

114 

115 

116@skip_no_h5py 

117def test_unsupported_dtype_raises_on_read(tmp_path: Path) -> None: 

118 """Verify read_array raises NotImplementedError when reading complex128.""" 

119 path = tmp_path / "test.sdf" 

120 with h5py.File(path, "w") as f: 

121 # Write directly with h5py, bypassing write_array's check. 

122 f.create_dataset("X", data=np.array([1.0], dtype=np.complex128)) 

123 with h5py.File(path, "r") as f: 

124 with pytest.raises(NotImplementedError): 

125 _hds.read_array(f["X"]) 

126 

127 

128@skip_no_h5py 

129def test_read_array_rejects_char_dataset(tmp_path: Path) -> None: 

130 """Verify read_array raises ValueError when given a char (string) 

131 dataset. 

132 """ 

133 path = tmp_path / "test.sdf" 

134 with h5py.File(path, "w") as f: 

135 _hds.write_char_array(f, "WCS", ["hello", "world"], width=16) 

136 with h5py.File(path, "r") as f: 

137 with pytest.raises(ValueError): 

138 _hds.read_array(f["WCS"]) 

139 

140 

141@skip_no_h5py 

142def test_char_array_round_trip(tmp_path: Path) -> None: 

143 """Verify string lines round-trip through write_char_array / 

144 read_char_array. 

145 """ 

146 path = tmp_path / "test.sdf" 

147 lines = ["Begin FrameSet", "Nframe = 5", "End FrameSet"] 

148 with h5py.File(path, "w") as f: 

149 _hds.write_char_array(f, "DATA", lines, width=80) 

150 with h5py.File(path, "r") as f: 

151 ds = f["DATA"] 

152 assert ds.dtype == np.dtype("|S80") 

153 assert ds.shape == (3,) 

154 assert dict(ds.attrs) == {} 

155 assert _hds.read_char_array(ds) == lines 

156 

157 

158@skip_no_h5py 

159def test_char_array_pads_and_strips(tmp_path: Path) -> None: 

160 """Verify write_char_array space-pads to width and read_char_array strips 

161 trailing spaces. 

162 """ 

163 path = tmp_path / "test.sdf" 

164 with h5py.File(path, "w") as f: 

165 _hds.write_char_array(f, "X", ["short"], width=80) 

166 with h5py.File(path, "r") as f: 

167 # Raw data should be space-padded to 80 characters. 

168 assert f["X"][0] == b"short" + b" " * 75 

169 # read_char_array strips trailing spaces. 

170 assert _hds.read_char_array(f["X"]) == ["short"] 

171 

172 

173@skip_no_h5py 

174def test_char_array_rejects_long_lines(tmp_path: Path) -> None: 

175 """Verify write_char_array raises ValueError when a line exceeds the 

176 width. 

177 """ 

178 path = tmp_path / "test.sdf" 

179 with h5py.File(path, "w") as f: 

180 with pytest.raises(ValueError): 

181 _hds.write_char_array(f, "X", ["too long"], width=3) 

182 

183 

184@skip_no_h5py 

185def test_char_array_rejects_non_ascii(tmp_path: Path) -> None: 

186 """Verify write_char_array raises ValueError for non-ASCII content.""" 

187 path = tmp_path / "test.sdf" 

188 with h5py.File(path, "w") as f: 

189 with pytest.raises(ValueError): 

190 _hds.write_char_array(f, "X", ["not ascii: \N{LATIN SMALL LETTER E WITH ACUTE}"], width=80) 

191 

192 

193@skip_no_h5py 

194def test_ndf_ast_data_encoding_uses_flagged_fixed_width_records() -> None: 

195 """Verify encode_ndf_ast_data produces fixed-width flagged records that 

196 decode correctly. 

197 """ 

198 text = ( 

199 ' Begin FrameSet\n# Title = "demo"\n VeryLongAttribute = 12345678901234567890\n End FrameSet\n' 

200 ) 

201 expected = 'Begin FrameSet\n# Title = "demo"\nVeryLongAttribute = 12345678901234567890\nEnd FrameSet\n' 

202 

203 records = _hds.encode_ndf_ast_data(text) 

204 

205 assert all(len(record) <= _hds.NDF_AST_DATA_WIDTH for record in records) 

206 assert all(record[0] in {" ", "+"} for record in records) 

207 assert ' # Title = "demo"' in records 

208 assert any(record.startswith("+") for record in records) 

209 assert _hds.decode_ndf_ast_data(records) == expected 

210 

211 

212@skip_no_h5py 

213def test_read_char_array_rejects_numeric_dataset(tmp_path: Path) -> None: 

214 """Verify read_char_array raises ValueError when given a numeric 

215 dataset. 

216 """ 

217 path = tmp_path / "test.sdf" 

218 with h5py.File(path, "w") as f: 

219 _hds.write_array(f, "DATA", np.zeros((2,), dtype=np.float32)) 

220 with h5py.File(path, "r") as f: 

221 with pytest.raises(ValueError): 

222 _hds.read_char_array(f["DATA"]) 

223 

224 

225@skip_no_h5py 

226def test_hds_type_for_dtype() -> None: 

227 """Verify hds_type_for_dtype maps numpy dtypes to correct HDS type 

228 strings. 

229 """ 

230 assert _hds.hds_type_for_dtype(np.dtype(np.bool_)) == "_LOGICAL" 

231 assert _hds.hds_type_for_dtype(np.dtype(np.float32)) == "_REAL" 

232 assert _hds.hds_type_for_dtype(np.dtype(np.float64)) == "_DOUBLE" 

233 assert _hds.hds_type_for_dtype(np.dtype(np.uint8)) == "_UBYTE" 

234 assert _hds.hds_type_for_dtype(np.dtype(np.int32)) == "_INTEGER" 

235 assert _hds.hds_type_for_dtype(np.dtype("|S80")) == "_CHAR*80" 

236 with pytest.raises(NotImplementedError): 

237 _hds.hds_type_for_dtype(np.dtype(np.complex128)) 

238 

239 

240@skip_no_h5py 

241def test_create_open_structure(tmp_path: Path) -> None: 

242 """Verify create_structure and open_structure round-trip CLASS and child 

243 names. 

244 """ 

245 path = tmp_path / "test.sdf" 

246 with h5py.File(path, "w") as f: 

247 ndf = _hds.create_structure(f, "ROOT", "NDF") 

248 _hds.create_structure(ndf, "DATA_ARRAY", "ARRAY") 

249 with h5py.File(path, "r") as f: 

250 root_obj = f["ROOT"] 

251 assert _attr_str(root_obj.attrs["CLASS"]) == "NDF" 

252 root, root_type = _hds.open_structure(f, "ROOT") 

253 assert root_type == "NDF" 

254 child_names = sorted(name for name, _ in _hds.iter_children(root)) 

255 assert child_names == ["DATA_ARRAY"] 

256 _, child_type = _hds.open_structure(root, "DATA_ARRAY") 

257 assert child_type == "ARRAY" 

258 

259 

260@skip_no_h5py 

261def test_open_structure_missing_class_raises(tmp_path: Path) -> None: 

262 """Verify open_structure raises ValueError when the CLASS attribute is 

263 absent. 

264 """ 

265 path = tmp_path / "test.sdf" 

266 with h5py.File(path, "w") as f: 

267 f.create_group("BAD") 

268 with h5py.File(path, "r") as f: 

269 with pytest.raises(ValueError): 

270 _hds.open_structure(f, "BAD") 

271 

272 

273@skip_no_h5py 

274def test_open_structure_accepts_legacy_hdstype(tmp_path: Path) -> None: 

275 """Verify open_structure accepts the legacy HDSTYPE attribute in place 

276 of CLASS. 

277 """ 

278 path = tmp_path / "test.sdf" 

279 with h5py.File(path, "w") as f: 

280 g = f.create_group("LEGACY") 

281 g.attrs["HDSTYPE"] = b"NDF" 

282 with h5py.File(path, "r") as f: 

283 _, t = _hds.open_structure(f, "LEGACY") 

284 assert t == "NDF" 

285 

286 

287@skip_no_h5py 

288def test_set_root_name(tmp_path: Path) -> None: 

289 """Verify set_root_name writes HDS_ROOT_NAME and CLASS to the root 

290 group. 

291 """ 

292 path = tmp_path / "test.sdf" 

293 with h5py.File(path, "w") as f: 

294 _hds.set_root_name(f, "MYNDF", "NDF") 

295 with h5py.File(path, "r") as f: 

296 assert _attr_str(f["/"].attrs["HDS_ROOT_NAME"]) == "MYNDF" 

297 assert _attr_str(f["/"].attrs["CLASS"]) == "NDF" 

298 

299 

300@skip_no_h5py 

301def test_root_is_ndf_with_root_name() -> None: 

302 """Verify the canonical example NDF has CLASS=NDF and the expected root 

303 name. 

304 """ 

305 with h5py.File(EXAMPLE, "r") as f: 

306 assert _attr_str(f["/"].attrs["CLASS"]) == "NDF" 

307 assert _attr_str(f["/"].attrs["HDS_ROOT_NAME"]) == "M57" 

308 

309 

310@skip_no_h5py 

311def test_data_array_is_array_structure() -> None: 

312 """Verify the example NDF's DATA_ARRAY is an ARRAY structure with 

313 correct dtype. 

314 """ 

315 with h5py.File(EXAMPLE, "r") as f: 

316 data_array, hds_type = _hds.open_structure(f, "DATA_ARRAY") 

317 assert hds_type == "ARRAY" 

318 data = data_array["DATA"] 

319 assert data.dtype == np.int16 

320 assert data.shape == (611, 609) 

321 assert _hds.hds_type_for_dtype(data.dtype) == "_WORD" 

322 arr = _hds.read_array(data) 

323 assert arr.shape == (611, 609) 

324 origin = _hds.read_array(data_array["ORIGIN"]) 

325 assert origin.dtype == np.int64 

326 assert origin.shape == (2,) 

327 

328 

329@skip_no_h5py 

330def test_wcs_is_structure_with_ast_text() -> None: 

331 """Verify the example NDF's WCS structure contains a valid AST FrameSet 

332 text dump. 

333 """ 

334 with h5py.File(EXAMPLE, "r") as f: 

335 wcs, hds_type = _hds.open_structure(f, "WCS") 

336 assert hds_type == "WCS" 

337 lines = _hds.read_char_array(wcs["DATA"]) 

338 text = _hds.decode_ndf_ast_data(lines) 

339 stripped = [line.lstrip() for line in text.splitlines()] 

340 assert any(s.startswith("Begin FrameSet") for s in stripped) 

341 assert any(s.startswith("End FrameSet") for s in stripped) 

342 

343 

344@skip_no_h5py 

345def test_more_fits_present() -> None: 

346 """Verify the example NDF's MORE/FITS extension contains FITS cards 

347 including NAXIS. 

348 """ 

349 with h5py.File(EXAMPLE, "r") as f: 

350 more, hds_type = _hds.open_structure(f, "MORE") 

351 assert hds_type == "EXT" 

352 cards = _hds.read_char_array(more["FITS"]) 

353 assert len(cards) > 0 

354 assert any(c.startswith("NAXIS") for c in cards)