Coverage for python/lsst/images/ndf/_model.py: 75%

302 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-07-04 10:14 -0700

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"""Python intermediate representation for NDF/HDS content.""" 

13 

14from __future__ import annotations 

15 

16__all__ = ( 

17 "HdsExtension", 

18 "HdsPrimitive", 

19 "HdsStructure", 

20 "Ndf", 

21 "NdfArray", 

22 "NdfContainer", 

23 "NdfDocument", 

24 "NdfQuality", 

25 "NdfWcs", 

26) 

27 

28from collections.abc import Iterable, Mapping, Sequence 

29from dataclasses import dataclass, field 

30from typing import Any, Self, cast 

31 

32import h5py 

33import numpy as np 

34 

35from . import _hds 

36 

37 

38def _decode_ascii_attr(value: Any) -> str | None: 

39 """Decode an HDS ASCII attribute value from h5py.""" 

40 if value is None: 

41 return None 

42 if isinstance(value, bytes): 

43 return value.decode("ascii") 

44 if isinstance(value, np.bytes_): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true

45 return bytes(value).decode("ascii") 

46 if isinstance(value, str): 46 ↛ 48line 46 didn't jump to line 48 because the condition on line 46 was always true

47 return value 

48 return str(value) 

49 

50 

51@dataclass 

52class HdsPrimitive: 

53 """An HDS primitive component.""" 

54 

55 data: np.ndarray | h5py.Dataset | None = None 

56 """Numeric/logical array data, or an open HDF5 dataset when this 

57 primitive was read lazily from an existing file. 

58 """ 

59 

60 char_lines: list[str] | None = None 

61 """Character data for ``_CHAR*N`` primitives. A scalar string is 

62 represented as a one-element list with ``char_scalar=True``. 

63 """ 

64 

65 char_width: int | None = None 

66 """Fixed HDS character width for ``char_lines``.""" 

67 

68 is_char_scalar: bool = False 

69 """If `True`, write ``char_lines`` as a scalar ``_CHAR*N`` primitive. 

70 Otherwise write them as a one-dimensional character array. 

71 """ 

72 

73 compression_options: dict[str, Any] = field(default_factory=dict) 

74 """Optional h5py compression options for numeric array data.""" 

75 

76 @classmethod 

77 def array( 

78 cls, 

79 data: np.ndarray | h5py.Dataset, 

80 *, 

81 compression_options: Mapping[str, Any] | None = None, 

82 ) -> Self: 

83 """Create a numeric/logical HDS primitive. 

84 

85 Parameters 

86 ---------- 

87 data 

88 Numeric or logical array data, or an open HDF5 dataset to 

89 wrap lazily. 

90 compression_options 

91 Optional h5py compression options to apply when writing the 

92 array. 

93 """ 

94 return cls( 

95 data=data if isinstance(data, h5py.Dataset) else np.asarray(data), 

96 compression_options=dict(compression_options) if compression_options else {}, 

97 ) 

98 

99 @classmethod 

100 def char_array(cls, lines: Sequence[str], *, width: int = 80) -> Self: 

101 """Create a one-dimensional HDS ``_CHAR*N`` primitive. 

102 

103 Parameters 

104 ---------- 

105 lines 

106 Character strings forming the one-dimensional array. 

107 width 

108 Fixed HDS character width for each string. 

109 """ 

110 return cls(char_lines=list(lines), char_width=width) 

111 

112 @classmethod 

113 def char_scalar(cls, text: str, *, width: int | None = None) -> Self: 

114 """Create a scalar HDS ``_CHAR*N`` primitive. 

115 

116 Parameters 

117 ---------- 

118 text 

119 Scalar string value to store. 

120 width 

121 Fixed HDS character width; if not given it is taken from the 

122 encoded length of ``text``. 

123 """ 

124 encoded = text.encode("ascii") 

125 return cls(char_lines=[text], char_width=max(width or 0, len(encoded), 1), is_char_scalar=True) 

126 

127 @classmethod 

128 def from_hdf5(cls, dataset: h5py.Dataset) -> Self: 

129 """Build an HDS primitive model from an open HDF5 dataset. 

130 

131 Parameters 

132 ---------- 

133 dataset 

134 Open HDF5 dataset to wrap as an HDS primitive. 

135 """ 

136 if dataset.dtype.kind == "S": 

137 if dataset.ndim == 0: 

138 raw = dataset[()] 

139 if isinstance(raw, np.bytes_): 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true

140 raw = bytes(raw) 

141 assert isinstance(raw, bytes) 

142 return cls( 

143 char_lines=[raw.decode("ascii").rstrip(" ")], 

144 char_width=dataset.dtype.itemsize, 

145 is_char_scalar=True, 

146 ) 

147 return cls( 

148 char_lines=_hds.read_char_array(dataset), 

149 char_width=dataset.dtype.itemsize, 

150 is_char_scalar=False, 

151 ) 

152 return cls(data=dataset) 

153 

154 def read_array(self) -> np.ndarray: 

155 """Return this primitive as a numpy array.""" 

156 if self.data is None: 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true

157 raise TypeError("Character HDS primitives cannot be read as numeric arrays.") 

158 if isinstance(self.data, h5py.Dataset): 

159 return _hds.read_array(self.data) 

160 return self.data 

161 

162 def read_char_array(self) -> list[str]: 

163 """Return this primitive as stripped ASCII strings.""" 

164 if self.char_lines is not None: 164 ↛ 166line 164 didn't jump to line 166 because the condition on line 164 was always true

165 return list(self.char_lines) 

166 if isinstance(self.data, h5py.Dataset) and self.data.dtype.kind == "S": 

167 if self.data.ndim == 0: 

168 raw = self.data[()] 

169 if isinstance(raw, np.bytes_): 

170 raw = bytes(raw) 

171 assert isinstance(raw, bytes) 

172 return [raw.decode("ascii").rstrip(" ")] 

173 return _hds.read_char_array(self.data) 

174 raise TypeError("Numeric HDS primitives cannot be read as character arrays.") 

175 

176 def write_to_hdf5(self, parent: h5py.Group, name: str) -> h5py.Dataset: 

177 """Write this primitive to an HDF5 group. 

178 

179 Parameters 

180 ---------- 

181 parent 

182 HDF5 group to create the dataset in. 

183 name 

184 Name of the dataset to create within ``parent``. 

185 """ 

186 if name in parent: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true

187 del parent[name] 

188 if self.char_lines is not None: 

189 width = self.char_width 

190 if width is None: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true

191 width = max((len(line.encode("ascii")) for line in self.char_lines), default=1) 

192 if self.is_char_scalar: 

193 if len(self.char_lines) != 1: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true

194 raise ValueError("Scalar _CHAR*N primitives require exactly one string.") 

195 line = self.char_lines[0] 

196 encoded = line.encode("ascii") 

197 if len(encoded) > width: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true

198 raise ValueError( 

199 f"Scalar _CHAR*N primitive {name!r} is {len(encoded)} bytes, " 

200 f"longer than width={width}." 

201 ) 

202 return parent.create_dataset(name, data=np.bytes_(encoded.ljust(width))) 

203 return _hds.write_char_array(parent, name, self.char_lines, width=width) 

204 data = self.read_array() 

205 return _hds.write_array( 

206 parent, 

207 name, 

208 data, 

209 compression=self.compression_options.get("compression"), 

210 compression_opts=self.compression_options.get("compression_opts"), 

211 ) 

212 

213 

214class HdsStructure: 

215 """An HDS structure component with named child components. 

216 

217 Parameters 

218 ---------- 

219 hds_type 

220 HDS type tag for this structure. 

221 children 

222 Initial child components keyed by name. 

223 """ 

224 

225 def __init__( 

226 self, 

227 hds_type: str, 

228 children: Mapping[str, HdsStructure | HdsPrimitive] | None = None, 

229 ) -> None: 

230 self.hds_type = hds_type 

231 self.children: dict[str, HdsStructure | HdsPrimitive] = dict(children) if children else {} 

232 

233 @classmethod 

234 def from_hdf5(cls, group: h5py.Group) -> HdsStructure: 

235 """Build a structure model from an open HDF5 group. 

236 

237 Parameters 

238 ---------- 

239 group 

240 Open HDF5 group to read the structure from. 

241 """ 

242 hds_type = _decode_ascii_attr(group.attrs.get(_hds.ATTR_CLASS)) 

243 if hds_type is None: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true

244 hds_type = _decode_ascii_attr(group.attrs.get("HDSTYPE")) or "EXT" 

245 structure = _new_structure(hds_type) 

246 for name, child in group.items(): 

247 if isinstance(child, h5py.Group): 

248 structure.children[name] = cls.from_hdf5(child) 

249 elif isinstance(child, h5py.Dataset): 249 ↛ 246line 249 didn't jump to line 246 because the condition on line 249 was always true

250 structure.children[name] = HdsPrimitive.from_hdf5(child) 

251 return structure 

252 

253 def __contains__(self, path: str) -> bool: 

254 try: 

255 self.get(path) 

256 except KeyError: 

257 return False 

258 return True 

259 

260 def __getitem__(self, path: str) -> HdsStructure | HdsPrimitive: 

261 return self.get(path) 

262 

263 def items(self) -> Iterable[tuple[str, HdsStructure | HdsPrimitive]]: 

264 """Iterate over direct child components.""" 

265 return self.children.items() 

266 

267 def get(self, path: str) -> HdsStructure | HdsPrimitive: 

268 """Return a child component by relative or absolute path. 

269 

270 Parameters 

271 ---------- 

272 path 

273 Relative or absolute path to the child component. 

274 """ 

275 if path in ("", "/"): 

276 return self 

277 cursor: HdsStructure | HdsPrimitive = self 

278 for part in _split_path(path): 

279 if not isinstance(cursor, HdsStructure): 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true

280 raise KeyError(path) 

281 cursor = cursor.children[part] 

282 return cursor 

283 

284 def get_structure(self, path: str) -> HdsStructure: 

285 """Return a child structure by relative or absolute path. 

286 

287 Parameters 

288 ---------- 

289 path 

290 Relative or absolute path to the child structure. 

291 """ 

292 node = self.get(path) 

293 if not isinstance(node, HdsStructure): 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true

294 raise KeyError(f"{path!r} is an HDS primitive, not a structure.") 

295 return node 

296 

297 def set(self, path: str, node: HdsStructure | HdsPrimitive) -> None: 

298 """Set a child component by relative or absolute path. 

299 

300 Parameters 

301 ---------- 

302 path 

303 Relative or absolute path at which to store the component. 

304 node 

305 Child structure or primitive to store at ``path``. 

306 """ 

307 parts = _split_path(path) 

308 if not parts: 

309 raise ValueError("Cannot replace an HDS structure with itself.") 

310 parent = self.ensure_structure("/".join(parts[:-1]), "EXT") 

311 parent.children[parts[-1]] = node 

312 

313 def delete(self, path: str) -> None: 

314 """Delete a child component if it exists. 

315 

316 Parameters 

317 ---------- 

318 path 

319 Relative or absolute path to the component to remove. 

320 """ 

321 parts = _split_path(path) 

322 if not parts: 

323 raise ValueError("Cannot delete an HDS structure root.") 

324 parent = self.get_structure("/".join(parts[:-1])) 

325 parent.children.pop(parts[-1], None) 

326 

327 def ensure_structure(self, path: str, hds_type: str | None = None) -> HdsStructure: 

328 """Return an existing structure or create it and its parents. 

329 

330 In HDS each structure carries a type tag distinct from its name. 

331 Newly-created structures here default to having the part name as 

332 their type (e.g. ``/MORE/LSST/PSF`` → types ``EXT``, ``LSST``, 

333 ``PSF``). The well-known NDF extension container ``MORE`` is the 

334 sole exception and defaults to ``EXT``. Pass ``hds_type`` to 

335 override the type on the leaf component only; existing structures 

336 retain their type unless overridden. 

337 

338 Parameters 

339 ---------- 

340 path 

341 Relative or absolute path to the structure to return or 

342 create. 

343 hds_type 

344 HDS type tag to assign to the leaf component when it is 

345 created, or to override on an existing leaf. 

346 """ 

347 if path in ("", "/"): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true

348 return self 

349 cursor: HdsStructure = self 

350 parts = _split_path(path) 

351 for index, part in enumerate(parts): 

352 existing = cursor.children.get(part) 

353 is_leaf = index == len(parts) - 1 

354 if existing is None: 

355 if is_leaf and hds_type is not None: 

356 child_type = hds_type 

357 else: 

358 child_type = _default_hds_type_for_name(part) 

359 existing = _new_structure(child_type) 

360 cursor.children[part] = existing 

361 if not isinstance(existing, HdsStructure): 361 ↛ 362line 361 didn't jump to line 362 because the condition on line 361 was never true

362 raise KeyError(f"{part!r} already exists as an HDS primitive.") 

363 cursor = existing 

364 if hds_type is not None and cursor.hds_type != hds_type: 364 ↛ 365line 364 didn't jump to line 365 because the condition on line 364 was never true

365 cursor.hds_type = hds_type 

366 return cursor 

367 

368 def write_to_hdf5(self, parent: h5py.Group, name: str | None = None) -> h5py.Group: 

369 """Write this structure to an HDF5 group. 

370 

371 Parameters 

372 ---------- 

373 parent 

374 HDF5 group to write into, or the group to populate directly 

375 when ``name`` is `None`. 

376 name 

377 Name of the child group to create under ``parent``; if `None` 

378 the structure is written into ``parent`` itself. 

379 """ 

380 if name is None: 

381 group = parent 

382 _clear_hdf5_group(group) 

383 _hds.set_ascii_attr(group, _hds.ATTR_CLASS, self.hds_type) 

384 else: 

385 if name in parent: 385 ↛ 386line 385 didn't jump to line 386 because the condition on line 385 was never true

386 del parent[name] 

387 group = _hds.create_structure(parent, name, self.hds_type) 

388 for child_name, child in self.children.items(): 

389 if isinstance(child, HdsStructure): 

390 child.write_to_hdf5(group, child_name) 

391 else: 

392 child.write_to_hdf5(group, child_name) 

393 return group 

394 

395 

396class HdsExtension(HdsStructure): 

397 """A general-purpose HDS extension structure. 

398 

399 Parameters 

400 ---------- 

401 children 

402 Initial child components keyed by name. 

403 """ 

404 

405 def __init__(self, children: Mapping[str, HdsStructure | HdsPrimitive] | None = None) -> None: 

406 super().__init__("EXT", children) 

407 

408 

409class NdfContainer(HdsStructure): 

410 """A top-level HDS container for multiple NDFs and shared metadata. 

411 

412 Parameters 

413 ---------- 

414 children 

415 Initial child components keyed by name. 

416 """ 

417 

418 def __init__(self, children: Mapping[str, HdsStructure | HdsPrimitive] | None = None) -> None: 

419 super().__init__("EXT", children) 

420 

421 def ensure_ndf(self, path: str) -> Ndf: 

422 """Return or create a child NDF at ``path``. 

423 

424 Parameters 

425 ---------- 

426 path 

427 Relative or absolute path to the NDF to return or create. 

428 """ 

429 structure = self.ensure_structure(path, "NDF") 

430 if not isinstance(structure, Ndf): 430 ↛ 431line 430 didn't jump to line 431 because the condition on line 430 was never true

431 structure.__class__ = Ndf 

432 return cast(Ndf, structure) 

433 

434 

435@dataclass 

436class NdfArray: 

437 """An NDF ARRAY component.""" 

438 

439 data: np.ndarray | h5py.Dataset 

440 origin: np.ndarray | Sequence[int] | None = None 

441 bad_pixel: bool | None = None 

442 compression_options: dict[str, Any] = field(default_factory=dict) 

443 

444 def to_hds_structure(self) -> HdsStructure: 

445 """Convert this array component to an HDS ``ARRAY`` structure.""" 

446 structure = HdsStructure("ARRAY") 

447 structure.children["DATA"] = HdsPrimitive.array( 

448 self.data, 

449 compression_options=self.compression_options, 

450 ) 

451 if self.origin is not None: 451 ↛ 453line 451 didn't jump to line 453 because the condition on line 451 was always true

452 structure.children["ORIGIN"] = HdsPrimitive.array(np.asarray(self.origin)) 

453 if self.bad_pixel is not None: 

454 structure.children["BAD_PIXEL"] = HdsPrimitive.array(np.array(self.bad_pixel, dtype=np.bool_)) 

455 return structure 

456 

457 @classmethod 

458 def from_hds_structure(cls, structure: HdsStructure) -> Self: 

459 """Build an `NdfArray` facade from an HDS ``ARRAY`` structure. 

460 

461 Parameters 

462 ---------- 

463 structure 

464 HDS ``ARRAY`` structure to read the array component from. 

465 """ 

466 data = structure.children["DATA"] 

467 if not isinstance(data, HdsPrimitive): 

468 raise TypeError("ARRAY.DATA must be an HDS primitive.") 

469 origin: np.ndarray | None = None 

470 if (origin_node := structure.children.get("ORIGIN")) is not None: 

471 if not isinstance(origin_node, HdsPrimitive): 

472 raise TypeError("ARRAY.ORIGIN must be an HDS primitive.") 

473 origin = origin_node.read_array() 

474 bad_pixel: bool | None = None 

475 if (bad_pixel_node := structure.children.get("BAD_PIXEL")) is not None: 

476 if not isinstance(bad_pixel_node, HdsPrimitive): 

477 raise TypeError("ARRAY.BAD_PIXEL must be an HDS primitive.") 

478 bad_pixel = bool(np.asarray(bad_pixel_node.read_array()).reshape(-1)[0]) 

479 array_data = data.data if data.data is not None else data.read_array() 

480 return cls(data=array_data, origin=origin, bad_pixel=bad_pixel) 

481 

482 

483@dataclass 

484class NdfQuality: 

485 """An NDF QUALITY component.""" 

486 

487 quality: NdfArray 

488 badbits: int = 255 

489 

490 def to_hds_structure(self) -> HdsStructure: 

491 """Convert this quality component to an HDS ``QUALITY`` structure.""" 

492 structure = HdsStructure("QUALITY") 

493 structure.children["BADBITS"] = HdsPrimitive.array(np.array(self.badbits, dtype=np.uint8)) 

494 structure.children["QUALITY"] = self.quality.to_hds_structure() 

495 return structure 

496 

497 

498@dataclass 

499class NdfWcs: 

500 """An NDF WCS component represented by encoded AST channel records.""" 

501 

502 lines: list[str] 

503 width: int = _hds.NDF_AST_DATA_WIDTH 

504 

505 def to_hds_structure(self) -> HdsStructure: 

506 """Convert this WCS component to an HDS ``WCS`` structure.""" 

507 structure = HdsStructure("WCS") 

508 structure.children["DATA"] = HdsPrimitive.char_array(self.lines, width=self.width) 

509 return structure 

510 

511 

512class Ndf(HdsStructure): 

513 """An NDF structure with convenience accessors for standard components. 

514 

515 Parameters 

516 ---------- 

517 children 

518 Initial child components keyed by name. 

519 """ 

520 

521 def __init__(self, children: Mapping[str, HdsStructure | HdsPrimitive] | None = None) -> None: 

522 super().__init__("NDF", children) 

523 

524 def set_array_component( 

525 self, 

526 name: str, 

527 data: np.ndarray | h5py.Dataset, 

528 *, 

529 origin: Sequence[int] | np.ndarray | None = None, 

530 bad_pixel: bool | None = None, 

531 compression_options: Mapping[str, Any] | None = None, 

532 ) -> None: 

533 """Set an NDF array-like component such as ``DATA_ARRAY``. 

534 

535 Parameters 

536 ---------- 

537 name 

538 Name of the NDF component to set, such as ``DATA_ARRAY``. 

539 data 

540 Pixel data for the component. 

541 origin 

542 Pixel origin of the array; `None` to omit the ``ORIGIN`` 

543 component. 

544 bad_pixel 

545 Value of the bad-pixel flag; `None` to omit the 

546 ``BAD_PIXEL`` component. 

547 compression_options 

548 HDF5 compression options applied to the stored array. 

549 """ 

550 self.children[name] = NdfArray( 

551 data, 

552 origin=origin, 

553 bad_pixel=bad_pixel, 

554 compression_options=dict(compression_options) if compression_options else {}, 

555 ).to_hds_structure() 

556 

557 def set_quality(self, quality: NdfQuality) -> None: 

558 """Set the NDF ``QUALITY`` component. 

559 

560 Parameters 

561 ---------- 

562 quality 

563 Quality component to store. 

564 """ 

565 self.children["QUALITY"] = quality.to_hds_structure() 

566 

567 def set_wcs(self, wcs: NdfWcs) -> None: 

568 """Set the NDF ``WCS`` component. 

569 

570 Parameters 

571 ---------- 

572 wcs 

573 WCS component to store. 

574 """ 

575 self.children["WCS"] = wcs.to_hds_structure() 

576 

577 def set_units(self, units: str | None) -> None: 

578 """Set or remove the NDF ``UNITS`` component. 

579 

580 Parameters 

581 ---------- 

582 units 

583 Units string to store, or `None` to remove the component. 

584 """ 

585 if units is None: 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true

586 self.children.pop("UNITS", None) 

587 else: 

588 self.children["UNITS"] = HdsPrimitive.char_scalar(units) 

589 

590 def get_units(self) -> str | None: 

591 """Return the NDF ``UNITS`` component, if present.""" 

592 node = self.children.get("UNITS") 

593 if node is None: 593 ↛ 594line 593 didn't jump to line 594 because the condition on line 593 was never true

594 return None 

595 if not isinstance(node, HdsPrimitive): 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true

596 raise TypeError("NDF.UNITS must be an HDS primitive.") 

597 lines = node.read_char_array() 

598 return lines[0] if lines else "" 

599 

600 def ensure_lsst_extension(self, *, base_path: str = "MORE/LSST") -> HdsStructure: 

601 """Return or create the LSST extension structure for this NDF. 

602 

603 Parameters 

604 ---------- 

605 base_path 

606 Path to the LSST extension structure within the NDF. 

607 """ 

608 return self.ensure_structure(base_path, "EXT") 

609 

610 

611@dataclass 

612class NdfDocument: 

613 """A complete HDS root object containing one NDF or an NDF container.""" 

614 

615 root: Ndf | NdfContainer = field(default_factory=Ndf) 

616 root_name: str | None = None 

617 

618 @classmethod 

619 def from_hdf5(cls, file: h5py.File) -> Self: 

620 """Read an NDF document model from an open HDF5 file. 

621 

622 Parameters 

623 ---------- 

624 file 

625 Open HDF5 file to read the document from. 

626 """ 

627 root = HdsStructure.from_hdf5(file["/"]) 

628 typed_root: Ndf | NdfContainer 

629 if isinstance(root, Ndf | NdfContainer): 

630 typed_root = root 

631 elif root.hds_type == "NDF": 631 ↛ 632line 631 didn't jump to line 632 because the condition on line 631 was never true

632 root.__class__ = Ndf 

633 typed_root = cast(Ndf, root) 

634 else: 

635 root.__class__ = NdfContainer 

636 typed_root = cast(NdfContainer, root) 

637 return cls(root=typed_root, root_name=_decode_ascii_attr(file["/"].attrs.get(_hds.ATTR_ROOT_NAME))) 

638 

639 def write_to_hdf5(self, file: h5py.File) -> None: 

640 """Write this document to an open HDF5 file. 

641 

642 Parameters 

643 ---------- 

644 file 

645 Open HDF5 file to write the document into. 

646 """ 

647 self.root.write_to_hdf5(file["/"]) 

648 if self.root_name is not None: 

649 _hds.set_root_name(file, self.root_name, self.root.hds_type) 

650 

651 def ensure_ndf(self, path: str = "/") -> Ndf: 

652 """Return or create an NDF at the requested absolute path. 

653 

654 Parameters 

655 ---------- 

656 path 

657 Absolute path to the NDF to return or create. 

658 """ 

659 if path in ("", "/"): 

660 if not isinstance(self.root, Ndf): 660 ↛ 661line 660 didn't jump to line 661 because the condition on line 660 was never true

661 raise TypeError("The document root is an NDF container, not an NDF.") 

662 return self.root 

663 if isinstance(self.root, NdfContainer): 

664 return self.root.ensure_ndf(path) 

665 structure = self.root.ensure_structure(path, "NDF") 

666 if not isinstance(structure, Ndf): 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true

667 structure.__class__ = Ndf 

668 return cast(Ndf, structure) 

669 

670 def get(self, path: str) -> HdsStructure | HdsPrimitive: 

671 """Return a component by absolute path. 

672 

673 Parameters 

674 ---------- 

675 path 

676 Absolute path to the component to return. 

677 """ 

678 return self.root.get(path) 

679 

680 

681def _new_structure(hds_type: str) -> HdsStructure: 

682 """Create a typed HDS structure model for an HDS type.""" 

683 if hds_type == "NDF": 

684 return Ndf() 

685 if hds_type == "EXT": 

686 return HdsExtension() 

687 return HdsStructure(hds_type) 

688 

689 

690def _default_hds_type_for_name(name: str) -> str: 

691 """Return the HDS type to use for a structure with the given name. 

692 

693 Mirrors the convention in real NDF files where each named structure 

694 carries a type tag describing what it is. ``MORE`` is the standard 

695 NDF extension container and is always ``EXT``; every other name is 

696 used verbatim as its own type. 

697 """ 

698 return "EXT" if name == "MORE" else name 

699 

700 

701def _split_path(path: str) -> list[str]: 

702 """Split an HDS/HDF5 path into component names.""" 

703 return [part for part in path.strip("/").split("/") if part] 

704 

705 

706def _clear_hdf5_group(group: h5py.Group) -> None: 

707 """Remove children and HDS root attributes before rewriting a group. 

708 

709 Parameters 

710 ---------- 

711 group 

712 HDF5 group to clear of child datasets and HDS attributes. 

713 """ 

714 for name in list(group.keys()): 

715 del group[name] 

716 for name in (_hds.ATTR_CLASS, _hds.ATTR_ROOT_NAME, _hds.ATTR_STRUCTURE_DIMS, "HDSTYPE"): 

717 if name in group.attrs: 

718 del group.attrs[name]