fix: remove unnecessary float casting when writing timestamps

Author: pyuxiang

src/fpfind/lib/parse_timestamps.py

@@ -668,11 +668,23 @@ def _consolidate_events(
     Note:
         float128 is needed, since float64 only encodes 53-bits of precision,
         while the high resolution timestamp has 54-bits precision.
+
+        Output format has the base 4ps resolution (finest), so no need to
+        perform a lossy conversion to floating if input timestamps are already
+        integers (which can only have coarser resolution).
     """
-    data = (
-        np.array(t, dtype=NP_PRECISEFLOAT) * (TSRES.PS4.value // resolution.value)
-    ).astype(np.uint64) << np.uint64(10)
-    data += np.array(p).astype(np.uint64)
+    # Convert to 4ps resolution then coerce as integer
+    if not np.issubdtype(t.dtype, np.integer):
+        data = np.array(t, dtype=NP_PRECISEFLOAT)  # copy with maximal accuracy
+    else:
+        data = np.array(t, dtype=np.uint64)
+    if resolution is not TSRES.PS4:  # short-circuit
+        data *= TSRES.PS4.value // resolution.value
+    data = np.asarray(data, dtype=np.uint64)
+
+    # Add detector pattern
+    data <<= np.uint64(10)
+    data += np.asarray(p, dtype=np.uint64)
     if sort:
         data = np.sort(data)
 

tests/parse_timestamps/test_correctness_write.py

@@ -5,6 +5,10 @@
 import fpfind.lib.parse_timestamps as parser
 from fpfind import TSRES
 
+LOAD_RAW = {
+    "resolution": TSRES.PS4,
+    "fractional": False,
+}
 
 def _write_a1_deprec(
     filename,
@@ -43,6 +47,22 @@ def test_write_a1(path_timestamp_a1, tmp_path):
     assert np.all(ps == _ps)
 
 
+def test_write_a1_4ps(path_timestamp_a1, tmp_path):
+    target = tmp_path / "ts"
+    ts, ps = parser.read_a1(path_timestamp_a1, legacy=True, **LOAD_RAW)
+
+    parser.write_a1(target, ts, ps, resolution=TSRES.PS4)
+    _ts, _ps = parser.read_a1(target, **LOAD_RAW)
+    assert np.all(ts == _ts)
+    assert np.all(ps == _ps)
+
+    # With legacy mode
+    parser.write_a1(target, ts, ps, legacy=True, resolution=TSRES.PS4)
+    _ts, _ps = parser.read_a1(target, legacy=True, **LOAD_RAW)
+    assert np.all(ts == _ts)
+    assert np.all(ps == _ps)
+
+
 def test_write_a1_deprec(path_timestamp_a1, tmp_path):
     target = tmp_path / "ts"
     target_deprec = tmp_path / "ts.deprec"