Coverage for python / lsst / daf / butler / dimensions / _coordinate.py: 33%

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 

28# 

29# Design notes for this module are in 

30# doc/lsst.daf.butler/dev/dataCoordinate.py. 

31# 

32 

33from __future__ import annotations 

34 

35__all__ = ( 

36 "DataCoordinate", 

37 "DataId", 

38 "DataIdKey", 

39 "DataIdValue", 

40 "SerializedDataCoordinate", 

41 "SerializedDataId", 

42) 

43 

44import numbers 

45from abc import abstractmethod 

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

47from typing import TYPE_CHECKING, Any, ClassVar, Self, TypeAlias, cast, overload 

48 

49import pydantic 

50 

51from lsst.sphgeom import IntersectionRegion, Region 

52 

53from .._exceptions import DimensionNameError 

54from .._timespan import Timespan 

55from ..json import from_json_pydantic, to_json_pydantic 

56from ..persistence_context import PersistenceContextVars 

57from ._group import DimensionGroup 

58from ._records import DataIdKey, DataIdValue, DimensionRecord, SerializedDimensionRecord 

59 

60if TYPE_CHECKING: # Imports needed only for type annotations; may be circular. 

61 from ..registry import Registry 

62 from ._universe import DimensionUniverse 

63 

64SerializedDataId: TypeAlias = dict[str, DataIdValue] 

65"""Simplified model for serializing the ``mapping`` property of 

66`DataCoordinate`. 

67""" 

68 

69 

70class SerializedDataCoordinate(pydantic.BaseModel): 

71 """Simplified model for serializing a `DataCoordinate`.""" 

72 

73 dataId: SerializedDataId 

74 records: dict[str, SerializedDimensionRecord] | None = None 

75 

76 @classmethod 

77 def direct(cls, *, dataId: SerializedDataId, records: dict[str, dict] | None) -> SerializedDataCoordinate: 

78 """Construct a `SerializedDataCoordinate` directly without validators. 

79 

80 Parameters 

81 ---------- 

82 dataId : `SerializedDataId` 

83 The data ID. 

84 records : `dict` or `None` 

85 The dimension records. 

86 

87 Notes 

88 ----- 

89 This differs from the pydantic "construct" method in that the arguments 

90 are explicitly what the model requires, and it will recurse through 

91 members, constructing them from their corresponding `direct` methods. 

92 

93 This method should only be called when the inputs are trusted. 

94 """ 

95 key = (frozenset(dataId.items()), records is not None) 

96 cache = PersistenceContextVars.serializedDataCoordinateMapping.get() 

97 if cache is not None and (result := cache.get(key)) is not None: 

98 return result 

99 

100 if records is None: 

101 serialized_records = None 

102 else: 

103 serialized_records = {k: SerializedDimensionRecord.direct(**v) for k, v in records.items()} 

104 

105 node = cls.model_construct(dataId=dataId, records=serialized_records) 

106 

107 if cache is not None: 

108 cache[key] = node 

109 return node 

110 

111 

112def _intersectRegions(*args: Region) -> Region | None: 

113 """Return the intersection of several regions. 

114 

115 For internal use by `ExpandedDataCoordinate` only. 

116 

117 If no regions are provided, returns `None`. 

118 """ 

119 if len(args) == 0: 

120 return None 

121 else: 

122 result = args[0] 

123 for n in range(1, len(args)): 

124 result = IntersectionRegion(result, args[n]) 

125 return result 

126 

127 

128class DataCoordinate: 

129 """A validated data ID. 

130 

131 DataCoordinate guarantees that its key-value pairs identify at least all 

132 required dimensions in a `DimensionGroup`. 

133 

134 Notes 

135 ----- 

136 `DataCoordinate` is an ABC, but it provides `staticmethod` factory 

137 functions for private concrete implementations that should be sufficient 

138 for most purposes. `standardize` is the most flexible and safe of these; 

139 the others (`make_empty`, `from_required_values`, and `from_full_values`) 

140 are more specialized and perform little or no checking of inputs. 

141 

142 Lookups for implied dimensions (those in ``self.dimensions.implied``) are 

143 supported if and only if `has_full_values` is `True`. This also sets the 

144 keys of the `mapping` attribute. This means that `DataCoordinate` equality 

145 is not the same as testing for equality on the `mapping` attribute 

146 (instead, it is the same as testing for equality on the `required` 

147 attribute). 

148 

149 See also :ref:`lsst.daf.butler-dimensions_data_ids` 

150 """ 

151 

152 __slots__ = () 

153 

154 _serializedType: ClassVar[type[pydantic.BaseModel]] = SerializedDataCoordinate 

155 

156 @staticmethod 

157 def standardize( 

158 mapping: Mapping[str, DataIdValue] | DataCoordinate | None = None, 

159 *, 

160 dimensions: Iterable[str] | DimensionGroup | None = None, 

161 universe: DimensionUniverse | None = None, 

162 defaults: DataCoordinate | None = None, 

163 **kwargs: Any, 

164 ) -> DataCoordinate: 

165 """Standardize the supplied dataId. 

166 

167 Adapts an arbitrary mapping and/or additional arguments into a true 

168 `DataCoordinate`, or augment an existing one. 

169 

170 Parameters 

171 ---------- 

172 mapping : `~collections.abc.Mapping` or `DataCoordinate`, optional 

173 An informal data ID that maps dimensions or dimension names to 

174 their primary key values (may also be a true `DataCoordinate`). 

175 dimensions : `~collections.abc.Iterable` [ `str` ], `DimensionGroup`, \ 

176 optional 

177 The dimensions to be identified by the new `DataCoordinate`. If not 

178 provided, will be inferred from the keys of ``mapping`` and 

179 ``**kwargs``, and ``universe`` must be provided unless ``mapping`` 

180 is already a `DataCoordinate`. 

181 universe : `DimensionUniverse` 

182 All known dimensions and their relationships; used to expand and 

183 validate dependencies when ``dimensions`` is not provided. 

184 defaults : `DataCoordinate`, optional 

185 Default dimension key-value pairs to use when needed. These are 

186 never used to infer ``group``, and are ignored if a different value 

187 is provided for the same key in ``mapping`` or `**kwargs``. 

188 **kwargs 

189 Additional keyword arguments are treated like additional key-value 

190 pairs in ``mapping``. 

191 

192 Returns 

193 ------- 

194 coordinate : `DataCoordinate` 

195 A validated `DataCoordinate` instance. 

196 

197 Raises 

198 ------ 

199 TypeError 

200 Raised if the set of optional arguments provided is not supported. 

201 DimensionNameError 

202 Raised if a key-value pair for a required dimension is missing. 

203 """ 

204 universe = universe or getattr(dimensions, "universe", None) or getattr(mapping, "universe", None) 

205 if universe is None: 

206 raise TypeError("universe must be provided, either directly or via dimensions or mapping.") 

207 if dimensions is not None: 

208 dimensions = universe.conform(dimensions) 

209 new_mapping: dict[str, DataIdValue] = {} 

210 if isinstance(mapping, DataCoordinate): 

211 if dimensions is None: 

212 if not kwargs: 

213 # Already standardized to exactly what we want. 

214 return mapping 

215 elif kwargs.keys().isdisjoint(dimensions.names): 

216 # User provided kwargs, but told us not to use them by 

217 # passing in dimensions that are disjoint from those kwargs. 

218 # This is not necessarily user error - it's a useful pattern 

219 # to pass in all of the key-value pairs you have and let the 

220 # code here pull out only what it needs. 

221 return mapping.subset(dimensions.names) 

222 new_mapping.update((name, mapping[name]) for name in mapping.dimensions.required) 

223 if mapping.hasFull(): 

224 new_mapping.update((name, mapping[name]) for name in mapping.dimensions.implied) 

225 elif mapping is not None: 

226 new_mapping.update(mapping) 

227 new_mapping.update(kwargs) 

228 if dimensions is None: 

229 if defaults is not None: 

230 universe = defaults.universe 

231 elif universe is None: 

232 raise TypeError("universe must be provided if dimensions is not.") 

233 dimensions = DimensionGroup(universe, new_mapping.keys()) 

234 if not dimensions: 

235 return DataCoordinate.make_empty(universe) 

236 # Some backends cannot handle numpy.int64 type which is a subclass of 

237 # numbers.Integral; convert that to int. 

238 for k, v in new_mapping.items(): 

239 if isinstance(v, numbers.Integral): # type: ignore 

240 new_mapping[k] = int(v) # type: ignore 

241 if defaults is not None: 

242 for k, v in defaults.mapping.items(): 

243 new_mapping.setdefault(k, v) 

244 if new_mapping.keys() >= dimensions.names: 

245 return DataCoordinate.from_full_values( 

246 dimensions, tuple(new_mapping[name] for name in dimensions.data_coordinate_keys) 

247 ) 

248 else: 

249 try: 

250 values = tuple(new_mapping[name] for name in dimensions.required) 

251 except KeyError as err: 

252 raise DimensionNameError( 

253 f"No value in data ID ({mapping}) for required dimension {err}." 

254 ) from err 

255 return DataCoordinate.from_required_values(dimensions, values) 

256 

257 @property 

258 @abstractmethod 

259 def mapping(self) -> Mapping[str, DataIdValue]: 

260 """A mapping view of the data ID with keys for all dimensions it has 

261 values for. 

262 """ 

263 raise NotImplementedError() 

264 

265 @property 

266 @abstractmethod 

267 def required(self) -> Mapping[str, DataIdValue]: 

268 """A mapping view of the data ID with keys for just its required 

269 dimensions. 

270 """ 

271 raise NotImplementedError() 

272 

273 @property 

274 @abstractmethod 

275 def required_values(self) -> tuple[DataIdValue, ...]: 

276 """The required values (only) of this data ID as a tuple. 

277 

278 Element order is consistent with `required`. 

279 

280 In contexts where all data IDs have the same dimensions, comparing and 

281 hashing these tuples can be much faster than comparing the original 

282 `DataCoordinate` instances. 

283 """ 

284 raise NotImplementedError() 

285 

286 @property 

287 def full_values(self) -> tuple[DataIdValue, ...]: 

288 """The full values (only) of this data ID as a tuple. 

289 

290 Element order is consistent with `DimensionGroup.data_coordinate_keys`, 

291 i.e. all required dimensions followed by all implied dimensions. 

292 """ 

293 raise ValueError(f"DataCoordinate {self} has only required values.") 

294 

295 @staticmethod 

296 def makeEmpty(universe: DimensionUniverse) -> DataCoordinate: 

297 """Return an empty `DataCoordinate`. 

298 

299 It identifies the null set of dimensions. 

300 

301 Parameters 

302 ---------- 

303 universe : `DimensionUniverse` 

304 Universe to which this null dimension set belongs. 

305 

306 Returns 

307 ------- 

308 dataId : `DataCoordinate` 

309 A data ID object that identifies no dimensions. `hasFull` and 

310 `hasRecords` are guaranteed to return `True`, because both `full` 

311 and `records` are just empty mappings. 

312 """ 

313 return DataCoordinate.make_empty(universe) 

314 

315 @staticmethod 

316 def make_empty(universe: DimensionUniverse) -> DataCoordinate: 

317 """Return an empty `DataCoordinate`. 

318 

319 It identifies the null set of dimensions. 

320 

321 Parameters 

322 ---------- 

323 universe : `DimensionUniverse` 

324 Universe to which this null dimension set belongs. 

325 

326 Returns 

327 ------- 

328 data_id : `DataCoordinate` 

329 A data ID object that identifies no dimensions. `hasFull` and 

330 `hasRecords` are guaranteed to return `True`, because both `full` 

331 and `records` are just empty mappings. 

332 """ 

333 return _ExpandedTupleDataCoordinate(universe.empty, (), {}) 

334 

335 @staticmethod 

336 def from_required_values(dimensions: DimensionGroup, values: tuple[DataIdValue, ...]) -> DataCoordinate: 

337 """Construct a `DataCoordinate` from required dimension values. 

338 

339 This is a low-level interface with at most assertion-level checking of 

340 inputs. Most callers should use `standardize` instead. 

341 

342 Parameters 

343 ---------- 

344 dimensions : `DimensionGroup` 

345 Dimensions this data ID will identify. 

346 values : `tuple` [ `int` or `str` ] 

347 Tuple of primary key values corresponding to 

348 ``dimensions.required``, in that order. 

349 

350 Returns 

351 ------- 

352 data_id : `DataCoordinate` 

353 A data ID object that identifies the given dimensions. 

354 ``dataId.hasFull()`` will return `True` only if 

355 ``dimensions.implied`` is empty. ``dataId.hasRecords()`` will 

356 return `True` if and only if ``dimensions`` is empty. 

357 """ 

358 assert len(dimensions.required) == len(values), ( 

359 f"Inconsistency between dimensions {dimensions.required} and required values {values}." 

360 ) 

361 if not dimensions: 

362 return DataCoordinate.make_empty(dimensions.universe) 

363 if not dimensions.implied: 

364 return _FullTupleDataCoordinate(dimensions, values) 

365 return _RequiredTupleDataCoordinate(dimensions, values) 

366 

367 @staticmethod 

368 def from_full_values(dimensions: DimensionGroup, values: tuple[DataIdValue, ...]) -> DataCoordinate: 

369 """Construct a `DataCoordinate` from all dimension values. 

370 

371 This is a low-level interface with at most assertion-level checking of 

372 inputs. Most callers should use `standardize` instead. 

373 

374 Parameters 

375 ---------- 

376 dimensions : `DimensionGroup` 

377 Dimensions this data ID will identify. 

378 values : `tuple` [ `int` or `str` ] 

379 Tuple of primary key values corresponding to 

380 ``itertools.chain(dimensions.required, dimensions.implied)``, in 

381 that order. Note that this is _not_ the same order as 

382 ``dimensions.names``, though these contain the same elements. 

383 

384 Returns 

385 ------- 

386 data_id : `DataCoordinate` 

387 A data ID object that identifies the given dimensions. 

388 ``dataId.hasFull()`` will always return `True`. 

389 ``dataId.hasRecords()`` will only return `True` if ``dimensions`` 

390 is empty. 

391 """ 

392 assert len(dimensions) == len(values), ( 

393 f"Inconsistency between dimensions {dimensions.data_coordinate_keys} and full values {values}." 

394 ) 

395 if not dimensions: 

396 return DataCoordinate.make_empty(dimensions.universe) 

397 return _FullTupleDataCoordinate(dimensions, values) 

398 

399 def __bool__(self) -> bool: 

400 return bool(self.dimensions) 

401 

402 def __hash__(self) -> int: 

403 return hash((self.dimensions,) + self.required_values) 

404 

405 def __eq__(self, other: Any) -> bool: 

406 if not isinstance(other, DataCoordinate): 

407 other = DataCoordinate.standardize(other, universe=self.universe) 

408 return self.dimensions == other.dimensions and self.required_values == other.required_values 

409 

410 @abstractmethod 

411 def __getitem__(self, key: str) -> DataIdValue: 

412 raise NotImplementedError() 

413 

414 def __contains__(self, key: str) -> bool: 

415 try: 

416 self.__getitem__(key) 

417 return True 

418 except KeyError: 

419 return False 

420 

421 @overload 

422 def get(self, key: str) -> DataIdValue | None: ... 422 ↛ exitline 422 didn't return from function 'get' because

423 

424 @overload 

425 def get(self, key: str, default: int) -> int: ... 425 ↛ exitline 425 didn't return from function 'get' because

426 

427 @overload 

428 def get(self, key: str, default: str) -> str: ... 428 ↛ exitline 428 didn't return from function 'get' because

429 

430 @overload 

431 def get(self, key: str, default: DataIdValue | None) -> DataIdValue | None: ... 431 ↛ exitline 431 didn't return from function 'get' because

432 

433 def get(self, key: str, default: DataIdValue | None = None) -> DataIdValue | None: 

434 try: 

435 return self.__getitem__(key) 

436 except KeyError: 

437 return default 

438 

439 def __repr__(self) -> str: 

440 # We can't make repr yield something that could be exec'd here without 

441 # printing out the whole DimensionUniverse. 

442 return str(self.mapping) 

443 

444 def __lt__(self, other: Any) -> bool: 

445 if not isinstance(other, DataCoordinate): 

446 return NotImplemented 

447 # Unlike repr() we only use required keys here to ensure that __eq__ 

448 # can not be true simultaneously with __lt__ being true. 

449 return self.required_values < other.required_values 

450 

451 @abstractmethod 

452 def subset(self, dimensions: DimensionGroup | Iterable[str]) -> DataCoordinate: 

453 """Return a `DataCoordinate` whose diensions are a subset of 

454 ``self.dimensions``. 

455 

456 Parameters 

457 ---------- 

458 dimensions : `DimensionGroup` or `~collections.abc.Iterable` [ `str` ] 

459 The dimensions identified by the returned `DataCoordinate`. 

460 

461 Returns 

462 ------- 

463 coordinate : `DataCoordinate` 

464 A `DataCoordinate` instance that identifies only the given 

465 dimensions. May be ``self`` if ``dimensions == self.dimensions``. 

466 

467 Raises 

468 ------ 

469 KeyError 

470 Raised if the primary key value for one or more required dimensions 

471 is unknown. This may happen even if the required subset of the new 

472 dimensions are not a subset of the dimensions actually known by 

473 this data ID.. As an example, consider trying to go from a data ID 

474 with dimensions {instrument, physical_filter, band} to just 

475 {instrument, band}; band is implied by physical_filter and hence 

476 would have no value in the original data ID if ``self.hasFull()`` 

477 is `False`. 

478 

479 Notes 

480 ----- 

481 If `hasFull` and `hasRecords` return `True` on ``self``, they will 

482 return `True` (respectively) on the returned `DataCoordinate` as well. 

483 The converse does not hold. 

484 """ 

485 raise NotImplementedError() 

486 

487 @abstractmethod 

488 def union(self, other: DataCoordinate) -> DataCoordinate: 

489 """Combine two data IDs. 

490 

491 Yields a new one that identifies all dimensions that either of them 

492 identify. 

493 

494 Parameters 

495 ---------- 

496 other : `DataCoordinate` 

497 Data ID to combine with ``self``. 

498 

499 Returns 

500 ------- 

501 unioned : `DataCoordinate` 

502 A `DataCoordinate` instance that satisfies 

503 ``unioned.dimensions == self.dimensions.union(other.dimensions)``. 

504 Will preserve ``hasFull`` and ``hasRecords`` whenever possible. 

505 

506 Notes 

507 ----- 

508 No checking for consistency is performed on values for keys that 

509 ``self`` and ``other`` have in common, and which value is included in 

510 the returned data ID is not specified. 

511 """ 

512 raise NotImplementedError() 

513 

514 @abstractmethod 

515 def expanded(self, records: Mapping[str, DimensionRecord | None]) -> DataCoordinate: 

516 """Return a `DataCoordinate` that holds the given records. 

517 

518 Guarantees that `hasRecords` returns `True`. 

519 

520 This is a low-level interface with at most assertion-level checking of 

521 inputs. Most callers should use `Registry.expandDataId` instead. 

522 

523 Parameters 

524 ---------- 

525 records : `~collections.abc.Mapping` [ `str`, `DimensionRecord` or \ 

526 `None` ] 

527 A`~collections.abc.Mapping` with `str` (dimension element name) 

528 keys and `DimensionRecord` values. Keys must cover all elements in 

529 ``self.dimensions.elements``. Values may be `None`, but only to 

530 reflect actual NULL values in the database, not just records that 

531 have not been fetched. 

532 """ 

533 raise NotImplementedError() 

534 

535 @property 

536 def universe(self) -> DimensionUniverse: 

537 """Universe that defines all known compatible dimensions. 

538 

539 The universe will be compatible with this coordinate 

540 (`DimensionUniverse`). 

541 """ 

542 return self.dimensions.universe 

543 

544 @property 

545 @abstractmethod 

546 def dimensions(self) -> DimensionGroup: 

547 """Dimensions identified by this data ID (`DimensionGroup`). 

548 

549 Note that values are only required to be present for dimensions in 

550 ``self.dimensions.required``; all others may be retrieved (from a 

551 `Registry`) given these. 

552 """ 

553 raise NotImplementedError() 

554 

555 @abstractmethod 

556 def hasFull(self) -> bool: 

557 """Whether this data ID contains implied and required values. 

558 

559 Returns 

560 ------- 

561 state : `bool` 

562 If `True`, `__getitem__`, `get`, and `__contains__` (but not 

563 `keys`!) will act as though the mapping includes key-value pairs 

564 for implied dimensions, and the `full` property may be used. If 

565 `False`, these operations only include key-value pairs for required 

566 dimensions, and accessing `full` is an error. Always `True` if 

567 there are no implied dimensions. 

568 """ 

569 raise NotImplementedError() 

570 

571 @abstractmethod 

572 def hasRecords(self) -> bool: 

573 """Whether this data ID contains records. 

574 

575 These are the records for all of the dimension elements it identifies. 

576 

577 Returns 

578 ------- 

579 state : `bool` 

580 If `True`, the following attributes may be accessed: 

581 

582 - `records` 

583 - `region` 

584 - `timespan` 

585 - `pack` 

586 

587 If `False`, accessing any of these is considered a logic error. 

588 """ 

589 raise NotImplementedError() 

590 

591 @property 

592 def records(self) -> Mapping[str, DimensionRecord | None]: 

593 """A mapping that contains `DimensionRecord` objects for all 

594 elements identified by this data ID. 

595 

596 Notes 

597 ----- 

598 The values of this mapping may be `None` if and only if there is no 

599 record for that element with these dimensions in the database (which 

600 means some foreign key field must have a NULL value). 

601 

602 Accessing this attribute if `hasRecords` returns `False` is a logic 

603 error that may raise an exception of unspecified type either 

604 immediately or when the returned mapping is used, depending on the 

605 implementation and whether assertions are enabled. 

606 """ 

607 assert self.hasRecords(), "records may only be accessed if hasRecords() returns True." 

608 return _DataCoordinateRecordsView(self) 

609 

610 @abstractmethod 

611 def _record(self, name: str) -> DimensionRecord | None: 

612 """Protected implementation hook that backs the ``records`` attribute. 

613 

614 Parameters 

615 ---------- 

616 name : `str` 

617 The name of a `DimensionElement`, guaranteed to be in 

618 ``self.dimensions.elements``. 

619 

620 Returns 

621 ------- 

622 record : `DimensionRecord` or `None` 

623 The dimension record for the given element identified by this 

624 data ID, or `None` if there is no such record. 

625 """ 

626 raise NotImplementedError() 

627 

628 @property 

629 def region(self) -> Region | None: 

630 """Spatial region associated with this data ID. 

631 

632 (`lsst.sphgeom.Region` or `None`). 

633 

634 This is `None` if and only if ``self.dimensions.spatial`` is empty. 

635 

636 Accessing this attribute if `hasRecords` returns `False` is a logic 

637 error that may or may not raise an exception, depending on the 

638 implementation and whether assertions are enabled. 

639 """ 

640 assert self.hasRecords(), "region may only be accessed if hasRecords() returns True." 

641 regions = [] 

642 for family in self.dimensions.spatial: 

643 element = family.choose(self.dimensions) 

644 record = self._record(element.name) 

645 if record is None or record.region is None: 

646 return None 

647 else: 

648 regions.append(record.region) 

649 return _intersectRegions(*regions) 

650 

651 @property 

652 def timespan(self) -> Timespan | None: 

653 """Temporal interval associated with this data ID. 

654 

655 (`Timespan` or `None`). 

656 

657 This is `None` if and only if ``self.dimensions.temporal`` is empty. 

658 

659 Accessing this attribute if `hasRecords` returns `False` is a logic 

660 error that may or may not raise an exception, depending on the 

661 implementation and whether assertions are enabled. 

662 """ 

663 assert self.hasRecords(), "timespan may only be accessed if hasRecords() returns True." 

664 timespans = [] 

665 for family in self.dimensions.temporal: 

666 element = family.choose(self.dimensions) 

667 record = self._record(element.name) 

668 # DimensionRecord subclasses for temporal elements always have 

669 # .timespan, but they're dynamic so this can't be type-checked. 

670 if record is None or record.timespan is None: 

671 return None 

672 else: 

673 timespans.append(record.timespan) 

674 if not timespans: 

675 return None 

676 elif len(timespans) == 1: 

677 return timespans[0] 

678 else: 

679 return Timespan.intersection(*timespans) 

680 

681 def to_simple(self, minimal: bool = False) -> SerializedDataCoordinate: 

682 """Convert this class to a simple python type. 

683 

684 This is suitable for serialization. 

685 

686 Parameters 

687 ---------- 

688 minimal : `bool`, optional 

689 Use minimal serialization. If set the records will not be attached. 

690 

691 Returns 

692 ------- 

693 simple : `SerializedDataCoordinate` 

694 The object converted to simple form. 

695 """ 

696 # Convert to a dict form 

697 records: dict[str, SerializedDimensionRecord] | None 

698 if not minimal and self.hasRecords(): 

699 records = { 

700 k: v.to_simple() for k in self.dimensions.elements if (v := self.records[k]) is not None 

701 } 

702 else: 

703 records = None 

704 

705 return SerializedDataCoordinate(dataId=dict(self.mapping), records=records) 

706 

707 @classmethod 

708 def from_simple( 

709 cls, 

710 simple: SerializedDataCoordinate, 

711 universe: DimensionUniverse | None = None, 

712 registry: Registry | None = None, 

713 ) -> DataCoordinate: 

714 """Construct a new object from the simplified form. 

715 

716 The data is assumed to be of the form returned from the `to_simple` 

717 method. 

718 

719 Parameters 

720 ---------- 

721 simple : `dict` of [`str`, `Any`] 

722 The `dict` returned by `to_simple()`. 

723 universe : `DimensionUniverse` 

724 Object that manages all known dimensions. 

725 registry : `lsst.daf.butler.Registry`, optional 

726 Registry from which a universe can be extracted. Can be `None` 

727 if universe is provided explicitly. 

728 

729 Returns 

730 ------- 

731 dataId : `DataCoordinate` 

732 Newly-constructed object. 

733 """ 

734 key = (frozenset(simple.dataId.items()), simple.records is not None) 

735 cache = PersistenceContextVars.dataCoordinates.get() 

736 if cache is not None and (result := cache.get(key)) is not None: 

737 return result 

738 if universe is None and registry is None: 

739 raise ValueError("One of universe or registry is required to convert a dict to a DataCoordinate") 

740 if universe is None and registry is not None: 

741 universe = registry.dimensions 

742 if universe is None: 

743 # this is for mypy 

744 raise ValueError("Unable to determine a usable universe") 

745 

746 dataId = cls.standardize(simple.dataId, universe=universe) 

747 if simple.records: 

748 dataId = dataId.expanded( 

749 {k: DimensionRecord.from_simple(v, universe=universe) for k, v in simple.records.items()} 

750 ) 

751 if cache is not None: 

752 cache[key] = dataId 

753 return dataId 

754 

755 to_json = to_json_pydantic 

756 from_json: ClassVar[Callable[..., Self]] = cast(Callable[..., Self], classmethod(from_json_pydantic)) 

757 

758 

759DataId = DataCoordinate | Mapping[str, Any] 

760"""A type-annotation alias for signatures that accept both informal data ID 

761dictionaries and validated `DataCoordinate` instances. 

762""" 

763 

764 

765class _DataCoordinateRecordsView(Mapping[str, DimensionRecord | None]): 

766 """View class for `DataCoordinate.records`. 

767 

768 Provides the default implementation for 

769 `DataCoordinate.records`. 

770 

771 Parameters 

772 ---------- 

773 target : `DataCoordinate` 

774 The `DataCoordinate` instance this object provides a view of. 

775 """ 

776 

777 def __init__(self, target: DataCoordinate): 

778 self._target = target 

779 

780 __slots__ = ("_target",) 

781 

782 def __repr__(self) -> str: 

783 terms = [f"{d}: {self[d]!r}" for d in self._target.dimensions.elements] 

784 return "{{{}}}".format(", ".join(terms)) 

785 

786 def __str__(self) -> str: 

787 return "\n".join(str(v) for v in self.values()) 

788 

789 def __getitem__(self, key: str) -> DimensionRecord | None: 

790 return self._target._record(key) 

791 

792 def __iter__(self) -> Iterator[str]: 

793 return iter(self._target.dimensions.elements) 

794 

795 def __len__(self) -> int: 

796 return len(self._target.dimensions.elements) 

797 

798 

799class _BasicTupleDataCoordinate(DataCoordinate): 

800 """Intermediate base class for the standard implementation of 

801 `DataCoordinate`. 

802 

803 This class should only be accessed outside this module via the 

804 `DataCoordinate` interface, and should only be constructed via the static 

805 methods there. 

806 

807 Parameters 

808 ---------- 

809 dimensions : `DimensionGroup` 

810 The dimensions to be identified. 

811 values : `tuple` [ `int` or `str` ] 

812 Data ID values, ordered to match 

813 ``dimensions.data_coordinate_keys``. May include values for just 

814 required dimensions (which always come first) or all dimensions 

815 (concrete subclasses implementations will care which). 

816 """ 

817 

818 def __init__(self, dimensions: DimensionGroup, values: tuple[DataIdValue, ...]): 

819 self._dimensions = dimensions 

820 self._values = values 

821 

822 __slots__ = ("_dimensions", "_values") 

823 

824 @property 

825 def dimensions(self) -> DimensionGroup: 

826 # Docstring inherited from DataCoordinate. 

827 return self._dimensions 

828 

829 @property 

830 def required(self) -> Mapping[str, DataIdValue]: 

831 # Docstring inherited from DataCoordinate. 

832 return _DataCoordinateRequiredMappingView(self) 

833 

834 def __getitem__(self, key: str) -> DataIdValue: 

835 # Docstring inherited from DataCoordinate. 

836 

837 index = self._dimensions._data_coordinate_indices[key] 

838 try: 

839 return self._values[index] 

840 except IndexError: 

841 # Caller asked for an implied dimension, but this object only has 

842 # values for the required ones. 

843 raise KeyError(key) from None 

844 

845 def hasRecords(self) -> bool: 

846 # Docstring inherited from DataCoordinate. 

847 return False 

848 

849 def _record(self, name: str) -> DimensionRecord | None: 

850 # Docstring inherited from DataCoordinate. 

851 raise AssertionError() 

852 

853 def __getattr__(self, name: str) -> Any: 

854 if name in self.dimensions.elements: 

855 raise AttributeError( 

856 f"Dimension record attribute {name!r} is only available on expanded DataCoordinates." 

857 ) 

858 raise AttributeError(name) 

859 

860 

861class _DataCoordinateRequiredMappingView(Mapping[str, DataIdValue]): 

862 """A DataCoordinate Mapping view class whose keys are just the required 

863 dimensions. 

864 """ 

865 

866 def __init__(self, target: DataCoordinate): 

867 self._target = target 

868 

869 __slots__ = ("_target",) 

870 

871 def __getitem__(self, key: str) -> DataIdValue: 

872 if key not in self._target.dimensions.required: 

873 raise KeyError(key) 

874 return self._target[key] 

875 

876 def __len__(self) -> int: 

877 return len(self._target.dimensions.required) 

878 

879 def __iter__(self) -> Iterator[str]: 

880 return iter(self._target.dimensions.required) 

881 

882 def __repr__(self) -> str: 

883 return f"{{{', '.join(f'{k}: {v!r}' for k, v in self.items())}}}" 

884 

885 

886class _DataCoordinateFullMappingView(Mapping[str, DataIdValue]): 

887 """A DataCoordinate Mapping view class whose keys are all dimensions.""" 

888 

889 def __init__(self, target: DataCoordinate): 

890 self._target = target 

891 

892 __slots__ = ("_target",) 

893 

894 def __getitem__(self, key: str) -> DataIdValue: 

895 return self._target[key] 

896 

897 def __len__(self) -> int: 

898 return len(self._target.dimensions) 

899 

900 def __iter__(self) -> Iterator[str]: 

901 return iter(self._target.dimensions.data_coordinate_keys) 

902 

903 def __repr__(self) -> str: 

904 return f"{{{', '.join(f'{k}: {v!r}' for k, v in self.items())}}}" 

905 

906 

907class _RequiredTupleDataCoordinate(_BasicTupleDataCoordinate): 

908 """A `DataCoordinate` implementation that has values for required 

909 dimensions only, when implied dimensions already exist. 

910 

911 Note that `_FullTupleDataCoordinate` should be used if there are no 

912 implied dimensions. 

913 

914 This class should only be accessed outside this module via the 

915 `DataCoordinate` interface, and should only be constructed via calls to 

916 `DataCoordinate.from_full_values`. 

917 """ 

918 

919 __slots__ = () 

920 

921 @property 

922 def mapping(self) -> Mapping[str, DataIdValue]: 

923 # Docstring inherited from DataCoordinate. 

924 return _DataCoordinateRequiredMappingView(self) 

925 

926 @property 

927 def required_values(self) -> tuple[DataIdValue, ...]: 

928 # Docstring inherited from DataCoordinate. 

929 return self._values 

930 

931 def subset(self, dimensions: DimensionGroup | Iterable[str]) -> DataCoordinate: 

932 # Docstring inherited from DataCoordinate. 

933 dimensions = self.universe.conform(dimensions) 

934 if self._dimensions == dimensions: 

935 return self 

936 elif self._dimensions.required >= dimensions.names: 

937 return DataCoordinate.from_full_values( 

938 dimensions, 

939 tuple(self[k] for k in dimensions.data_coordinate_keys), 

940 ) 

941 else: 

942 return DataCoordinate.from_required_values( 

943 dimensions, tuple(self[k] for k in dimensions.required) 

944 ) 

945 

946 def union(self, other: DataCoordinate) -> DataCoordinate: 

947 # Docstring inherited from DataCoordinate. 

948 dimensions = self.dimensions.union(other.dimensions) 

949 # See if the other one is already what we want to return. We don't