Coverage for python / lsst / daf / butler / time_utils.py: 37%
95 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-18 08:43 +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/>.
27from __future__ import annotations
29__all__ = ("TimeConverter",)
31import logging
32import warnings
33from typing import Any, ClassVar
35import astropy.time
36import astropy.time.formats
37import astropy.utils.exceptions
38import numpy
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 Singleton
50_LOG = logging.getLogger(__name__)
53class _FastTimeUnixTai(astropy.time.formats.TimeUnixTai):
54 """Special astropy time format that skips some checks of the parameters.
55 This format is only used internally by TimeConverter so we can trust that
56 it passes correct values to astropy Time class.
57 """
59 # number of seconds in a day
60 _SEC_PER_DAY: ClassVar[int] = 24 * 3600
62 # Name of this format, it is registered in astropy formats registry.
63 name = "unix_tai_fast"
65 def _check_val_type(self, val1: Any, val2: Any) -> tuple:
66 # We trust everything that is passed to us.
67 return val1, val2
69 def set_jds(self, val1: numpy.ndarray, val2: numpy.ndarray | None) -> None:
70 # Epoch time format is TimeISO with scalar jd1/jd2 arrays, convert them
71 # to floats to speed things up.
72 epoch = self._epoch._time
73 jd1, jd2 = float(epoch._jd1), float(epoch._jd2)
75 assert val1.ndim == 0, "Expect scalar"
76 whole_days, seconds = divmod(float(val1), self._SEC_PER_DAY)
77 if val2 is not None:
78 assert val2.ndim == 0, "Expect scalar"
79 seconds += float(val2)
81 jd1 += whole_days
82 jd2 += seconds / self._SEC_PER_DAY
83 while jd2 > 0.5:
84 jd2 -= 1.0
85 jd1 += 1.0
87 self._jd1, self._jd2 = numpy.array(jd1), numpy.array(jd2)
90class TimeConverter(metaclass=Singleton):
91 """A singleton for mapping TAI times to integer nanoseconds.
93 This class allows some time calculations to be deferred until first use,
94 rather than forcing them to happen at module import time.
95 """
97 def __init__(self) -> None:
98 # EPOCH is used to convert from nanoseconds; its precision is used by
99 # all timestamps returned by nsec_to_astropy, and we set it to 1
100 # microsecond.
101 self.epoch = astropy.time.Time("1970-01-01 00:00:00", format="iso", scale="tai", precision=6)
102 self.max_time = astropy.time.Time("2100-01-01 00:00:00", format="iso", scale="tai")
103 self.min_nsec = 0
104 self.max_nsec = self.astropy_to_nsec(self.max_time)
106 def astropy_to_nsec(self, astropy_time: astropy.time.Time) -> int:
107 """Convert astropy time to nanoseconds since epoch.
109 Input time is converted to TAI scale before conversion to
110 nanoseconds.
112 Parameters
113 ----------
114 astropy_time : `astropy.time.Time`
115 Time to be converted.
117 Returns
118 -------
119 time_nsec : `int`
120 Nanoseconds since epoch.
122 Notes
123 -----
124 Only the limited range of input times is supported by this method as it
125 is defined useful in the context of Butler and Registry. If input time
126 is earlier than ``min_time`` then this method returns ``min_nsec``. If
127 input time comes after ``max_time`` then it returns ``max_nsec``.
128 """
129 # sometimes comparison produces warnings if input value is in UTC
130 # scale, transform it to TAI before doing anything but also trap
131 # warnings in case we are dealing with simulated data from the future
132 with warnings.catch_warnings():
133 warnings.simplefilter("ignore", category=astropy.utils.exceptions.AstropyWarning)
134 if erfa is not None:
135 warnings.simplefilter("ignore", category=erfa.ErfaWarning)
136 value = astropy_time.tai
137 # anything before epoch or after max_time is truncated
138 if value < self.epoch:
139 _LOG.warning(
140 "'%s' is earlier than epoch time '%s', epoch time will be used instead",
141 astropy_time,
142 self.epoch,
143 )
144 value = self.epoch
145 elif value > self.max_time:
146 _LOG.warning(
147 "'%s' is later than max. time '%s', max. time time will be used instead", value, self.max_time
148 )
149 value = self.max_time
151 delta = value - self.epoch
152 # Special care needed to preserve nanosecond precision.
153 # Usually jd1 has no fractional part but just in case.
154 # Can use "internal" ._time.jd1 interface because we know that both
155 # are TAI. This is a few percent faster than using .jd1.
156 jd1, extra_jd2 = divmod(delta._time.jd1, 1)
157 value = int(jd1) * self._NSEC_PER_DAY + int(round((delta._time.jd2 + extra_jd2) * self._NSEC_PER_DAY))
158 return value
160 def nsec_to_astropy(self, time_nsec: int) -> astropy.time.Time:
161 """Convert nanoseconds since epoch to astropy time.
163 Parameters
164 ----------
165 time_nsec : `int`
166 Nanoseconds since epoch.
168 Returns
169 -------
170 astropy_time : `astropy.time.Time`
171 Time to be converted.
173 Notes
174 -----
175 Usually the input time for this method is the number returned from
176 `astropy_to_nsec` which has a limited range. This method does not check
177 that the number falls in the supported range and can produce output
178 time that is outside of that range.
179 """
180 jd1, jd2 = divmod(time_nsec, 1_000_000_000)
181 time = astropy.time.Time(
182 float(jd1), jd2 / 1_000_000_000, format="unix_tai_fast", scale="tai", precision=6
183 )
184 # Force the format to be something more obvious to external users.
185 # There is a small overhead doing this but it does avoid confusion.
186 time.format = "jd"
187 return time
189 def times_equal(
190 self, time1: astropy.time.Time, time2: astropy.time.Time, precision_nsec: float = 1.0
191 ) -> bool:
192 """Check that times are equal within specified precision.
194 Parameters
195 ----------
196 time1, time2 : `astropy.time.Time`
197 Times to compare.
198 precision_nsec : `float`, optional
199 Precision to use for comparison in nanoseconds, default is one
200 nanosecond which is larger that round-trip error for conversion
201 to/from integer nanoseconds.
202 """
203 # To compare we need them in common scale, for simplicity just
204 # bring them both to TAI scale
205 # Hide any warnings from this conversion since they are not relevant
206 # to the equality
207 with warnings.catch_warnings():
208 warnings.simplefilter("ignore", category=astropy.utils.exceptions.AstropyWarning)
209 if erfa is not None:
210 warnings.simplefilter("ignore", category=erfa.ErfaWarning)
211 time1 = time1.tai
212 time2 = time2.tai
213 delta = (time2.jd1 - time1.jd1) + (time2.jd2 - time1.jd2)
214 delta *= self._NSEC_PER_DAY
215 return abs(delta) < precision_nsec
217 # number of nanoseconds in a day
218 _NSEC_PER_DAY: ClassVar[int] = 1_000_000_000 * 24 * 3600
220 epoch: astropy.time.Time
221 """Epoch for calculating time delta, this is the minimum time that can be
222 stored in the database.
223 """
225 max_time: astropy.time.Time
226 """Maximum time value that the converter can handle (`astropy.time.Time`).
228 Assuming 64-bit integer field we can actually store higher values but we
229 intentionally limit it to arbitrary but reasonably high value. Note that
230 this value will be stored in registry database for eternity, so it should
231 not be changed without proper consideration.
232 """
234 min_nsec: int
235 """Minimum value returned by `astropy_to_nsec`, corresponding to
236 `epoch` (`int`).
237 """
239 max_nsec: int
240 """Maximum value returned by `astropy_to_nsec`, corresponding to
241 `max_time` (`int`).
242 """
245class _AstropyTimeToYAML:
246 """Handle conversion of astropy Time to/from YAML representation.
248 This class defines methods that convert astropy Time instances to or from
249 YAML representation. On output it converts time to string ISO format in
250 TAI scale with maximum precision defining special YAML tag for it. On
251 input it does inverse transformation. The methods need to be registered
252 with YAML dumper and loader classes.
254 Notes
255 -----
256 Python ``yaml`` module defines special helper base class ``YAMLObject``
257 that provides similar functionality but its use is complicated by the need
258 to convert ``Time`` instances to instances of ``YAMLObject`` sub-class
259 before saving them to YAML. This class avoids this intermediate step but
260 it requires separate regisration step.
261 """
263 yaml_tag = "!butler_time/tai/iso" # YAML tag name for Time class
265 @classmethod
266 def to_yaml(cls, dumper: yaml.Dumper, data: astropy.time.Time) -> Any:
267 """Convert astropy Time object into YAML format.
269 Parameters
270 ----------
271 dumper : `yaml.Dumper`
272 YAML dumper instance.
273 data : `astropy.time.Time`
274 Data to be converted.
275 """
276 if data is not None:
277 # we store time in ISO format but we need full nanosecond
278 # precision so we have to construct intermediate instance to make
279 # sure its precision is set correctly.
280 data = astropy.time.Time(data.tai, precision=9)
281 data = data.to_value("iso")
282 return dumper.represent_scalar(cls.yaml_tag, data)
284 @classmethod
285 def from_yaml(cls, loader: yaml.SafeLoader, node: yaml.ScalarNode) -> astropy.time.Time:
286 """Convert YAML node into astropy time.
288 Parameters
289 ----------
290 loader : `yaml.SafeLoader`
291 Instance of YAML loader class.
292 node : `yaml.ScalarNode`
293 YAML node.
295 Returns
296 -------
297 time : `astropy.time.Time`
298 Time instance, can be ``None``.
299 """
300 if node.value is not None:
301 return astropy.time.Time(node.value, format="iso", scale="tai")
304# Register Time -> YAML conversion method with Dumper class
305yaml.Dumper.add_representer(astropy.time.Time, _AstropyTimeToYAML.to_yaml)
307# Register YAML -> Time conversion method with Loader, for our use case we
308# only need SafeLoader.
309yaml.SafeLoader.add_constructor(_AstropyTimeToYAML.yaml_tag, _AstropyTimeToYAML.from_yaml)