Upcast timing columns to float32 in Arrow export to avoid halffloat interop issues (#43)

* Upcast timing columns to float32 in Arrow export to avoid halffloat interop issues

* add fix for issue 38

* add comment for current limitation

---------

Co-authored-by: Stefan Taubert <23339395+stefantaubert@users.noreply.github.com>

Author: fegue

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Bugfixes
 
 - Fixed acoustic inference session being aborted on macOS when stats were enabled: hardened parent/child memory tracking against `psutil.AccessDenied`, and replaced the two tracked semaphores with a wrapper that mirrors the count into shared memory so `get_value()` works on macOS (#39)
+- Fixed float16 quantization of segment timestamps in prediction results, which caused up to ±0.05 s drift in CSV/DataFrame/Parquet output (#38, #42). Also closed an analogous hole in encoding results where a hop duration that is exactly representable in float16 (e.g. hop=1.5) could still produce drifting accumulated timestamps. Timestamps are now always materialized at >= float32 precision at the source.
 
 ## [0.2.15] - 2026-05-02
 

docs/general.rst

@@ -26,3 +26,25 @@ A *Producer* loads only as much audio as the buffer can hold, keeping RAM usage
 * **Buffer Size** – By default, the buffer is set to twice the *Worker* count, ensuring that every *Worker* always has a pre-loaded batch to process and thus avoids idle time.
 * **Model Backends** – Each worker loads its own instance of the inference model. On the CPU, both **TFLite** and **Protocol Buffers** (Protobuf) models can be used; Protobuf models can optionally run on the GPU.
 * **Best Practice for CPU Inference** – For CPU-only execution on Linux, the number of *Worker* processes should not exceed the number of physical cores, as oversubscription typically leads to reduced performance. When running TFLite, keep the batch size to one (1); larger batches offer no throughput benefit.
+
+Known limitations
+----
+
+**End-time precision on the last segment of short files (≤ ~34 minutes).**
+For memory efficiency, per-file durations are stored in the smallest float
+dtype that covers their magnitude: ``float16`` for files up to 2\ :sup:`11` ≈
+2048 s, ``float32`` for files up to 2\ :sup:`24` s (~194 days), ``float64``
+beyond. The stored duration is used as the upper clamp when computing the
+``end_time`` of the *last* segment of each file. Inside the float16 range
+this rounding is visible: the largest representable float16 below ``X`` may
+differ from ``X`` by up to one ULP — about 0.06 s near 128 s, 0.25 s near
+1024 s, and 0.5 s near 2048 s. The error appears only on the very last
+segment per file and only when the actual file duration is not exactly
+representable in float16 (integer-second durations up to 2048 s are
+exact). For files of one hour or longer the storage dtype is float32, where
+the equivalent ULP is below 4 ms even at 12 h, so the effect is not
+observable in practice.
+
+All other timestamps (``start_time`` and ``end_time`` of every segment that
+does not hit the clamp) are computed at ≥ float32 precision regardless of
+file length.

src/birdnet/acoustic/inference/core/encoding/encoding_result.py

@@ -21,7 +21,6 @@
   format_input_for_csv,
   get_uint_dtype,
   hms_centis_fast,
-  upgrade_float_dtype_for_value,
 )
 
 if TYPE_CHECKING:
@@ -135,11 +134,11 @@ def to_structured_array(self) -> np.ndarray:
     embeddings_selected = self.embeddings[valid_file_idx, valid_seg_idx]
 
     hop_duration_s = self.hop_duration_s
-    # Upgrade the storage dtype for the output if it cannot represent hop
-    # exactly, otherwise rounding accumulates across segments.
-    time_dtype = upgrade_float_dtype_for_value(
-      self._input_durations.dtype, hop_duration_s
-    )
+    # Force at least float32 for timing columns. The bulk _input_durations
+    # array is stored in a magnitude-based dtype (float16 for files <= 2**11 s),
+    # which is too coarse for accumulated i*hop products and would also produce
+    # Arrow halffloat that some implementations (e.g. R) cannot read.
+    time_dtype = np.result_type(self._input_durations.dtype, np.float32)
 
     dtype = [
       (VAR_INPUT, self._input_dtype),

src/birdnet/acoustic/inference/core/prediction/prediction_result.py

@@ -153,10 +153,15 @@ def to_structured_array(self) -> np.ndarray:
     del valid_mask
 
     n_predictions = len(valid_indices[0])
+    # Force at least float32 for timing columns. The bulk _input_durations
+    # array is stored in a magnitude-based dtype (float16 for files <= 2**11 s),
+    # which is too coarse for accumulated i*hop products and would also produce
+    # Arrow halffloat that some implementations (e.g. R) cannot read.
+    time_dtype = np.result_type(self._input_durations.dtype, np.float32)
     dtype = [
       (VAR_INPUT, self._input_dtype),
-      (VAR_START_TIME, self._input_durations.dtype),
-      (VAR_END_TIME, self._input_durations.dtype),
+      (VAR_START_TIME, time_dtype),
+      (VAR_END_TIME, time_dtype),
       (VAR_SPECIES_NAME, object),
       (VAR_CONFIDENCE, self._species_probs.dtype),
     ]
@@ -191,15 +196,15 @@ def to_structured_array(self) -> np.ndarray:
     del sort_indices
 
     hop_duration_s = self.hop_duration_s
-    start_times = chunk_idx_flat.astype(self._input_durations.dtype) * hop_duration_s
+    start_times = chunk_idx_flat.astype(time_dtype) * hop_duration_s
     del hop_duration_s
     del chunk_idx_flat
 
     structured_array[VAR_START_TIME] = start_times
     structured_array[VAR_END_TIME] = np.minimum(
       start_times
       + apply_speed_to_duration(self._segment_duration_s[0], self._speed[0]),
-      self._input_durations[file_idx_flat],
+      self._input_durations[file_idx_flat].astype(time_dtype),
     )
     del start_times
     structured_array[VAR_INPUT] = self._inputs[file_idx_flat]

src/birdnet_tests/acoustic_models/inference/encoding/encoding_result_py/test_encoding_to_parquet.py

@@ -0,0 +1,158 @@
+from pathlib import Path
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from birdnet.acoustic.inference.core.encoding.encoding_result import (
+  AcousticFileEncodingResult,
+)
+from birdnet_tests.acoustic_models.inference.encoding.encoding_result_py.test_encoding_to_structured_array import (  # noqa: E501
+  create_file_encoding_result,
+)
+
+
+def _create_result_with_float16_durations() -> AcousticFileEncodingResult:
+  """Create an encoding result whose input_durations are float16."""
+  result = create_file_encoding_result(
+    n_files=2,
+    duration_s=12,
+    segment_duration_s=3.0,
+    overlap_duration_s=0.0,
+  )
+  assert result.input_durations.dtype == np.float16
+  return result
+
+
+def _create_result_with_float32_durations() -> AcousticFileEncodingResult:
+  """Create an encoding result whose input_durations are float32."""
+  result = create_file_encoding_result(
+    n_files=1,
+    duration_s=5000,
+    segment_duration_s=3.0,
+    overlap_duration_s=0.0,
+  )
+  assert result.input_durations.dtype == np.float32
+  return result
+
+
+def _create_result_with_float64_durations() -> AcousticFileEncodingResult:
+  """Create an encoding result whose input_durations are float64.
+
+  Uses a small duration for speed, then coerces dtype to float64 to exercise
+  the Arrow type-promotion path without creating millions of segments.
+  """
+  result = create_file_encoding_result(
+    n_files=1,
+    duration_s=12,
+    segment_duration_s=3.0,
+    overlap_duration_s=0.0,
+  )
+  result._input_durations = result._input_durations.astype(np.float64)
+  assert result.input_durations.dtype == np.float64
+  return result
+
+
+def test_arrow_table_time_columns_are_float32_when_durations_float16() -> None:
+  result = _create_result_with_float16_durations()
+  table = result.to_arrow_table()
+
+  assert table.schema.field("start_time").type == pa.float32()
+  assert table.schema.field("end_time").type == pa.float32()
+
+
+def test_arrow_table_time_columns_are_float32_when_durations_float32() -> None:
+  result = _create_result_with_float32_durations()
+  table = result.to_arrow_table()
+
+  assert table.schema.field("start_time").type == pa.float32()
+  assert table.schema.field("end_time").type == pa.float32()
+
+
+def test_arrow_table_time_columns_are_float64_when_durations_float64() -> None:
+  result = _create_result_with_float64_durations()
+  table = result.to_arrow_table()
+
+  assert table.schema.field("start_time").type == pa.float64()
+  assert table.schema.field("end_time").type == pa.float64()
+
+
+def test_parquet_roundtrip_schema_float16(tmp_path: Path) -> None:
+  result = _create_result_with_float16_durations()
+  out = tmp_path / "result.parquet"
+
+  result.to_parquet(out, silent=True)
+  table = pq.read_table(out)
+
+  assert table.schema.field("start_time").type == pa.float32()
+  assert table.schema.field("end_time").type == pa.float32()
+
+
+def test_parquet_roundtrip_values_float16(tmp_path: Path) -> None:
+  result = _create_result_with_float16_durations()
+  structured = result.to_structured_array()
+  out = tmp_path / "result.parquet"
+
+  expected_start = np.array(structured["start_time"], dtype=np.float64)
+  expected_end = np.array(structured["end_time"], dtype=np.float64)
+
+  result.to_parquet(out, silent=True)
+  table = pq.read_table(out)
+
+  actual_start = np.array(table.column("start_time").to_pylist(), dtype=np.float64)
+  actual_end = np.array(table.column("end_time").to_pylist(), dtype=np.float64)
+
+  np.testing.assert_allclose(expected_start, actual_start, rtol=1e-3)
+  np.testing.assert_allclose(expected_end, actual_end, rtol=1e-3)
+
+
+def test_parquet_roundtrip_values_float32(tmp_path: Path) -> None:
+  result = _create_result_with_float32_durations()
+  structured = result.to_structured_array()
+  out = tmp_path / "result.parquet"
+
+  expected_start = np.array(structured["start_time"], dtype=np.float64)
+  expected_end = np.array(structured["end_time"], dtype=np.float64)
+
+  result.to_parquet(out, silent=True)
+  table = pq.read_table(out)
+
+  actual_start = np.array(table.column("start_time").to_pylist(), dtype=np.float64)
+  actual_end = np.array(table.column("end_time").to_pylist(), dtype=np.float64)
+
+  np.testing.assert_allclose(expected_start, actual_start, rtol=1e-6)
+  np.testing.assert_allclose(expected_end, actual_end, rtol=1e-6)
+
+
+def test_parquet_roundtrip_values_float64(tmp_path: Path) -> None:
+  result = _create_result_with_float64_durations()
+  structured = result.to_structured_array()
+  out = tmp_path / "result.parquet"
+
+  expected_start = np.array(structured["start_time"], dtype=np.float64)
+  expected_end = np.array(structured["end_time"], dtype=np.float64)
+
+  result.to_parquet(out, silent=True)
+  table = pq.read_table(out)
+
+  actual_start = np.array(table.column("start_time").to_pylist(), dtype=np.float64)
+  actual_end = np.array(table.column("end_time").to_pylist(), dtype=np.float64)
+
+  np.testing.assert_allclose(expected_start, actual_start, rtol=1e-9)
+  np.testing.assert_allclose(expected_end, actual_end, rtol=1e-9)
+
+
+def test_parquet_time_columns_no_halffloat(tmp_path: Path) -> None:
+  """Ensure start_time and end_time never use halffloat in Parquet."""
+  result = _create_result_with_float16_durations()
+  out = tmp_path / "result.parquet"
+
+  result.to_parquet(out, silent=True)
+  table = pq.read_table(out)
+
+  for col_name in ("start_time", "end_time"):
+    field = table.schema.field(col_name)
+    assert field.type != pa.float16(), (
+      f"Column '{col_name}' uses halffloat (float16), "
+      f"which is not interoperable across Arrow implementations"
+    )

src/birdnet_tests/acoustic_models/inference/encoding/encoding_result_py/test_encoding_to_structured_array.py

@@ -366,7 +366,38 @@ def test_dtype_structure() -> None:
     "embedding",
   )
   assert structured.dtype["input"] == np.dtype("O")
-  assert structured.dtype["start_time"] == result._input_durations.dtype
-  assert structured.dtype["end_time"] == result._input_durations.dtype
+  expected_time_dtype = np.result_type(result._input_durations.dtype, np.float32)
+  assert structured.dtype["start_time"] == expected_time_dtype
+  assert structured.dtype["end_time"] == expected_time_dtype
   assert embedding_dtype.shape == (DEFAULT_EMBEDDING_DIM,)
   assert embedding_dtype.base == np.dtype(np.float32)
+
+
+def test_time_calculations_hop_exact_in_float16_but_products_drift() -> None:
+  # hop=1.5 is exactly representable in float16, but i*1.5 in the 1024..2048
+  # range only has step size 1, so e.g. 1365*1.5 = 2047.5 rounds. The previous
+  # upgrade_float_dtype_for_value heuristic only checked the scalar hop and
+  # missed this; the source fix forces >= float32 unconditionally.
+  duration = 1500.0
+  segment_duration = 3.0
+  overlap_duration = 1.5
+  speed = 1.0
+  result = create_file_encoding_result(
+    n_files=1,
+    duration_s=duration,
+    segment_duration_s=segment_duration,
+    overlap_duration_s=overlap_duration,
+    speed=speed,
+  )
+
+  structured = result.to_structured_array()
+  hop = get_hop_duration_s(segment_duration, overlap_duration, speed)
+  expected_starts = np.arange(len(structured)) * hop
+  expected_ends = np.minimum(
+    expected_starts + segment_duration * speed, result.input_durations[0]
+  )
+
+  np.testing.assert_allclose(structured["start_time"], expected_starts)
+  np.testing.assert_allclose(structured["end_time"], expected_ends)
+  assert structured.dtype["start_time"] != np.float16
+  assert structured.dtype["end_time"] != np.float16

src/birdnet_tests/acoustic_models/inference/predictions/prediction_result_py/test_prediction_to_structured_array.py

@@ -14,6 +14,7 @@
 from birdnet.model_loader import load
 from birdnet.utils.helper import (
   get_float_dtype,
+  get_hop_duration_s,
   get_n_segments_speed,
 )
 from birdnet_tests.test_files import TEST_FILE_LONG
@@ -481,8 +482,9 @@ def test_dtype_structure() -> None:
   ]
   assert structured.dtype.names == tuple(expected_fields)
   assert structured.dtype["input"] == np.dtype("O")
-  assert structured.dtype["start_time"] == result._input_durations.dtype
-  assert structured.dtype["end_time"] == result._input_durations.dtype
+  expected_time_dtype = np.result_type(result._input_durations.dtype, np.float32)
+  assert structured.dtype["start_time"] == expected_time_dtype
+  assert structured.dtype["end_time"] == expected_time_dtype
   assert structured.dtype["species_name"] == np.dtype("O")
   assert structured.dtype["confidence"] == result._species_probs.dtype
 
@@ -525,3 +527,33 @@ def test_full_pipeline_np() -> None:
     res = session.run_arrays(sf_read)
   structured = res.to_structured_array()
   assert len(structured) == 80
+
+
+def test_time_calculations_issue_38_long_file_with_overlap_and_slowdown() -> None:
+  # Reproduces issue #38: a 120s file with speed=0.3 and overlap=0.7 forces
+  # the float16 input_durations dtype, and accumulated i*hop products were
+  # quantized into ~0.05s drift on later segments before the source-level fix.
+  duration = 120.0
+  segment_duration = 3.0
+  overlap_duration = 0.7
+  speed = 0.3
+  result = create_file_prediction_result(
+    n_files=1,
+    duration_s=duration,
+    top_k=1,
+    segment_duration_s=segment_duration,
+    overlap_duration_s=overlap_duration,
+    speed=speed,
+  )
+
+  structured = result.to_structured_array()
+  hop = get_hop_duration_s(segment_duration, overlap_duration, speed)
+  expected_starts = np.arange(len(structured)) * hop
+  expected_ends = np.minimum(
+    expected_starts + segment_duration * speed, result.input_durations[0]
+  )
+
+  np.testing.assert_allclose(structured["start_time"], expected_starts)
+  np.testing.assert_allclose(structured["end_time"], expected_ends)
+  assert structured.dtype["start_time"] != np.float16
+  assert structured.dtype["end_time"] != np.float16