Centralize numeric range validation

src/hflow/_field_guards.py

@@ -1,4 +1,4 @@
-"""Shared numeric-field type guards for caller-constructed settings.
+"""Shared numeric-field type and range guards for caller-constructed settings.
 
 Range checks alone (``0 <= x <= 100``) do not reject the wrong *type*: ``bool``
 subclasses ``int``, so ``True``/``False`` satisfy every numeric comparison and
@@ -11,20 +11,89 @@
 output, so unlike ``catalog.py``'s NumPy-scalar coercion, no NumPy handling is
 added here: a caller building a ``np.float64`` threshold can call ``.item()``
 itself, the same way any other non-native-Python value would need to.
+
+Range guards compose the type guards so callers can state the whole invariant
+without repeating the bool exclusion. The Real guard preserves batching's
+broader numeric contract without adding coercion to the native-number guards.
+Each guard returns the validated value with its refined type, without coercion.
 """
 
+import math
+from numbers import Real
+
 
-def require_int(value: object, name: str) -> None:
+def require_int(value: object, name: str) -> int:
     """Refuse anything but a plain ``int``, ``bool`` included."""
     if isinstance(value, bool) or not isinstance(value, int):
         raise ValueError(f"{name} must be an int, got {type(value).__name__}")
+    return value
 
 
-def require_float(value: object, name: str) -> None:
+def require_float(value: object, name: str) -> int | float:
     """Refuse anything but a plain ``int`` or ``float``, ``bool`` included.
 
     An ``int`` is accepted for a float-declared field: it is a perfectly good
     float value (``0`` is a real, falsy value that must still pass).
     """
     if isinstance(value, bool) or not isinstance(value, int | float):
         raise ValueError(f"{name} must be an int or float, got {type(value).__name__}")
+    return value
+
+
+def require_positive_int(value: object, name: str) -> int:
+    """Refuse anything but a strictly positive int, excluding bool."""
+    number = require_int(value, name)
+    if number <= 0:
+        raise ValueError(f"{name} must be > 0, got {number}")
+    return number
+
+
+def require_non_negative_int(value: object, name: str) -> int:
+    """Refuse anything but a non-negative int, excluding bool."""
+    number = require_int(value, name)
+    if number < 0:
+        raise ValueError(f"{name} must be >= 0, got {number}")
+    return number
+
+
+def require_int_in_range(value: object, name: str, *, minimum: int, maximum: int) -> int:
+    """Refuse anything but an int within the inclusive bounds, excluding bool."""
+    number = require_int(value, name)
+    if not minimum <= number <= maximum:
+        raise ValueError(f"{name} must be in [{minimum}, {maximum}], got {number}")
+    return number
+
+
+def require_finite_float(value: object, name: str) -> int | float:
+    """Refuse anything but a finite int or float, excluding bool."""
+    number = require_float(value, name)
+    if not math.isfinite(number):
+        raise ValueError(f"{name} must be finite, got {number}")
+    return number
+
+
+def require_positive_float(value: object, name: str) -> int | float:
+    """Refuse anything but a finite, strictly positive int or float."""
+    number = require_finite_float(value, name)
+    if number <= 0:
+        raise ValueError(f"{name} must be > 0, got {number}")
+    return number
+
+
+def require_non_negative_float(value: object, name: str) -> int | float:
+    """Refuse anything but a finite, non-negative int or float."""
+    number = require_finite_float(value, name)
+    if number < 0:
+        raise ValueError(f"{name} must be >= 0, got {number}")
+    return number
+
+
+def require_non_negative_real(value: object, name: str) -> Real:
+    """Preserve batching's finite, non-negative Real contract, excluding bool."""
+    if isinstance(value, bool) or not isinstance(value, Real):
+        raise ValueError(f"{name} must be a real number, got {type(value).__name__}")
+    if not math.isfinite(value):
+        raise ValueError(f"{name} must be finite, got {value}")
+    if value < 0:
+        raise ValueError(f"{name} must be >= 0, got {value}")
+    return value

src/hflow/batching.py

@@ -22,12 +22,16 @@
 """
 
 import heapq
-import math
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
-from numbers import Real
 from pathlib import Path
 
+from hflow._field_guards import (
+    require_non_negative_int,
+    require_non_negative_real,
+    require_positive_int,
+)
+
 
 @dataclass(frozen=True)
 class PlannedBatch:
@@ -54,34 +58,14 @@ def plan_batches(
     if (batch_count is None) == (target_batch_bytes is None):
         raise ValueError("pass exactly one of batch_count or target_batch_bytes")
     for uri, size_bytes in item_sizes.items():
-        if not isinstance(size_bytes, int) or isinstance(size_bytes, bool):
-            raise ValueError(
-                f"item {uri!r} has type {type(size_bytes).__name__}, expected int bytes"
-            )
-        if size_bytes < 0:
-            raise ValueError(f"item {uri!r} has negative size {size_bytes}")
-
-    if not isinstance(stagger_interval_s, Real) or isinstance(stagger_interval_s, bool):
-        raise ValueError(
-            f"stagger_interval_s must be a number, got {type(stagger_interval_s).__name__}"
-        )
-    if not math.isfinite(stagger_interval_s):
-        raise ValueError(f"stagger_interval_s must be finite, got {stagger_interval_s}")
-    if stagger_interval_s < 0:
-        raise ValueError(f"stagger_interval_s must be nonnegative, got {stagger_interval_s}")
+        require_non_negative_int(size_bytes, f"item {uri!r} size_bytes")
+
+    require_non_negative_real(stagger_interval_s, "stagger_interval_s")
 
     if batch_count is not None:
-        if not isinstance(batch_count, int) or isinstance(batch_count, bool):
-            raise ValueError(f"batch_count must be an int, got {type(batch_count).__name__}")
-        if batch_count < 1:
-            raise ValueError(f"batch_count must be >= 1, got {batch_count}")
+        require_positive_int(batch_count, "batch_count")
     else:
-        if not isinstance(target_batch_bytes, int) or isinstance(target_batch_bytes, bool):
-            raise ValueError(
-                f"target_batch_bytes must be an int, got {type(target_batch_bytes).__name__}"
-            )
-        if target_batch_bytes < 1:
-            raise ValueError(f"target_batch_bytes must be >= 1, got {target_batch_bytes}")
+        require_positive_int(target_batch_bytes, "target_batch_bytes")
 
     if not item_sizes:
         return []

src/hflow/build_ai_vlm_checks.py

@@ -45,6 +45,13 @@
 
 import httpx2
 
+from hflow._field_guards import (
+    require_finite_float,
+    require_non_negative_float,
+    require_non_negative_int,
+    require_positive_float,
+    require_positive_int,
+)
 from hflow._version import __version__
 from hflow._video_measurement_toolchain import measure_video_frame_statistics_for_hflow
 from hflow._video_measurements import FrameStatisticsSettings
@@ -215,16 +222,10 @@ def __post_init__(self) -> None:
             raise ValueError(
                 "api_key_environment_variable must be a valid environment variable name"
             )
-        if not isinstance(self.max_tokens, int) or isinstance(self.max_tokens, bool):
-            raise ValueError("max_tokens must be an integer")
-        if self.max_tokens <= 0:
-            raise ValueError("max_tokens must be greater than zero")
-        if not isinstance(self.max_retries, int) or isinstance(self.max_retries, bool):
-            raise ValueError("max_retries must be an integer")
-        if self.max_retries < 0:
-            raise ValueError("max_retries must not be negative")
-        if self.temperature is not None and not math.isfinite(self.temperature):
-            raise ValueError("temperature must be finite")
+        require_positive_int(self.max_tokens, "max_tokens")
+        require_non_negative_int(self.max_retries, "max_retries")
+        if self.temperature is not None:
+            require_finite_float(self.temperature, "temperature")
 
 
 @dataclass(frozen=True)
@@ -250,24 +251,12 @@ class HFlowHostedExecution:
 
     def __post_init__(self) -> None:
         _require_absolute_http_url(self.base_url, name="base_url")
-        if not isinstance(self.max_retries, int) or isinstance(self.max_retries, bool):
-            raise ValueError("max_retries must be an integer")
-        if self.max_retries < 0:
-            raise ValueError("max_retries must not be negative")
+        require_non_negative_int(self.max_retries, "max_retries")
         parsed_base_url = urlsplit(self.base_url)
         if parsed_base_url.query or parsed_base_url.fragment:
             raise ValueError("base_url must not contain a query string or fragment")
-        if not isinstance(self.check_version, int) or isinstance(self.check_version, bool):
-            raise ValueError("check_version must be an integer")
-        if self.check_version <= 0:
-            raise ValueError("check_version must be greater than zero")
-        if (
-            isinstance(self.request_timeout_seconds, bool)
-            or not isinstance(self.request_timeout_seconds, int | float)
-            or not math.isfinite(self.request_timeout_seconds)
-            or self.request_timeout_seconds <= 0
-        ):
-            raise ValueError("request_timeout_seconds must be finite and greater than zero")
+        require_positive_int(self.check_version, "check_version")
+        require_positive_float(self.request_timeout_seconds, "request_timeout_seconds")
 
 
 BuildAIExecution = OpenAICompatibleExecution | HFlowHostedExecution
@@ -299,16 +288,12 @@ class FrameSampling:
     def __post_init__(self) -> None:
         if not isinstance(self.skip_black_frames, bool):
             raise ValueError("skip_black_frames must be a bool")
-        if isinstance(self.fps, bool) or not math.isfinite(self.fps) or self.fps <= 0:
-            raise ValueError("fps must be finite and greater than zero")
-        if isinstance(self.start_s, bool) or not math.isfinite(self.start_s) or self.start_s < 0:
-            raise ValueError("start_s must be finite and non-negative")
-        if self.end_s is not None and (
-            isinstance(self.end_s, bool)
-            or not math.isfinite(self.end_s)
-            or self.end_s <= self.start_s
-        ):
-            raise ValueError("end_s must be finite and greater than start_s")
+        require_positive_float(self.fps, "fps")
+        require_non_negative_float(self.start_s, "start_s")
+        if self.end_s is not None:
+            require_finite_float(self.end_s, "end_s")
+            if self.end_s <= self.start_s:
+                raise ValueError("end_s must be greater than start_s")
 
 
 @dataclass(frozen=True)
@@ -337,10 +322,7 @@ def __post_init__(self) -> None:
                 "HFlowHostedExecution uses the hosted check's fixed prompt and does not support "
                 "prompt overrides"
             )
-        if isinstance(self.frame_time_seconds, bool) or not math.isfinite(self.frame_time_seconds):
-            raise ValueError("frame_time_seconds must be finite and non-negative")
-        if self.frame_time_seconds < 0:
-            raise ValueError("frame_time_seconds must be finite and non-negative")
+        require_non_negative_float(self.frame_time_seconds, "frame_time_seconds")
         if self.camera == "":
             raise ValueError("camera must be None or a non-empty topic name")
 
@@ -536,8 +518,9 @@ def model_output_check_result(
             outcome.response_metadata.response_model
         )
     for usage_name, usage_value in outcome.response_metadata.usage.items():
-        if isinstance(usage_value, int | float) and not isinstance(usage_value, bool):
-            measurements[f"{measurement_prefix}/usage/{usage_name}"] = usage_value
+        if isinstance(usage_value, bool) or not isinstance(usage_value, int | float):
+            continue
+        measurements[f"{measurement_prefix}/usage/{usage_name}"] = usage_value
 
     observation_values: dict[str, MeasurementValue] = {
         "task": task.value,

tests/test_batching.py

@@ -1,5 +1,6 @@
 """Byte-balanced bin-packing and staggered batch starts."""
 
+from fractions import Fraction
 from pathlib import Path
 from typing import Any
 
@@ -53,11 +54,11 @@ def test_argument_validation() -> None:
         plan_batches({"a": 1})
     with pytest.raises(ValueError, match="exactly one"):
         plan_batches({"a": 1}, batch_count=1, target_batch_bytes=1)
-    with pytest.raises(ValueError, match="negative"):
+    with pytest.raises(ValueError, match="item 'a' size_bytes must be >= 0"):
         plan_batches({"a": -1}, batch_count=1)
-    with pytest.raises(ValueError, match="batch_count must be >= 1, got 0"):
+    with pytest.raises(ValueError, match="batch_count must be > 0, got 0"):
         plan_batches({"a": 1}, batch_count=0)
-    with pytest.raises(ValueError, match="target_batch_bytes must be >= 1, got 0"):
+    with pytest.raises(ValueError, match="target_batch_bytes must be > 0, got 0"):
         plan_batches({"a": 1}, target_batch_bytes=0)
     assert plan_batches({}, batch_count=3) == []
 
@@ -79,8 +80,8 @@ def test_batch_parameters_require_real_integers(kwargs: dict[str, Any], message:
 @pytest.mark.parametrize(
     ("size", "message"),
     [
-        (True, "item 'a' has type bool, expected int bytes"),
-        (1.0, "item 'a' has type float, expected int bytes"),
+        (True, "item 'a' size_bytes must be an int, got bool"),
+        (1.0, "item 'a' size_bytes must be an int, got float"),
     ],
 )
 def test_item_sizes_require_real_integers(size: Any, message: str) -> None:
@@ -91,9 +92,9 @@ def test_item_sizes_require_real_integers(size: Any, message: str) -> None:
 @pytest.mark.parametrize(
     ("stagger", "message"),
     [
-        (True, "stagger_interval_s must be a number, got bool"),
-        ("1.0", "stagger_interval_s must be a number, got str"),
-        (-1.0, "stagger_interval_s must be nonnegative, got -1.0"),
+        (True, "stagger_interval_s must be a real number, got bool"),
+        ("1.0", "stagger_interval_s must be a real number, got str"),
+        (-1.0, "stagger_interval_s must be >= 0, got -1.0"),
         (float("nan"), "stagger_interval_s must be finite, got nan"),
         (float("inf"), "stagger_interval_s must be finite, got inf"),
     ],
@@ -103,8 +104,9 @@ def test_stagger_interval_requires_a_finite_nonnegative_number(stagger: Any, mes
         plan_batches({"a": 1}, batch_count=1, stagger_interval_s=stagger)  # type: ignore[arg-type]
 
 
-def test_zero_size_items_and_zero_stagger_remain_valid() -> None:
-    batches = plan_batches({"empty": 0}, batch_count=1, stagger_interval_s=0)
+@pytest.mark.parametrize("stagger", [0, 0.0, Fraction(0)])
+def test_zero_size_items_and_zero_stagger_remain_valid(stagger: Any) -> None:
+    batches = plan_batches({"empty": 0}, batch_count=1, stagger_interval_s=stagger)
     assert batches == [PlannedBatch(items=("empty",), total_bytes=0, start_delay_s=0)]
 
 
@@ -123,3 +125,16 @@ def test_plan_from_real_files(tmp_path: Path) -> None:
     batches = plan_batches_from_files([small, large], batch_count=2)
     assert batches[0].total_bytes == 10_000
     assert batches[0].items == (str(large),)
+
+
+def test_fractional_stagger_preserves_exact_delays() -> None:
+    batches = plan_batches(
+        {"a": 1, "b": 1, "c": 1},
+        batch_count=3,
+        stagger_interval_s=Fraction(1, 3),  # ty: ignore[invalid-argument-type]
+    )
+    assert [batch.start_delay_s for batch in batches] == [
+        Fraction(0),
+        Fraction(1, 3),
+        Fraction(2, 3),
+    ]