Coverage for python/lsst/cp/pipe/utilsEfd.py: 86%

1# This file is part of cp_pipe.

3# Developed for the LSST Data Management System.

4# This product includes software developed by the LSST Project

5# (https://www.lsst.org).

6# See the COPYRIGHT file at the top-level directory of this distribution

7# for details of code ownership.

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 <https://www.gnu.org/licenses/>.

21#

23__all__ = ["CpEfdClient"]

25import logging

26import numpy as np

27import re

28import requests

29import os

30from pathlib import Path

31import json

33from astropy.table import Table

34from astropy.time import Time

37class CpEfdClient():

38 """An EFD client to retrieve calibration results.

40 Parameters

41 ----------

42 efdInstance : `str`, optional

43 EFD instance name to connect to.

44 log : `logging.Logger`, optional

45 Log to write messages to.

46 """

48 def __init__(self, efdInstance="usdf_efd", dieOnSearch=False, log=None):

49 self.log = log if log else logging.getLogger(__name__)

50 self.dieOnSearch = dieOnSearch

52 authDict = self._getAuth(efdInstance)

53 self._auth = (authDict["username"], authDict["password"])

54 self._databaseName = authDict["database"]

55 self._databaseUrl = authDict["url"]

57 self.checkConnection()

59 def _getAuth(self, instanceAlias):

60 """Get authorization credentials.

62 Parameters

63 ----------

64 instanceAlias : `str`

65 EFD instance to get credentials for.

67 Returns

68 -------

69 credentials : `dict` [`str`, `str`]

70 A dictionary of authorization credentials, including at

71 least these key/value pairs:

73 ``"username"``

74 Login username.

75 ``"password"``

76 Login passwords.

77 ``"url"``

78 URL to connect to.

80 Raises

81 ------

82 RuntimeError :

83 Raised if the HTTPS request fails.

84 """

85 if os.getenv("EFDAUTH"): 85 ↛ 94line 85 didn't jump to line 94 because the condition on line 85 was always true

86 auth_path = Path(os.environ["EFDAUTH"])

87 auth_data = json.loads(auth_path.read_text())

89 if instanceAlias not in auth_data: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true

90 raise RuntimeError(f"Could not find {instanceAlias} in {auth_path}")

92 return auth_data[instanceAlias]

93 else:

94 raise RuntimeError("Could not find EFDAUTH environment variable.")

96 def checkConnection(self):

97 """Check the connection to the EFD.

99 Raises

100 ------

101 RuntimeError :

102 Raised if the connection check fails.

103 """

104 # The ping command will return 204 (No Content) on success.

105

106 params = {"wait_for_leader": "5s"}

107

108 response = requests.get(f"{self._databaseUrl}/ping", params=params, auth=self._auth)

109 response.raise_for_status()

110 if response.status_code != 204: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 raise RuntimeError(f"Connection check failed for {self._databaseUrl}")

112

113 def getSchemaDtype(self, topicName):

114 """Get datatypes for a topic.

115

116 Parameters

117 ----------

118 topicName : `str`

119 Topic to get datatypes for

120

121 Returns

122 -------

123 datatypes : `list` [`tuple` [`str`, `str`]]

124 List of tuples of field names and data types.

125 """

126 query = f"SHOW FIELD KEYS FROM \"{topicName}\""

127 data = self.query(query)

128

129 values = data["results"][0]["series"][0]["values"]

130

131 dtype = [("time", "str")]

132 for (fieldName, fieldType) in values:

133 if fieldType == "float":

134 fieldDtype = np.float64

135 elif fieldType == "integer":

136 fieldDtype = np.int64

137 elif fieldType == "string": 137 ↛ 139line 137 didn't jump to line 139 because the condition on line 137 was always true

138 fieldDtype = "str"

139 dtype.append((fieldName, fieldDtype))

140 return dtype

141

142 def query(self, query):

143 """Execute an EFD query.

144

145 Parameters

146 ----------

147 query : `str`

148 Query to run.

149

150 Returns

151 -------

152 results : `dict`

153 Dictionary of results returned.

154

155 Raises

156 ------

157 RuntimeError :

158 Raised if the the database could not be read from.

159 """

160 params = {

161 "db": self._databaseName,

162 "q": query,

163 }

164

165 try:

166 response = requests.get(f"{self._databaseUrl}/query", params=params, auth=self._auth)

167 response.raise_for_status()

168 return response.json()

169 except requests.exceptions.RequestException as e:

170 raise RuntimeError(f"Could not read data from database with query: {query}") from e

171

172 def selectTimeSeries(self, topicName, fields=[], startDate=None, endDate=None):

173 """Query a topic for a time series.

174

175 Parameters

176 ----------

177 topicName : `str`

178 Database "topic" to query.

179 fields : `list`, optional

180 List of fields to return. If empty, all fields are

181 returned.

182 startDate : `astropy.time.Time`, optional

183 Start date to limit the results returned.

184 endDate : `astropy.time.Time`, optional

185 End date to limit the results returned.

186

187 Returns

188 -------

189 table : `astropy.table.Table`

190 A table containing the fields requested, with each row

191 corresponding to one date (available in the ``"time"``

192 column).

193 """

194 query = "SELECT "

195

196 if not fields:

197 query += "*"

198 else:

199 query += ",".join(fields)

200

201 query += f" FROM \"{topicName}\""

202

203 if startDate is not None or endDate is not None: 203 ↛ 206line 203 didn't jump to line 206 because the condition on line 203 was always true

204 query += " WHERE"

205

206 if startDate is not None: 206 ↛ 210line 206 didn't jump to line 210 because the condition on line 206 was always true

207 query += f" time >= '{startDate.utc.isot}Z'"

208 if endDate is not None: 208 ↛ 210line 208 didn't jump to line 210 because the condition on line 208 was always true

209 query += " AND"

210 if endDate is not None: 210 ↛ 213line 210 didn't jump to line 213 because the condition on line 210 was always true

211 query += f" time <= '{endDate.utc.isot}Z'"

212

213 data = self.query(query)

214

215 # data is a dictionary with one key, "results"

216 results = data["results"][0]

217 if "series" in results: 217 ↛ 220line 217 didn't jump to line 220 because the condition on line 217 was always true

218 series = results["series"][0]

219 else:

220 raise RuntimeError(f"No results found for query: {query}")

221

222 schemaDtype = self.getSchemaDtype(topicName)

223 tableDtype = []

224 for dtype in schemaDtype:

225 if dtype[0] in series["columns"]:

226 tableDtype.append(dtype[1])

227

228 # The value stored in "time" may not be consistent and

229 # monotonic (and is in UTC). "private_sndStamp" comes from

230 # the device itself, and is therefore preferred.

231 table = Table(rows=series["values"], names=series["columns"], dtype=tableDtype)

232 table["time"] = Time(table["time"], scale="utc").tai

233 table.sort("time")

234 if "private_sndStamp" in table.columns: 234 ↛ 238line 234 didn't jump to line 238 because the condition on line 234 was always true

235 table["private_sndStamp"] = Time(table["private_sndStamp"], format="unix_tai")

236 table.sort("private_sndStamp")

237

238 return table

239

240 def searchResults(self, data, dateStr):

241 """Find the row entry in ``data`` immediately preceding the specified

242 date.

243

244 Parameters

245 ----------

246 data : `astropy.table.Table`

247 The table of results from the EFD.

248 dateStr : `str`

249 The date (in TAI) to look up in the status for.

250

251 Returns

252 -------

253 result = `astropy.table.Row`

254 The row of the data table corresponding to ``dateStr``.

255 """

256 dateValue = Time(dateStr, scale='tai', format="isot")

257 # Table is now sorted on "time", which is in TAI.

258

259 # Check that the date we want to consider is contained in the

260 # EFD data.

261 if data["time"][0] > dateValue or data["time"][-1] < dateValue: 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true

262 msg = f"Requested date {dateStr} outside of data range {data['time'][0]}-{data['time'][-1]}"

263 if self.dieOnSearch:

264 raise RuntimeError(msg)

265 else:

266 # Return the start, as we're more likely to have

267 # errors in that direction.

268 self.log.warning(msg)

269 return data[0], np.nan

270

271 # Binary search through the EFD entries in date, until the

272 # most recent monochromator state update prior to the spectrum

273 # in question is found.

274 low = 0

275 high = len(data)

276 idx = (high + low) // 2

277 found = False

278 iteration = 0

279 while not found:

280 if idx < 0 or idx > len(data) or iteration > 20: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 raise RuntimeError(f"Search for date failed: {dateValue} {idx} {iteration}.")

282

283 myTime = data["private_sndStamp"][idx]

284 if myTime <= dateValue:

285 low = idx

286 elif myTime > dateValue: 286 ↛ 289line 286 didn't jump to line 289 because the condition on line 286 was always true

287 high = idx

288

289 idx = (high + low) // 2

290 iteration += 1

291 if high - low == 1:

292 found = True

293 self.log.debug("parse search %d %d %d %d %s %s",

294 low, high, idx, found, myTime, dateValue)

295

296 # End binary search.

297 return data[idx], idx

298

299 def getEfdMonochromatorData(self, dataSeries=None, dateMin=None, dateMax=None):

300 """Retrieve Monochromator data from the EFD.

301

302 Parameters

303 ----------

304 dataSeries : `str`, optional

305 Data series to request from the EFD.

306 dateMin : `str`, optional

307 Minimum date (in TAI) to retrieve from EFD.

308 dateMax : `str`, optional

309 Maximum date (in TAI) to retrieve from EFD.

310

311 Returns

312 -------

313 results : `astropy.table.Table`

314 The table of results returned from the EFD.

315 """

316 # This is currently the only monochromator available.

317 dataSeries = dataSeries if dataSeries else "lsst.sal.ATMonochromator.logevent_wavelength"

318

319 if dateMin: 319 ↛ 322line 319 didn't jump to line 322 because the condition on line 319 was always true

320 startDate = Time(dateMin, format="isot", scale="tai")

321 else:

322 startDate = None

323 if dateMax: 323 ↛ 326line 323 didn't jump to line 326 because the condition on line 323 was always true

324 stopDate = Time(dateMax, format="isot", scale="tai")

325 else:

326 stopDate = None

327

328 results = self.selectTimeSeries(dataSeries, ["wavelength", "private_sndStamp"],

329 startDate, stopDate)

330 return results

331

332 def parseMonochromatorStatus(self, data, dateStr):

333 """Determine monochromator status for a specific date.

334

335 Parameters

336 ----------

337 data : `astropy.table.Table`

338 The dataframe of monochromator results from the EFD.

339 dateStr : `str`

340 The date (in TAI) to look up in the status for.

341

342 Returns

343 -------

344 indexDate : `str`

345 Date string (in TAI) indicating the monochromator state change.

346 wavelength : `float`

347 Monochromator commanded peak.

348 """

349 result, _ = self.searchResults(data, dateStr)

350 myTime = result["private_sndStamp"]

351 return myTime.strftime("%Y-%m-%dT%H:%M:%S.%f"), result["wavelength"]

352

353 def getEfdElectrometerData(self, dataSeries=None, dateMin=None, dateMax=None):

354 """Retrieve Electrometer data from the EFD.

355

356 Parameters

357 ----------

358 dataSeries : `str`, optional

359 Data series to request from the EFD.

360 dateMin : `str`, optional

361 Minimum date (in TAI) to retrieve from EFD.

362 dateMax : `str`, optional

363 Maximum date (in TAI) to retrieve from EFD.

364

365 Returns

366 -------

367 results : `astropy.table.Table`

368 The table of results returned from the EFD.

369 """

370 # All electrometer data gets written to the same series.

371 defaultSeries = "lsst.sal.Electrometer.logevent_intensity"

372 alternateSeries = "lsst.sal.Electrometer.logevent_logMessage"

373 dataSeries = dataSeries if dataSeries else defaultSeries

374

375 if dateMin:

376 startDate = Time(dateMin, format="isot", scale="tai")

377 else:

378 startDate = None

379 if dateMax:

380 stopDate = Time(dateMax, format="isot", scale="tai")

381 else:

382 stopDate = None

383

384 if (dataSeries == alternateSeries and (startDate is None or stopDate is None)):

385 raise RuntimeError("Cannot query logevent_logMessage without dates to limit memory issues.")

386

387 results = self.selectTimeSeries(dataSeries, [],

388 startDate, stopDate)

389 if dataSeries == "lsst.sal.Electrometer.logevent_logMessage":

390 results = self.rewriteElectrometerStatus(results)

391 return results

392

393 def rewriteElectrometerStatus(self, inResults):

394 """Rewrite intermediate electrometer data extracted from the EFD

395 logEvents.

396

397 Parameters

398 ----------

399 inResults : `astropy.table.Table`

400 The table of results returned from the EFD.

401

402 Returns

403 -------

404 outResults : `astropy.table.Table`

405 The rewritten table containing only electrometer summary

406 status events.

407 """

408 # This is fragile against upstream changes.

409 # Ignore all entries that are not the ones we care about.

410 outResults = inResults[inResults["functionName"] == "write_fits_file"]

411 outResults = outResults[outResults["level"] == 20]

412

413 # These will be new columns

414 intensityMean = []

415 intensityStdev = []

416 intensityTime = []

417 intensityFile = []

418

419 for row in outResults:

420 # Fallback values in case the regexp fails

421 mean = np.nan

422 stdev = np.nan

423 time_mean = np.nan

424 filename = "REGEXP_FAIL"

425

426 # Find the last "filename\.extension" before the first newline;

427 # the last [] grouped values before the second newline;

428 # the last [] grouped values before the end of the string.

429 magic = re.findall(r"\b\w+.+?(\w+?\.\w+)\n\b\w+.+\[(.*?)\]\n\b\w+.+\[(.*?)\]$",

430 row["message"])

431 if len(magic) != 0: 431 ↛ 445line 431 didn't jump to line 445 because the condition on line 431 was always true

432 # If we matched, split the grouped values, cast them to floats.

433 filename, intensity_str, time_str = magic[0]

434 mean, median, stdev = intensity_str.split(",")

435 time_mean, time_median = time_str.split(",")

436 mean = float(mean)

437 median = float(median)

438 stdev = float(stdev)

439 time_mean = float(time_mean)

440 time_median = float(time_median)

441 # Censor the saturated points so plots look nice.

442 if np.abs(mean) > 1e37: 442 ↛ 443line 442 didn't jump to line 443 because the condition on line 442 was never true

443 mean = np.nan

444

445 intensityMean.append(mean)

446 intensityStdev.append(stdev)

447 intensityFile.append(filename)

448 intensityTime.append(time_mean)

449

450 # Add our new columns at the start of the column list

451 outResults.add_column(intensityMean, name="intensity", index=1)

452 outResults.add_column(intensityStdev, name="intensityStd", index=2)

453 outResults.add_column(intensityTime, name="intensityTimeMean", index=3)

454 outResults.add_column(intensityFile, name="expectedLfaFile", index=4)

455 return outResults

456

457 def parseElectrometerStatus(self, data, dateStr, dateEnd=None,

458 doIntegrateSamples=False, index=201):

459 """Determine electrometer status for a specific date.

460

461 Parameters

462 ----------

463 data : `astropy.table.Table`

464 The dataframe of electrometer results from the EFD.

465 dateStr : `str`

466 The date (in TAI) to look up in the status for.

467 dateEnd : `str`

468 The end date (in TAI) to look in the status for.

469 doIntegrateSamples: `bool`

470 If true, take the average of all samples between

471 ``dateStr`` and ``dateEnd``.

472 index : `int`

473 The salIndex of the device we want to read. For LATISS,

474 this should be 201. For the main telescope, 101.

475

476 Returns

477 -------

478 indexDate : `str`

479 Date string (in TAI) indicating the electrometer state

480 change.

481 intensity: `float`

482 Average electrometer intensity.

483 """

484 if index is not None: 484 ↛ 489line 484 didn't jump to line 489 because the condition on line 484 was always true

485 mask = (data["salIndex"] == index)

486 data = data[mask]

487

488 # searchResults returns the first entry prior to this date

489 result, idx = self.searchResults(data, dateStr)

490

491 myTime = result["private_sndStamp"]

492 myIntensity = result["intensity"]

493 myEndTime = None

494

495 if doIntegrateSamples:

496 myEndResult, myEndIdx = self.searchResults(data, dateEnd)

497 if myEndIdx != idx: 497 ↛ 501line 497 didn't jump to line 501 because the condition on line 497 was always true

498 myEndTime = myEndResult["private_sndStamp"]

499 myIntensity = np.mean(data[idx+1:myEndIdx]["intensity"])

500

501 return (myTime.strftime("%Y-%m-%dT%H:%M:%S.%f"),

502 myIntensity,

503 myEndTime.strftime("%Y-%m-%dT%H:%M:%S.%f") if myEndTime else None)