Fix the ITS ramp-up shift rounding and the Pylint errors on master

This fixes the millisecond rounding of the TF-aligned ITS ramp-up shift, updates the ramp-up test to the new signature and clears the remaining Pylint error.

- The ramp-up is now rounded up to whole milliseconds before it is rounded up to whole TFs, so the shifted timestamp can no longer lie inside the ramp.
- test_anchoring_rampup.py passes the TF length to shift_anchor_past_ITS_rampup and has a regression test for a ramp ending just below a TF boundary.
- ccdb_cross_check in getCCDBTimeMachineTimestamp.py keeps the pinned object in a local variable, which Pylint can follow.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Author: sawenzel

GRID/utils/getCCDBTimeMachineTimestamp.py

@@ -86,11 +86,9 @@ def ccdb_cross_check(path, not_after, timeout=15):
 
     as_of = [o for o in objects if o.get("Created", 0) <= not_after]
     newest = max(objects, key=lambda o: o.get("Created", 0))
-    result = {"newest": newest, "as_of": None, "outdated": False}
-    if as_of:
-        result["as_of"] = max(as_of, key=lambda o: o.get("Created", 0))
-        result["outdated"] = newest["Created"] > result["as_of"]["Created"]
-    return result
+    pinned = max(as_of, key=lambda o: o.get("Created", 0)) if as_of else None
+    outdated = pinned is not None and newest["Created"] > pinned["Created"]
+    return {"newest": newest, "as_of": pinned, "outdated": outdated}
 
 
 def describe_object(o):

MC/bin/o2dpg_sim_workflow_anchored.py

@@ -336,9 +336,9 @@ def shift_anchor_past_ITS_rampup(run_start, first_orbit, orbitsPerTF, ITS_rampup
     timestamp leaves a job at production offset 0 inside the ramp, where the ITS
     time-dead map masks every chip.
     """
-    # convert ITS_rampup to orbits multiple to orbitsPerTF
-    rampupOrbits = ((milliseconds_to_orbits(ITS_rampup) + orbitsPerTF - 1) // orbitsPerTF ) * orbitsPerTF
-    
+    # round the ramp-up up to whole milliseconds, then to a whole number of timeframes;
+    # the truncated millisecond shift then still ends after the ramp and before the orbit
+    rampupOrbits = -(-milliseconds_to_orbits(math.ceil(ITS_rampup)) // orbitsPerTF) * orbitsPerTF
     return run_start + int(rampupOrbits * LHCOrbitMUS / 1000.), first_orbit + rampupOrbits
 
 def retrieve_MinBias_CTPScaler_Rate(raw_rate_at, finaltime, trig_eff_arg, NBunches, ColSystem, eCM, run_number = -1):

MC/bin/tests/test_anchoring_rampup.py

@@ -21,6 +21,7 @@
 FIRST_ORBIT = 20505888
 SOR = 1778806732526
 ITS_RAMPUP_MS = 5000
+ORBITS_PER_TF = 128
 FIRST_ALIVE_ORBIT = 20539968
 
 
@@ -36,32 +37,40 @@ def test_orbits_from_milliseconds(self):
 
     def test_both_coordinates_move(self):
         """A ramp-up of a few seconds must move the orbit as well as the timestamp."""
-        start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ITS_RAMPUP_MS)
-        self.assertEqual(start, SOR + ITS_RAMPUP_MS)
+        start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, ITS_RAMPUP_MS)
+        self.assertGreaterEqual(start, SOR + ITS_RAMPUP_MS)
         self.assertGreater(orbit, FIRST_ORBIT)
+        self.assertEqual((orbit - FIRST_ORBIT) % ORBITS_PER_TF, 0)
 
     def test_nothing_moves_without_a_ramp(self):
-        self.assertEqual(anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, 0),
+        self.assertEqual(anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, 0),
                          (SOR, FIRST_ORBIT))
 
     def test_shifted_orbit_is_never_inside_the_ramp(self):
         """The shifted orbit must sit at or after the shifted timestamp, never before."""
         for ramp_ms in (0, 1, 500, ITS_RAMPUP_MS, 30000):
-            start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ramp_ms)
+            start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, ramp_ms)
             time_of_orbit = SOR + (orbit - FIRST_ORBIT) * anchored.LHCOrbitMUS / 1000.
             self.assertGreaterEqual(time_of_orbit, start,
                                     f"orbit shift falls short of the ramp for {ramp_ms} ms")
 
     def test_shift_agrees_with_the_timestamp_to_orbit_conversion(self):
         """Closure: the shifted orbit is what main() derives from the shifted timestamp."""
-        start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ITS_RAMPUP_MS)
+        start, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, ITS_RAMPUP_MS)
         # this is the conversion main() uses for the exclude_timestamp() check
         derived = FIRST_ORBIT + int((start - SOR) / (anchored.LHCOrbitMUS / 1000.))
-        self.assertLessEqual(abs(orbit - derived), 1)
+        # the timestamp has millisecond resolution, which is about 11 orbits
+        self.assertLessEqual(abs(orbit - derived), anchored.milliseconds_to_orbits(1))
+
+    def test_timestamp_is_not_inside_the_ramp_at_a_tf_boundary(self):
+        """Regression: a ramp ending just below a TF boundary must not be cut short by the ms truncation."""
+        ramp_ms = 12928 * anchored.LHCOrbitMUS / 1000. - 0.001
+        start, _ = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, ramp_ms)
+        self.assertGreaterEqual(start - SOR, ramp_ms)
 
     def test_split_id_one_clears_the_its_dead_window(self):
         """Regression: the first job of a production must not sample the dead window."""
-        _, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ITS_RAMPUP_MS)
+        _, orbit = anchored.shift_anchor_past_ITS_rampup(SOR, FIRST_ORBIT, ORBITS_PER_TF, ITS_RAMPUP_MS)
         self.assertGreater(orbit, FIRST_ALIVE_ORBIT)