Coverage for middle_layer/solicit/domain_layer/entities/capability.py: 94.37%

1# Copyright 2024 Associated Universities, Inc. 

2# 

3# This file is part of Telescope Time Allocation Tools (TTAT). 

4# 

5# TTAT is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# any later version. 

9# 

10# TTAT is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

15# You should have received a copy of the GNU General Public License 

16# along with TTAT. If not, see <https://www.gnu.org/licenses/>. 

17# 

18# pyright: reportImportCycles=false 

19from collections.abc import Sequence 

20from enum import Enum 

21from typing import TYPE_CHECKING, TypedDict, TypeVar, override 

22 

23import sqlalchemy as sa 

24from sqlalchemy import Column, ForeignKey, Integer, Table 

25from sqlalchemy.orm import Mapped, mapped_column, relationship 

26 

27from common.domain_layer import JSON_OBJECT 

28from common.domain_layer.entities.base import Base 

29from solicit.domain_layer.entities.array_configuration import ArrayConfiguration 

30from solicit.domain_layer.entities.backend import Backend 

31from solicit.domain_layer.entities.external_joint_parameter import ExternalJointParameter 

32from solicit.domain_layer.entities.frontend import Frontend 

33from solicit.domain_layer.entities.station import Station 

34 

35if TYPE_CHECKING: 

36 from allocate.domain_layer.entities.allocation_version import AllocationVersion 

37 from allocate.domain_layer.entities.available_time_model_version import AvailableTimeModelVersion 

38 from closeout.domain_layer.entities.prototype_project import PrototypeProject 

39 from solicit.domain_layer.entities.solicitation_facility_capability import SolicitationFacilityCapability 

40 

41 

42T = TypeVar("T", bound=Base) 

43 

44 

45external_joint_required_facilities = Table( 

46 "external_joint_required_facilities", 

47 Base.metadata, 

48 Column( 

49 "external_joint_parameter_id", 

50 Integer, 

51 ForeignKey("external_joint_parameters.external_joint_parameter_id", ondelete="cascade"), 

52 nullable=False, 

53 ), 

54 Column("facility_id", Integer, ForeignKey("facilities.facility_id", ondelete="cascade"), nullable=False), 

55 extend_existing=True, 

56) 

57 

58 

59class Facility(Base): 

60 """ 

61 A resource that the NRAO makes available for Authors to write Proposals to use 

62 Relative to a Solicitation, for the moment 

63 """ 

64 

65 __tablename__: str = "facilities" 

66 

67 # Unique identifier from backend 

68 facility_id: Mapped[int] = mapped_column(primary_key=True) 

69 facility_name: Mapped[str] = mapped_column(nullable=False) 

70 is_active: Mapped[bool] = mapped_column(nullable=False, default=True) 

71 description: Mapped[str] = mapped_column(nullable=True) 

72 requested_amount_unit: Mapped[str] = mapped_column() 

73 

74 allocation_versions: Mapped[list["AllocationVersion"]] = relationship( 

75 "AllocationVersion", back_populates="facility", cascade="all, delete" 

76 ) 

77 allocation_requests: Mapped[list["AllocationVersion"]] = relationship( 

78 "AllocationRequest", back_populates="facility", cascade="all, delete" 

79 ) 

80 available_time_model_versions: Mapped[list["AvailableTimeModelVersion"]] = relationship( 

81 "AvailableTimeModelVersion", back_populates="facility", cascade="all, delete" 

82 ) 

83 prototype_projects: Mapped[list["PrototypeProject"]] = relationship( 

84 "PrototypeProject", back_populates="facility", cascade="all, delete", passive_deletes=True 

85 ) 

86 solicitation_facility_capabilities: Mapped[list["SolicitationFacilityCapability"]] = relationship( 

87 "SolicitationFacilityCapability", back_populates="facility", cascade="all, delete" 

88 ) 

89 frontends: Mapped[list[Frontend]] = relationship( 

90 back_populates="facility", cascade="all, delete-orphan", passive_deletes=True 

91 ) 

92 backends: Mapped[list[Backend]] = relationship( 

93 back_populates="facility", cascade="all, delete-orphan", passive_deletes=True 

94 ) 

95 stations: Mapped[list[Station]] = relationship( 

96 back_populates="facility", cascade="all, delete-orphan", passive_deletes=True 

97 ) 

98 array_configurations: Mapped[list[ArrayConfiguration]] = relationship( 

99 back_populates="facility", cascade="all, delete-orphan", passive_deletes=True 

100 ) 

101 external_joint_parameters: Mapped[list[ExternalJointParameter]] = relationship( 

102 back_populates="required_facilities", 

103 secondary=external_joint_required_facilities, 

104 ) 

105 

106 def __init__(self, name: str) -> None: 

107 super().__init__(facility_name=name) 

108 

109 @override 

110 def __repr__(self) -> str: 

111 return f"<Facility: {self.facility_name}>" 

112 

113 @override 

114 def __json__(self) -> JSON_OBJECT: 

115 facility_json: JSON_OBJECT = super().__json__() 

116 

117 # map attributes to JSON keys 

118 attribute_map: dict[str, Sequence[Base]] = { 

119 "frontends": self.frontends, 

120 "backends": self.backends, 

121 "arrayConfigurations": self.array_configurations, 

122 "stations": self.stations, 

123 } 

124 for js_attr, python_attr in attribute_map.items(): 

125 if python_attr: 

126 facility_json[js_attr] = [f.__json__() for f in python_attr] 

127 return facility_json 

128 

129 @override 

130 def __eq__(self, other: object) -> bool: 

131 return ( 

132 isinstance(other, self.__class__) 

133 and self.facility_id == other.facility_id 

134 and self.facility_name == other.facility_name 

135 and self.is_active == other.is_active 

136 ) 

137 

138 @override 

139 def __hash__(self): 

140 return hash((self.facility_id, self.facility_name, self.is_active)) 

141 

142 

143class RequestSpecificationType(Enum): 

144 OBSERVATION_SPECIFICATION = "Observation Specification" 

145 RESOURCE_SPECIFICATION = "Resource Specification" 

146 DATA_PROCESSING_SPECIFICATION = "Data Processing Specification" 

147 

148 

149class Capability(Base): 

150 __tablename__: str = "capabilities" 

151 """ 

152 A template for a SolicitationCapability. 

153 The different ways a Facility may be operated and the resources available. 

154 

155 Solicitations cannot share a capability, and capabilities must be copied if applied to another Solicitation. 

156 This is so changes (e.g. CapabilityParameterSpecifications) can be adjusted independently for each capability. 

157 """ 

158 

159 capability_id: Mapped[int] = mapped_column(primary_key=True) 

160 capability_name: Mapped[str] = mapped_column() 

161 is_active: Mapped[bool] = mapped_column() 

162 description: Mapped[str] = mapped_column(nullable=True) 

163 request_specification_type: Mapped[RequestSpecificationType] = mapped_column( 

164 sa.Enum(RequestSpecificationType, name="request_specification_type"), 

165 nullable=False, 

166 default=RequestSpecificationType.OBSERVATION_SPECIFICATION, 

167 ) 

168 solicitation_facility_capabilities: Mapped[list["SolicitationFacilityCapability"]] = relationship( 

169 "SolicitationFacilityCapability", back_populates="capability", cascade="all, delete" 

170 ) 

171 

172 @property 

173 def is_external_joint(self) -> bool: 

174 return self.capability_name == "External Joint" 

175 

176 def __init__( 

177 self, 

178 name: str, 

179 is_active: bool = True, 

180 request_specification_type: RequestSpecificationType = RequestSpecificationType.OBSERVATION_SPECIFICATION, 

181 cps_list: list["SolicitationFacilityCapability"] | None = None, 

182 ): 

183 super().__init__( 

184 capability_name=name, 

185 is_active=is_active, 

186 request_specification_type=request_specification_type, 

187 ) 

188 

189 @override 

190 def __repr__(self): 

191 return f"<Capability: {self.capability_name} at {hex(id(self))}>" 

192 

193 

194class RequiredCPS(TypedDict): 

195 name: str 

196 children: list["RequiredCPS"] 

197 

198 

199REQUIRED_CPSS_VLA_CONTINUUM: list[RequiredCPS] = [ 

200 { 

201 "name": "Field Sources", 

202 "children": [ 

203 {"name": "Right Ascension", "children": []}, 

204 {"name": "Declination", "children": []}, 

205 {"name": "Name", "children": []}, 

206 ], 

207 }, 

208 { 

209 "name": "Spectral Specifications", 

210 "children": [ 

211 {"name": "Center Frequency", "children": []}, 

212 {"name": "Bandwidth", "children": []}, 

213 {"name": "Name", "children": []}, 

214 ], 

215 }, 

216 {"name": "RMS Sensitivity", "children": []}, 

217] 

218REQUIRED_CPSS_GBT_SPECTRAL_LINE: list[RequiredCPS] = [ 

219 { 

220 "name": "Field Sources", 

221 "children": [ 

222 {"name": "Right Ascension", "children": []}, 

223 {"name": "Declination", "children": []}, 

224 {"name": "Name", "children": []}, 

225 ], 

226 }, 

227 { 

228 "name": "Spectral Specifications", 

229 "children": [ 

230 {"name": "Center Frequency", "children": []}, 

231 {"name": "Bandwidth", "children": []}, 

232 {"name": "Name", "children": []}, 

233 ], 

234 }, 

235 {"name": "RMS Sensitivity", "children": []}, 

236]