Coverage for python / lsst / dax / apdb / apdbAdmin.py: 100%

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

21 

22from __future__ import annotations 

23 

24__all__ = ["ApdbAdmin", "DiaForcedSourceLocator", "DiaObjectLocator", "DiaSourceLocator"] 

25 

26from abc import ABC, abstractmethod 

27from collections.abc import Iterable 

28from typing import NamedTuple 

29 

30 

31class DiaObjectLocator(NamedTuple): 

32 """Collection of facts necessary to locate DiaObject.""" 

33 

34 diaObjectId: int 

35 ra: float # degrees 

36 dec: float # degrees 

37 

38 

39class DiaSourceLocator(NamedTuple): 

40 """Collection of facts necessary to locate DiaSource.""" 

41 

42 diaSourceId: int 

43 diaObjectId: int # This is not identifying, but needed for other reasons. 

44 ra: float # degrees 

45 dec: float # degrees 

46 midpointMjdTai: float 

47 

48 

49class DiaForcedSourceLocator(NamedTuple): 

50 """Collection of facts necessary to locate DiaForcedSource.""" 

51 

52 diaObjectId: int 

53 visit: int 

54 detector: int 

55 ra: float # degrees 

56 dec: float # degrees 

57 midpointMjdTai: float 

58 

59 

60class ApdbAdmin(ABC): 

61 """Abstract interface for administrative tasks for APDB.""" 

62 

63 @abstractmethod 

64 def apdb_part(self, ra_deg: float, dec_deg: float) -> int: 

65 """Return spatial partition index for given coordinates. 

66 

67 Parameters 

68 ---------- 

69 ra_deg, dec_deg : `float` 

70 Spatial coordinates, in degrees. 

71 

72 Returns 

73 ------- 

74 index : `int` 

75 Spatial partition index. 

76 """ 

77 raise NotImplementedError() 

78 

79 @abstractmethod 

80 def apdb_time_part(self, midpointMjdTai: float) -> int: 

81 """Return temporal partition index for given time. 

82 

83 Parameters 

84 ---------- 

85 midpointMjdTai : `float` 

86 Time in MJD. 

87 

88 Returns 

89 ------- 

90 index : `int` 

91 Temporal partition index. 

92 """ 

93 raise NotImplementedError() 

94 

95 @abstractmethod 

96 def delete_records( 

97 self, 

98 objects: Iterable[DiaObjectLocator], 

99 sources: Iterable[DiaSourceLocator], 

100 forced_sources: Iterable[DiaForcedSourceLocator], 

101 ) -> None: 

102 """Remove DiaObjects and all their associated DiaSources and 

103 DiaForcedSources from database. 

104 

105 Parameters 

106 ---------- 

107 objects : `~collections.abc.Iterable` [`DiaObjectLocator`] 

108 Locators for DiaObject records to remove. 

109 sources : `~collections.abc.Iterable` [`DiaSourceLocator`] 

110 Locators for DiaSource records to remove. 

111 forced_sources : `~collections.abc.Iterable` [`DiaSourceLocator`] 

112 Locators for DiaForcedSource records to remove. 

113 

114 Notes 

115 ----- 

116 The list of sources has to include all sources associated with objects 

117 in ``objects`` list. Sources that are not associated with any object 

118 from ``objects`` are not removed. 

119 """ 

120 raise NotImplementedError()