DM-40948: Add TMA eventMaker function for getting a specific event

python/lsst/summit/utils/tmaUtils.py

@@ -1002,6 +1002,41 @@ def _mergeData(self, data):
 
         return merged
 
+    def getEvent(self, dayObs, seqNum):
+        """Get a specific event for a given dayObs and seqNum.
+
+        Repeated calls for the same ``dayObs`` will use the cached data if the
+        day is in the past, and so will be much quicker. If the ``dayObs`` is
+        the current day then the EFD will be queried for new data for each
+        call, so a call which returns ```None` on the first try might return an
+        event on the next, if the TMA is still moving and thus generating
+        events.
+
+        Parameters
+        ----------
+        dayObs : `int`
+            The dayObs to get the event for.
+        seqNum : `int`
+            The sequence number of the event to get.
+
+        Returns
+        -------
+        event : `lsst.summit.utils.tmaUtils.TMAEvent`
+            The event for the specified dayObs and seqNum, or `None` if the
+            event was not found.
+        """
+        events = self.getEvents(dayObs)
+        if seqNum <= len(events):
+            event = events[seqNum]
+            if event.seqNum != seqNum:
+                # it's zero-indexed and contiguous so this must be true but
+                # a sanity check doesn't hurt.
+                raise AssertionError(f"Event sequence number mismatch: {event.seqNum} != {seqNum}")
+            return event
+        else:
+            self.log.warning(f"Event {seqNum} not found for {dayObs}")
+            return None
+
     def getEvents(self, dayObs):
         """Get the TMA events for the specified dayObs.
 

tests/test_tmaUtils.py

@@ -321,10 +321,13 @@ def test_endToEnd(self):
     def test_noDataBehaviour(self):
         eventMaker = self.tmaEventMaker
         noDataDayObs = 19500101  # do not use 19700101 - there is data for that day!
-        with self.assertWarns(Warning, msg=f"No EFD data found for dayObs={noDataDayObs}"):
+        with self.assertLogs(level='WARNING') as cm:
+            correctMsg = f"No EFD data found for dayObs={noDataDayObs}"
             events = eventMaker.getEvents(noDataDayObs)
             self.assertIsInstance(events, list)
             self.assertEqual(len(events), 0)
+            msg = cm.output[0]
+            self.assertIn(correctMsg, msg)
 
     @vcr.use_cassette()
     def test_helperFunctions(self):
@@ -338,6 +341,25 @@ def test_helperFunctions(self):
         self.assertEqual(slews, foundSlews)
         self.assertEqual(tracks, foundTracks)
 
+    @vcr.use_cassette()
+    def test_getEvent(self):
+        # test the singular event getter, and what happens if the event doesn't
+        # exist for the day
+        eventMaker = self.tmaEventMaker
+        events = eventMaker.getEvents(self.dayObs)
+        nEvents = len(events)
+
+        event = eventMaker.getEvent(self.dayObs, 0)
+        self.assertIsInstance(event, TMAEvent)
+        event = eventMaker.getEvent(self.dayObs, 100)
+        self.assertIsInstance(event, TMAEvent)
+
+        with self.assertLogs(level='WARNING') as cm:
+            correctMsg = f"Event {nEvents+1} not found for {self.dayObs}"
+            event = eventMaker.getEvent(self.dayObs, nEvents+1)
+            msg = cm.output[0]
+            self.assertIn(correctMsg, msg)
+
     @vcr.use_cassette()
     def test_printing(self):
         eventMaker = self.tmaEventMaker