Coverage for python/lsst/daf/butler/

1# This file is part of daf_butler.

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.

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/>.

27from __future__ import annotations

29__all__ = ("Timespan",)

31import enum

32import warnings

33from collections.abc import Generator

34from typing import Any, ClassVar, TypeAlias

36import astropy.time

37import astropy.utils.exceptions

38import pydantic

39import yaml

41# As of astropy 4.2, the erfa interface is shipped independently and

42# ErfaWarning is no longer an AstropyWarning

43try:

44 import erfa

45except ImportError:

46 erfa = None

48from lsst.utils.classes import cached_getter

50from .time_utils import TimeConverter

52_ONE_DAY = astropy.time.TimeDelta("1d", scale="tai")

55class _SpecialTimespanBound(enum.Enum):

56 """Enumeration to provide a singleton value for empty timespan bounds.

58 This enum's only member should generally be accessed via the

59 `Timespan.EMPTY` alias.

60 """

62 EMPTY = enum.auto()

63 """The value used for both `Timespan.begin` and `Timespan.end` for empty

64 Timespans that contain no points.

65 """

68TimespanBound: TypeAlias = astropy.time.Time | _SpecialTimespanBound | None

71class Timespan(pydantic.BaseModel):

72 """A half-open time interval with nanosecond precision.

74 Parameters

75 ----------

76 begin : `astropy.time.Time`, `Timespan.EMPTY`, or `None`

77 Minimum timestamp in the interval (inclusive). `None` indicates that

78 the timespan has no lower bound. `Timespan.EMPTY` indicates that the

79 timespan contains no times; if this is used as either bound, the other

80 bound is ignored.

81 end : `astropy.time.Time`, `SpecialTimespanBound`, or `None`

82 Maximum timestamp in the interval (exclusive). `None` indicates that

83 the timespan has no upper bound. As with ``begin``, `Timespan.EMPTY`

84 creates an empty timespan.

85 padInstantaneous : `bool`, optional

86 If `True` (default) and ``begin == end`` *after discretization to

87 integer nanoseconds*, extend ``end`` by one nanosecond to yield a

88 finite-duration timespan. If `False`, ``begin == end`` evaluates to

89 the empty timespan.

90 _nsec : `tuple` of `int`, optional

91 Integer nanosecond representation, for internal use by `Timespan` and

92 `TimespanDatabaseRepresentation` implementation only. If provided,

93 all other arguments are are ignored.

95 Raises

96 ------

97 TypeError

98 Raised if ``begin`` or ``end`` has a type other than

99 `astropy.time.Time`, and is not `None` or `Timespan.EMPTY`.

100 ValueError

101 Raised if ``begin`` or ``end`` exceeds the minimum or maximum times

102 supported by this class.

103

104 Notes

105 -----

106 Timespans are half-open intervals, i.e. ``[begin, end)``.

107

108 Any timespan with ``begin > end`` after nanosecond discretization

109 (``begin >= end`` if ``padInstantaneous`` is `False`), or with either bound

110 set to `Timespan.EMPTY`, is transformed into the empty timespan, with both

111 bounds set to `Timespan.EMPTY`. The empty timespan is equal to itself, and

112 contained by all other timespans (including itself). It is also disjoint

113 with all timespans (including itself), and hence does not overlap any

114 timespan - this is the only case where ``contains`` does not imply

115 ``overlaps``.

116

117 Finite timespan bounds are represented internally as integer nanoseconds,

118 and hence construction from `astropy.time.Time` (which has picosecond

119 accuracy) can involve a loss of precision. This is of course

120 deterministic, so any `astropy.time.Time` value is always mapped

121 to the exact same timespan bound, but if ``padInstantaneous`` is `True`,

122 timespans that are empty at full precision (``begin > end``,

123 ``begin - end < 1ns``) may be finite after discretization. In all other

124 cases, the relationships between full-precision timespans should be

125 preserved even if the values are not.

126

127 The `astropy.time.Time` bounds that can be obtained after construction from

128 `Timespan.begin` and `Timespan.end` are also guaranteed to round-trip

129 exactly when used to construct other `Timespan` instances.

130 """

131

132 def __init__(

133 self,

134 begin: TimespanBound,

135 end: TimespanBound,

136 padInstantaneous: bool = True,

137 _nsec: tuple[int, int] | None = None,

138 ):

139 converter = TimeConverter()

140 if _nsec is None:

141 begin_nsec: int

142 if begin is None:

143 begin_nsec = converter.min_nsec

144 elif begin is self.EMPTY:

145 begin_nsec = converter.max_nsec

146 elif isinstance(begin, astropy.time.Time):

147 begin_nsec = converter.astropy_to_nsec(begin)

148 else:

149 raise TypeError(

150 f"Unexpected value of type {type(begin).__name__} for Timespan.begin: {begin!r}."

151 )

152 end_nsec: int

153 if end is None:

154 end_nsec = converter.max_nsec

155 elif end is self.EMPTY:

156 end_nsec = converter.min_nsec

157 elif isinstance(end, astropy.time.Time):

158 end_nsec = converter.astropy_to_nsec(end)

159 else:

160 raise TypeError(f"Unexpected value of type {type(end).__name__} for Timespan.end: {end!r}.")

161 if begin_nsec == end_nsec:

162 if begin_nsec == converter.max_nsec or end_nsec == converter.min_nsec:

163 with warnings.catch_warnings():

164 warnings.simplefilter("ignore", category=astropy.utils.exceptions.AstropyWarning)

165 if erfa is not None:

166 warnings.simplefilter("ignore", category=erfa.ErfaWarning)

167 if begin is not None and begin < converter.epoch:

168 raise ValueError(f"Timespan.begin may not be earlier than {converter.epoch}.")

169 if end is not None and end > converter.max_time:

170 raise ValueError(f"Timespan.end may not be later than {converter.max_time}.")

171 raise ValueError("Infinite instantaneous timespans are not supported.")

172 elif padInstantaneous:

173 end_nsec += 1

174 if end_nsec == converter.max_nsec:

175 raise ValueError(

176 f"Cannot construct near-instantaneous timespan at {end}; "

177 "within one ns of maximum time."

178 )

179 _nsec = (begin_nsec, end_nsec)

180 if _nsec[0] >= _nsec[1]:

181 # Standardizing all empty timespans to the same underlying values

182 # here simplifies all other operations (including interactions

183 # with TimespanDatabaseRepresentation implementations).

184 _nsec = (converter.max_nsec, converter.min_nsec)

185 super().__init__(nsec=_nsec)

186

187 nsec: tuple[int, int] = pydantic.Field(frozen=True)

188

189 model_config = pydantic.ConfigDict(

190 json_schema_extra={

191 "description": (

192 "A [begin, end) TAI timespan with bounds as integer nanoseconds since 1970-01-01 00:00:00."

193 )

194 }

195 )

196

197 EMPTY: ClassVar[_SpecialTimespanBound] = _SpecialTimespanBound.EMPTY

198

199 # YAML tag name for Timespan

200 yaml_tag: ClassVar[str] = "!lsst.daf.butler.Timespan"

201

202 @classmethod

203 def makeEmpty(cls) -> Timespan:

204 """Construct an empty timespan.

205

206 Returns

207 -------

208 empty : `Timespan`

209 A timespan that is contained by all timespans (including itself)

210 and overlaps no other timespans (including itself).

211 """

212 converter = TimeConverter()

213 return Timespan(None, None, _nsec=(converter.max_nsec, converter.min_nsec))

214

215 @classmethod

216 def fromInstant(cls, time: astropy.time.Time) -> Timespan:

217 """Construct a timespan that approximates an instant in time.

218

219 This is done by constructing a minimum-possible (1 ns) duration

220 timespan.

221

222 This is equivalent to ``Timespan(time, time, padInstantaneous=True)``,

223 but may be slightly more efficient.

224

225 Parameters

226 ----------

227 time : `astropy.time.Time`

228 Time to use for the lower bound.

229

230 Returns

231 -------

232 instant : `Timespan`

233 A ``[time, time + 1ns)`` timespan.

234 """

235 converter = TimeConverter()

236 nsec = converter.astropy_to_nsec(time)

237 if nsec == converter.max_nsec - 1:

238 raise ValueError(

239 f"Cannot construct near-instantaneous timespan at {time}; within one ns of maximum time."

240 )

241 return Timespan(None, None, _nsec=(nsec, nsec + 1))

242

243 @classmethod

244 def from_day_obs(cls, day_obs: int, offset: int = 0) -> Timespan:

245 """Construct a timespan for a 24-hour period based on the day of

246 observation.

247

248 Parameters

249 ----------

250 day_obs : `int`

251 The day of observation as an integer of the form YYYYMMDD.

252 The year must be at least 1970 since these are converted to TAI.

253 offset : `int`, optional

254 Offset in seconds from TAI midnight to be applied.

255

256 Returns

257 -------

258 day_span : `Timespan`

259 A timespan corresponding to a full day of observing.

260

261 Notes

262 -----

263 If the observing day is 20240229 and the offset is 12 hours the

264 resulting time span will be 2024-02-29T12:00 to 2024-03-01T12:00.

265 """

266 if day_obs < 1970_00_00 or day_obs > 1_0000_00_00:

267 raise ValueError(f"day_obs must be in form yyyyMMDD and be newer than 1970, not {day_obs}.")

268

269 ymd = str(day_obs)

270 t1 = astropy.time.Time(f"{ymd[0:4]}-{ymd[4:6]}-{ymd[6:8]}T00:00:00", format="isot", scale="tai")

271

272 if offset != 0:

273 t_delta = astropy.time.TimeDelta(offset, format="sec", scale="tai")

274 t1 += t_delta

275

276 t2 = t1 + _ONE_DAY

277

278 return Timespan(t1, t2)

279

280 @property

281 @cached_getter

282 def begin(self) -> TimespanBound:

283 """Minimum timestamp in the interval, inclusive.

284

285 If this bound is finite, this is an `astropy.time.Time` instance.

286 If the timespan is unbounded from below, this is `None`.

287 If the timespan is empty, this is the special value `Timespan.EMPTY`.

288 """

289 if self.isEmpty():

290 return self.EMPTY

291 elif self.nsec[0] == TimeConverter().min_nsec:

292 return None

293 else:

294 return TimeConverter().nsec_to_astropy(self.nsec[0])

295

296 @property

297 @cached_getter

298 def end(self) -> TimespanBound:

299 """Maximum timestamp in the interval, exclusive.

300

301 If this bound is finite, this is an `astropy.time.Time` instance.

302 If the timespan is unbounded from above, this is `None`.

303 If the timespan is empty, this is the special value `Timespan.EMPTY`.

304 """

305 if self.isEmpty():

306 return self.EMPTY

307 elif self.nsec[1] == TimeConverter().max_nsec:

308 return None

309 else:

310 return TimeConverter().nsec_to_astropy(self.nsec[1])

311

312 def isEmpty(self) -> bool:

313 """Test whether ``self`` is the empty timespan (`bool`)."""

314 return self.nsec[0] >= self.nsec[1]

315

316 def __str__(self) -> str:

317 if self.isEmpty():

318 return "(empty)"

319 fmt = "%Y-%m-%dT%H:%M:%S"

320 # Trap dubious year warnings in case we have timespans from

321 # simulated data in the future

322 with warnings.catch_warnings():

323 warnings.simplefilter("ignore", category=astropy.utils.exceptions.AstropyWarning)

324 if erfa is not None:

325 warnings.simplefilter("ignore", category=erfa.ErfaWarning)

326 if self.begin is None:

327 head = "(-∞, "

328 else:

329 assert isinstance(self.begin, astropy.time.Time), "guaranteed by earlier checks and ctor"

330 head = f"[{self.begin.tai.strftime(fmt)}, "

331 if self.end is None:

332 tail = "∞)"

333 else:

334 assert isinstance(self.end, astropy.time.Time), "guaranteed by earlier checks and ctor"

335 tail = f"{self.end.tai.strftime(fmt)})"

336 return head + tail

337

338 def __repr_astropy__(self, t: astropy.time.Time | None) -> str:

339 # Provide our own repr for astropy time.

340 # For JD times we want to use jd1 and jd2 to maintain precision.

341 if isinstance(t, astropy.time.Time):

342 if t.format == "jd":

343 return f"astropy.time.Time({t.jd1}, {t.jd2}, scale='{t.scale}', format='{t.format}')"

344 else:

345 return f"astropy.time.Time('{t}', scale='{t.scale}', format='{t.format}')"

346 return str(t)

347

348 def __repr__(self) -> str:

349 # astropy.time.Time doesn't have an eval-friendly __repr__, so we

350 # simulate our own here to make Timespan's __repr__ eval-friendly.

351 # Interestingly, enum.Enum has an eval-friendly __str__, but not an

352 # eval-friendly __repr__.

353 begin = self.__repr_astropy__(self.begin)

354 end = self.__repr_astropy__(self.end)

355 return f"Timespan(begin={begin}, end={end})"

356

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

358 if not isinstance(other, Timespan):

359 return False

360 # Correctness of this simple implementation depends on __init__

361 # standardizing all empty timespans to a single value.

362 return self.nsec == other.nsec

363

364 def __hash__(self) -> int:

365 # Correctness of this simple implementation depends on __init__

366 # standardizing all empty timespans to a single value.

367 return hash(self.nsec)

368

369 def __reduce__(self) -> tuple:

370 return (Timespan, (None, None, False, self.nsec))

371

372 def __lt__(self, other: astropy.time.Time | Timespan) -> bool:

373 """Test if a Timespan's bounds are strictly less than the given time.

374

375 Parameters

376 ----------

377 other : `Timespan` or `astropy.time.Time`.

378 Timespan or instant in time to relate to ``self``.

379

380 Returns

381 -------

382 less : `bool`

383 The result of the less-than test. `False` if either operand is

384 empty.

385 """

386 # First term in each expression below is the "normal" one; the second

387 # ensures correct behavior for empty timespans. It's important that

388 # the second uses a strict inequality to make sure inf == inf isn't in

389 # play, and it's okay for the second to use a strict inequality only

390 # because we know non-empty Timespans have nonzero duration, and hence

391 # the second term is never false for non-empty timespans unless the

392 # first term is also false.

393 if isinstance(other, astropy.time.Time):

394 nsec = TimeConverter().astropy_to_nsec(other)

395 return self.nsec[1] <= nsec and self.nsec[0] < nsec

396 else:

397 return self.nsec[1] <= other.nsec[0] and self.nsec[0] < other.nsec[1]

398

399 def __gt__(self, other: astropy.time.Time | Timespan) -> bool:

400 """Test if a Timespan's bounds are strictly greater than given time.

401

402 Parameters

403 ----------

404 other : `Timespan` or `astropy.time.Time`.

405 Timespan or instant in time to relate to ``self``.

406

407 Returns

408 -------

409 greater : `bool`

410 The result of the greater-than test. `False` if either operand is

411 empty.

412 """

413 # First term in each expression below is the "normal" one; the second

414 # ensures correct behavior for empty timespans. It's important that

415 # the second uses a strict inequality to make sure inf == inf isn't in

416 # play, and it's okay for the second to use a strict inequality only

417 # because we know non-empty Timespans have nonzero duration, and hence

418 # the second term is never false for non-empty timespans unless the

419 # first term is also false.

420 if isinstance(other, astropy.time.Time):

421 nsec = TimeConverter().astropy_to_nsec(other)

422 return self.nsec[0] > nsec and self.nsec[1] > nsec

423 else:

424 return self.nsec[0] >= other.nsec[1] and self.nsec[1] > other.nsec[0]

425

426 def overlaps(self, other: Timespan | astropy.time.Time) -> bool:

427 """Test if the intersection of this Timespan with another is empty.

428

429 Parameters

430 ----------

431 other : `Timespan` or `astropy.time.Time`

432 Timespan or time to relate to ``self``. If a single time, this is

433 a synonym for `contains`.

434

435 Returns

436 -------

437 overlaps : `bool`

438 The result of the overlap test.

439

440 Notes

441 -----

442 If either ``self`` or ``other`` is empty, the result is always `False`.

443 In all other cases, ``self.contains(other)`` being `True` implies that

444 ``self.overlaps(other)`` is also `True`.

445 """

446 if isinstance(other, astropy.time.Time):

447 return self.contains(other)

448 return self.nsec[1] > other.nsec[0] and other.nsec[1] > self.nsec[0]

449

450 def contains(self, other: astropy.time.Time | Timespan) -> bool:

451 """Test if the supplied timespan is within this one.

452

453 Tests whether the intersection of this timespan with another timespan

454 or point is equal to the other one.

455

456 Parameters

457 ----------

458 other : `Timespan` or `astropy.time.Time`

459 Timespan or instant in time to relate to ``self``.

460

461 Returns

462 -------

463 overlaps : `bool`

464 The result of the contains test.

465

466 Notes

467 -----

468 If ``other`` is empty, `True` is always returned. In all other cases,

469 ``self.contains(other)`` being `True` implies that

470 ``self.overlaps(other)`` is also `True`.

471

472 Testing whether an instantaneous `astropy.time.Time` value is contained

473 in a timespan is not equivalent to testing a timespan constructed via

474 `Timespan.fromInstant`, because Timespan cannot exactly represent

475 zero-duration intervals. In particular, ``[a, b)`` contains the time

476 ``b``, but not the timespan ``[b, b + 1ns)`` that would be returned

477 by ``Timespan.fromInstant(b)``.

478 """

479 if isinstance(other, astropy.time.Time):

480 nsec = TimeConverter().astropy_to_nsec(other)

481 return self.nsec[0] <= nsec and self.nsec[1] > nsec

482 else:

483 return self.nsec[0] <= other.nsec[0] and self.nsec[1] >= other.nsec[1]

484

485 def intersection(self, *args: Timespan) -> Timespan:

486 """Return a new `Timespan` that is contained by all of the given ones.

487

488 Parameters

489 ----------

490 *args

491 All positional arguments are `Timespan` instances.

492

493 Returns

494 -------

495 intersection : `Timespan`

496 The intersection timespan.

497 """

498 if not args:

499 return self

500 lowers = [self.nsec[0]]

501 lowers.extend(ts.nsec[0] for ts in args)

502 uppers = [self.nsec[1]]

503 uppers.extend(ts.nsec[1] for ts in args)

504 nsec = (max(*lowers), min(*uppers))

505 return Timespan(begin=None, end=None, _nsec=nsec)

506

507 def difference(self, other: Timespan) -> Generator[Timespan, None, None]:

508 """Return the one or two timespans that cover the interval(s).

509

510 The interval is defined as one that is in ``self`` but not ``other``.

511

512 This is implemented as a generator because the result may be zero, one,

513 or two `Timespan` objects, depending on the relationship between the

514 operands.

515

516 Parameters

517 ----------

518 other : `Timespan`

519 Timespan to subtract.

520

521 Yields

522 ------

523 result : `Timespan`

524 A `Timespan` that is contained by ``self`` but does not overlap

525 ``other``. Guaranteed not to be empty.

526 """

527 intersection = self.intersection(other)

528 if intersection.isEmpty():

529 yield self

530 elif intersection == self:

531 yield from ()

532 else:

533 if intersection.nsec[0] > self.nsec[0]:

534 yield Timespan(None, None, _nsec=(self.nsec[0], intersection.nsec[0]))

535 if intersection.nsec[1] < self.nsec[1]:

536 yield Timespan(None, None, _nsec=(intersection.nsec[1], self.nsec[1]))

537

538 @classmethod

539 def to_yaml(cls, dumper: yaml.Dumper, timespan: Timespan) -> Any:

540 """Convert Timespan into YAML format.

541

542 This produces a scalar node with a tag "!_SpecialTimespanBound" and

543 value being a name of _SpecialTimespanBound enum.

544

545 Parameters

546 ----------

547 dumper : `yaml.Dumper`

548 YAML dumper instance.

549 timespan : `Timespan`

550 Data to be converted.

551 """

552 if timespan.isEmpty():

553 return dumper.represent_scalar(cls.yaml_tag, "EMPTY")

554 else:

555 return dumper.represent_mapping(

556 cls.yaml_tag,

557 dict(begin=timespan.begin, end=timespan.end),

558 )

559

560 @classmethod

561 def from_yaml(cls, loader: yaml.SafeLoader, node: yaml.MappingNode) -> Timespan | None:

562 """Convert YAML node into _SpecialTimespanBound.

563

564 Parameters

565 ----------

566 loader : `yaml.SafeLoader`

567 Instance of YAML loader class.

568 node : `yaml.ScalarNode`

569 YAML node.

570

571 Returns

572 -------

573 value : `Timespan`

574 Timespan instance, can be ``None``.

575 """

576 if node.value is None:

577 return None

578 elif node.value == "EMPTY":

579 return Timespan.makeEmpty()

580 else:

581 d = loader.construct_mapping(node)

582 return Timespan(d["begin"], d["end"])

583

584 @pydantic.model_validator(mode="before")

585 @classmethod

586 def _validate(cls, value: Any) -> Any:

587 if isinstance(value, Timespan):

588 return value

589 if isinstance(value, dict):

590 return value

591 return {"nsec": value}

592

593 @pydantic.model_serializer(mode="plain")

594 def _serialize(self) -> tuple[int, int]:

595 return self.nsec

596

597

598# Register Timespan -> YAML conversion method with Dumper class

599yaml.Dumper.add_representer(Timespan, Timespan.to_yaml)

600

601# Register YAML -> Timespan conversion method with Loader, for our use case we

602# only need SafeLoader.

603yaml.SafeLoader.add_constructor(Timespan.yaml_tag, Timespan.from_yaml)

Coverage for python / lsst / daf / butler / _timespan.py: 21%

201 statements