Make it possible to configure batcher explicitly via python APIs (#10657)

### What
We changed the behavior of the batcher in:
#10620

However, the only way to override this behavior was to set environment
variables, which can be tricky/annoying in some circumstances. This
plumbs through the configuration settings so that this can be specified
explicitly now.

---------

Co-authored-by: Clement Rey <[email protected]>

Author: jleibs

rerun_py/rerun_bindings/rerun_bindings.pyi

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import os
 from collections.abc import Iterable, Iterator, Sequence
-from datetime import datetime
+from datetime import datetime, timedelta
 from enum import Enum
 from typing import Any, Callable, Optional, Self
 
@@ -614,6 +614,120 @@ class PyRecordingStream:
         Calling operations such as flush or set_sink will result in an error.
         """
 
+class ChunkBatcherConfig:
+    """Defines the different batching thresholds used within the RecordingStream."""
+
+    def __init__(
+        self,
+        flush_tick: int | float | timedelta | None = None,
+        flush_num_bytes: int | None = None,
+        flush_num_rows: int | None = None,
+        chunk_max_rows_if_unsorted: int | None = None,
+    ) -> None:
+        """
+        Initialize the chunk batcher configuration.
+
+        Parameters
+        ----------
+        flush_tick : int | float | timedelta | None
+            Duration of the periodic tick, by default `None`.
+            Equivalent to setting: `RERUN_FLUSH_TICK_SECS` environment variable.
+
+        flush_num_bytes : int | None
+            Flush if the accumulated payload has a size in bytes equal or greater than this, by default `None`.
+            Equivalent to setting: `RERUN_FLUSH_NUM_BYTES` environment variable.
+
+        flush_num_rows : int | None
+            Flush if the accumulated payload has a number of rows equal or greater than this, by default `None`.
+            Equivalent to setting: `RERUN_FLUSH_NUM_ROWS` environment variable.
+
+        chunk_max_rows_if_unsorted : int | None
+            Split a chunk if it contains >= rows than this threshold and one or more of its timelines are unsorted,
+            by default `None`.
+            Equivalent to setting: `RERUN_CHUNK_MAX_ROWS_IF_UNSORTED` environment variable.
+
+        """
+
+    @property
+    def flush_tick(self) -> timedelta:
+        """
+        Duration of the periodic tick.
+
+        Equivalent to setting: `RERUN_FLUSH_TICK_SECS` environment variable.
+        """
+
+    @flush_tick.setter
+    def flush_tick(self, value: float | int | timedelta) -> None:
+        """
+        Duration of the periodic tick.
+
+        Equivalent to setting: `RERUN_FLUSH_TICK_SECS` environment variable.
+        """
+
+    @property
+    def flush_num_bytes(self) -> int:
+        """
+        Flush if the accumulated payload has a size in bytes equal or greater than this.
+
+        Equivalent to setting: `RERUN_FLUSH_NUM_BYTES` environment variable.
+        """
+
+    @flush_num_bytes.setter
+    def flush_num_bytes(self, value: int) -> None:
+        """
+        Flush if the accumulated payload has a size in bytes equal or greater than this.
+
+        Equivalent to setting: `RERUN_FLUSH_NUM_BYTES` environment variable.
+        """
+
+    @property
+    def flush_num_rows(self) -> int:
+        """
+        Flush if the accumulated payload has a number of rows equal or greater than this.
+
+        Equivalent to setting: `RERUN_FLUSH_NUM_ROWS` environment variable.
+        """
+
+    @flush_num_rows.setter
+    def flush_num_rows(self, value: int) -> None:
+        """
+        Flush if the accumulated payload has a number of rows equal or greater than this.
+
+        Equivalent to setting: `RERUN_FLUSH_NUM_ROWS` environment variable.
+        """
+
+    @property
+    def chunk_max_rows_if_unsorted(self) -> int:
+        """
+        Split a chunk if it contains >= rows than this threshold and one or more of its timelines are unsorted.
+
+        Equivalent to setting: `RERUN_CHUNK_MAX_ROWS_IF_UNSORTED` environment variable.
+        """
+
+    @chunk_max_rows_if_unsorted.setter
+    def chunk_max_rows_if_unsorted(self, value: int) -> None:
+        """
+        Split a chunk if it contains >= rows than this threshold and one or more of its timelines are unsorted.
+
+        Equivalent to setting: `RERUN_CHUNK_MAX_ROWS_IF_UNSORTED` environment variable.
+        """
+
+    @staticmethod
+    def DEFAULT() -> ChunkBatcherConfig:
+        """Default configuration, applicable to most use cases."""
+
+    @staticmethod
+    def LOW_LATENCY() -> ChunkBatcherConfig:
+        """Low-latency configuration, preferred when streaming directly to a viewer."""
+
+    @staticmethod
+    def ALWAYS() -> ChunkBatcherConfig:
+        """Always flushes ASAP."""
+
+    @staticmethod
+    def NEVER() -> ChunkBatcherConfig:
+        """Never flushes unless manually told to (or hitting one the builtin invariants)."""
+
 class PyMemorySinkStorage:
     def concat_as_bytes(self, concat: Optional[PyMemorySinkStorage] = None) -> bytes:
         """
@@ -658,6 +772,7 @@ def new_recording(
     make_thread_default: bool = True,
     default_enabled: bool = True,
     send_properties: bool = True,
+    batcher_config: Optional[ChunkBatcherConfig] = None,
 ) -> PyRecordingStream:
     """Create a new recording stream."""
 

rerun_py/rerun_sdk/rerun/__init__.py

@@ -153,6 +153,7 @@
 )
 from .recording_stream import (
     BinaryStream as BinaryStream,
+    ChunkBatcherConfig as ChunkBatcherConfig,
     RecordingStream as RecordingStream,
     binary_stream as binary_stream,
     get_application_id as get_application_id,

rerun_py/rerun_sdk/rerun/recording_stream.py

@@ -11,6 +11,9 @@
 from typing import TYPE_CHECKING, Any, Callable, TypeVar, overload
 
 import numpy as np
+from rerun_bindings import (
+    ChunkBatcherConfig as ChunkBatcherConfig,
+)
 from typing_extensions import deprecated
 
 import rerun as rr
@@ -39,6 +42,7 @@ def new_recording(
     make_thread_default: bool = False,
     spawn: bool = False,
     default_enabled: bool = True,
+    batcher_config: ChunkBatcherConfig | None = None,
 ) -> RecordingStream:
     """
     Creates a new recording with a user-chosen application id (name) that can be used to log data.
@@ -103,6 +107,8 @@ def new_recording(
     default_enabled
         Should Rerun logging be on by default?
         Can be overridden with the RERUN env-var, e.g. `RERUN=on` or `RERUN=off`.
+    batcher_config
+        Optional configuration for the chunk batcher.
 
     Returns
     -------
@@ -117,6 +123,7 @@ def new_recording(
         make_default=make_default,
         make_thread_default=make_thread_default,
         default_enabled=default_enabled,
+        batcher_config=batcher_config,
     )
 
     if spawn:
@@ -302,7 +309,8 @@ class RecordingStream:
     Micro-batching using both space and time triggers (whichever comes first) is done automatically
     in a dedicated background thread.
 
-    You can configure the frequency of the batches using the following environment variables:
+    You can configure the frequency of the batches using the `batcher_config` parameter when creating
+    the RecordingStream, or via the following environment variables:
 
     - `RERUN_FLUSH_TICK_SECS`:
         Flush frequency in seconds (default: `0.2` (200ms)).
@@ -322,6 +330,7 @@ def __init__(
         make_thread_default: bool = False,
         default_enabled: bool = True,
         send_properties: bool = True,
+        batcher_config: ChunkBatcherConfig | None = None,
     ) -> None:
         """
         Creates a new recording stream with a user-chosen application id (name) that can be used to log data.
@@ -383,6 +392,8 @@ def __init__(
             Can be overridden with the RERUN env-var, e.g. `RERUN=on` or `RERUN=off`.
         send_properties
             Immediately send the recording properties to the viewer (default: True)
+        batcher_config
+            Optional configuration for the chunk batcher.
 
         Returns
         -------
@@ -411,6 +422,7 @@ def __init__(
             make_thread_default=make_thread_default,
             default_enabled=default_enabled,
             send_properties=send_properties,
+            batcher_config=batcher_config,
         )
 
         self._prev: RecordingStream | None = None