Coverage for python/lsst/daf/butler/_formatter.py: 83%

533 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-24 02:19 +0000

1# This file is part of daf_butler. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <http://www.gnu.org/licenses/>. 

27 

28from __future__ import annotations 

29 

30__all__ = ( 

31 "FileIntegrityError", 

32 "Formatter", 

33 "FormatterFactory", 

34 "FormatterNotImplementedError", 

35 "FormatterParameter", 

36 "FormatterV1inV2", 

37 "FormatterV2", 

38) 

39 

40import contextlib 

41import copy 

42import logging 

43import os 

44import zipfile 

45from abc import ABCMeta, abstractmethod 

46from collections.abc import Callable, Iterator, Mapping, Set 

47from typing import TYPE_CHECKING, Any, BinaryIO, ClassVar, TypeAlias, final 

48 

49from lsst.resources import ResourceHandleProtocol, ResourcePath 

50from lsst.utils.introspection import get_full_type_name 

51from lsst.utils.timer import time_this 

52 

53from ._config import Config 

54from ._config_support import LookupKey, processLookupConfigs 

55from ._file_descriptor import FileDescriptor 

56from ._location import Location 

57from ._rubin.temporary_for_ingest import TemporaryForIngest 

58from .dimensions import DataCoordinate, DimensionUniverse 

59from .mapping_factory import MappingFactory 

60 

61log = logging.getLogger(__name__) 

62 

63if TYPE_CHECKING: 

64 from ._dataset_provenance import DatasetProvenance 

65 from ._dataset_ref import DatasetRef 

66 from ._dataset_type import DatasetType 

67 from ._storage_class import StorageClass 

68 from .datastore.cache_manager import AbstractDatastoreCacheManager 

69 

70 # Define a new special type for functions that take "entity" 

71 Entity: TypeAlias = DatasetType | DatasetRef | StorageClass | str 

72 

73 

74class FileIntegrityError(RuntimeError): 

75 """The file metadata is inconsistent with the metadata supplied by 

76 the datastore. 

77 """ 

78 

79 

80class FormatterNotImplementedError(NotImplementedError): 

81 """Formatter does not implement the specific read or write method 

82 that is being requested. 

83 """ 

84 

85 

86class FormatterV2: 

87 """Interface for reading and writing datasets using URIs. 

88 

89 The formatters are associated with a particular `StorageClass`. 

90 

91 Parameters 

92 ---------- 

93 file_descriptor : `FileDescriptor`, optional 

94 Identifies the file to read or write, and the associated storage 

95 classes and parameter information. 

96 ref : `DatasetRef` 

97 The dataset associated with this formatter. Should not be a component 

98 dataset ref. 

99 write_parameters : `dict`, optional 

100 Parameters to control how the dataset is serialized. 

101 write_recipes : `dict`, optional 

102 Detailed write recipes indexed by recipe name. 

103 

104 **kwargs 

105 Additional arguments that will be ignored but allow for 

106 `Formatter` V1 parameters to be given. 

107 

108 Notes 

109 ----- 

110 A `FormatterV2` author should not override the default `read` or `write` 

111 method. Instead for read the formatter author should implement one or all 

112 of `read_from_stream`, `read_from_uri`, or `read_from_local_file`. The 

113 method `read_from_uri` will always be attempted first and could be more 

114 efficient (since it allows the possibility for a subset of the data file to 

115 be accessed remotely when parameters or components are specified) but it 

116 will not update the local cache. If the entire contents of the remote file 

117 are being accessed (no component or parameters defined) and the dataset 

118 would be cached, `read_from_uri` will be called with a local file. If the 

119 file is remote and the parameters that have been included are known to be 

120 more efficiently handled with a local file, the `read_from_uri` method can 

121 return `NotImplemented` to indicate that a local file should be given 

122 instead. 

123 

124 Similarly for writes, the `write` method can not be subclassed. Instead 

125 the formatter author should implement `to_bytes` or `write_local_file`. 

126 For local URIs the system will always call `write_local_file` first (which 

127 by default will call `to_bytes`) to ensure atomic writes are implemented. 

128 For remote URIs with local caching disabled, `to_bytes` will be called 

129 first and the remote updated directly. If the dataset should be cached 

130 it will always be written locally first. 

131 """ 

132 

133 unsupported_parameters: ClassVar[Set[str] | None] = frozenset() 

134 """Set of read parameters not understood by this `Formatter`. An empty set 

135 means all parameters are supported. `None` indicates that no parameters 

136 are supported. These parameters should match those defined in the storage 

137 class definition. (`frozenset`). 

138 """ 

139 

140 supported_write_parameters: ClassVar[Set[str] | None] = None 

141 """Parameters understood by this formatter that can be used to control 

142 how a dataset is serialized. `None` indicates that no parameters are 

143 supported.""" 

144 

145 default_extension: ClassVar[str | None] = None 

146 """Default extension to use when writing a file. 

147 

148 Can be `None` if the extension is determined dynamically. Use the 

149 `get_write_extension` method to get the actual extension to use. 

150 """ 

151 

152 supported_extensions: ClassVar[Set[str]] = frozenset() 

153 """Set of all extensions supported by this formatter. 

154 

155 Any extension assigned to the ``default_extension`` property will be 

156 automatically included in the list of supported extensions. 

157 """ 

158 

159 can_read_from_uri: ClassVar[bool] = False 

160 """Declare whether `read_from_uri` is available to this formatter.""" 

161 

162 can_read_from_stream: ClassVar[bool] = False 

163 """Declare whether `read_from_stream` is available to this formatter.""" 

164 

165 can_read_from_local_file: ClassVar[bool] = False 

166 """Declare whether `read_from_local_file` is available to this 

167 formatter.""" 

168 

169 def __init__( 

170 self, 

171 file_descriptor: FileDescriptor, 

172 *, 

173 ref: DatasetRef, 

174 write_parameters: Mapping[str, Any] | None = None, 

175 write_recipes: Mapping[str, Any] | None = None, 

176 # Compatibility parameters. Unused in v2. 

177 **kwargs: Any, 

178 ): 

179 if not isinstance(file_descriptor, FileDescriptor): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 raise TypeError("File descriptor must be a FileDescriptor") 

181 

182 self._file_descriptor = file_descriptor 

183 

184 if ref.isComponent(): 

185 # It is a component ref for disassembled composites. 

186 ref = ref.makeCompositeRef() 

187 self._dataset_ref = ref 

188 

189 # Check that the write parameters are allowed 

190 if write_parameters: 

191 if self.supported_write_parameters is None: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 raise ValueError( 

193 f"This formatter does not accept any write parameters. Got: {', '.join(write_parameters)}" 

194 ) 

195 else: 

196 given = set(write_parameters) 

197 unknown = given - self.supported_write_parameters 

198 if unknown: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true

199 s = "s" if len(unknown) != 1 else "" 

200 unknownStr = ", ".join(f"'{u}'" for u in unknown) 

201 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}") 

202 

203 self._write_parameters = write_parameters 

204 self._write_recipes = self.validate_write_recipes(write_recipes) 

205 

206 def __str__(self) -> str: 

207 return f"{self.name()}@{self.file_descriptor.location.uri}" 

208 

209 def __repr__(self) -> str: 

210 return f"{self.name()}({self.file_descriptor!r})" 

211 

212 @property 

213 def file_descriptor(self) -> FileDescriptor: 

214 """File descriptor associated with this formatter 

215 (`FileDescriptor`). 

216 """ 

217 return self._file_descriptor 

218 

219 @property 

220 def data_id(self) -> DataCoordinate: 

221 """Return Data ID associated with this formatter (`DataCoordinate`).""" 

222 return self._dataset_ref.dataId 

223 

224 @property 

225 def dataset_ref(self) -> DatasetRef: 

226 """Return Dataset Ref associated with this formatter (`DatasetRef`).""" 

227 return self._dataset_ref 

228 

229 @property 

230 def write_parameters(self) -> Mapping[str, Any]: 

231 """Parameters to use when writing out datasets.""" 

232 if self._write_parameters is not None: 

233 return self._write_parameters 

234 return {} 

235 

236 @property 

237 def write_recipes(self) -> Mapping[str, Any]: 

238 """Detailed write Recipes indexed by recipe name.""" 

239 if self._write_recipes is not None: 

240 return self._write_recipes 

241 return {} 

242 

243 def get_write_extension(self) -> str: 

244 """Extension to use when writing a file.""" 

245 default_extension = self.default_extension 

246 extension = default_extension if default_extension is not None else "" 

247 return extension 

248 

249 def can_accept(self, in_memory_dataset: Any) -> bool: 

250 """Indicate whether this formatter can accept the specified 

251 storage class directly. 

252 

253 Parameters 

254 ---------- 

255 in_memory_dataset : `object` 

256 The dataset that is to be written. 

257 

258 Returns 

259 ------- 

260 accepts : `bool` 

261 If `True` the formatter can write data of this type without 

262 requiring datastore to convert it. If `False` the datastore 

263 will attempt to convert before writing. 

264 

265 Notes 

266 ----- 

267 The base class always returns `False` even if the given type is an 

268 instance of the storage class type. This will result in a storage 

269 class conversion no-op but also allows mocks with mocked storage 

270 classes to work properly. 

271 """ 

272 return False 

273 

274 @classmethod 

275 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None: 

276 """Validate supplied recipes for this formatter. 

277 

278 The recipes are supplemented with default values where appropriate. 

279 

280 Parameters 

281 ---------- 

282 recipes : `dict` 

283 Recipes to validate. 

284 

285 Returns 

286 ------- 

287 validated : `dict` 

288 Validated recipes. 

289 

290 Raises 

291 ------ 

292 RuntimeError 

293 Raised if validation fails. The default implementation raises 

294 if any recipes are given. 

295 """ 

296 if recipes: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true

297 raise RuntimeError(f"This formatter does not understand these write recipes: {recipes}") 

298 return recipes 

299 

300 @classmethod 

301 def name(cls) -> str: 

302 """Return the fully qualified name of the formatter. 

303 

304 Returns 

305 ------- 

306 name : `str` 

307 Fully-qualified name of formatter class. 

308 """ 

309 return get_full_type_name(cls) 

310 

311 def _is_disassembled(self) -> bool: 

312 """Return `True` if this formatter is looking at a disassembled 

313 component. 

314 """ 

315 return self.file_descriptor.component is not None 

316 

317 def _check_resource_size(self, uri: ResourcePath, recorded_size: int, resource_size: int) -> None: 

318 """Compare the recorded size with the resource size. 

319 

320 The given URI will not be accessed. 

321 """ 

322 if recorded_size >= 0 and resource_size != recorded_size: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true

323 raise FileIntegrityError( 

324 "Integrity failure in Datastore. " 

325 f"Size of file {uri} ({resource_size}) " 

326 f"does not match size recorded in registry of {recorded_size}" 

327 ) 

328 

329 def _get_cache_ref(self) -> DatasetRef: 

330 """Get the `DatasetRef` to use for cache look ups. 

331 

332 Returns 

333 ------- 

334 ref : `lsst.daf.butler.DatasetRef` 

335 The dataset ref to use when looking in the cache. 

336 For single-file dataset this will be the dataset ref directly. 

337 If this is disassembled we need the component and the component 

338 will be in the `FileDescriptor`. 

339 """ 

340 if self.file_descriptor.component is None: 

341 cache_ref = self.dataset_ref 

342 else: 

343 cache_ref = self.dataset_ref.makeComponentRef(self.file_descriptor.component) 

344 return cache_ref 

345 

346 def _ensure_cache( 

347 self, cache_manager: AbstractDatastoreCacheManager | None = None 

348 ) -> AbstractDatastoreCacheManager: 

349 """Return the cache if given else return a null cache.""" 

350 if cache_manager is None: 350 ↛ 352line 350 didn't jump to line 352 because the condition on line 350 was never true

351 # Circular import avoidance. 

352 from .datastore.cache_manager import DatastoreDisabledCacheManager 

353 

354 cache_manager = DatastoreDisabledCacheManager(None, None) 

355 return cache_manager 

356 

357 def read( 

358 self, 

359 component: str | None = None, 

360 expected_size: int = -1, 

361 cache_manager: AbstractDatastoreCacheManager | None = None, 

362 ) -> Any: 

363 """Read a Dataset. 

364 

365 Parameters 

366 ---------- 

367 component : `str`, optional 

368 Component to read from the file. Only used if the `StorageClass` 

369 for reading differed from the `StorageClass` used to write the 

370 file. 

371 expected_size : `int`, optional 

372 If known, the expected size of the resource to read. This can be 

373 used for verification or to decide whether to do a direct read or a 

374 file download. ``-1`` indicates the file size is not known. 

375 cache_manager : `AbstractDatastoreCacheManager` 

376 A cache manager to use to allow a formatter to cache a remote file 

377 locally or read a cached file that is already local. 

378 

379 Returns 

380 ------- 

381 in_memory_dataset : `object` 

382 The requested Dataset. 

383 

384 Notes 

385 ----- 

386 This method should not be subclassed. Instead formatter subclasses 

387 should re-implement the specific ``read_from_*`` methods as 

388 appropriate. Each of these methods has a corresponding class property 

389 that must be `True` for the method to be called. 

390 

391 The priority for reading is: 

392 

393 * `read_from_uri` 

394 * `read_from_stream` 

395 * `read_from_local_file` 

396 * `read_from_uri` (but with a local file) 

397 

398 Any of these methods can return `NotImplemented` if there is a desire 

399 to skip to the next one in the list. If a dataset is being requested 

400 with no component, no parameters, and it should also be added to the 

401 local cache, the first two calls will be skipped (unless 

402 `read_from_stream` is the only implemented read method) such that a 

403 local file will be used. 

404 

405 A Formatter can also read a file from within a Zip file if the 

406 URI associated with the `FileDescriptor` corresponds to a file with 

407 a `.zip` extension and a URI fragment of the form 

408 ``zip-path={path_in_zip}``. When reading a file from within a Zip 

409 file the priority for reading is: 

410 

411 * `read_from_stream` 

412 * `read_from_local_file` 

413 * `read_from_uri` 

414 

415 There are multiple cases that must be handled for reading: 

416 

417 For a single file: 

418 

419 * No component requested, read the whole file. 

420 * Component requested, optionally read the component efficiently, 

421 else read the whole file and extract the component. 

422 * Derived component requested, read whole file or read relevant 

423 component and derive. 

424 

425 Disassembled Composite: 

426 

427 * The file to read here is the component itself. Formatter only knows 

428 about this one component file. Should be no component specified 

429 in the ``read`` call but the `FileDescriptor` will know which 

430 component this is. 

431 * A derived component. The file to read is a component but not the 

432 specified component. The caching needs the component from which 

433 it's derived. 

434 

435 Raises 

436 ------ 

437 lsst.daf.butler.FormatterNotImplementedError 

438 Raised if no implementations were found that could read this 

439 resource. 

440 """ 

441 # If the file to read is a ZIP file with a fragment requesting 

442 # a file within the ZIP file, it is no longer possible to use the 

443 # direct read from URI option and the contents of the Zip file must 

444 # be extracted. 

445 uri = self.file_descriptor.location.uri 

446 if uri.fragment and uri.unquoted_fragment.startswith("zip-path="): 

447 _, _, path_in_zip = uri.unquoted_fragment.partition("=") 

448 

449 # Open the Zip file using ResourcePath. 

450 with uri.open("rb") as fd: 

451 with zipfile.ZipFile(fd) as zf: # type: ignore 

452 if self.can_read_from_stream: 

453 with contextlib.closing(zf.open(path_in_zip)) as zip_fd: 

454 result = self.read_from_stream(zip_fd, component, expected_size=expected_size) 

455 

456 if result is not NotImplemented: 456 ↛ 461line 456 didn't jump to line 461 because the condition on line 456 was always true

457 return result 

458 

459 # For now for both URI and local file options we retrieve 

460 # the bytes to a temporary local and use that. 

461 _, suffix = os.path.splitext(path_in_zip) 

462 with ResourcePath.temporary_uri(suffix=suffix) as tmp_uri: 

463 tmp_uri.write(zf.read(path_in_zip)) 

464 

465 if self.can_read_from_local_file: 

466 result = self.read_from_local_file( 

467 tmp_uri.ospath, component, expected_size=expected_size 

468 ) 

469 if result is not NotImplemented: 469 ↛ 471line 469 didn't jump to line 471 because the condition on line 469 was always true

470 return result 

471 if self.can_read_from_uri: 471 ↛ 476line 471 didn't jump to line 476

472 result = self.read_from_uri(tmp_uri, component, expected_size=expected_size) 

473 if result is not NotImplemented: 473 ↛ 476line 473 didn't jump to line 476

474 return result 

475 

476 raise FormatterNotImplementedError( 

477 f"Formatter {self.name()} could not read the file at {uri} using any method." 

478 ) 

479 

480 # If the there are no parameters, no component request, and 

481 # the file should be cached, it is better to defer the read_from_uri 

482 # to use the local file and populate the cache. 

483 prefer_local = False 

484 if ( 

485 component is None 

486 and not self.file_descriptor.parameters 

487 and self._ensure_cache(cache_manager).should_be_cached(self._get_cache_ref()) 

488 ): 

489 prefer_local = True 

490 

491 # First see if the formatter can support direct remote read from 

492 # a URI. This can be called later from the local path. If it returns 

493 # NotImplemented the formatter decided on its own that local 

494 # reads are preferred. 

495 if not prefer_local and self.can_read_from_uri: 

496 result = self.read_directly_from_possibly_cached_uri( 

497 component, expected_size, cache_manager=cache_manager 

498 ) 

499 if result is not NotImplemented: 499 ↛ 506line 499 didn't jump to line 506 because the condition on line 499 was always true

500 return result 

501 

502 # Some formatters might want to be able to read directly from 

503 # an open file stream. This is preferred over forcing a download 

504 # to local file system unless a file read option is available and we 

505 # want to store it in the cache because the whole file is being read. 

506 if self.can_read_from_stream and not ( 

507 prefer_local and (self.can_read_from_uri or self.can_read_from_local_file) 

508 ): 

509 result = self.read_from_possibly_cached_stream( 

510 component, expected_size, cache_manager=cache_manager 

511 ) 

512 if result is not NotImplemented: 512 ↛ 516line 512 didn't jump to line 516 because the condition on line 512 was always true

513 return result 

514 

515 # Finally, try to read the local file. 

516 if self.can_read_from_local_file or self.can_read_from_uri: 

517 result = self.read_from_possibly_cached_local_file( 

518 component, expected_size, cache_manager=cache_manager 

519 ) 

520 if result is not NotImplemented: 520 ↛ 523line 520 didn't jump to line 523 because the condition on line 520 was always true

521 return result 

522 

523 raise FormatterNotImplementedError( 

524 f"Formatter {self.name()} could not read the file at {uri} using any method." 

525 ) 

526 

527 def _read_from_possibly_cached_location_no_cache_write( 

528 self, 

529 callback: Callable[[ResourcePath, str | None, int], Any], 

530 component: str | None = None, 

531 expected_size: int = -1, 

532 *, 

533 cache_manager: AbstractDatastoreCacheManager | None = None, 

534 ) -> Any: 

535 """Read from the cache and call payload without writing to cache.""" 

536 cache_manager = self._ensure_cache(cache_manager) 

537 

538 uri = self.file_descriptor.location.uri 

539 cache_ref = self._get_cache_ref() 

540 

541 # The component for log messages is either the component requested 

542 # explicitly or the component from the file descriptor. 

543 log_component = component if component is not None else self.file_descriptor.component 

544 

545 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file: 

546 if cached_file is not None: 

547 desired_uri = cached_file 

548 msg = f" (cached version of {uri})" 

549 else: 

550 desired_uri = uri 

551 msg = "" 

552 

553 if desired_uri.isLocal: 

554 # Do not spend the time doing a slow size() call to a remote 

555 # resource. 

556 self._check_resource_size(desired_uri, expected_size, desired_uri.size()) 

557 

558 with time_this( 

559 log, 

560 msg="Reading%s from file handle %s%s with formatter %s", 

561 args=( 

562 f" component {log_component}" if log_component else "", 

563 desired_uri, 

564 msg, 

565 self.name(), 

566 ), 

567 ): 

568 return callback(desired_uri, component, expected_size) 

569 

570 def read_from_possibly_cached_stream( 

571 self, 

572 component: str | None = None, 

573 expected_size: int = -1, 

574 *, 

575 cache_manager: AbstractDatastoreCacheManager | None = None, 

576 ) -> Any: 

577 """Read from a stream, checking for possible presence in local cache. 

578 

579 Parameters 

580 ---------- 

581 component : `str`, optional 

582 Component to read from the file. Only used if the `StorageClass` 

583 for reading differed from the `StorageClass` used to write the 

584 file. 

585 expected_size : `int`, optional 

586 If known, the expected size of the resource to read. This can be 

587 used for verification or to decide whether to do a direct read or a 

588 file download. ``-1`` indicates the file size is not known. 

589 cache_manager : `AbstractDatastoreCacheManager` 

590 A cache manager to use to allow a formatter to check if there is 

591 a copy of the file in the local cache. 

592 

593 Returns 

594 ------- 

595 in_memory_dataset : `object` or `NotImplemented` 

596 The requested Dataset or an indication that the read mode was 

597 not implemented. 

598 

599 Notes 

600 ----- 

601 Calls `read_from_stream` but will first check the datastore cache 

602 in case the file is present locally. This method will not download 

603 a file to the local cache. 

604 """ 

605 

606 def _open_stream(uri: ResourcePath, comp: str | None, size: int = -1) -> Any: 

607 with uri.open("rb") as fd: 

608 return self.read_from_stream(fd, comp, expected_size=size) 

609 

610 return self._read_from_possibly_cached_location_no_cache_write( 

611 _open_stream, component, expected_size=expected_size, cache_manager=cache_manager 

612 ) 

613 

614 def read_directly_from_possibly_cached_uri( 

615 self, 

616 component: str | None = None, 

617 expected_size: int = -1, 

618 *, 

619 cache_manager: AbstractDatastoreCacheManager | None = None, 

620 ) -> Any: 

621 """Read from arbitrary URI, checking for possible presence in local 

622 cache. 

623 

624 Parameters 

625 ---------- 

626 component : `str`, optional 

627 Component to read from the file. Only used if the `StorageClass` 

628 for reading differed from the `StorageClass` used to write the 

629 file. 

630 expected_size : `int`, optional 

631 If known, the expected size of the resource to read. This can be 

632 ``-1`` indicates the file size is not known. 

633 cache_manager : `AbstractDatastoreCacheManager` 

634 A cache manager to use to allow a formatter to check if there is 

635 a copy of the file in the local cache. 

636 

637 Returns 

638 ------- 

639 in_memory_dataset : `object` or `NotImplemented` 

640 The requested Dataset or an indication that the read mode was 

641 not implemented. 

642 

643 Notes 

644 ----- 

645 This method will first check the datastore cache 

646 in case the file is present locally. This method will not cache a 

647 remote dataset and will only do a size check for local files to avoid 

648 unnecessary round trips to a remote server. 

649 

650 The URI will be read by calling `read_from_uri`. 

651 """ 

652 

653 def _open_uri(uri: ResourcePath, comp: str | None, size: int = -1) -> Any: 

654 return self.read_from_uri(uri, comp, expected_size=size) 

655 

656 return self._read_from_possibly_cached_location_no_cache_write( 

657 _open_uri, component, expected_size=expected_size, cache_manager=cache_manager 

658 ) 

659 

660 def read_from_possibly_cached_local_file( 

661 self, 

662 component: str | None = None, 

663 expected_size: int = -1, 

664 *, 

665 cache_manager: AbstractDatastoreCacheManager | None = None, 

666 ) -> Any: 

667 """Read a dataset ensuring that a local file is used, checking the 

668 cache for it. 

669 

670 Parameters 

671 ---------- 

672 component : `str`, optional 

673 Component to read from the file. Only used if the `StorageClass` 

674 for reading differed from the `StorageClass` used to write the 

675 file. 

676 expected_size : `int`, optional 

677 If known, the expected size of the resource to read. This can be 

678 used for verification or to decide whether to do a direct read or a 

679 file download. ``-1`` indicates the file size is not known. 

680 cache_manager : `AbstractDatastoreCacheManager` 

681 A cache manager to use to allow a formatter to cache a remote file 

682 locally or read a cached file that is already local. 

683 

684 Returns 

685 ------- 

686 in_memory_dataset : `object` or `NotImplemented` 

687 The requested Dataset or an indication that the read mode was 

688 not implemented. 

689 

690 Notes 

691 ----- 

692 The file will be downloaded and cached if it is a remote resource. 

693 The file contents will be read using `read_from_local_file` or 

694 `read_from_uri`, with preference given to the former. 

695 """ 

696 cache_manager = self._ensure_cache(cache_manager) 

697 uri = self.file_descriptor.location.uri 

698 

699 # Need to have something we can look up in the cache. 

700 cache_ref = self._get_cache_ref() 

701 

702 # The component for log messages is either the component requested 

703 # explicitly or the component from the file descriptor. 

704 log_component = component if component is not None else self.file_descriptor.component 

705 

706 result = NotImplemented 

707 

708 # Ensure we have a local file. 

709 with cache_manager.find_in_cache(cache_ref, uri.getExtension()) as cached_file: 

710 if cached_file is not None: 

711 msg = f"(via cache read of remote file {uri})" 

712 uri = cached_file 

713 else: 

714 msg = "" 

715 

716 # If the file is remote and destined for the cache, download it 

717 # directly onto the cache file system. This makes the subsequent 

718 # move into the cache an atomic rename rather than a 

719 # cross-file-system copy, which is what happens when the default 

720 # temporary directory and the cache directory are on different 

721 # file systems. 

722 should_cache = not uri.isLocal and cache_manager.should_be_cached(cache_ref) 

723 download_dir = cache_manager.download_directory if should_cache else None 

724 

725 with uri.as_local(tmpdir=download_dir) as local_uri: 

726 self._check_resource_size(self.file_descriptor.location.uri, expected_size, local_uri.size()) 

727 can_be_cached = False 

728 if uri != local_uri: 

729 # URI was remote and file was downloaded 

730 cache_msg = "" 

731 

732 if should_cache: 732 ↛ 744line 732 didn't jump to line 744 because the condition on line 732 was always true

733 # In this scenario we want to ask if the downloaded 

734 # file should be cached but we should not cache 

735 # it until after we've used it (to ensure it can't 

736 # be expired whilst we are using it). 

737 can_be_cached = True 

738 

739 # Say that it is "likely" to be cached because 

740 # if the formatter read fails we will not be 

741 # caching this file. 

742 cache_msg = " and likely cached" 

743 

744 msg = f"(via download to local file{cache_msg})" 

745 

746 with time_this( 

747 log, 

748 msg="Reading%s from location %s %s with formatter %s", 

749 args=( 

750 f" component {log_component}" if log_component else "", 

751 uri, 

752 msg, 

753 self.name(), 

754 ), 

755 ): 

756 if self.can_read_from_local_file: 

757 result = self.read_from_local_file( 

758 local_uri.ospath, component=component, expected_size=expected_size 

759 ) 

760 if result is NotImplemented and self.can_read_from_uri: 

761 # If the direct URI reader was skipped earlier and 

762 # there is no explicit local file implementation, pass 

763 # in the guaranteed local URI to the generic reader. 

764 result = self.read_from_uri( 

765 local_uri, component=component, expected_size=expected_size 

766 ) 

767 

768 # File was read successfully so can move to cache. 

769 # Also move to cache even if NotImplemented was returned. 

770 if can_be_cached: 

771 cache_manager.move_to_cache(local_uri, cache_ref) 

772 

773 return result 

774 

775 def read_from_uri(self, uri: ResourcePath, component: str | None = None, expected_size: int = -1) -> Any: 

776 """Read a dataset from a URI that can be local or remote. 

777 

778 Parameters 

779 ---------- 

780 uri : `lsst.resources.ResourcePath` 

781 URI to use to read the dataset. This URI can be local or remote 

782 and can refer to the actual resource or to a locally cached file. 

783 component : `str` or `None`, optional 

784 The component to be read from the dataset. 

785 expected_size : `int`, optional 

786 If known, the expected size of the resource to read. This can be 

787 ``-1`` indicates the file size is not known. 

788 

789 Returns 

790 ------- 

791 in_memory_dataset : `object` or `NotImplemented` 

792 The Python object read from the resource or `NotImplemented`. 

793 

794 Raises 

795 ------ 

796 lsst.daf.butler.FormatterNotImplementedError 

797 Raised if there is no support for direct reads from a, possibly, 

798 remote URI. 

799 

800 Notes 

801 ----- 

802 This method is only called if the class property 

803 ``can_read_from_uri`` is set to `True`. 

804 

805 It is possible that a cached local file will be given to this method 

806 even if it was originally a remote URI. This can happen if the original 

807 write resulted in the file being added to the local cache. 

808 

809 If the full file is being read this file will not be added to the 

810 local cache. Consider returning `NotImplemented` in 

811 this situation, for example if there are no parameters or component 

812 specified, and allowing the system to fall back to calling 

813 `read_from_local_file` (which will populate the cache if configured 

814 to do so). 

815 """ 

816 return NotImplemented 

817 

818 def read_from_stream( 

819 self, stream: BinaryIO | ResourceHandleProtocol, component: str | None = None, expected_size: int = -1 

820 ) -> Any: 

821 """Read from an open file descriptor. 

822 

823 Parameters 

824 ---------- 

825 stream : `lsst.resources.ResourceHandleProtocol` or \ 

826 `typing.BinaryIO` 

827 File stream to use to read the dataset. 

828 component : `str` or `None`, optional 

829 The component to be read from the dataset. 

830 expected_size : `int`, optional 

831 If known, the expected size of the resource to read. This can be 

832 ``-1`` indicates the file size is not known. 

833 

834 Returns 

835 ------- 

836 in_memory_dataset : `object` or `NotImplemented` 

837 The Python object read from the stream or `NotImplemented`. 

838 

839 Notes 

840 ----- 

841 Only called if the class property ``can_read_from_stream`` is `True`. 

842 """ 

843 return NotImplemented 

844 

845 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any: 

846 """Read a dataset from a URI guaranteed to refer to the local file 

847 system. 

848 

849 Parameters 

850 ---------- 

851 path : `str` 

852 Path to a local file that should be read. 

853 component : `str` or `None`, optional 

854 The component to be read from the dataset. 

855 expected_size : `int`, optional 

856 If known, the expected size of the resource to read. This can be 

857 ``-1`` indicates the file size is not known. 

858 

859 Returns 

860 ------- 

861 in_memory_dataset : `object` or `NotImplemented` 

862 The Python object read from the resource or `NotImplemented`. 

863 

864 Raises 

865 ------ 

866 lsst.daf.butler.FormatterNotImplementedError 

867 Raised if there is no implementation written to read data 

868 from a local file. 

869 

870 Notes 

871 ----- 

872 This method will only be called if the class property 

873 ``can_read_from_local_file`` is `True` and other options were not 

874 used. 

875 """ 

876 return NotImplemented 

877 

878 def add_provenance( 

879 self, in_memory_dataset: Any, /, *, provenance: DatasetProvenance | None = None 

880 ) -> Any: 

881 """Add provenance to the dataset. 

882 

883 Parameters 

884 ---------- 

885 in_memory_dataset : `object` 

886 The dataset to serialize. 

887 provenance : `~lsst.daf.butler.DatasetProvenance` or `None`, optional 

888 Provenance to attach to dataset. 

889 

890 Returns 

891 ------- 

892 dataset_to_write : `object` 

893 The dataset to use for serialization. Can be the same object as 

894 given. 

895 

896 Notes 

897 ----- 

898 The base class implementation returns the given object unchanged. 

899 """ 

900 return in_memory_dataset 

901 

902 @final 

903 def write( 

904 self, 

905 in_memory_dataset: Any, 

906 /, 

907 *, 

908 cache_manager: AbstractDatastoreCacheManager | None = None, 

909 provenance: DatasetProvenance | None = None, 

910 ) -> None: 

911 """Write a Dataset. 

912 

913 Parameters 

914 ---------- 

915 in_memory_dataset : `object` 

916 The Dataset to serialize. 

917 cache_manager : `AbstractDatastoreCacheManager` 

918 A cache manager to use to allow a formatter to cache the written 

919 file. 

920 provenance : `DatasetProvenance` | `None`, optional 

921 Provenance to attach to the file being written. 

922 

923 Returns 

924 ------- 

925 None 

926 

927 Raises 

928 ------ 

929 lsst.daf.butler.FormatterNotImplementedError 

930 Raised if the formatter subclass has not implemented 

931 `write_local_file` and `to_bytes` was not called. 

932 Exception 

933 Raised if there is an error serializing the dataset to disk. 

934 

935 Notes 

936 ----- 

937 The intent is for subclasses to implement either `to_bytes` or 

938 `write_local_file` or both and not to subclass this method. 

939 """ 

940 # Ensure we are using the correct file extension. 

941 uri = self.file_descriptor.location.uri.updatedExtension(self.get_write_extension()) 

942 

943 # Attach any provenance to the dataset. This could involve returning 

944 # a different object. 

945 in_memory_dataset = self.add_provenance(in_memory_dataset, provenance=provenance) 

946 

947 written = self.write_direct(in_memory_dataset, uri, cache_manager) 

948 if not written: 948 ↛ exitline 948 didn't return from function 'write' because the condition on line 948 was always true

949 self.write_locally_then_move(in_memory_dataset, uri, cache_manager) 

950 

951 def write_direct( 

952 self, 

953 in_memory_dataset: Any, 

954 uri: ResourcePath, 

955 cache_manager: AbstractDatastoreCacheManager | None = None, 

956 ) -> bool: 

957 """Serialize and write directly to final location. 

958 

959 Parameters 

960 ---------- 

961 in_memory_dataset : `object` 

962 The Dataset to serialize. 

963 uri : `lsst.resources.ResourcePath` 

964 URI to use when writing the serialized dataset. 

965 cache_manager : `AbstractDatastoreCacheManager` 

966 A cache manager to use to allow a formatter to cache the written 

967 file. 

968 

969 Returns 

970 ------- 

971 written : `bool` 

972 Flag to indicate whether the direct write did happen. 

973 

974 Raises 

975 ------ 

976 Exception 

977 Raised if there was a failure from serializing to bytes that 

978 was not `~lsst.daf.butler.FormatterNotImplementedError`. 

979 

980 Notes 

981 ----- 

982 This method will call `to_bytes` to serialize the in-memory dataset 

983 and then will call the `~lsst.resources.ResourcePath.write` method 

984 directly. 

985 

986 If the dataset should be cached or is local the file will not be 

987 written and the method will return `False`. This is because local URIs 

988 should be written to a temporary file name and then renamed to allow 

989 atomic writes. That path is handled by `write_locally_then_move` 

990 through `write_local_file`) and is preferred over this method being 

991 subclassed and the atomic write re-implemented. 

992 """ 

993 cache_manager = self._ensure_cache(cache_manager) 

994 

995 # For remote URIs some datasets can be serialized directly 

996 # to bytes and sent to the remote datastore without writing a 

997 # file. If the dataset is intended to be saved to the cache 

998 # a file is always written and direct write to the remote 

999 # datastore is bypassed. 

1000 data_written = False 

1001 if not uri.isLocal and not cache_manager.should_be_cached(self._get_cache_ref()): 1001 ↛ 1003line 1001 didn't jump to line 1003 because the condition on line 1001 was never true

1002 # Remote URI that is not cached so can write directly. 

1003 try: 

1004 serialized_dataset = self.to_bytes(in_memory_dataset) 

1005 except FormatterNotImplementedError: 

1006 # Fallback to the file writing option. 

1007 pass 

1008 except Exception as e: 

1009 e.add_note( 

1010 f"Failed to serialize dataset {self.dataset_ref} of " 

1011 f"type {get_full_type_name(in_memory_dataset)} to bytes." 

1012 ) 

1013 raise 

1014 else: 

1015 log.debug("Writing bytes directly to %s", uri) 

1016 uri.write(serialized_dataset, overwrite=True) 

1017 log.debug("Successfully wrote bytes directly to %s", uri) 

1018 data_written = True 

1019 return data_written 

1020 

1021 def write_locally_then_move( 

1022 self, 

1023 in_memory_dataset: Any, 

1024 uri: ResourcePath, 

1025 cache_manager: AbstractDatastoreCacheManager | None = None, 

1026 ) -> None: 

1027 """Write file to file system and then move to final location. 

1028 

1029 Parameters 

1030 ---------- 

1031 in_memory_dataset : `object` 

1032 The Dataset to serialize. 

1033 uri : `lsst.resources.ResourcePath` 

1034 URI to use when writing the serialized dataset. 

1035 cache_manager : `AbstractDatastoreCacheManager` 

1036 A cache manager to use to allow a formatter to cache the written 

1037 file. 

1038 

1039 Raises 

1040 ------ 

1041 lsst.daf.butler.FormatterNotImplementedError 

1042 Raised if the formatter subclass has not implemented 

1043 `write_local_file`. 

1044 Exception 

1045 Raised if there is an error serializing the dataset to disk. 

1046 """ 

1047 cache_manager = self._ensure_cache(cache_manager) 

1048 

1049 with TemporaryForIngest.make_path(uri) as temporary_uri: 

1050 # Need to configure the formatter to write to a different 

1051 # location and that needs us to overwrite internals 

1052 log.debug("Writing dataset to temporary location at %s", temporary_uri) 

1053 

1054 # Assumes that if write_local_file is not subclassed that 

1055 # to_bytes will be called using the base class definition. 

1056 try: 

1057 self.write_local_file(in_memory_dataset, temporary_uri) 

1058 except Exception as e: 

1059 e.add_note( 

1060 f"Failed to serialize dataset {self.dataset_ref} of type" 

1061 f" {get_full_type_name(in_memory_dataset)} to " 

1062 f"temporary location {temporary_uri}." 

1063 ) 

1064 raise 

1065 

1066 # Use move for a local file since that becomes an efficient 

1067 # os.rename. For remote resources we use copy to allow the 

1068 # file to be cached afterwards. 

1069 transfer = "move" if uri.isLocal else "copy" 

1070 

1071 uri.transfer_from(temporary_uri, transfer=transfer, overwrite=True) 

1072 

1073 if transfer == "copy": 

1074 # Cache if required 

1075 cache_manager.move_to_cache(temporary_uri, self._get_cache_ref()) 

1076 

1077 log.debug("Successfully wrote dataset to %s via a temporary file.", uri) 

1078 

1079 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None: 

1080 """Serialize the in-memory dataset to a local file. 

1081 

1082 Parameters 

1083 ---------- 

1084 in_memory_dataset : `object` 

1085 The Python object to serialize. 

1086 uri : `~lsst.resources.ResourcePath` 

1087 The URI to use when writing the file. 

1088 

1089 Notes 

1090 ----- 

1091 By default this method will attempt to call `to_bytes` and then 

1092 write these bytes to the file. 

1093 

1094 Raises 

1095 ------ 

1096 lsst.daf.butler.FormatterNotImplementedError 

1097 Raised if the formatter subclass has not implemented this method 

1098 or has failed to implement the `to_bytes` method. 

1099 """ 

1100 log.debug("Writing bytes directly to %s", uri) 

1101 uri.write(self.to_bytes(in_memory_dataset)) 

1102 log.debug("Successfully wrote bytes directly to %s", uri) 

1103 

1104 def to_bytes(self, in_memory_dataset: Any) -> bytes: 

1105 """Serialize the in-memory dataset to bytes. 

1106 

1107 Parameters 

1108 ---------- 

1109 in_memory_dataset : `object` 

1110 The Python object to serialize. 

1111 

1112 Returns 

1113 ------- 

1114 serialized_dataset : `bytes` 

1115 Bytes representing the serialized dataset. 

1116 

1117 Raises 

1118 ------ 

1119 lsst.daf.butler.FormatterNotImplementedError 

1120 Raised if the formatter has not implemented the method. This will 

1121 not cause a problem if `write_local_file` has been implemented. 

1122 """ 

1123 raise FormatterNotImplementedError( 

1124 f"This formatter can not convert {get_full_type_name(in_memory_dataset)} directly to bytes." 

1125 ) 

1126 

1127 def make_updated_location(self, location: Location) -> Location: 

1128 """Return a new `Location` updated with this formatter's extension. 

1129 

1130 Parameters 

1131 ---------- 

1132 location : `Location` 

1133 The location to update. 

1134 

1135 Returns 

1136 ------- 

1137 updated : `Location` 

1138 A new `Location` with a new file extension applied. 

1139 """ 

1140 location = location.clone() 

1141 # If the extension is "" the extension will be removed. 

1142 location.updateExtension(self.get_write_extension()) 

1143 return location 

1144 

1145 @classmethod 

1146 def validate_extension(cls, location: Location) -> None: 

1147 """Check the extension of the provided location for compatibility. 

1148 

1149 Parameters 

1150 ---------- 

1151 location : `Location` 

1152 Location from which to extract a file extension. 

1153 

1154 Returns 

1155 ------- 

1156 None 

1157 

1158 Raises 

1159 ------ 

1160 ValueError 

1161 Raised if the formatter does not understand this extension. 

1162 """ 

1163 supported = set(cls.supported_extensions) 

1164 default = cls.default_extension # type: ignore 

1165 

1166 # If extension is implemented as an instance property it won't return 

1167 # a string when called as a class property. Assume that 

1168 # the supported extensions class property is complete. 

1169 if default is not None and isinstance(default, str): 1169 ↛ 1173line 1169 didn't jump to line 1173 because the condition on line 1169 was always true

1170 supported.add(default) 

1171 

1172 # Get the file name from the uri 

1173 file = location.uri.basename() 

1174 

1175 # Check that this file name ends with one of the supported extensions. 

1176 # This is less prone to confusion than asking the location for 

1177 # its extension and then doing a set comparison 

1178 for ext in supported: 1178 ↛ 1182line 1178 didn't jump to line 1182 because the loop on line 1178 didn't complete

1179 if file.endswith(ext): 

1180 return 

1181 

1182 raise ValueError( 

1183 f"Extension '{location.getExtension()}' on '{location}' " 

1184 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})" 

1185 ) 

1186 

1187 def predict_path(self) -> str: 

1188 """Return the path that would be returned by write. 

1189 

1190 Does not write any data file. 

1191 

1192 Uses the `FileDescriptor` associated with the instance. 

1193 

1194 Returns 

1195 ------- 

1196 path : `str` 

1197 Path within datastore that would be associated with the location 

1198 stored in this `Formatter`. 

1199 """ 

1200 updated = self.make_updated_location(self.file_descriptor.location) 

1201 return updated.pathInStore.path 

1202 

1203 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]: 

1204 """Segregate the supplied parameters. 

1205 

1206 This splits the parameters into those understood by the 

1207 formatter and those not understood by the formatter. 

1208 

1209 Any unsupported parameters are assumed to be usable by associated 

1210 assemblers. 

1211 

1212 Parameters 

1213 ---------- 

1214 parameters : `dict`, optional 

1215 Parameters with values that have been supplied by the caller 

1216 and which might be relevant for the formatter. If `None` 

1217 parameters will be read from the registered `FileDescriptor`. 

1218 

1219 Returns 

1220 ------- 

1221 supported : `dict` 

1222 Those parameters supported by this formatter. 

1223 unsupported : `dict` 

1224 Those parameters not supported by this formatter. 

1225 """ 

1226 if parameters is None: 1226 ↛ 1229line 1226 didn't jump to line 1229 because the condition on line 1226 was always true

1227 parameters = self.file_descriptor.parameters 

1228 

1229 if parameters is None: 

1230 return {}, {} 

1231 

1232 if self.unsupported_parameters is None: 

1233 # Support none of the parameters 

1234 return {}, parameters.copy() 

1235 

1236 # Start by assuming all are supported 

1237 supported = parameters.copy() 

1238 unsupported = {} 

1239 

1240 # And remove any we know are not supported 

1241 for p in set(supported): 

1242 if p in self.unsupported_parameters: 1242 ↛ 1243line 1242 didn't jump to line 1243 because the condition on line 1242 was never true

1243 unsupported[p] = supported.pop(p) 

1244 

1245 return supported, unsupported 

1246 

1247 

1248class Formatter(metaclass=ABCMeta): 

1249 """Interface for reading and writing Datasets. 

1250 

1251 The formatters are associated with a particular `StorageClass`. 

1252 

1253 Parameters 

1254 ---------- 

1255 fileDescriptor : `FileDescriptor`, optional 

1256 Identifies the file to read or write, and the associated storage 

1257 classes and parameter information. 

1258 dataId : `DataCoordinate` 

1259 Data ID associated with this formatter. 

1260 writeParameters : `dict`, optional 

1261 Parameters to control how the dataset is serialized. 

1262 writeRecipes : `dict`, optional 

1263 Detailed write recipes indexed by recipe name. 

1264 **kwargs 

1265 Additional parameters that can allow parameters 

1266 from `FormatterV2` to be provided. 

1267 

1268 Notes 

1269 ----- 

1270 All Formatter subclasses should share the base class's constructor 

1271 signature. 

1272 """ 

1273 

1274 # Now assuming that Formatter v1 can only refer to files so can add 

1275 # this property for compatibility with v2. 

1276 extension: str | None = None 

1277 """Default file extension to use for writing files. None means that no 

1278 modifications will be made to the supplied file extension. (`str`)""" 

1279 

1280 unsupportedParameters: ClassVar[Set[str] | None] = frozenset() 

1281 """Set of read parameters not understood by this `Formatter`. An empty set 

1282 means all parameters are supported. `None` indicates that no parameters 

1283 are supported. These parameters should match those defined in the storage 

1284 class definition. (`frozenset`). 

1285 """ 

1286 

1287 supportedWriteParameters: ClassVar[Set[str] | None] = None 

1288 """Parameters understood by this formatter that can be used to control 

1289 how a dataset is serialized. `None` indicates that no parameters are 

1290 supported.""" 

1291 

1292 supportedExtensions: ClassVar[Set[str]] = frozenset() 

1293 """Set of all extensions supported by this formatter. 

1294 

1295 Only expected to be populated by Formatters that write files. Any extension 

1296 assigned to the ``extension`` property will be automatically included in 

1297 the list of supported extensions.""" 

1298 

1299 def __init__( 

1300 self, 

1301 fileDescriptor: FileDescriptor, 

1302 *, 

1303 dataId: DataCoordinate | None = None, 

1304 writeParameters: Mapping[str, Any] | None = None, 

1305 writeRecipes: Mapping[str, Any] | None = None, 

1306 # Allow FormatterV2 parameters to be dropped. 

1307 **kwargs: Any, 

1308 ): 

1309 if not isinstance(fileDescriptor, FileDescriptor): 1309 ↛ 1310line 1309 didn't jump to line 1310 because the condition on line 1309 was never true

1310 raise TypeError("File descriptor must be a FileDescriptor") 

1311 self._fileDescriptor = fileDescriptor 

1312 

1313 if dataId is None: 

1314 raise RuntimeError("dataId is now required for formatter initialization") 

1315 if not isinstance(dataId, DataCoordinate): 1315 ↛ 1316line 1315 didn't jump to line 1316 because the condition on line 1315 was never true

1316 raise TypeError(f"DataId is required to be a DataCoordinate but got {type(dataId)}.") 

1317 self._dataId = dataId 

1318 

1319 # V2 compatibility. 

1320 if writeParameters is None: 1320 ↛ 1322line 1320 didn't jump to line 1322 because the condition on line 1320 was always true

1321 writeParameters = kwargs.get("write_parameters") 

1322 if writeRecipes is None: 1322 ↛ 1326line 1322 didn't jump to line 1326 because the condition on line 1322 was always true

1323 writeRecipes = kwargs.get("write_recipes") 

1324 

1325 # Check that the write parameters are allowed 

1326 if writeParameters: 

1327 if self.supportedWriteParameters is None: 

1328 raise ValueError( 

1329 f"This formatter does not accept any write parameters. Got: {', '.join(writeParameters)}" 

1330 ) 

1331 else: 

1332 given = set(writeParameters) 

1333 unknown = given - self.supportedWriteParameters 

1334 if unknown: 

1335 s = "s" if len(unknown) != 1 else "" 

1336 unknownStr = ", ".join(f"'{u}'" for u in unknown) 

1337 raise ValueError(f"This formatter does not accept parameter{s} {unknownStr}") 

1338 

1339 self._writeParameters = writeParameters 

1340 self._writeRecipes = self.validate_write_recipes(writeRecipes) 

1341 

1342 def __str__(self) -> str: 

1343 return f"{self.name()}@{self.fileDescriptor.location.path}" 

1344 

1345 def __repr__(self) -> str: 

1346 return f"{self.name()}({self.fileDescriptor!r})" 

1347 

1348 @property 

1349 def fileDescriptor(self) -> FileDescriptor: 

1350 """File descriptor associated with this formatter 

1351 (`FileDescriptor`). 

1352 """ 

1353 return self._fileDescriptor 

1354 

1355 @property 

1356 def dataId(self) -> DataCoordinate: 

1357 """Return Data ID associated with this formatter (`DataCoordinate`).""" 

1358 return self._dataId 

1359 

1360 @property 

1361 def writeParameters(self) -> Mapping[str, Any]: 

1362 """Parameters to use when writing out datasets.""" 

1363 if self._writeParameters is not None: 

1364 return self._writeParameters 

1365 return {} 

1366 

1367 @property 

1368 def writeRecipes(self) -> Mapping[str, Any]: 

1369 """Detailed write Recipes indexed by recipe name.""" 

1370 if self._writeRecipes is not None: 

1371 return self._writeRecipes 

1372 return {} 

1373 

1374 def can_accept(self, in_memory_dataset: Any) -> bool: 

1375 """Indicate whether this formatter can accept the specified 

1376 storage class directly. 

1377 

1378 Parameters 

1379 ---------- 

1380 in_memory_dataset : `object` 

1381 The dataset that is to be written. 

1382 

1383 Returns 

1384 ------- 

1385 accepts : `bool` 

1386 If `True` the formatter can write data of this type without 

1387 requiring datastore to convert it. If `False` the datastore 

1388 will attempt to convert before writing. 

1389 

1390 Notes 

1391 ----- 

1392 The base class checks that the given python type matches 

1393 the python type specified for this formatter when 

1394 constructed. 

1395 """ 

1396 return isinstance(in_memory_dataset, self.file_descriptor.storageClass.pytype) 

1397 

1398 @classmethod 

1399 def validateWriteRecipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None: 

1400 """Validate supplied recipes for this formatter. 

1401 

1402 The recipes are supplemented with default values where appropriate. 

1403 

1404 Parameters 

1405 ---------- 

1406 recipes : `dict` 

1407 Recipes to validate. 

1408 

1409 Returns 

1410 ------- 

1411 validated : `dict` 

1412 Validated recipes. 

1413 

1414 Raises 

1415 ------ 

1416 RuntimeError 

1417 Raised if validation fails. The default implementation raises 

1418 if any recipes are given. 

1419 """ 

1420 if recipes: 1420 ↛ 1421line 1420 didn't jump to line 1421 because the condition on line 1420 was never true

1421 raise RuntimeError(f"This formatter does not understand these writeRecipes: {recipes}") 

1422 return recipes 

1423 

1424 @classmethod 

1425 def validate_write_recipes(cls, recipes: Mapping[str, Any] | None) -> Mapping[str, Any] | None: 

1426 return cls.validateWriteRecipes(recipes) 

1427 

1428 @classmethod 

1429 def name(cls) -> str: 

1430 """Return the fully qualified name of the formatter. 

1431 

1432 Returns 

1433 ------- 

1434 name : `str` 

1435 Fully-qualified name of formatter class. 

1436 """ 

1437 return get_full_type_name(cls) 

1438 

1439 @abstractmethod 

1440 def read(self, component: str | None = None) -> Any: 

1441 """Read a Dataset. 

1442 

1443 Parameters 

1444 ---------- 

1445 component : `str`, optional 

1446 Component to read from the file. Only used if the `StorageClass` 

1447 for reading differed from the `StorageClass` used to write the 

1448 file. 

1449 

1450 Returns 

1451 ------- 

1452 inMemoryDataset : `object` 

1453 The requested Dataset. 

1454 """ 

1455 raise FormatterNotImplementedError("Type does not support reading") 

1456 

1457 @abstractmethod 

1458 def write(self, inMemoryDataset: Any) -> None: 

1459 """Write a Dataset. 

1460 

1461 Parameters 

1462 ---------- 

1463 inMemoryDataset : `object` 

1464 The Dataset to store. 

1465 """ 

1466 raise FormatterNotImplementedError("Type does not support writing") 

1467 

1468 @classmethod 

1469 def can_read_bytes(cls) -> bool: 

1470 """Indicate if this formatter can format from bytes. 

1471 

1472 Returns 

1473 ------- 

1474 can : `bool` 

1475 `True` if the `fromBytes` method is implemented. 

1476 """ 

1477 # We have no property to read so instead try to format from a byte 

1478 # and see what happens 

1479 try: 

1480 # We know the arguments are incompatible 

1481 cls.fromBytes(cls, b"") # type: ignore 

1482 except FormatterNotImplementedError: 

1483 return False 

1484 except Exception: 

1485 # There will be problems with the bytes we are supplying so ignore 

1486 pass 

1487 return True 

1488 

1489 def fromBytes(self, serializedDataset: bytes, component: str | None = None) -> object: 

1490 """Read serialized data into a Dataset or its component. 

1491 

1492 Parameters 

1493 ---------- 

1494 serializedDataset : `bytes` 

1495 Bytes object to unserialize. 

1496 component : `str`, optional 

1497 Component to read from the Dataset. Only used if the `StorageClass` 

1498 for reading differed from the `StorageClass` used to write the 

1499 file. 

1500 

1501 Returns 

1502 ------- 

1503 inMemoryDataset : `object` 

1504 The requested data as a Python object. The type of object 

1505 is controlled by the specific formatter. 

1506 """ 

1507 raise FormatterNotImplementedError("Type does not support reading from bytes.") 

1508 

1509 def toBytes(self, inMemoryDataset: Any) -> bytes: 

1510 """Serialize the Dataset to bytes based on formatter. 

1511 

1512 Parameters 

1513 ---------- 

1514 inMemoryDataset : `object` 

1515 The Python object to serialize. 

1516 

1517 Returns 

1518 ------- 

1519 serializedDataset : `bytes` 

1520 Bytes representing the serialized dataset. 

1521 """ 

1522 raise FormatterNotImplementedError("Type does not support writing to bytes.") 

1523 

1524 @contextlib.contextmanager 

1525 def _updateLocation(self, location: Location | None) -> Iterator[Location]: 

1526 """Temporarily replace the location associated with this formatter. 

1527 

1528 Parameters 

1529 ---------- 

1530 location : `Location` 

1531 New location to use for this formatter. If `None` the 

1532 formatter will not change but it will still return 

1533 the old location. This allows it to be used in a code 

1534 path where the location may not need to be updated 

1535 but the with block is still convenient. 

1536 

1537 Yields 

1538 ------ 

1539 old : `Location` 

1540 The old location that will be restored. 

1541 

1542 Notes 

1543 ----- 

1544 This is an internal method that should be used with care. 

1545 It may change in the future. Should be used as a context 

1546 manager to restore the location when the temporary is no 

1547 longer required. 

1548 """ 

1549 old = self._fileDescriptor.location 

1550 try: 

1551 if location is not None: 1551 ↛ 1553line 1551 didn't jump to line 1553 because the condition on line 1551 was always true

1552 self._fileDescriptor.location = location 

1553 yield old 

1554 finally: 

1555 if location is not None: 1555 ↛ exitline 1555 didn't return from function '_updateLocation' because the condition on line 1555 was always true

1556 self._fileDescriptor.location = old 

1557 

1558 def make_updated_location(self, location: Location) -> Location: 

1559 """Return a new `Location` updated with this formatter's extension. 

1560 

1561 Parameters 

1562 ---------- 

1563 location : `Location` 

1564 The location to update. 

1565 

1566 Returns 

1567 ------- 

1568 updated : `Location` 

1569 A new `Location` with a new file extension applied. 

1570 

1571 Raises 

1572 ------ 

1573 NotImplementedError 

1574 Raised if there is no ``extension`` attribute associated with 

1575 this formatter. 

1576 

1577 Notes 

1578 ----- 

1579 This method is available to all Formatters but might not be 

1580 implemented by all formatters. It requires that a formatter set 

1581 an ``extension`` attribute containing the file extension used when 

1582 writing files. If ``extension`` is `None` the supplied file will 

1583 not be updated. Not all formatters write files so this is not 

1584 defined in the base class. 

1585 """ 

1586 location = location.clone() 

1587 try: 

1588 # We are deliberately allowing extension to be undefined by 

1589 # default in the base class and mypy complains. 

1590 location.updateExtension(self.extension) # type:ignore 

1591 except AttributeError: 

1592 raise NotImplementedError("No file extension registered with this formatter") from None 

1593 return location 

1594 

1595 @classmethod 

1596 def validate_extension(cls, location: Location) -> None: 

1597 """Check the extension of the provided location for compatibility. 

1598 

1599 Parameters 

1600 ---------- 

1601 location : `Location` 

1602 Location from which to extract a file extension. 

1603 

1604 Returns 

1605 ------- 

1606 None 

1607 

1608 Raises 

1609 ------ 

1610 NotImplementedError 

1611 Raised if file extensions are a concept not understood by this 

1612 formatter. 

1613 ValueError 

1614 Raised if the formatter does not understand this extension. 

1615 

1616 Notes 

1617 ----- 

1618 This method is available to all Formatters but might not be 

1619 implemented by all formatters. It requires that a formatter set 

1620 an ``extension`` attribute containing the file extension used when 

1621 writing files. If ``extension`` is `None` only the set of supported 

1622 extensions will be examined. 

1623 """ 

1624 supported = set(cls.supportedExtensions) 

1625 

1626 try: 

1627 # We are deliberately allowing extension to be undefined by 

1628 # default in the base class and mypy complains. 

1629 default = cls.extension # type: ignore 

1630 except AttributeError: 

1631 raise NotImplementedError("No file extension registered with this formatter") from None 

1632 

1633 # If extension is implemented as an instance property it won't return 

1634 # a string when called as a class property. Assume that 

1635 # the supported extensions class property is complete. 

1636 if default is not None and isinstance(default, str): 1636 ↛ 1640line 1636 didn't jump to line 1640 because the condition on line 1636 was always true

1637 supported.add(default) 

1638 

1639 # Get the file name from the uri 

1640 file = location.uri.basename() 

1641 

1642 # Check that this file name ends with one of the supported extensions. 

1643 # This is less prone to confusion than asking the location for 

1644 # its extension and then doing a set comparison 

1645 for ext in supported: 

1646 if file.endswith(ext): 

1647 return 

1648 

1649 raise ValueError( 

1650 f"Extension '{location.getExtension()}' on '{location}' " 

1651 f"is not supported by Formatter '{cls.__name__}' (supports: {supported})" 

1652 ) 

1653 

1654 def predict_path(self) -> str: 

1655 """Return the path that would be returned by write. 

1656 

1657 Does not write any data file. 

1658 

1659 Uses the `FileDescriptor` associated with the instance. 

1660 

1661 Returns 

1662 ------- 

1663 path : `str` 

1664 Path within datastore that would be associated with the location 

1665 stored in this `Formatter`. 

1666 """ 

1667 updated = self.make_updated_location(self.fileDescriptor.location) 

1668 return updated.pathInStore.path 

1669 

1670 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]: 

1671 """Segregate the supplied parameters. 

1672 

1673 This splits the parameters into those understood by the 

1674 formatter and those not understood by the formatter. 

1675 

1676 Any unsupported parameters are assumed to be usable by associated 

1677 assemblers. 

1678 

1679 Parameters 

1680 ---------- 

1681 parameters : `dict`, optional 

1682 Parameters with values that have been supplied by the caller 

1683 and which might be relevant for the formatter. If `None` 

1684 parameters will be read from the registered `FileDescriptor`. 

1685 

1686 Returns 

1687 ------- 

1688 supported : `dict` 

1689 Those parameters supported by this formatter. 

1690 unsupported : `dict` 

1691 Those parameters not supported by this formatter. 

1692 """ 

1693 if parameters is None: 1693 ↛ 1696line 1693 didn't jump to line 1696 because the condition on line 1693 was always true

1694 parameters = self.fileDescriptor.parameters 

1695 

1696 if parameters is None: 

1697 return {}, {} 

1698 

1699 if self.unsupportedParameters is None: 1699 ↛ 1704line 1699 didn't jump to line 1704 because the condition on line 1699 was always true

1700 # Support none of the parameters 

1701 return {}, parameters.copy() 

1702 

1703 # Start by assuming all are supported 

1704 supported = parameters.copy() 

1705 unsupported = {} 

1706 

1707 # And remove any we know are not supported 

1708 for p in set(supported): 

1709 if p in self.unsupportedParameters: 

1710 unsupported[p] = supported.pop(p) 

1711 

1712 return supported, unsupported 

1713 

1714 # Support classic V1 interface. 

1715 makeUpdatedLocation = make_updated_location 

1716 validateExtension = validate_extension 

1717 segregateParameters = segregate_parameters 

1718 predictPath = predict_path 

1719 

1720 # Compatibility with V2 properties. 

1721 @property 

1722 def write_parameters(self) -> Mapping[str, Any]: 

1723 return self.writeParameters 

1724 

1725 @property 

1726 def write_recipes(self) -> Mapping[str, Any]: 

1727 return self.writeRecipes 

1728 

1729 @property 

1730 def file_descriptor(self) -> FileDescriptor: 

1731 return self.fileDescriptor 

1732 

1733 @property 

1734 def data_id(self) -> DataCoordinate: 

1735 return self.dataId 

1736 

1737 

1738class FormatterFactory: 

1739 """Factory for `Formatter` instances.""" 

1740 

1741 defaultKey = LookupKey("default") 

1742 """Configuration key associated with default write parameter settings.""" 

1743 

1744 writeRecipesKey = LookupKey("write_recipes") 

1745 """Configuration key associated with write recipes.""" 

1746 

1747 def __init__(self) -> None: 

1748 self._mappingFactory = MappingFactory(Formatter) 

1749 

1750 def __contains__(self, key: LookupKey | str) -> bool: 

1751 """Indicate whether the supplied key is present in the factory. 

1752 

1753 Parameters 

1754 ---------- 

1755 key : `LookupKey`, `str` or objects with ``name`` attribute 

1756 Key to use to lookup in the factory whether a corresponding 

1757 formatter is present. 

1758 

1759 Returns 

1760 ------- 

1761 in : `bool` 

1762 `True` if the supplied key is present in the factory. 

1763 """ 

1764 return key in self._mappingFactory 

1765 

1766 def registerFormatters(self, config: Config, *, universe: DimensionUniverse) -> None: 

1767 """Bulk register formatters from a config. 

1768 

1769 Parameters 

1770 ---------- 

1771 config : `Config` 

1772 ``formatters`` section of a configuration. 

1773 universe : `DimensionUniverse`, optional 

1774 Set of all known dimensions, used to expand and validate any used 

1775 in lookup keys. 

1776 

1777 Notes 

1778 ----- 

1779 The configuration can include one level of hierarchy where an 

1780 instrument-specific section can be defined to override more general 

1781 template specifications. This is represented in YAML using a 

1782 key of form ``instrument<name>`` which can then define templates 

1783 that will be returned if a `DatasetRef` contains a matching instrument 

1784 name in the data ID. 

1785 

1786 The config is parsed using the function 

1787 `~lsst.daf.butler.configSubset.processLookupConfigs`. 

1788 

1789 The values for formatter entries can be either a simple string 

1790 referring to a python type or a dict representing the formatter and 

1791 parameters to be hard-coded into the formatter constructor. For 

1792 the dict case the following keys are supported: 

1793 

1794 - formatter: The python type to be used as the formatter class. 

1795 - parameters: A further dict to be passed directly to the 

1796 ``write_parameters`` Formatter constructor to seed it. 

1797 These parameters are validated at instance creation and not at 

1798 configuration. 

1799 

1800 Additionally, a special ``default`` section can be defined that 

1801 uses the formatter type (class) name as the keys and specifies 

1802 default write parameters that should be used whenever an instance 

1803 of that class is constructed. 

1804 

1805 .. code-block:: yaml 

1806 

1807 formatters: 

1808 default: 

1809 lsst.daf.butler.formatters.example.ExampleFormatter: 

1810 max: 10 

1811 min: 2 

1812 comment: Default comment 

1813 calexp: lsst.daf.butler.formatters.example.ExampleFormatter 

1814 coadd: 

1815 formatter: lsst.daf.butler.formatters.example.ExampleFormatter 

1816 parameters: 

1817 max: 5 

1818 

1819 Any time an ``ExampleFormatter`` is constructed it will use those 

1820 parameters. If an explicit entry later in the configuration specifies 

1821 a different set of parameters, the two will be merged with the later 

1822 entry taking priority. In the example above ``calexp`` will use 

1823 the default parameters but ``coadd`` will override the value for 

1824 ``max``. 

1825 

1826 Formatter configuration can also include a special section describing 

1827 collections of write parameters that can be accessed through a 

1828 simple label. This allows common collections of options to be 

1829 specified in one place in the configuration and reused later. 

1830 The ``write_recipes`` section is indexed by Formatter class name 

1831 and each key is the label to associate with the parameters. 

1832 

1833 .. code-block:: yaml 

1834 

1835 formatters: 

1836 write_recipes: 

1837 lsst.obs.base.formatters.fitsExposure.FixExposureFormatter: 

1838 lossless: 

1839 ... 

1840 noCompression: 

1841 ... 

1842 

1843 By convention a formatter that uses write recipes will support a 

1844 ``recipe`` write parameter that will refer to a recipe name in 

1845 the ``write_recipes`` component. The `Formatter` will be constructed 

1846 in the `FormatterFactory` with all the relevant recipes and 

1847 will not attempt to filter by looking at ``write_parameters`` in 

1848 advance. See the specific formatter documentation for details on 

1849 acceptable recipe options. 

1850 """ 

1851 allowed_keys = {"formatter", "parameters"} 

1852 

1853 contents = processLookupConfigs(config, allow_hierarchy=True, universe=universe) 

1854 

1855 # Extract any default parameter settings 

1856 defaultParameters = contents.get(self.defaultKey, {}) 

1857 if not isinstance(defaultParameters, Mapping): 1857 ↛ 1858line 1857 didn't jump to line 1858 because the condition on line 1857 was never true

1858 raise RuntimeError( 

1859 "Default formatter parameters in config can not be a single string" 

1860 f" (got: {type(defaultParameters)})" 

1861 ) 

1862 

1863 # Extract any global write recipes -- these are indexed by 

1864 # Formatter class name. 

1865 writeRecipes = contents.get(self.writeRecipesKey, {}) 

1866 if isinstance(writeRecipes, str): 1866 ↛ 1867line 1866 didn't jump to line 1867 because the condition on line 1866 was never true

1867 raise RuntimeError( 

1868 f"The formatters.{self.writeRecipesKey} section must refer to a dict not '{writeRecipes}'" 

1869 ) 

1870 

1871 for key, f in contents.items(): 

1872 # default is handled in a special way 

1873 if key == self.defaultKey: 

1874 continue 

1875 if key == self.writeRecipesKey: 

1876 continue 

1877 

1878 # Can be a str or a dict. 

1879 specificWriteParameters = {} 

1880 if isinstance(f, str): 

1881 formatter = f 

1882 elif isinstance(f, Mapping): 1882 ↛ 1893line 1882 didn't jump to line 1893 because the condition on line 1882 was always true

1883 all_keys = set(f) 

1884 unexpected_keys = all_keys - allowed_keys 

1885 if unexpected_keys: 1885 ↛ 1886line 1885 didn't jump to line 1886 because the condition on line 1885 was never true

1886 raise ValueError(f"Formatter {key} uses unexpected keys {unexpected_keys} in config") 

1887 if "formatter" not in f: 1887 ↛ 1888line 1887 didn't jump to line 1888 because the condition on line 1887 was never true

1888 raise ValueError(f"Mandatory 'formatter' key missing for formatter key {key}") 

1889 formatter = f["formatter"] 

1890 if "parameters" in f: 1890 ↛ 1896line 1890 didn't jump to line 1896 because the condition on line 1890 was always true

1891 specificWriteParameters = f["parameters"] 

1892 else: 

1893 raise ValueError(f"Formatter for key {key} has unexpected value: '{f}'") 

1894 

1895 # Apply any default parameters for this formatter 

1896 writeParameters = copy.deepcopy(defaultParameters.get(formatter, {})) 

1897 writeParameters.update(specificWriteParameters) 

1898 

1899 kwargs: dict[str, Any] = {} 

1900 if writeParameters: 

1901 kwargs["write_parameters"] = writeParameters 

1902 

1903 if formatter in writeRecipes: 

1904 kwargs["write_recipes"] = writeRecipes[formatter] 

1905 

1906 self.registerFormatter(key, formatter, **kwargs) 

1907 

1908 def getLookupKeys(self) -> set[LookupKey]: 

1909 """Retrieve the look up keys for all the registry entries. 

1910 

1911 Returns 

1912 ------- 

1913 keys : `set` of `LookupKey` 

1914 The keys available for matching in the registry. 

1915 """ 

1916 return self._mappingFactory.getLookupKeys() 

1917 

1918 def getFormatterClassWithMatch( 

1919 self, entity: Entity 

1920 ) -> tuple[LookupKey, type[Formatter | FormatterV2], dict[str, Any]]: 

1921 """Get the matching formatter class along with the registry key. 

1922 

1923 Parameters 

1924 ---------- 

1925 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str` 

1926 Entity to use to determine the formatter to return. 

1927 `StorageClass` will be used as a last resort if `DatasetRef` 

1928 or `DatasetType` instance is provided. Supports instrument 

1929 override if a `DatasetRef` is provided configured with an 

1930 ``instrument`` value for the data ID. 

1931 

1932 Returns 

1933 ------- 

1934 matchKey : `LookupKey` 

1935 The key that resulted in the successful match. 

1936 formatter : `type` 

1937 The class of the registered formatter. 

1938 formatter_kwargs : `dict` 

1939 Keyword arguments that are associated with this formatter entry. 

1940 """ 

1941 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames() 

1942 matchKey, formatter, formatter_kwargs = self._mappingFactory.getClassFromRegistryWithMatch(names) 

1943 log.debug( 

1944 "Retrieved formatter %s from key '%s' for entity '%s'", 

1945 get_full_type_name(formatter), 

1946 matchKey, 

1947 entity, 

1948 ) 

1949 

1950 return matchKey, formatter, formatter_kwargs 

1951 

1952 def getFormatterClass(self, entity: Entity) -> type: 

1953 """Get the matching formatter class. 

1954 

1955 Parameters 

1956 ---------- 

1957 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str` 

1958 Entity to use to determine the formatter to return. 

1959 `StorageClass` will be used as a last resort if `DatasetRef` 

1960 or `DatasetType` instance is provided. Supports instrument 

1961 override if a `DatasetRef` is provided configured with an 

1962 ``instrument`` value for the data ID. 

1963 

1964 Returns 

1965 ------- 

1966 formatter : `type` 

1967 The class of the registered formatter. 

1968 """ 

1969 _, formatter, _ = self.getFormatterClassWithMatch(entity) 

1970 return formatter 

1971 

1972 def getFormatterWithMatch( 

1973 self, entity: Entity, *args: Any, **kwargs: Any 

1974 ) -> tuple[LookupKey, Formatter | FormatterV2]: 

1975 """Get a new formatter instance along with the matching registry key. 

1976 

1977 Parameters 

1978 ---------- 

1979 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str` 

1980 Entity to use to determine the formatter to return. 

1981 `StorageClass` will be used as a last resort if `DatasetRef` 

1982 or `DatasetType` instance is provided. Supports instrument 

1983 override if a `DatasetRef` is provided configured with an 

1984 ``instrument`` value for the data ID. 

1985 *args : `tuple` 

1986 Positional arguments to use pass to the object constructor. 

1987 **kwargs 

1988 Keyword arguments to pass to object constructor. 

1989 

1990 Returns 

1991 ------- 

1992 matchKey : `LookupKey` 

1993 The key that resulted in the successful match. 

1994 formatter : `Formatter` 

1995 An instance of the registered formatter. 

1996 """ 

1997 names = (LookupKey(name=entity),) if isinstance(entity, str) else entity._lookupNames() 

1998 matchKey, formatter = self._mappingFactory.getFromRegistryWithMatch(names, *args, **kwargs) 

1999 log.debug( 

2000 "Retrieved formatter %s from key '%s' for entity '%s'", 

2001 get_full_type_name(formatter), 

2002 matchKey, 

2003 entity, 

2004 ) 

2005 

2006 return matchKey, formatter 

2007 

2008 def getFormatter(self, entity: Entity, *args: Any, **kwargs: Any) -> Formatter | FormatterV2: 

2009 """Get a new formatter instance. 

2010 

2011 Parameters 

2012 ---------- 

2013 entity : `DatasetRef`, `DatasetType`, `StorageClass`, or `str` 

2014 Entity to use to determine the formatter to return. 

2015 `StorageClass` will be used as a last resort if `DatasetRef` 

2016 or `DatasetType` instance is provided. Supports instrument 

2017 override if a `DatasetRef` is provided configured with an 

2018 ``instrument`` value for the data ID. 

2019 *args : `tuple` 

2020 Positional arguments to use pass to the object constructor. 

2021 **kwargs 

2022 Keyword arguments to pass to object constructor. 

2023 

2024 Returns 

2025 ------- 

2026 formatter : `Formatter` 

2027 An instance of the registered formatter. 

2028 """ 

2029 _, formatter = self.getFormatterWithMatch(entity, *args, **kwargs) 

2030 return formatter 

2031 

2032 def registerFormatter( 

2033 self, 

2034 type_: LookupKey | str | StorageClass | DatasetType, 

2035 formatter: str, 

2036 *, 

2037 overwrite: bool = False, 

2038 **kwargs: Any, 

2039 ) -> None: 

2040 """Register a `Formatter`. 

2041 

2042 Parameters 

2043 ---------- 

2044 type_ : `LookupKey`, `str`, `StorageClass` or `DatasetType` 

2045 Type for which this formatter is to be used. If a `LookupKey` 

2046 is not provided, one will be constructed from the supplied string 

2047 or by using the ``name`` property of the supplied entity. 

2048 formatter : `str` or class of type `Formatter` 

2049 Identifies a `Formatter` subclass to use for reading and writing 

2050 Datasets of this type. Can be a `Formatter` class. 

2051 overwrite : `bool`, optional 

2052 If `True` an existing entry will be replaced by the new value. 

2053 Default is `False`. 

2054 **kwargs 

2055 Keyword arguments to always pass to object constructor when 

2056 retrieved. 

2057 

2058 Raises 

2059 ------ 

2060 ValueError 

2061 Raised if the formatter does not name a valid formatter type and 

2062 ``overwrite`` is `False`. 

2063 """ 

2064 self._mappingFactory.placeInRegistry(type_, formatter, overwrite=overwrite, **kwargs) 

2065 

2066 

2067class FormatterV1inV2(FormatterV2): 

2068 """An implementation of a V2 formatter that provides a compatibility 

2069 interface for V1 formatters. 

2070 

2071 Parameters 

2072 ---------- 

2073 file_descriptor : `FileDescriptor`, optional 

2074 Identifies the file to read or write, and the associated storage 

2075 classes and parameter information. Its value can be `None` if the 

2076 caller will never call `Formatter.read` or `Formatter.write`. 

2077 ref : `DatasetRef` 

2078 The dataset associated with this formatter. Should not be a component 

2079 dataset ref. 

2080 formatter : `Formatter` 

2081 A version 1 `Formatter` instance. The V2 formatter layer forwards calls 

2082 to this formatter. 

2083 write_parameters : `dict`, optional 

2084 Any parameters to be hard-coded into this instance to control how 

2085 the dataset is serialized. 

2086 write_recipes : `dict`, optional 

2087 Detailed write Recipes indexed by recipe name. 

2088 **kwargs 

2089 Additional arguments that will be ignored but allow for 

2090 `Formatter` V1 parameters to be given. 

2091 """ 

2092 

2093 can_read_from_local_file = True 

2094 """This formatter can read from a local file.""" 

2095 

2096 def __init__( 

2097 self, 

2098 file_descriptor: FileDescriptor, 

2099 *, 

2100 ref: DatasetRef, 

2101 formatter: Formatter, 

2102 write_parameters: Mapping[str, Any] | None = None, 

2103 write_recipes: Mapping[str, Any] | None = None, 

2104 # Compatibility parameters. Unused in v2. 

2105 **kwargs: Any, 

2106 ): 

2107 if not isinstance(formatter, Formatter): 2107 ↛ 2108line 2107 didn't jump to line 2108 because the condition on line 2107 was never true

2108 raise TypeError(f"Formatter parameter was not a V1 formatter (was {type(formatter)})") 

2109 

2110 # Replace the class property with instance values from this 

2111 # V1 formatter so that the V2 __init__ will be able to validate. 

2112 self.supported_write_parameters = formatter.supportedWriteParameters # type: ignore 

2113 self._formatter = formatter 

2114 

2115 super().__init__( 

2116 file_descriptor, ref=ref, write_parameters=write_parameters, write_recipes=write_recipes 

2117 ) 

2118 

2119 def get_write_extension(self) -> str: 

2120 ext = self._formatter.extension 

2121 return ext if ext is not None else "" 

2122 

2123 def segregate_parameters(self, parameters: dict[str, Any] | None = None) -> tuple[dict, dict]: 

2124 return self._formatter.segregate_parameters(parameters) 

2125 

2126 def validate_write_recipes( # type: ignore 

2127 self, 

2128 recipes: Mapping[str, Any] | None, 

2129 ) -> Mapping[str, Any] | None: 

2130 # This should be a class method but a class method can not work 

2131 # for a dynamic shim. Luckily the shim is only used as an instance. 

2132 return self._formatter.validate_write_recipes(recipes) 

2133 

2134 def read_from_local_file(self, path: str, component: str | None = None, expected_size: int = -1) -> Any: 

2135 # Need to temporarily override the location since the V1 formatter 

2136 # will not know anything about this local file. 

2137 

2138 # V2 does not have a fromBytes equivalent. 

2139 if self._formatter.can_read_bytes(): 2139 ↛ 2140line 2139 didn't jump to line 2140 because the condition on line 2139 was never true

2140 with open(path, "rb") as fd: 

2141 serialized_dataset = fd.read() 

2142 return self._formatter.fromBytes(serialized_dataset, component=component) 

2143 

2144 location = Location(None, path) 

2145 with self._formatter._updateLocation(location): 

2146 try: 

2147 result = self._formatter.read(component=component) 

2148 except NotImplementedError: 

2149 # V1 raises NotImplementedError but V2 is expecting something 

2150 # slightly different. 

2151 return NotImplemented 

2152 return result 

2153 

2154 def to_bytes(self, in_memory_dataset: Any) -> bytes: 

2155 try: 

2156 return self._formatter.toBytes(in_memory_dataset) 

2157 except NotImplementedError as e: 

2158 # V1 raises NotImplementedError but V2 is expecting something 

2159 # slightly different. 

2160 raise FormatterNotImplementedError(str(e)) from e 

2161 

2162 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None: 

2163 with self._formatter._updateLocation(Location(None, uri)): 

2164 try: 

2165 self._formatter.write(in_memory_dataset) 

2166 except NotImplementedError as e: 

2167 # V1 raises NotImplementedError but V2 is expecting something 

2168 # slightly different. 

2169 raise FormatterNotImplementedError(str(e)) from e 

2170 

2171 

2172# Type to use when allowing a Formatter or its class name 

2173FormatterParameter: TypeAlias = str | type[Formatter] | Formatter | FormatterV2 | type[FormatterV2]