Coverage for middlelayer/ngect/api.py: 66.53%

1# -*- coding: utf-8 -*-

4# This file is part of ngVLA Exposure Calculator Tool (ngECT).

6# ngECT is free software: you can redistribute it and/or modify

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

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

9# any later version.

10#

11# ngECT is distributed in the hope that it will be useful,

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

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

14# GNU General Public License for more details.

15#

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

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

19"""API Backend"""

20import importlib

21import io

22import json

23import logging

24import platform

25import time

26import traceback

27from contextlib import asynccontextmanager

28from datetime import datetime

29from typing import Optional

31import uvicorn

32from astropy import units as u

33from astropy.utils.misc import JsonCustomEncoder

34from fastapi import FastAPI, HTTPException, Request

35from fastapi.middleware.cors import CORSMiddleware

36from fastapi.responses import HTMLResponse

37from fastapi.templating import Jinja2Templates

38from pydantic import BaseModel

40from . import __formatter__, __logfile__, __version__, engine, system

42logger = logging.getLogger(__name__)

43startup_time = time.time()

45# initialize the FastAPI app

46@asynccontextmanager

47async def lifespan(app: FastAPI):

48 uvicorn_logger = logging.getLogger("uvicorn")

49 logfile_handler = logging.FileHandler(__logfile__, mode="a")

50 logfile_handler.setFormatter(__formatter__)

51 logfile_handler.setLevel("TRACE")

52 uvicorn_logger.addHandler(logfile_handler)

53 logging.getLogger("uvicorn.access").propagate = True

54 logging.getLogger("uvicorn.error").propagate = True

55 yield

58app = FastAPI(lifespan=lifespan)

59# add the CORS middleware and configure its policy

60# see https://fastapi.tiangolo.com/tutorial/cors/

61# note: We can update the policy to restrict the javascript frontend access from a short

62# whitelist. Alternatively, we use 'wildcard' to ensure that the integration between

63# the frontend and backend works properly in a wide range of settings.

64# In the future, we need to revisit this security measure for production instances.

65origins = [

66 r"http://localhost",

67 r"http://localhost:3333",

68 r"http://localhost:3334",

69 r"http://localhost:3335",

70 r"http://localhost:4200",

71 r"https://.*\.nrao\.edu'",

72 r"http://.*\.nrao\.edu'",

73]

74app.add_middleware(

75 CORSMiddleware,

76 allow_origins=origins,

77 allow_credentials=True,

78 allow_methods=["*"],

79 allow_headers=["*"],

80 expose_headers=["X-ngect-suggestion"],

81)

83# Since the list of choices will not change after the application is first initialized,

84# we simply create all the possible options here at bootup

85SUBARRAY_CHOICES: dict[str, list[str]] = {}

86for revision in system.available_subarray_revisions():

87 SUBARRAY_CHOICES[revision] = list(system.read_subarray_data(revision).keys())

88DEFAULT_SUBARRAY_CHOICES = SUBARRAY_CHOICES[system.default_subarray_revision()]

91class Quantity(BaseModel):

92 value: float

93 unit: str

94 desc: Optional[str] = None

97class SubArray(BaseModel):

98 value: str = "core"

99 value_select: list[str] = DEFAULT_SUBARRAY_CHOICES

100 desc: str = "Subarray"

101

102

103class ConfigRev(BaseModel):

104 value: str = system.default_subarray_revision()

105 value_select: list[str] = system.available_subarray_revisions()

106 desc: str = "Configuration revision"

107

108

109class RxConfig(BaseModel):

110 value: str = "Baseline"

111 desc: str = "Receiver Configuration"

112 value_select: dict = system.read_receiver_data()

113

114

115class PolNum(BaseModel):

116 value: int = 2

117 value_select: list[int] = [1, 2]

118 desc: str = "Polarization setup: 1=single, 2=dual"

119

120

121class Frequency(BaseModel):

122 value: float = 24.0

123 unit: str = "GHz"

124 unit_select: list[str] = ["GHz", "MHz", "kHz", "Hz"]

125 desc: str = "Representative Frequency"

126

127

128class SpecMode(BaseModel):

129 value: str = "line"

130 value_select: list[str] = ["line", "cont"]

131 desc: str = ""

132

133

134class Chanwidth(BaseModel):

135 value: float = 1e4

136 unit: str = "m/s"

137 unit_select: list[str] = ["GHz", "MHz", "kHz", "Hz", "km/s", "m/s"]

138 desc: str = "Chanwidth in velocity or frequency"

139

140

141class Theta(BaseModel):

142 value: float = -1

143 unit: str = "arcsec"

144 unit_select: list[str] = ["arcsec", "mas"]

145 desc: str = ""

146

147

148class Pwv(BaseModel):

149 # value: float = Field(6., desciption='Value of pwv')

150 value: str = "6 mm"

151 value_select: list[str] = ["1 mm", "6 mm", "13mm"]

152 desc: str = "Preciptable Water Vapor (not implemented)"

153 # current model assumption: 1 mm PWV for band-6 and 6 mm PWV for the other bands.

154

155

156class Tos(BaseModel):

157 value: float = 60

158 unit: str = "s"

159 unit_select: list[str] = ["hr", "min", "s"]

160 desc: str = "Time on Source"

161

162

163class Sigma(Quantity):

164 value: float = 10

165 unit: str = "K"

166 unit_select: list[str] = ["nJy/beam", "uJy/beam", "mJy/beam", "Jy/beam", "mK", "K"]

167 desc: str = ""

168

169

170class RxName(BaseModel):

171 value: str

172 desc: str = "Receiver Band Name"

173

174

175class EctMode(BaseModel):

176 value: str = "tos2rms"

177 value_select: list[str] = ["tos2rms", "rms2tos"]

178 desc: str = "Calculator Model: tos2rms or rms2tos"

179

180

181class NumAntenna(BaseModel):

182 value: int

183 desc: str = "Number of Antennas"

184

185

186class DigitalSamplers(BaseModel):

187 value: str = "8-bit"

188 desc: str = "Digital Samplers (not implemented)"

189

190

191class Fov(BaseModel):

192 value: float

193 unit: str

194 desc: str = "Field of View"

195

196

197class ApertureEta(BaseModel):

198 value: float

199 desc: str = "Aperture Efficiency"

200

201

202class EffArea(BaseModel):

203 value: float

204 unit: str

205 desc: str = "Effective Area"

206

207

208class ConfusionLevel(BaseModel):

209 value: float

210 unit: str

211 desc: str = "Confusion Level (not implemented)"

212

213

214class Tsys(BaseModel):

215 value: float

216 unit: str

217 desc: str = "System Temeperature"

218

219

220class BandwidthMax(BaseModel):

221 value: float | None = None

222 unit: str | None = None

223 desc: str = "Maximum Instantaneous Bandwidth"

224

225

226class Elevation(BaseModel):

227 value: float = 45.0

228 value_range: tuple = (0, 90)

229 unit: str = "degree"

230 unit_select: list = ["degree"]

231 desc: str = "Elevation (not implemented)"

232

233

234class InputModel(BaseModel):

235 freq: Frequency

236 theta: Theta

237 rx_config: RxConfig

238 subarray: SubArray

239 configrev: ConfigRev

240 specmode: SpecMode

241 chanwidth: Chanwidth

242 npol: PolNum

243 tos: Tos

244 sigma: Sigma

245 elevation: Elevation

246 ectmode: EctMode

247 pwv: Pwv

248

249

250class OutputModel(BaseModel):

251 freq: Frequency

252 rx_name: RxName

253 fov: Fov

254 max_bw: BandwidthMax

255 t_sys: Tsys

256 chanwidth_freq: Chanwidth

257 chanwidth_velo: Chanwidth

258 tos: Tos

259 sigma_fl: Sigma

260 sigma_tb: Sigma

261 samplers: DigitalSamplers

262 num_ant: NumAntenna

263 theta: Theta

264 eta_A: ApertureEta

265 total_eff_A: EffArea

266 cf_level: ConfusionLevel

267

268

269@app.post(

270 "/ect",

271 response_model=OutputModel,

272 response_model_exclude_unset=False,

273 response_model_exclude_defaults=False,

274 response_model_exclude_none=False,

275)

276async def ect(input: InputModel):

277 """

278 Calculate ngVLA exposure time and sensitivity parameters.

279

280 This endpoint performs exposure calculator computations for the next-generation

281 Very Large Array (ngVLA), calculating sensitivity and observing time parameters

282 based on input observational requirements.

283

284 The endpoint processes input parameters, performs array performance calculations,

285 and returns detailed results including sensitivity metrics, system temperatures,

286 and observing recommendations.

287

288 Parameters

289 ----------

290 input : InputModel

291 Input model containing observational parameters including:

292

293 * frequency specifications

294 * array configuration

295 * observing mode settings

296 * target sensitivity requirements

297 * channel width and spectral settings

298

299 Returns

300 -------

301 dict

302 Dictionary containing calculated exposure parameters with the following structure:

303

304 * **freq** : Observing frequency with units

305 * **sigma_fl** : Point source flux density sensitivity

306 * **sigma_tb** : Brightness temperature sensitivity

307 * **tos** : Required on-source observing time

308 * **rx_name** : Selected receiver band name

309 * **rx_freq_min/max** : Receiver frequency range

310 * **max_bw** : Maximum available bandwidth

311 * **fov** : Field of view

312 * **eta_A** : Antenna efficiency

313 * **total_eff_A** : Total effective area

314 * **t_sys** : System temperature

315 * **sefd** : System equivalent flux density

316 * **theta_max** : Maximum angular resolution

317 * **theta** : Synthesized beam size

318 * **chanwidth_freq/velo** : Channel width in frequency/velocity units

319 * **eta_W** : Weighting efficiency factor

320 * **LAS** : Largest angular scale

321 * **num_ant** : Number of antennas

322 * **samplers** : Digitizer configuration

323 * **cf_level** : Confusion limit

324 * **info** : Processing log messages

325

326 Raises

327 ------

328 HTTPException

329 Status 400 for invalid input parameters or calculation errors.

330

331 For out-of-range frequency errors, includes an ``X-ngect-suggestion``

332 header containing JSON with the nearest supported frequency recommendation.

333

334 Notes

335 -----

336 The endpoint automatically:

337

338 * Converts input units using astropy.units

339 * Selects appropriate receiver bands based on frequency

340 * Performs continuum and/or spectral line calculations

341 * Provides detailed logging of calculation steps

342 * Handles frequency validation with helpful suggestions

343

344 Examples

345 --------

346 A typical successful response includes sensitivity calculations for both

347 continuum and spectral line observing modes, with all physical quantities

348 properly unit-tagged for scientific applications.

349

350 For invalid frequency inputs, the error response includes machine-readable

351 suggestions for the nearest supported frequency via response headers.

352

353 See Also

354 --------

355 InputModel : Request model schema

356 OutputModel : Response model schema

357 ArrayPerformance.ect : Core calculation engine

358 """

359 # decoding

360 input_dict = input.model_dump()

361 for k, v in input_dict.items():

362 if "value" in v and "unit" in v:

363 input_dict[k] = v["value"] * u.Unit(v["unit"])

364 elif "value" in v:

365 input_dict[k] = v["value"]

366

367 ap = engine.ArrayPerformance(input.configrev.value)

368

369 request_log_string = io.StringIO()

370 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")

371 request_handler = logging.StreamHandler(request_log_string)

372 request_handler.setFormatter(formatter)

373 request_handler.setLevel("DEBUG")

374 app_logger = logging.getLogger("ngect")

375 app_logger.addHandler(request_handler)

376

377 try:

378 output_dict = ap.ect(input_dict)

379

380 request_msg = request_log_string.getvalue()

381 request_log_string.close()

382 app_logger.removeHandler(request_handler)

383

384 # encoding

385 for key, value in output_dict.items():

386 if not isinstance(value, u.Quantity):

387 output_dict[key] = dict(value=value)

388

389 # placeholder

390 output_dict["samplers"] = dict(value="8-bit")

391 output_dict["cf_level"] = dict(value=-1, unit="mJy/beam")

392 output_dict["info"] = dict(value=request_msg)

393

394 output_json = json.dumps(output_dict, ensure_ascii=False, indent=4, sort_keys=True, cls=JsonCustomEncoder)

395 # return Response(content=output_json, media_type="application/json")

396 # return jsonable_encoder(output_dict['frequency'])

397

398 return json.loads(output_json)

399

400 except Exception as ex:

401 logger.error(f"Exception from ngect_api: {ex}")

402 tb_msg = traceback.format_exc()

403 logger.error(f"Traceback from ngect_api: {tb_msg}")

404

405 app_logger.removeHandler(request_handler)

406

407 detail_str = str(ex)

408

409 # If this is the known out-of-range frequency error, attach a machine-readable

410 # suggestion via response headers to avoid breaking clients that expect a string.

411 headers = None

412 if "outside of the range of the frequencies supported by the ngVLA" in detail_str:

413 try:

414 # Use the same ArrayPerformance instance/config to compute nearest suggestion

415 freq_q = input_dict.get("freq")

416 suggestion = ap.nearest_supported_frequency(freq_q)

417 headers = {"X-ngect-suggestion": json.dumps(suggestion)}

418 except Exception:

419 pass

420

421 raise HTTPException(status_code=400, detail=detail_str, headers=headers)

422

423

424@app.get("/ect_select")

425async def get_ect_valid():

426 """Return valid selections/ranges for the ECT input schema.

427

428 Builds a lightweight dictionary describing frontend-selectable options

429 for each field in the :class:`InputModel`, including descriptions,

430 unit choices, value ranges, and enumerations. Also appends static

431 configuration such as available subarray choices and default revision.

432

433 Returns

434 -------

435 dict

436 Mapping of input field names to metadata suitable for populating

437 UI controls, plus ``subarray_choices`` and default ``configrev``.

438 """

439 ect_valid = {}

440 ect_input_schema = InputModel.model_json_schema()

441 for k, v in ect_input_schema["properties"].items():

442 input_cls_name = v["$ref"].split("/")[-1]

443

444 field_list = ["desc", "unit_select", "value_range", "value_select"]

445 ect_valid_one = dict()

446 for field_name in field_list:

447 try:

448 field_shortname = (

449 field_name.replace("unit_select", "unit")

450 .replace("value_select", "value")

451 .replace("value_range", "range")

452 )

453 ect_valid_one[field_shortname] = ect_input_schema["$defs"][input_cls_name]["properties"][field_name][

454 "default"

455 ]

456 except KeyError:

457 pass

458 ect_valid[k] = ect_valid_one

459

460 # append the revisions and subarray choices here

461 # this is not really something that needs to come back as an input value, so

462 # we bypass the validation

463 ect_valid["subarray_choices"] = SUBARRAY_CHOICES

464 ect_valid["configrev"]["default"] = system.default_subarray_revision()

465

466 # optional enhancement: include receiver band metadata so FE can proactively snap

467 try:

468 rx_dt = system.read_receiver_data()

469 rx_config_value = {}

470 for rx_name, rx_specs in rx_dt.items():

471 try:

472 band_freqs = rx_specs.get("freq", [])

473 if band_freqs:

474 f_low = float(min(band_freqs))

475 f_high = float(max(band_freqs))

476 rx_config_value[rx_name] = {"name": rx_name, "low": f_low, "hi": f_high}

477 except Exception:

478 # skip malformed band entries

479 pass

480 if rx_config_value:

481 ect_valid["rx_config"] = {"unit": "GHz", "value": rx_config_value}

482 except Exception:

483 # If receiver data cannot be read, omit rx_config silently

484 pass

485

486 return ect_valid

487

488

489@app.get("/status", response_class=HTMLResponse)

490def status(request: Request):

491 """Render a simple service status page.

492

493 Parameters

494 ----------

495 request : fastapi.Request

496 Incoming request object used by the Jinja template renderer.

497

498 Returns

499 -------

500 fastapi.responses.HTMLResponse

501 HTML page with a few runtime details (time, hostname, uptime, version).

502 """

503 items = {

504 "localtime": datetime.now().strftime("%H:%M:%S") + " " + time.tzname[1],

505 "hostname": platform.node(),

506 "uptime": f"{time.time() - startup_time:.0f}s",

507 "version": f"{__version__}",

508 }

509 return templates.TemplateResponse(request, "index.html", {"items": items})

510

511

512with importlib.resources.path(__name__, "templates") as template_path:

513 templates = Jinja2Templates(directory=template_path)

514

515

516@app.get("/", response_class=HTMLResponse)

517def home(request: Request):

518 """Root handler that delegates to the status page."""

519 return status(request)

520

521

522def start_api():

523 logger.info("start the ngect api backend")

524 logger.info("the equivalent cli command: $ uvicorn ngect:app --host 0.0.0.0 --port 3335 --reload")

525 # here the backend will listen on all IPv4 interfaces on the host system and accessible

526 # to clients from the network.

527 uvicorn.run("ngect:app", host="0.0.0.0", port=3335, reload=True)

Coverage for middlelayer / ngect / api.py: 66.53%

251 statements