Coverage for python/lsst/dax/apdb/apdb.py: 100%
69 statements
« prev ^ index » next coverage.py v7.17.2, created at 2026-11-07 09:42 +0000
1# This file is part of dax_apdb.
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 program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ["Apdb", "ApdbConfig"]
26import logging
27from abc import ABC, abstractmethod
28from collections.abc import Iterable, Mapping
29from typing import TYPE_CHECKING
31import astropy.time
32import pandas
34from lsst.resources import ResourcePathExpression
35from lsst.sphgeom import Region
37from .apdbSchema import ApdbSchema, ApdbTables
38from .config import ApdbConfig
39from .factory import make_apdb
40from .recordIds import DiaForcedSourceId, DiaObjectId, DiaSourceId
41from .schema_model import Table
43if TYPE_CHECKING:
44 from .apdbAdmin import ApdbAdmin
45 from .apdbMetadata import ApdbMetadata
47_LOG = logging.getLogger(__name__)
50class Apdb(ABC):
51 """Abstract interface for APDB."""
53 @classmethod
54 def from_config(cls, config: ApdbConfig) -> Apdb:
55 """Create Ppdb instance from configuration object.
57 Parameters
58 ----------
59 config : `ApdbConfig`
60 Configuration object, type of this object determines type of the
61 Apdb implementation.
63 Returns
64 -------
65 apdb : `apdb`
66 Instance of `Apdb` class.
67 """
68 return make_apdb(config)
70 @classmethod
71 def from_uri(cls, uri: ResourcePathExpression) -> Apdb:
72 """Make Apdb instance from a serialized configuration.
74 Parameters
75 ----------
76 uri : `~lsst.resources.ResourcePathExpression`
77 URI or local file path pointing to a file with serialized
78 configuration, or a string with a "label:" prefix. In the latter
79 case, the configuration will be looked up from an APDB index file
80 using the label name that follows the prefix. The APDB index file's
81 location is determined by the ``DAX_APDB_INDEX_URI`` environment
82 variable.
84 Returns
85 -------
86 apdb : `apdb`
87 Instance of `Apdb` class, the type of the returned instance is
88 determined by configuration.
89 """
90 config = ApdbConfig.from_uri(uri)
91 return make_apdb(config)
93 @abstractmethod
94 def close(self) -> None:
95 """Release all resources associated with this Apdb instance. The
96 instance may no longer be used after this is called.
97 """
98 raise NotImplementedError()
100 @abstractmethod
101 def getConfig(self) -> ApdbConfig:
102 """Return APDB configuration for this instance, including any updates
103 that may be read from database.
105 Returns
106 -------
107 config : `ApdbConfig`
108 APDB configuration.
109 """
110 raise NotImplementedError()
112 @abstractmethod
113 def tableDef(self, table: ApdbTables) -> Table | None:
114 """Return table schema definition for a given table.
116 Parameters
117 ----------
118 table : `ApdbTables`
119 One of the known APDB tables.
121 Returns
122 -------
123 tableSchema : `.schema_model.Table` or `None`
124 Table schema description, `None` is returned if table is not
125 defined by this implementation.
126 """
127 raise NotImplementedError()
129 @abstractmethod
130 def getDiaObjects(self, region: Region) -> pandas.DataFrame:
131 """Return catalog of DiaObject instances from a given region.
133 This method returns only the last version of each DiaObject,
134 and may return only the subset of the DiaObject columns needed
135 for AP association. Some
136 records in a returned catalog may be outside the specified region, it
137 is up to a client to ignore those records or cleanup the catalog before
138 futher use.
140 Parameters
141 ----------
142 region : `lsst.sphgeom.Region`
143 Region to search for DIAObjects.
145 Returns
146 -------
147 catalog : `pandas.DataFrame`
148 Catalog containing DiaObject records for a region that may be a
149 superset of the specified region.
150 """
151 raise NotImplementedError()
153 @abstractmethod
154 def getDiaSources(
155 self,
156 region: Region,
157 object_ids: Iterable[int] | None,
158 visit_time: astropy.time.Time,
159 start_time: astropy.time.Time | None = None,
160 ) -> pandas.DataFrame | None:
161 """Return catalog of DiaSource instances from a given region.
163 Parameters
164 ----------
165 region : `lsst.sphgeom.Region`
166 Region to search for DIASources.
167 object_ids : iterable [ `int` ], optional
168 List of DiaObject IDs to further constrain the set of returned
169 sources. If `None` then returned sources are not constrained. If
170 list is empty then empty catalog is returned with a correct
171 schema.
172 visit_time : `astropy.time.Time`
173 Time of the current visit. If APDB contains records later than this
174 time they may also be returned.
175 start_time : `astropy.time.Time`, optional
176 Lower bound of time window for the query. If not specified then
177 it is calculated using ``visit_time`` and
178 ``read_forced_sources_months`` configuration parameter.
180 Returns
181 -------
182 catalog : `pandas.DataFrame`, or `None`
183 Catalog containing DiaSource records. `None` is returned if
184 ``start_time`` is not specified and ``read_sources_months``
185 configuration parameter is set to 0.
187 Notes
188 -----
189 This method returns DiaSource catalog for a region with additional
190 filtering based on DiaObject IDs. Only a subset of DiaSource history
191 is returned limited by ``read_sources_months`` config parameter, w.r.t.
192 ``visit_time``. If ``object_ids`` is empty then an empty catalog is
193 always returned with the correct schema (columns/types). If
194 ``object_ids`` is `None` then no filtering is performed and some of the
195 returned records may be outside the specified region.
196 """
197 raise NotImplementedError()
199 @abstractmethod
200 def getDiaForcedSources(
201 self,
202 region: Region,
203 object_ids: Iterable[int] | None,
204 visit_time: astropy.time.Time,
205 start_time: astropy.time.Time | None = None,
206 ) -> pandas.DataFrame | None:
207 """Return catalog of DiaForcedSource instances from a given region.
209 Parameters
210 ----------
211 region : `lsst.sphgeom.Region`
212 Region to search for DIASources.
213 object_ids : iterable [ `int` ], optional
214 List of DiaObject IDs to further constrain the set of returned
215 sources. If list is empty then empty catalog is returned with a
216 correct schema. If `None` then returned sources are not
217 constrained.
218 visit_time : `astropy.time.Time`
219 Time of the current visit. If APDB contains records later than this
220 time they may also be returned.
221 start_time : `astropy.time.Time`, optional
222 Lower bound of time window for the query. If not specified then
223 it is calculated using ``visit_time`` and
224 ``read_forced_sources_months`` configuration parameter.
226 Returns
227 -------
228 catalog : `pandas.DataFrame`, or `None`
229 Catalog containing DiaForcedSource records. `None` is returned if
230 ``start_time`` is not specified and ``read_forced_sources_months``
231 configuration parameter is set to 0.
233 Raises
234 ------
235 NotImplementedError
236 May be raised by some implementations if ``object_ids`` is `None`.
238 Notes
239 -----
240 This method returns DiaForcedSource catalog for a region with
241 additional filtering based on DiaObject IDs. Only a subset of DiaSource
242 history is returned limited by ``read_forced_sources_months`` config
243 parameter, w.r.t. ``visit_time``. If ``object_ids`` is empty then an
244 empty catalog is always returned with the correct schema
245 (columns/types). If ``object_ids`` is `None` then no filtering is
246 performed and some of the returned records may be outside the specified
247 region.
248 """
249 raise NotImplementedError()
251 @abstractmethod
252 def getDiaObjectsForDedup(self, since: astropy.time.Time | None = None) -> pandas.DataFrame:
253 """Return catalog of DiaObject stored in APDB since specified time.
255 This method should be used by deduplication algorithm to retrieve
256 DiaObject records added to APDB since previous deduplication (typically
257 during previous night). Returned catalog will have only a small subset
258 of DiaObject attributes required by deduplication algorithm.
260 Parameters
261 ----------
262 since : `astropy.time.Time`, optional
263 Starting search time (time of previous deduplication). If not
264 provided the time of the last deduplication stored in metadata
265 by `resetDedup` method is used.
267 Returns
268 -------
269 catalog : `pandas.DataFrame`
270 Catalog containing DiaObject records, only a subset of attributes
271 will be returned.
272 """
273 raise NotImplementedError()
275 @abstractmethod
276 def getDiaSourcesForDiaObjects(
277 self, objects: list[DiaObjectId], start_time: astropy.time.Time, max_dist_arcsec: float = 1.0
278 ) -> pandas.DataFrame:
279 """Return catalog of DiaSources associated with given DiaObjects.
281 Parameters
282 ----------
283 objects : `list` [`DiaObjectId`]
284 DiaObjects associated with returned DiaSources.
285 start_time : `astropy.time.Time`
286 Lower bound for ``midpointMjdTai`` for returned DiaSources.
287 max_dist_arcsec : `float`
288 Maximum expected distance in arcsec between DiaSource and
289 DiaObject. This parameter is used to optimize spatial queries in
290 cases when DiaObject is located near the partition boundary. If the
291 distance from DiaObject to the boundary is smaller than
292 ``max_dist_arcsec``, then the neighbor partition will be included
293 in search too.
295 Returns
296 -------
297 catalog : `pandas.DataFrame`
298 Catalog containing DiaSource records associated to given
299 DiaObjects.
301 Notes
302 -----
303 Primary purpose of this method is to support deduplication algorithm.
304 Its implementation is likely to be very slow and inefficient, it should
305 not be used for regular queries.
306 """
307 raise NotImplementedError()
309 @abstractmethod
310 def containsVisitDetector(
311 self,
312 visit: int,
313 detector: int,
314 region: Region | None = None,
315 visit_time: astropy.time.Time | None = None,
316 ) -> bool:
317 """Test whether any sources for a given visit-detector are present in
318 the APDB.
320 Parameters
321 ----------
322 visit, detector : `int`
323 The ID of the visit-detector to search for.
324 region : `lsst.sphgeom.Region`, optional
325 Deprecated - parameter is not used. Region corresponding to the
326 visit/detector combination.
327 visit_time : `astropy.time.Time`, optional
328 Deprecated - parameter is not used. Visit time (as opposed to visit
329 processing time). This can be any timestamp in the visit timespan,
330 e.g. its begin or end time.
332 Returns
333 -------
334 present : `bool`
335 `True` if given visit/detector combination was stored in APDB,
336 `False` otherwise.
337 """
338 raise NotImplementedError()
340 @abstractmethod
341 def store(
342 self,
343 visit_time: astropy.time.Time,
344 objects: pandas.DataFrame,
345 sources: pandas.DataFrame | None = None,
346 forced_sources: pandas.DataFrame | None = None,
347 ) -> None:
348 """Store all three types of catalogs in the database.
350 Parameters
351 ----------
352 visit_time : `astropy.time.Time`
353 Time of the visit.
354 objects : `pandas.DataFrame`
355 Catalog with DiaObject records.
356 sources : `pandas.DataFrame`, optional
357 Catalog with DiaSource records.
358 forced_sources : `pandas.DataFrame`, optional
359 Catalog with DiaForcedSource records.
361 Notes
362 -----
363 This methods takes DataFrame catalogs, their schema must be
364 compatible with the schema of APDB table:
366 - column names must correspond to database table columns
367 - types and units of the columns must match database definitions,
368 no unit conversion is performed presently
369 - columns that have default values in database schema can be
370 omitted from catalog
371 - this method knows how to fill interval-related columns of DiaObject
372 (validityStart, validityEnd) they do not need to appear in a
373 catalog
374 - source catalogs have ``diaObjectId`` column associating sources
375 with objects
377 This operation need not be atomic, but DiaSources and DiaForcedSources
378 will not be stored until all DiaObjects are stored.
379 """
380 raise NotImplementedError()
382 @abstractmethod
383 def reassignDiaSourcesToDiaObjects(
384 self,
385 idMap: Mapping[DiaSourceId, int],
386 *,
387 increment_nDiaSources: bool = True,
388 decrement_nDiaSources: bool = True,
389 ) -> None:
390 """Re-assign DiaSources from one DiaObject to another, typically
391 during deduplication.
393 Parameters
394 ----------
395 idMap : `~collections.abc.Mapping` [`DiaSourceId`, `int`]
396 Mapping from DiaSource to their new ``diaObjectId``.
397 increment_nDiaSources : `bool`, optional
398 If `True` then increment the value of ``nDiaSources`` in DiaObjects
399 that DiaSources are reassigned to.
400 decrement_nDiaSources : `bool`, optional
401 If `True` then decrement the value of ``nDiaSources`` in DiaObjects
402 that DiaSources are reassigned from.
404 Raises
405 ------
406 LookupError
407 Raised if some of DiaSources or DiaObjects are not found.
409 Notes
410 -----
411 DiaSources initially could be associated with SSObjects. This method
412 needs to be called before `setValidityEnd`.
413 """
414 raise NotImplementedError()
416 @abstractmethod
417 def setValidityEnd(
418 self, objects: list[DiaObjectId], validityEnd: astropy.time.Time, raise_on_missing_id: bool = False
419 ) -> int:
420 """Close validity interval for specified DiaObjects.
422 Parameters
423 ----------
424 objects : `list` [`DiaObjectId`]
425 DiaObjects which will have their validityEnd updated, if their
426 current validityEnd is NULL.
427 validityEnd : `astropy.time.Time`
428 Value for validityEnd.
429 raise_on_missing_id : `bool`, optional
430 If `True` then `LookupError` will be raised if any object in the
431 list is missing from the database.
433 Returns
434 -------
435 count : `int`
436 Actual number of records for which validityEnd was updated.
438 Raises
439 ------
440 LookupError
441 Raised if ``raise_on_missing_id`` is `True` and some of the
442 specified DiaObjects could not be found in the database.
444 Notes
445 -----
446 This method has to be called after `reassignDiaSourcesToDiaObjects`.
447 """
448 raise NotImplementedError()
450 @abstractmethod
451 def resetDedup(self, dedup_time: astropy.time.Time | None = None) -> None:
452 """Delete deduplication-related data and remember deduplication time.
453 Deduplication data generated before ``dedup_time`` will be erased.
455 Parameters
456 ----------
457 dedup_time : `astropy.time.Time`, optional
458 Time of the last deduplication, current time is used if not
459 provided.
460 """
461 raise NotImplementedError()
463 @abstractmethod
464 def reassignDiaSources(self, idMap: Mapping[int, int]) -> None:
465 """Associate DiaSources with SSObjects, dis-associating them
466 from DiaObjects.
468 Parameters
469 ----------
470 idMap : `Mapping`
471 Maps DiaSource IDs to their new SSObject IDs.
473 Raises
474 ------
475 ValueError
476 Raised if DiaSource ID does not exist in the database.
477 """
478 raise NotImplementedError()
480 @abstractmethod
481 def withdrawDiaSources(
482 self,
483 diaSourceIds: Iterable[DiaSourceId],
484 *,
485 timeWithdrawn: astropy.time.Time | None = None,
486 ) -> None:
487 """Withdraw DiaSources.
489 Parameters
490 ----------
491 diaSourceIds : `~collections.abc.Iterable` [`DiaSourceId`]
492 Identifiers of DiaSources to withdraw.
493 timeWithdrawn : `astropy.time.Time`, optional
494 Set the value of ``time_withdrawn`` column to this time, current
495 time by default.
497 Raises
498 ------
499 LookupError
500 Raised if some of DiaSources are not found.
501 """
502 raise NotImplementedError()
504 @abstractmethod
505 def withdrawDiaForcedSources(
506 self,
507 diaForcedSourceIds: Iterable[DiaForcedSourceId],
508 *,
509 timeWithdrawn: astropy.time.Time | None = None,
510 ) -> None:
511 """Withdraw DiaForcedSources.
513 Parameters
514 ----------
515 diaForcedSourceIds : `~collections.abc.Iterable` [`DiaForcedSourceId`]
516 Identifiers of DiaForcedSources to withdraw.
517 timeWithdrawn : `astropy.time.Time`, optional
518 Set the value of ``time_withdrawn`` column to this time, current
519 time by default.
521 Raises
522 ------
523 LookupError
524 Raised if some of DiaForcedSources are not found.
525 """
526 raise NotImplementedError()
528 @abstractmethod
529 def countUnassociatedObjects(self) -> int:
530 """Return the number of DiaObjects that have only one DiaSource
531 associated with them.
533 Used as part of ap_verify metrics.
535 Returns
536 -------
537 count : `int`
538 Number of DiaObjects with exactly one associated DiaSource.
540 Notes
541 -----
542 This method can be very inefficient or slow in some implementations.
543 """
544 raise NotImplementedError()
546 @property
547 @abstractmethod
548 def schema(self) -> ApdbSchema:
549 """APDB table schema from ``sdm_schemas`` (`ApdbSchema`)."""
550 raise NotImplementedError()
552 @property
553 @abstractmethod
554 def metadata(self) -> ApdbMetadata:
555 """Object controlling access to APDB metadata (`ApdbMetadata`)."""
556 raise NotImplementedError()
558 @property
559 @abstractmethod
560 def admin(self) -> ApdbAdmin:
561 """Object providing adminitrative interface for APDB (`ApdbAdmin`)."""
562 raise NotImplementedError()
564 def _current_time(self) -> astropy.time.Time:
565 """Return current system time.
567 Returns
568 -------
569 current_time : `astropy.time.Time`
570 Current time.
572 Notes
573 -----
574 This method exists primarily for testing purposes, it can be
575 monkey-patched in unit tests to return something else than current
576 system time, if necessary.
577 """
578 return astropy.time.Time.now()