Coverage for middle_layer/solicit/domain_layer/services/configure_solicitation_service.py: 94.02%
117 statements
« prev ^ index » next coverage.py v7.10.5, created at 2026-03-09 06:13 +0000
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#
18import itertools
20from common import parse_iso_8601_strings
21from common.application_layer.orm_repositories.orm_repository import ORMRepository
22from common.application_layer.rest_api import make_expected_params_message
23from solicit.domain_layer.entities import solicitation_facility_capability
24from solicit.domain_layer.entities.array_configuration import ArrayConfiguration, ArrayConfigurationConfiguration
25from solicit.domain_layer.entities.backend import Backend, BackendConfiguration
26from solicit.domain_layer.entities.capability import Capability, Facility
27from solicit.domain_layer.entities.default_instruction import DefaultInstruction, Instruction
28from solicit.domain_layer.entities.frontend import Frontend, FrontendConfiguration
29from solicit.domain_layer.entities.parameter_configuration import InputUnitGroups, ParameterConfiguration
30from solicit.domain_layer.entities.solicitation import (
31 BAND_FREQUENCY_MAP_ALL,
32 Band,
33 Solicitation,
34 SolicitationProposalClass,
35 SolicitationProposalProcess,
36)
37from solicit.domain_layer.entities.solicitation_config import (
38 SolicitationCapabilityConfig,
39 SolicitationCapabilityParameterSpecificationConfig,
40 SolicitationConfig,
41)
42from solicit.domain_layer.entities.solicitation_facility_capability import SolicitationFacilityCapability
43from solicit.domain_layer.services.default_parameters_service import configure_parameters_for_sfc
46def configure_solicitation(config: dict[str, dict], repo: ORMRepository) -> Solicitation:
47 """Configure a Solicitation based on a JSON object.
48 This service aids in the "Configure Solicitation" use case
49 satisfied by the 'add_solicitation_from_config_file' route.
50 **NB:** The JSON object described below differs from that accepted by
51 api.views.solicitation.construct_solicitation_from_json
52 The config file should follow the formatting of
53 `smoke_testing_helpers/both_capabilities_config.json`
54 If 'observing_types' is included in the JSON entry of a Facility,
55 the service will query the appropriate Capability and create a
56 SolicitationCapability for the observing type. If only 'name' is defined
57 for the Facility, a SolicitationCapability will not be added.
59 :param config: configuration dictionary used to configure a Solicitation and SolicitationCapabilities
60 :param repo: ORMRepository from which to request full (ProposalProcess, Facility) objects from the given names
61 :raises: KeyError if dictionary has improper keys or formatting
62 :return: The configured Solicitation
63 """
64 solicitation_config = config["solicitation"]
65 expected_keys = [
66 "name",
67 "proposalCodePrefix",
68 "callPeriod",
69 "executionPeriod",
70 "gracePeriod",
71 "testType",
72 "scienceCategories",
73 "proposalProcess",
74 "facilities",
75 "dispositionLetterTemplate",
76 ]
77 expected_keys_call_period = ["start", "end"]
78 expected_keys_execution_period = ["start", "end"]
79 if not all([expected in solicitation_config.keys() for expected in expected_keys]):
80 raise KeyError(make_expected_params_message(expected_keys, solicitation_config.keys()))
81 if not all([expected in solicitation_config["callPeriod"].keys() for expected in expected_keys_call_period]):
82 raise KeyError(
83 f"Expected JSON object to have keys {expected_keys_call_period} at 'callPeriod'. "
84 f"Received {[key for key in solicitation_config['callPeriod'].keys()]}."
85 )
86 if not all(
87 [expected in solicitation_config["executionPeriod"].keys() for expected in expected_keys_execution_period]
88 ):
89 raise KeyError(
90 f"Expected JSON object to have keys {expected_keys_execution_period} at 'executionPeriod'. "
91 f"Received {[key for key in solicitation_config['executionPeriod'].keys()]}."
92 )
93 if repo.solicitation_repo.by_proposal_code_prefix(solicitation_config["proposalCodePrefix"]):
94 raise ValueError(
95 f"Solicitation with Proposal ID prefix {solicitation_config['proposalCodePrefix']} already exists."
96 )
97 call_period_start = parse_iso_8601_strings(solicitation_config["callPeriod"]["start"])
98 call_period_end = parse_iso_8601_strings(solicitation_config["callPeriod"]["end"])
99 execution_period_start = parse_iso_8601_strings(solicitation_config["executionPeriod"]["start"])
100 execution_period_end = parse_iso_8601_strings(solicitation_config["executionPeriod"]["end"])
101 grace_period = int(solicitation_config["gracePeriod"])
102 test_type = (
103 solicitation_config["testType"]
104 if solicitation_config["testType"] == "Test" or solicitation_config["testType"] == "Demo"
105 else "None"
106 )
107 pp_config = solicitation_config["proposalProcess"]
108 legacy = False
109 if isinstance(pp_config, str): # old-style config file. TODO: get rid of these post refactor
110 legacy = True
111 proposal_process = repo.proposal_process_repo.by_name(pp_config)
112 else:
113 proposal_process = repo.proposal_process_repo.by_name(pp_config["proposalProcessName"])
115 disposition_letter_template = solicitation_config["dispositionLetterTemplate"]
117 # using arbitrary notification group
118 solicitation = Solicitation(
119 solicitation_config["name"],
120 solicitation_config["proposalCodePrefix"],
121 call_period_start,
122 call_period_end,
123 execution_period_start,
124 execution_period_end,
125 grace_period,
126 proposal_process,
127 repo.notification_group_repo.by_id(1),
128 test_type,
129 disposition_letter_template=disposition_letter_template,
130 )
131 # Try to read Facilities, Science Categories, and SolicitationCapabilities from solicitation_config
132 facilities = []
133 science_categories = []
134 default_instructions = {di.shortcode: di for di in repo.session.query(DefaultInstruction).all()}
135 if not legacy:
136 for instruction in pp_config["defaultInstructions"]:
137 # see if this default instruction already exists
138 if instruction["shortcode"] in default_instructions:
139 di = default_instructions[instruction["shortcode"]]
140 else:
141 di = DefaultInstruction()
142 di.shortcode = instruction["shortcode"]
143 di.dropdown_name = instruction["dropdownName"]
144 di.instruction_text = instruction["instructionText"]
145 default_instructions[di.shortcode] = di
146 proposal_process.default_instructions.append(di)
148 spp = (
149 SolicitationProposalProcess(solicitation=solicitation, proposal_process=proposal_process)
150 if not legacy
151 else None
152 )
153 repo.solicitation_repo.add(solicitation)
154 for fac_config in solicitation_config["facilities"]:
155 facility_name = fac_config["name"]
156 facility = repo.facility_repo.by_name(facility_name)
157 facilities.append(facility)
159 # fetch the constant sets from the DB
160 fes = {
161 fe.frontend_name: fe
162 for fe in repo.session.query(Frontend)
163 .join(Frontend.facility)
164 .where(Facility.facility_name == facility_name)
165 .all()
166 }
167 bes = {
168 be.backend_name: be
169 for be in repo.session.query(Backend)
170 .join(Backend.facility)
171 .where(Facility.facility_name == facility_name)
172 .all()
173 }
174 acs = {
175 ac.array_configuration_name: ac
176 for ac in repo.session.query(ArrayConfiguration)
177 .join(ArrayConfiguration.facility)
178 .where(Facility.facility_name == facility_name)
179 .all()
180 }
182 for cap_config in fac_config["observingTypes"]:
183 cap_name = cap_config["name"]
184 sfc = SolicitationFacilityCapability(
185 solicitation=solicitation,
186 facility=facility,
187 allow_triggered=cap_config["allowTriggered"],
188 allow_monitored=cap_config["allowMonitored"],
189 allow_fixed_date=cap_config["allowFixedDate"],
190 regular_max_seconds=cap_config["regularMaxSeconds"],
191 capability=repo.capability_repo.by_name(cap_name),
192 execution_period_start=fac_config["executionPeriod"]["start"],
193 execution_period_end=fac_config["executionPeriod"]["end"],
194 )
195 configure_parameters_for_sfc(sfc)
197 # Create all the ParameterConfigurations for this SFC
198 json_collections = {
199 "fieldSourceConfigurations": "field_source_configurations",
200 "calibrationConfigurations": "calibration_configurations",
201 "spectralSpecConfigurations": "spectral_spec_configurations",
202 "performanceParameterConfigurations": "performance_parameter_configurations",
203 }
204 for json_field, attribute_name in json_collections.items():
205 collection = getattr(sfc, attribute_name)
206 for fs_config in cap_config[json_field]:
207 config = next(
208 (pc for pc in collection if pc.parameter.lower() == fs_config["parameter"].lower()), None
209 )
210 if not config:
211 config = ParameterConfiguration(
212 solicitation_facility_capability=sfc, parameter=fs_config["parameter"]
213 )
214 collection.append(config)
215 config.interactivity_mode = fs_config["interactivityMode"]
216 if "lowerBound" in fs_config.keys():
217 config.lower_bound = fs_config["lowerBound"]
218 if "upperBound" in fs_config.keys():
219 config.upper_bound = fs_config["upperBound"]
220 if "fieldDefault" in fs_config.keys():
221 config.field_default = fs_config["fieldDefault"]
222 if "fieldType" in fs_config.keys() and fs_config["fieldType"]:
223 config.field_type = InputUnitGroups(fs_config["fieldType"])
225 frontend_configurations = [
226 FrontendConfiguration(
227 solicitation_facility_capability=sfc,
228 frontend=fes[fe_config["frontendName"]],
229 tuning_range_min=fe_config["tuningRange"]["min"],
230 tuning_range_max=fe_config["tuningRange"]["max"],
231 weather_conditions=fe_config["weatherCondition"],
232 )
233 for fe_config in cap_config["frontendConfigurations"]
234 ]
236 backend_configurations = [
237 BackendConfiguration(
238 solicitation_facility_capability=sfc,
239 backend=bes[be_config["backendName"]],
240 )
241 for be_config in cap_config["backendConfigurations"]
242 ]
244 array_configuration_configurations = (
245 [
246 ArrayConfigurationConfiguration(
247 solicitation_facility_capability=sfc,
248 array_configuration=acs[ac_config["arrayConfigurationName"]],
249 baseline_min=ac_config["baseline"]["min"],
250 baseline_max=ac_config["baseline"]["max"],
251 )
252 for ac_config in cap_config["arrayConfigurationConfigurations"]
253 ]
254 if "arrayConfigurationConfigurations" in cap_config.keys()
255 else []
256 )
258 solicitation.solicitation_facility_capabilities.append(sfc)
259 else:
260 repo.session.add(solicitation)
262 validation_errors = validate_science_categories(solicitation_config["scienceCategories"], repo)
263 if len(validation_errors):
264 raise ValueError(
265 f"Science category validation errors found in solicitation config: {', '.join(validation_errors)}"
266 )
268 for scicat in solicitation_config["scienceCategories"]:
269 science_categories.append(repo.science_category_repo.by_name(scicat["name"]))
271 if "proposalClasses" in solicitation_config.keys():
272 for pc_json in solicitation_config["proposalClasses"]:
273 spc = SolicitationProposalClass()
274 spc.proposal_class = repo.proposal_class_repo.by_name(pc_json["proposalClassName"])
275 spc.title_max_characters = pc_json["titleMaxCharacters"]
276 spc.title_min_characters = pc_json["titleMinCharacters"]
277 spc.abstract_max_characters = pc_json["abstractMaxCharacters"]
278 spc.abstract_min_characters = pc_json["abstractMinCharacters"]
279 spc.scientific_justification_max_pages = pc_json["scientificJustificationMaxPages"]
280 spc.scientific_justification_min_pages = pc_json["scientificJustificationMinPages"]
281 solicitation.solicitation_proposal_classes.append(spc)
283 solicitation.science_categories = science_categories
284 return solicitation
287def validate_science_categories(science_categories: list, repo: ORMRepository) -> set[str]:
288 """Validate the science categories supplied in a solicitation configuration
290 :param science_categories: list of science categories to validate
291 :param repo: the repository to use
292 :returns: The complete set of errors that need to be addressed for science categories to match against master list
293 of science categories
294 """
295 validation_errors: set[str] = set()
297 for scicat in science_categories:
298 scicat_name = scicat["name"]
299 try:
300 repo.science_category_repo.by_name(scicat["name"])
301 except ValueError as e:
302 validation_errors.add(str(e))
304 return validation_errors