Coverage for python/lsst/summit/utils/efdUtils.py: 13%

1# This file is part of summit_utils.

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

21from __future__ import annotations

23import asyncio

24import datetime

25import logging

26import re

27from typing import TYPE_CHECKING, Any, Callable

29import astropy

30import pandas as pd

31from astropy import units as u

32from astropy.time import Time, TimeDelta

33from deprecated.sphinx import deprecated

35from lsst.utils.iteration import ensure_iterable

37if TYPE_CHECKING:

38 from .tmaUtils import TMAEvent

39 from lsst.daf.butler import DimensionRecord

40 from pandas import DataFrame, Series, Timestamp

42from .utils import getSite

44HAS_EFD_CLIENT = True

45try:

46 from lsst_efd_client import EfdClient

47except ImportError:

48 HAS_EFD_CLIENT = False

51__all__ = [

52 "getEfdData",

53 "getMostRecentRowWithDataBefore",

54 "makeEfdClient",

55 "expRecordToTimespan",

56 "efdTimestampToAstropy",

57 "astropyToEfdTimestamp",

58 "clipDataToEvent",

59 "calcNextDay",

60 "getDayObsStartTime",

61 "getDayObsEndTime",

62 "getDayObsForTime",

63 "getSubTopics", # deprecated, being removed in w_2023_50

64 "getTopics",

65 "getCommands",

66]

69COMMAND_ALIASES = {

70 "raDecTarget": "lsst.sal.MTPtg.command_raDecTarget",

71 "moveToTarget": "lsst.sal.MTMount.command_moveToTarget",

72 "startTracking": "lsst.sal.MTMount.command_startTracking",

73 "stopTracking": "lsst.sal.MTMount.command_stopTracking",

74 "trackTarget": "lsst.sal.MTMount.command_trackTarget", # issued at 20Hz - don't plot

75}

77# When looking backwards in time to find the most recent state event, look back

78# in chunks of this size. Too small, and there will be too many queries, too

79# large and there will be too much data returned unnecessarily, as we only need

80# one row by definition. Will tune this parameters in consultation with SQuaRE.

81TIME_CHUNKING = datetime.timedelta(minutes=15)

84def _getBeginEnd(

85 dayObs: int | None = None,

86 begin: Time | None = None,

87 end: Time | None = None,

88 timespan: TimeDelta | None = None,

89 event: TMAEvent | None = None,

90 expRecord: DimensionRecord | None = None,

91) -> tuple[Time, Time]:

92 """Calculate the begin and end times to pass to _getEfdData, given the

93 kwargs passed to getEfdData.

95 Parameters

96 ----------

97 dayObs : `int`

98 The dayObs to query. If specified, this is used to determine the begin

99 and end times.

100 begin : `astropy.Time`

101 The begin time for the query. If specified, either an end time or a

102 timespan must be supplied.

103 end : `astropy.Time`

104 The end time for the query. If specified, a begin time must also be

105 supplied.

106 timespan : `astropy.TimeDelta`

107 The timespan for the query. If specified, a begin time must also be

108 supplied.

109 event : `lsst.summit.utils.efdUtils.TMAEvent`

110 The event to query. If specified, this is used to determine the begin

111 and end times, and all other options are disallowed.

112 expRecord : `lsst.daf.butler.dimensions.DimensionRecord`

113 The exposure record containing the timespan to query. If specified, all

114 other options are disallowed.

115

116 Returns

117 -------

118 begin : `astropy.Time`

119 The begin time for the query.

120 end : `astropy.Time`

121 The end time for the query.

122 """

123 if expRecord is not None:

124 forbiddenOpts = [event, begin, end, timespan, dayObs]

125 if any(x is not None for x in forbiddenOpts):

126 raise ValueError("You can't specify both an expRecord and a begin/end or timespan or dayObs")

127 begin = expRecord.timespan.begin

128 end = expRecord.timespan.end

129 return begin, end

130

131 if event is not None:

132 forbiddenOpts = [begin, end, timespan, dayObs]

133 if any(x is not None for x in forbiddenOpts):

134 raise ValueError("You can't specify both an event and a begin/end or timespan or dayObs")

135 begin = event.begin

136 end = event.end

137 return begin, end

138

139 # check for dayObs, and that other options aren't inconsistently specified

140 if dayObs is not None:

141 forbiddenOpts = [begin, end, timespan]

142 if any(x is not None for x in forbiddenOpts):

143 raise ValueError("You can't specify both a dayObs and a begin/end or timespan")

144 begin = getDayObsStartTime(dayObs)

145 end = getDayObsEndTime(dayObs)

146 return begin, end

147 # can now disregard dayObs entirely

148

149 if begin is None:

150 raise ValueError("You must specify either a dayObs or a begin/end or begin/timespan")

151 # can now rely on begin, so just need to deal with end/timespan

152

153 if end is None and timespan is None:

154 raise ValueError("If you specify a begin, you must specify either a end or a timespan")

155 if end is not None and timespan is not None:

156 raise ValueError("You can't specify both a end and a timespan")

157 if end is None:

158 assert timespan is not None

159 if timespan > datetime.timedelta(minutes=0):

160 end = begin + timespan # the normal case

161 else:

162 end = begin # the case where timespan is negative

163 begin = begin + timespan # adding the negative to the start, i.e. subtracting it to bring back

164

165 assert begin is not None

166 assert end is not None

167 return begin, end

168

169

170def getEfdData(

171 client: EfdClient,

172 topic: str,

173 *,

174 columns: list[str] | None = None,

175 prePadding: float = 0,

176 postPadding: float = 0,

177 dayObs: int | None = None,

178 begin: Time | None = None,

179 end: Time | None = None,

180 timespan: TimeDelta | None = None,

181 event: TMAEvent | None = None,

182 expRecord: DimensionRecord | None = None,

183 warn: bool = True,

184 raiseIfTopicNotInSchema: bool = True,

185) -> DataFrame:

186 """Get one or more EFD topics over a time range, synchronously.

187

188 The time range can be specified as either:

189 * a dayObs, in which case the full 24 hour period is used,

190 * a begin point and a end point,

191 * a begin point and a timespan.

192 * a mount event

193 * an exposure record

194 If it is desired to use an end time with a timespan, just specify it as the

195 begin time and use a negative timespan.

196

197 The results from all topics are merged into a single dataframe.

198

199 `raiseIfTopicNotInSchema` should only be set to `False` when running on the

200 summit or in utility code for topics which might have had no data taken

201 within the last <data_retention_period> (nominally 30 days). Once a topic

202 is in the schema at USDF it will always be there, and thus users there

203 never need worry about this, always using `False` will be fine. However, at

204 the summit things are a little less predictable, so something missing from

205 the schema doesn't necessarily mean a typo, and utility code shouldn't

206 raise when data has been expunged.

207

208 Parameters

209 ----------

210 client : `lsst_efd_client.efd_helper.EfdClient`

211 The EFD client to use.

212 topic : `str`

213 The topic to query.

214 columns : `list` of `str`, optional

215 The columns to query. If not specified, all columns are queried.

216 prePadding : `float`

217 The amount of time before the nominal start of the query to include, in

218 seconds.

219 postPadding : `float`

220 The amount of extra time after the nominal end of the query to include,

221 in seconds.

222 dayObs : `int`, optional

223 The dayObs to query. If specified, this is used to determine the begin

224 and end times.

225 begin : `astropy.Time`, optional

226 The begin time for the query. If specified, either a end time or a

227 timespan must be supplied.

228 end : `astropy.Time`, optional

229 The end time for the query. If specified, a begin time must also be

230 supplied.

231 timespan : `astropy.TimeDelta`, optional

232 The timespan for the query. If specified, a begin time must also be

233 supplied.

234 event : `lsst.summit.utils.efdUtils.TMAEvent`, optional

235 The event to query. If specified, this is used to determine the begin

236 and end times, and all other options are disallowed.

237 expRecord : `lsst.daf.butler.dimensions.DimensionRecord`, optional

238 The exposure record containing the timespan to query. If specified, all

239 other options are disallowed.

240 warn : bool, optional

241 If ``True``, warn when no data is found. Exists so that utility code

242 can disable warnings when checking for data, and therefore defaults to

243 ``True``.

244 raiseIfTopicNotInSchema : `bool`, optional

245 Whether to raise an error if the topic is not in the EFD schema.

246

247 Returns

248 -------

249 data : `pd.DataFrame`

250 The merged data from all topics.

251

252 Raises

253 ------

254 ValueError:

255 If the topics are not in the EFD schema.

256 ValueError:

257 If both a dayObs and a begin/end or timespan are specified.

258 ValueError:

259 If a begin time is specified but no end time or timespan.

260

261 """

262 # TODO: DM-40100 ideally should calls mpts as necessary so that users

263 # needn't care if things are packed

264

265 # supports aliases so that you can query with them. If there is no entry in

266 # the alias dict then it queries with the supplied key. The fact the schema

267 # is now being checked means this shouldn't be a problem now.

268

269 # TODO: RFC-948 Move this import back to top of file once is implemented.

270 import nest_asyncio

271

272 begin, end = _getBeginEnd(dayObs, begin, end, timespan, event, expRecord)

273 begin -= TimeDelta(prePadding, format="sec")

274 end += TimeDelta(postPadding, format="sec")

275

276 nest_asyncio.apply()

277 loop = asyncio.get_event_loop()

278 ret = loop.run_until_complete(

279 _getEfdData(

280 client=client,

281 topic=topic,

282 begin=begin,

283 end=end,

284 columns=columns,

285 raiseIfTopicNotInSchema=raiseIfTopicNotInSchema,

286 )

287 )

288 if ret.empty and warn:

289 log = logging.getLogger(__name__)

290 msg = ""

291 if raiseIfTopicNotInSchema:

292 f"Topic {topic} is in the schema, but "

293 msg += "no data was returned by the query for the specified time range"

294 log.warning(msg)

295 return ret

296

297

298async def _getEfdData(

299 client: EfdClient,

300 topic: str,

301 begin: Time,

302 end: Time,

303 columns: list[str] | None = None,

304 raiseIfTopicNotInSchema: bool = True,

305) -> DataFrame:

306 """Get data for a topic from the EFD over the specified time range.

307

308 Parameters

309 ----------

310 client : `lsst_efd_client.efd_helper.EfdClient`

311 The EFD client to use.

312 topic : `str`

313 The topic to query.

314 begin : `astropy.Time`

315 The begin time for the query.

316 end : `astropy.Time`

317 The end time for the query.

318 columns : `list` of `str`, optional

319 The columns to query. If not specified, all columns are returned.

320 raiseIfTopicNotInSchema : `bool`, optional

321 Whether to raise an error if the topic is not in the EFD schema.

322

323 Returns

324 -------

325 data : `pd.DataFrame`

326 The data from the query.

327 """

328 if columns is None:

329 columns = ["*"]

330 columns = list(ensure_iterable(columns))

331

332 availableTopics = await client.get_topics()

333

334 if topic not in availableTopics:

335 if raiseIfTopicNotInSchema:

336 raise ValueError(f"Topic {topic} not in EFD schema")

337 else:

338 log = logging.getLogger(__name__)

339 log.debug(f"Topic {topic} not in EFD schema, returning empty DataFrame")

340 return pd.DataFrame()

341

342 data = await client.select_time_series(topic, columns, begin.utc, end.utc)

343

344 return data

345

346

347def getMostRecentRowWithDataBefore(

348 client: EfdClient,

349 topic: str,

350 timeToLookBefore: Time,

351 warnStaleAfterNMinutes: float | int = 60 * 12,

352 maxSearchNMinutes: float | int | None = None,

353 where: Callable[[DataFrame], list[bool]] | None = None,

354 raiseIfTopicNotInSchema: bool = True,

355) -> Series:

356 """Get the most recent row of data for a topic before a given time.

357

358 Parameters

359 ----------

360 client : `lsst_efd_client.efd_helper.EfdClient`

361 The EFD client to use.

362 topic : `str`

363 The topic to query.

364 timeToLookBefore : `astropy.Time`

365 The time to look before.

366 warnStaleAfterNMinutes : `float`, optional

367 The number of minutes after which to consider the data stale and issue

368 a warning.

369 maxSearchNMinutes: `float` or None, optional

370 Maximum number of minutes to search before raising ValueError.

371 where: `Callable` or None, optional

372 A callable taking a single pd.Dataframe argument and returning a

373 boolean list indicating rows to consider.

374 raiseIfTopicNotInSchema : `bool`, optional

375 Whether to raise an error if the topic is not in the EFD schema.

376

377 Returns

378 -------

379 row : `pd.Series`

380 The row of data from the EFD containing the most recent data before the

381 specified time.

382

383 Raises

384 ------

385 ValueError:

386 If the topic is not in the EFD schema.

387 """

388 staleAge = datetime.timedelta(warnStaleAfterNMinutes)

389

390 earliest = getDayObsStartTime(20190101)

391

392 if timeToLookBefore < earliest:

393 raise ValueError(f"Requested time {timeToLookBefore} is before any data was put in the EFD")

394

395 if maxSearchNMinutes is not None:

396 earliest = max(earliest, timeToLookBefore - maxSearchNMinutes * u.min)

397

398 df = pd.DataFrame()

399 beginTime = timeToLookBefore

400 while df.empty and beginTime > earliest:

401 df = getEfdData(

402 client,

403 topic,

404 begin=beginTime,

405 timespan=-TIME_CHUNKING,

406 warn=False,

407 raiseIfTopicNotInSchema=raiseIfTopicNotInSchema,

408 )

409 beginTime -= TIME_CHUNKING

410 if where is not None and not df.empty:

411 df = df[where(df)]

412

413 if beginTime < earliest and df.empty: # search ended early

414 out = f"EFD searched backwards from {timeToLookBefore} to {earliest} and no data "

415 if where is not None:

416 out += "consistent with `where` predicate "

417 out += f"was found in {topic=}"

418 raise ValueError(out)

419

420 lastRow = df.iloc[-1]

421 commandTime = efdTimestampToAstropy(lastRow["private_efdStamp"])

422

423 commandAge = timeToLookBefore - commandTime

424 if commandAge > staleAge:

425 log = logging.getLogger(__name__)

426 log.warning(

427 f"Component {topic} was last set {commandAge.sec / 60:.1} minutes before the requested time"

428 )

429

430 return lastRow

431

432

433def makeEfdClient(testing: bool | None = False, databaseName: str | None = None) -> EfdClient:

434 """Automatically create an EFD client based on the site.

435

436 Parameters

437 ----------

438 testing : `bool`

439 Set to ``True`` if running in a test suite. This will default to using

440 the USDF EFD, for which data has been recorded for replay by the ``vcr`

441 package. Note data must be re-recorded to ``vcr`` from both inside and

442 outside the USDF when the package/data changes, due to the use of a

443 proxy meaning that the web requests are different depending on whether

444 the EFD is being contacted from inside and outside the USDF.

445 databaseName : `str`, optional

446 Name of the database within influxDB to query. If not provided, the

447 default specified by EfdClient() is used.

448

449 Returns

450 -------

451 efdClient : `lsst_efd_client.efd_helper.EfdClient`, optional

452 The EFD client to use for the current site.

453 """

454 efdKwargs: dict[str, Any] = {}

455 if databaseName is not None:

456 efdKwargs["db_name"] = databaseName

457

458 if not HAS_EFD_CLIENT:

459 raise RuntimeError("Could not create EFD client because importing lsst_efd_client failed.")

460

461 if testing:

462 return EfdClient("usdf_efd", **efdKwargs)

463

464 site = getSite()

465 if site == "UNKNOWN":

466 raise RuntimeError("Could not create EFD client as the site could not be determined")

467

468 if site == "summit":

469 return EfdClient("summit_efd", **efdKwargs)

470 if site == "tucson":

471 return EfdClient("tucson_teststand_efd", **efdKwargs)

472 if site == "base":

473 return EfdClient("base_efd", **efdKwargs)

474 if site in ["staff-rsp", "rubin-devl", "usdf-k8s"]:

475 return EfdClient("usdf_efd", **efdKwargs)

476 if site == "jenkins":

477 return EfdClient("usdf_efd", **efdKwargs)

478

479 raise RuntimeError(f"Could not create EFD client as the {site=} is not recognized")

480

481

482def expRecordToTimespan(expRecord: DimensionRecord) -> dict:

483 """Get the timespan from an exposure record.

484

485 Returns the timespan in a format where it can be used to directly unpack

486 into a efdClient.select_time_series() call.

487

488 Parameters

489 ----------

490 expRecord : `lsst.daf.butler.DimensionRecord`

491 The exposure record.

492

493 Returns

494 -------

495 timespanDict : `dict`

496 The timespan in a format that can be used to directly unpack into a

497 efdClient.select_time_series() call.

498 """

499 return {

500 "begin": expRecord.timespan.begin.utc,

501 "end": expRecord.timespan.end.utc,

502 }

503

504

505def efdTimestampToAstropy(timestamp: float) -> Time:

506 """Get an efd timestamp as an astropy.time.Time object.

507

508 Parameters

509 ----------

510 timestamp : `float`

511 The timestamp, as a float.

512

513 Returns

514 -------

515 time : `astropy.time.Time`

516 The timestamp as an astropy.time.Time object.

517 """

518 return Time(timestamp, format="unix")

519

520

521def astropyToEfdTimestamp(time: Time) -> float:

522 """Get astropy Time object as an efd timestamp

523

524 Parameters

525 ----------

526 time : `astropy.time.Time`

527 The time as an astropy.time.Time object.

528

529 Returns

530 -------

531 timestamp : `float`

532 The timestamp, in UTC, in unix seconds.

533 """

534

535 return time.utc.unix

536

537

538def clipDataToEvent(

539 df: DataFrame,

540 event: TMAEvent,

541 prePadding: float = 0,

542 postPadding: float = 0,

543 logger: logging.Logger | None = None,

544) -> DataFrame:

545 """Clip a padded dataframe to an event.

546

547 Parameters

548 ----------

549 df : `pd.DataFrame`

550 The dataframe to clip.

551 event : `lsst.summit.utils.efdUtils.TMAEvent`

552 The event to clip to.

553 prePadding : `float`, optional

554 The amount of time before the nominal start of the event to include, in

555 seconds.

556 postPadding : `float`, optional

557 The amount of extra time after the nominal end of the event to include,

558 in seconds.

559 logger : `logging.Logger`, optional

560 The logger to use. If not specified, a new one is created.

561

562 Returns

563 -------

564 clipped : `pd.DataFrame`

565 The clipped dataframe.

566 """

567 begin = event.begin.value - prePadding

568 end = event.end.value + postPadding

569

570 if logger is None:

571 logger = logging.getLogger(__name__)

572

573 if begin < df["private_efdStamp"].min():

574 logger.warning(f"Requested begin time {begin} is before the start of the data")

575 if end > df["private_efdStamp"].max():

576 logger.warning(f"Requested end time {end} is after the end of the data")

577

578 mask = (df["private_efdStamp"] >= begin) & (df["private_efdStamp"] <= end)

579 clipped_df = df.loc[mask].copy()

580 return clipped_df

581

582

583def offsetDayObs(dayObs: int, nDays: int) -> int:

584 """Offset a dayObs by a given number of days.

585

586 Parameters

587 ----------

588 dayObs : `int`

589 The dayObs, as an integer, e.g. 20231225

590 nDays : `int`

591 The number of days to offset the dayObs by.

592

593 Returns

594 -------

595 newDayObs : `int`

596 The new dayObs, as an integer, e.g. 20231225

597 """

598 d1 = datetime.datetime.strptime(str(dayObs), "%Y%m%d")

599 oneDay = datetime.timedelta(days=nDays)

600 return int((d1 + oneDay).strftime("%Y%m%d"))

601

602

603def calcNextDay(dayObs: int) -> int:

604 """Given an integer dayObs, calculate the next integer dayObs.

605

606 Integers are used for dayObs, but dayObs values are therefore not

607 contiguous due to month/year ends etc, so this utility provides a robust

608 way to get the integer dayObs which follows the one specified.

609

610 Parameters

611 ----------

612 dayObs : `int`

613 The dayObs, as an integer, e.g. 20231231

614

615 Returns

616 -------

617 nextDayObs : `int`

618 The next dayObs, as an integer, e.g. 20240101

619 """

620 return offsetDayObs(dayObs, 1)

621

622

623def calcPreviousDay(dayObs: int) -> int:

624 """Given an integer dayObs, calculate the next integer dayObs.

625

626 Integers are used for dayObs, but dayObs values are therefore not

627 contiguous due to month/year ends etc, so this utility provides a robust

628 way to get the integer dayObs which follows the one specified.

629

630 Parameters

631 ----------

632 dayObs : `int`

633 The dayObs, as an integer, e.g. 20231231

634

635 Returns

636 -------

637 nextDayObs : `int`

638 The next dayObs, as an integer, e.g. 20240101

639 """

640 return offsetDayObs(dayObs, -1)

641

642

643def getDayObsStartTime(dayObs: int) -> astropy.time.Time:

644 """Get the start of the given dayObs as an astropy.time.Time object.

645

646 The observatory rolls the date over at UTC-12.

647

648 Parameters

649 ----------

650 dayObs : `int`

651 The dayObs, as an integer, e.g. 20231225

652

653 Returns

654 -------

655 time : `astropy.time.Time`

656 The start of the dayObs as an astropy.time.Time object.

657 """

658 pythonDateTime = datetime.datetime.strptime(str(dayObs), "%Y%m%d")

659 return Time(pythonDateTime) + 12 * u.hour

660

661

662def getDayObsEndTime(dayObs: int) -> Time:

663 """Get the end of the given dayObs as an astropy.time.Time object.

664

665 Parameters

666 ----------

667 dayObs : `int`

668 The dayObs, as an integer, e.g. 20231225

669

670 Returns

671 -------

672 time : `astropy.time.Time`

673 The end of the dayObs as an astropy.time.Time object.

674 """

675 return getDayObsStartTime(dayObs) + 24 * u.hour

676

677

678def getDayObsForTime(time: Time) -> int:

679 """Get the dayObs in which an astropy.time.Time object falls.

680

681 Parameters

682 ----------

683 time : `astropy.time.Time`

684 The time.

685

686 Returns

687 -------

688 dayObs : `int`

689 The dayObs, as an integer, e.g. 20231225

690 """

691 twelveHours = datetime.timedelta(hours=-12)

692 offset = TimeDelta(twelveHours, format="datetime")

693 return int((time + offset).utc.isot[:10].replace("-", ""))

694

695

696@deprecated(

697 reason="getSubTopics() has been replaced by getTopics() and using wildcards. "

698 "Will be removed after w_2023_50.",

699 version="w_2023_40",

700 category=FutureWarning,

701)

702def getSubTopics(client: EfdClient, topic: str) -> list[str]:

703 """Get all the sub topics within a given topic.

704

705 Note that the topic need not be a complete one, for example, rather than

706 doing `getSubTopics(client, 'lsst.sal.ATMCS')` to get all the topics for

707 the AuxTel Mount Control System, you can do `getSubTopics(client,

708 'lsst.sal.AT')` to get all which relate to the AuxTel in general.

709

710 Parameters

711 ----------

712 client : `lsst_efd_client.efd_helper.EfdClient`

713 The EFD client to use.

714 topic : `str`

715 The topic to query.

716

717 Returns

718 -------

719 subTopics : `list` of `str`

720 The sub topics.

721 """

722 loop = asyncio.get_event_loop()

723 topics = loop.run_until_complete(client.get_topics())

724 return sorted([t for t in topics if t.startswith(topic)])

725

726

727def getTopics(client: EfdClient, toFind: str, caseSensitive: bool = False) -> list[str]:

728 """Return all the strings in topics which match the topic query string.

729

730 Supports wildcards, which are denoted as `*``, as per shell globs.

731

732 Example:

733 >>> # assume topics are ['apple', 'banana', 'grape']

734 >>> getTopics(, 'a*p*')

735 ['apple', 'grape']

736

737 Parameters

738 ----------

739 client : `lsst_efd_client.efd_helper.EfdClient`

740 The EFD client to use.

741 toFind : `str`

742 The query string, with optional wildcards denoted as *.

743 caseSensitive : `bool`, optional

744 If ``True``, the query is case sensitive. Defaults to ``False``.

745

746 Returns

747 -------

748 matches : `list` of `str`

749 The list of matching topics.

750 """

751 loop = asyncio.get_event_loop()

752 topics = loop.run_until_complete(client.get_topics())

753

754 # Replace wildcard with regex equivalent

755 pattern = toFind.replace("*", ".*")

756 flags = re.IGNORECASE if not caseSensitive else 0

757

758 matches = []

759 for topic in topics:

760 if re.match(pattern, topic, flags):

761 matches.append(topic)

762

763 return matches

764

765

766def getCommands(

767 client: EfdClient,

768 commands: list[str],

769 begin: Time,

770 end: Time,

771 prePadding: float,

772 postPadding: float,

773 timeFormat: str = "python",

774) -> dict[Time | Timestamp | datetime.datetime, str]:

775 """Retrieve the commands issued within a specified time range.

776

777 Parameters

778 ----------

779 client : `EfdClient`

780 The client object used to retrieve EFD data.

781 commands : `list`

782 A list of commands to retrieve.

783 begin : `astropy.time.Time`

784 The start time of the time range.

785 end : `astropy.time.Time`

786 The end time of the time range.

787 prePadding : `float`

788 The amount of time to pad before the begin time.

789 postPadding : `float`

790 The amount of time to pad after the end time.

791 timeFormat : `str`

792 One of 'pandas' or 'astropy' or 'python'. If 'pandas', the dictionary

793 keys will be pandas timestamps, if 'astropy' they will be astropy times

794 and if 'python' they will be python datetimes.

795

796 Returns

797 -------

798 commandTimes : `dict` [`time`, `str`]

799 A dictionary of the times at which the commands where issued. The type

800 that `time` takes is determined by the format key, and defaults to

801 python datetime.

802

803 Raises

804 ------

805 ValueError

806 Raise if there is already a command at a timestamp in the dictionary,

807 i.e. there is a collision.

808 """

809 if timeFormat not in ["pandas", "astropy", "python"]:

810 raise ValueError(f"format must be one of 'pandas', 'astropy' or 'python', not {timeFormat=}")

811

812 commands = list(ensure_iterable(commands))

813

814 commandTimes: dict[Time | Timestamp | datetime.datetime, str] = {}

815 for command in commands:

816 data = getEfdData(

817 client,

818 command,

819 begin=begin,

820 end=end,

821 prePadding=prePadding,

822 postPadding=postPadding,

823 warn=False, # most commands will not be issue so we expect many empty queries

824 raiseIfTopicNotInSchema=False,

825 )

826 for time, _ in data.iterrows():

827 # this is much the most simple data structure, and the chance

828 # of commands being *exactly* simultaneous is minimal so try

829 # it like this, and just raise if we get collisions for now. So

830 # far in testing this seems to be just fine.

831

832 timeKey = None

833 match timeFormat:

834 case "pandas":

835 timeKey = time

836 case "astropy":

837 timeKey = Time(time)

838 case "python":

839 assert isinstance(time, pd.Timestamp)

840 timeKey = time.to_pydatetime()

841

842 if timeKey in commandTimes:

843 msg = f"There is already a command at {timeKey=} - make a better data structure!"

844 msg += f"Colliding commands = {commandTimes[timeKey]} and {command}"

845 raise ValueError(msg)

846 commandTimes[timeKey] = command

847 return commandTimes