Benchmarkers

hgp_lib.benchmarkers.gp_benchmarker.GPBenchmarker

Benchmarker for Boolean GP: runs multiple independent experiments with stratified train/test split and k-fold CV per run, then aggregates results.

Data flow per run:

1. Stratified train/test split (using test_size).
2. For each of the n_folds folds:

a. A fresh copy of the binarizer is fitted on the training fold. b. The validation fold is transformed using the same fitted binarizer. c. Binarized data is converted to boolean numpy arrays and used for GP training.

1. The best fold (highest validation score) is selected. Its binarizer is used to transform the held-out test set for final evaluation.

Raw (non-binarized) data should be passed as a pandas.DataFrame in BenchmarkerConfig.data. Binarization happens internally per fold to prevent data leakage.

Parameters:

Name	Type	Description	Default
config	BenchmarkerConfig	Configuration with data, labels, binarizer, trainer_config template, and benchmarker-specific options. See BenchmarkerConfig for more details.	required

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
>>> from hgp_lib.benchmarkers import GPBenchmarker
>>> from hgp_lib.rules import Rule
>>> data = pd.DataFrame({
...     "f1": [True, False, True, False, True, False, True, False],
...     "f2": [False, True, True, False, False, True, True, False],
... })
>>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
>>> gp_config = BooleanGPConfig(optimize_scorer=False)
>>> trainer_config = TrainerConfig(gp_config=gp_config, num_epochs=3, progress_bar=False)
>>> config = BenchmarkerConfig(
...     data=data, labels=labels, trainer_config=trainer_config,
...     num_runs=2, n_folds=2, n_jobs=1,
... )
>>> benchmarker = GPBenchmarker(config)
>>> result = benchmarker.fit()
>>> len(result.runs)
2
>>> isinstance(result.best_rule, Rule)
True

Source code in src/hgp_lib/benchmarkers/gp_benchmarker.py

class GPBenchmarker:
    """
    Benchmarker for Boolean GP: runs multiple independent experiments with
    stratified train/test split and k-fold CV per run, then aggregates results.

    Data flow per run:

    1. Stratified train/test split (using `test_size`).
    2. For each of the `n_folds` folds:

       a. A fresh copy of the `binarizer` is fitted on the training fold.
       b. The validation fold is transformed using the same fitted binarizer.
       c. Binarized data is converted to boolean numpy arrays and used for GP
          training.

    3. The best fold (highest validation score) is selected. Its binarizer is
       used to transform the held-out test set for final evaluation.

    Raw (non-binarized) data should be passed as a
    `pandas.DataFrame` in `BenchmarkerConfig.data`. Binarization happens
    internally per fold to prevent data leakage.

    Args:
        config (BenchmarkerConfig): Configuration with data,
            labels, binarizer, trainer_config template, and benchmarker-specific
            options. See `BenchmarkerConfig` for more details.

    Examples:
        >>> import numpy as np
        >>> import pandas as pd
        >>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
        >>> from hgp_lib.benchmarkers import GPBenchmarker
        >>> from hgp_lib.rules import Rule
        >>> data = pd.DataFrame({
        ...     "f1": [True, False, True, False, True, False, True, False],
        ...     "f2": [False, True, True, False, False, True, True, False],
        ... })
        >>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
        >>> gp_config = BooleanGPConfig(optimize_scorer=False)
        >>> trainer_config = TrainerConfig(gp_config=gp_config, num_epochs=3, progress_bar=False)
        >>> config = BenchmarkerConfig(
        ...     data=data, labels=labels, trainer_config=trainer_config,
        ...     num_runs=2, n_folds=2, n_jobs=1,
        ... )
        >>> benchmarker = GPBenchmarker(config)
        >>> result = benchmarker.fit()
        >>> len(result.runs)
        2
        >>> isinstance(result.best_rule, Rule)
        True
    """

    def __init__(self, config: BenchmarkerConfig):
        validate_benchmarker_config(config)
        self.config = config
        if self.config.binarizer is None:
            self.config.binarizer = StandardBinarizer()
        self._run_results: ExperimentResult | None = None

    def _effective_n_jobs(self) -> int:
        """
        Compute the effective number of parallel jobs to use.

        Returns:
            int: The number of parallel workers to use (always >= 1).

        Examples:
            >>> import numpy as np
            >>> import pandas as pd
            >>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
            >>> from hgp_lib.benchmarkers import GPBenchmarker
            >>> data = pd.DataFrame({"f": [True, False, True, False, True, False, True, False]})
            >>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
            >>> gp = BooleanGPConfig(optimize_scorer=False)
            >>> tc = TrainerConfig(gp_config=gp, num_epochs=1, progress_bar=False)
            >>> cfg = BenchmarkerConfig(data=data, labels=labels, trainer_config=tc,
            ...     num_runs=3, n_folds=2, n_jobs=1)
            >>> GPBenchmarker(cfg)._effective_n_jobs()
            1
        """
        n_jobs = self.config.n_jobs
        if n_jobs < 0:
            n_jobs = os.cpu_count() or 1
        return max(1, min(n_jobs, self.config.num_runs))

    def _run_sequential(self) -> ExperimentResult:
        """Run all benchmark runs sequentially with nested progress bars."""
        run_results: list[RunResult] = []

        if self.config.show_run_progress:
            self.config.trainer_config.leave_progress_bar = False
            self.config.binarizer.leave_progress_bar = False

        for run_id in tqdm(
            range(self.config.num_runs),
            desc="Benchmark Runs",
            disable=not self.config.show_run_progress,
        ):
            result = execute_single_run(
                run_id, self.config.base_seed + run_id, self.config
            )
            run_results.append(result)

        return ExperimentResult(runs=run_results)

    def _run_parallel(self, n_jobs: int) -> ExperimentResult:
        """Run all benchmark runs in parallel with centralized progress bars."""
        show_progress = self.config.trainer_config.progress_bar
        show_run_progress = show_progress and self.config.show_run_progress
        show_fold_progress = show_progress and self.config.show_fold_progress
        show_epoch_progress = show_progress and self.config.show_epoch_progress
        any_progress = show_run_progress or show_fold_progress or show_epoch_progress

        total_runs = self.config.num_runs
        total_folds = total_runs * self.config.n_folds
        total_epochs = total_folds * self.config.trainer_config.num_epochs

        progress_config = ProgressConfig(
            total_runs=total_runs,
            total_folds=total_folds,
            total_epochs=total_epochs,
            show_run_progress=show_run_progress,
            show_fold_progress=show_fold_progress,
            show_epoch_progress=show_epoch_progress,
        )

        # LIFO cleanup: pool first (so workers stop emitting progress), then the listener, then the manager
        with contextlib.ExitStack() as stack:
            reporter = ProgressReporter()
            if any_progress:
                manager = stack.enter_context(multiprocessing.Manager())
                queue = manager.Queue()
                reporter = ProgressReporter(queue)
                listener = ProgressListener(queue, progress_config)
                listener.start()
                stack.callback(listener.stop)

            run_args = [
                (run_id, self.config.base_seed + run_id, self.config, reporter)
                for run_id in range(self.config.num_runs)
            ]

            pool = stack.enter_context(_graceful_pool(n_jobs))
            run_results = pool.map(single_run_wrapper, run_args)

        return ExperimentResult(runs=run_results)

    def fit(self) -> ExperimentResult:
        """
        Run all benchmark runs (parallel or sequential) and aggregate results.

        Each run performs a stratified train/test split, k-fold CV with per-fold
        binarization, and test-set evaluation. Results across runs are aggregated
        into an ExperimentResult.

        Returns:
            ExperimentResult: Contains all run results with methods to get
                best_run, best_rule, test scores statistics, etc.
        """
        effective_n_jobs = self._effective_n_jobs()

        if effective_n_jobs == 1:
            run_results = self._run_sequential()
        else:
            run_results = self._run_parallel(effective_n_jobs)

        self._run_results = run_results
        return run_results

    def predict(self, data: pd.DataFrame) -> np.ndarray:
        """
        Predict labels for raw ``data`` using the best rule found across all runs.

        This mirrors the scikit-learn ``predict`` API so a fitted ``GPBenchmarker``
        can be used where an estimator is expected. It must be called after ``fit``.
        The raw ``DataFrame`` is binarized with the best run's fitted binarizer (the
        one from its best fold), then the best rule is evaluated on it.

        Args:
            data (pd.DataFrame):
                Raw (non-binarized) features with the same columns, in the same order
                and dtypes, as the data used to fit the benchmarker.

        Returns:
            np.ndarray: 1-D boolean array with one prediction per input row.

        Raises:
            NotFittedError: If called before ``fit`` (no results are available yet).

        Examples:
            >>> import numpy as np
            >>> import pandas as pd
            >>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
            >>> from hgp_lib.benchmarkers import GPBenchmarker
            >>> data = pd.DataFrame({
            ...     "f1": [True, False, True, False, True, False, True, False],
            ...     "f2": [False, True, True, False, False, True, True, False],
            ... })
            >>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
            >>> gp_config = BooleanGPConfig(optimize_scorer=False)
            >>> trainer_config = TrainerConfig(gp_config=gp_config, num_epochs=3, progress_bar=False)
            >>> config = BenchmarkerConfig(
            ...     data=data, labels=labels, trainer_config=trainer_config,
            ...     num_runs=2, n_folds=2, n_jobs=1,
            ... )
            >>> benchmarker = GPBenchmarker(config)
            >>> _ = benchmarker.fit()
            >>> predictions = benchmarker.predict(data)
            >>> predictions.shape
            (8,)
        """
        if self._run_results is None:
            raise NotFittedError("GPBenchmarker must be fit before calling predict")

        best_run = self._run_results.best_run
        binarized = best_run.binarizer.transform(data).to_numpy(dtype=bool)
        return best_run.best_rule.evaluate(binarized)

fit()

Run all benchmark runs (parallel or sequential) and aggregate results.

Each run performs a stratified train/test split, k-fold CV with per-fold binarization, and test-set evaluation. Results across runs are aggregated into an ExperimentResult.

Returns:

Name	Type	Description
ExperimentResult	ExperimentResult	Contains all run results with methods to get best_run, best_rule, test scores statistics, etc.

Source code in src/hgp_lib/benchmarkers/gp_benchmarker.py

def fit(self) -> ExperimentResult:
    """
    Run all benchmark runs (parallel or sequential) and aggregate results.

    Each run performs a stratified train/test split, k-fold CV with per-fold
    binarization, and test-set evaluation. Results across runs are aggregated
    into an ExperimentResult.

    Returns:
        ExperimentResult: Contains all run results with methods to get
            best_run, best_rule, test scores statistics, etc.
    """
    effective_n_jobs = self._effective_n_jobs()

    if effective_n_jobs == 1:
        run_results = self._run_sequential()
    else:
        run_results = self._run_parallel(effective_n_jobs)

    self._run_results = run_results
    return run_results

predict(data)

Predict labels for raw data using the best rule found across all runs.

This mirrors the scikit-learn predict API so a fitted GPBenchmarker can be used where an estimator is expected. It must be called after fit. The raw DataFrame is binarized with the best run's fitted binarizer (the one from its best fold), then the best rule is evaluated on it.

Parameters:

Name	Type	Description	Default
data	DataFrame	Raw (non-binarized) features with the same columns, in the same order and dtypes, as the data used to fit the benchmarker.	required

Returns:

Type	Description
ndarray	np.ndarray: 1-D boolean array with one prediction per input row.

Raises:

Type	Description
NotFittedError	If called before fit (no results are available yet).

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
>>> from hgp_lib.benchmarkers import GPBenchmarker
>>> data = pd.DataFrame({
...     "f1": [True, False, True, False, True, False, True, False],
...     "f2": [False, True, True, False, False, True, True, False],
... })
>>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
>>> gp_config = BooleanGPConfig(optimize_scorer=False)
>>> trainer_config = TrainerConfig(gp_config=gp_config, num_epochs=3, progress_bar=False)
>>> config = BenchmarkerConfig(
...     data=data, labels=labels, trainer_config=trainer_config,
...     num_runs=2, n_folds=2, n_jobs=1,
... )
>>> benchmarker = GPBenchmarker(config)
>>> _ = benchmarker.fit()
>>> predictions = benchmarker.predict(data)
>>> predictions.shape
(8,)

Source code in src/hgp_lib/benchmarkers/gp_benchmarker.py

def predict(self, data: pd.DataFrame) -> np.ndarray:
    """
    Predict labels for raw ``data`` using the best rule found across all runs.

    This mirrors the scikit-learn ``predict`` API so a fitted ``GPBenchmarker``
    can be used where an estimator is expected. It must be called after ``fit``.
    The raw ``DataFrame`` is binarized with the best run's fitted binarizer (the
    one from its best fold), then the best rule is evaluated on it.

    Args:
        data (pd.DataFrame):
            Raw (non-binarized) features with the same columns, in the same order
            and dtypes, as the data used to fit the benchmarker.

    Returns:
        np.ndarray: 1-D boolean array with one prediction per input row.

    Raises:
        NotFittedError: If called before ``fit`` (no results are available yet).

    Examples:
        >>> import numpy as np
        >>> import pandas as pd
        >>> from hgp_lib.configs import BooleanGPConfig, TrainerConfig, BenchmarkerConfig
        >>> from hgp_lib.benchmarkers import GPBenchmarker
        >>> data = pd.DataFrame({
        ...     "f1": [True, False, True, False, True, False, True, False],
        ...     "f2": [False, True, True, False, False, True, True, False],
        ... })
        >>> labels = np.array([1, 0, 1, 0, 1, 0, 1, 0])
        >>> gp_config = BooleanGPConfig(optimize_scorer=False)
        >>> trainer_config = TrainerConfig(gp_config=gp_config, num_epochs=3, progress_bar=False)
        >>> config = BenchmarkerConfig(
        ...     data=data, labels=labels, trainer_config=trainer_config,
        ...     num_runs=2, n_folds=2, n_jobs=1,
        ... )
        >>> benchmarker = GPBenchmarker(config)
        >>> _ = benchmarker.fit()
        >>> predictions = benchmarker.predict(data)
        >>> predictions.shape
        (8,)
    """
    if self._run_results is None:
        raise NotFittedError("GPBenchmarker must be fit before calling predict")

    best_run = self._run_results.best_run
    binarized = best_run.binarizer.transform(data).to_numpy(dtype=bool)
    return best_run.best_rule.evaluate(binarized)

hgp_lib.benchmarkers.runner.execute_single_run(run_id, seed, config, reporter=None)

Execute one benchmark run: stratified train/test split, per-fold binarization, k-fold CV training, best-fold selection, and test-set evaluation.

This is a module-level function so it can be pickled for multiprocessing.

Per-fold binarization: For each fold a fresh clone of the configured binarizer is fitted on the training fold (with labels, enabling supervised binning for numerical features) and used to transform the validation fold. After selecting the best fold, its binarizer transforms the held-out test data. This prevents data leakage across folds and between train/test sets.

Parameters:

Name	Type	Description	Default
run_id	int	Index of the run (0-based).	required
seed	int	Random seed for stratified split and k-fold.	required
config	BenchmarkerConfig	Configuration for the benchmark run.	required
reporter	ProgressReporter | None	Where to report epoch/fold/run progress. Default: an inert reporter that discards updates.	None

Returns:

Name	Type	Description
RunResult	RunResult	Contains run_id, seed, folds, test_score, best_rule, feature_names.

Raises:

Type	Description
RuntimeError	If no best rule is available after training a fold.

Source code in src/hgp_lib/benchmarkers/runner.py

def execute_single_run(
    run_id: int,
    seed: int,
    config: BenchmarkerConfig,
    reporter: ProgressReporter | None = None,
) -> RunResult:
    """
    Execute one benchmark run: stratified train/test split, per-fold binarization,
    k-fold CV training, best-fold selection, and test-set evaluation.

    This is a module-level function so it can be pickled for `multiprocessing`.

    **Per-fold binarization:** For each fold a fresh `clone` of the configured
    binarizer is fitted on the training fold (with labels, enabling supervised
    binning for numerical features) and used to transform the validation fold.
    After selecting the best fold, its binarizer transforms the held-out test
    data. This prevents data leakage across folds and between train/test sets.

    Args:
        run_id (int): Index of the run (0-based).
        seed (int): Random seed for stratified split and k-fold.
        config (BenchmarkerConfig): Configuration for the benchmark run.
        reporter (ProgressReporter | None): Where to report epoch/fold/run
            progress. Default: an inert reporter that discards updates.

    Returns:
        RunResult: Contains run_id, seed, folds, test_score, best_rule, feature_names.

    Raises:
        RuntimeError: If no best rule is available after training a fold.
    """
    trainer_template = config.trainer_config
    gp_template = trainer_template.gp_config

    if reporter is None:
        reporter = ProgressReporter()

    # Local bars would collide with the centralized ones the listener draws.
    show_local_bars = trainer_template.progress_bar and not reporter.enabled
    show_folds = config.show_fold_progress and show_local_bars
    show_epochs = config.show_epoch_progress and show_local_bars
    config.binarizer.progress_bar = show_epochs

    np.random.seed(seed)
    random.seed(seed)

    train_data, test_data, train_labels, test_labels = train_test_split(
        config.data,
        config.labels,
        test_size=config.test_size,
        stratify=config.labels,
        random_state=seed,
    )

    skf = StratifiedKFold(n_splits=config.n_folds, shuffle=True, random_state=seed)

    folds: list[PopulationHistory] = []
    binarizers = []
    feature_names_per_binarizer = []

    fold_splits = skf.split(train_data, train_labels)
    if show_folds:
        fold_splits = tqdm(fold_splits, total=config.n_folds, desc="Folds", leave=False)

    epoch_callback = reporter.epoch if reporter.enabled else None

    best_fold_idx = 0
    best_fold_score = -float("inf")
    for fold_idx, (train_idx, val_idx) in enumerate(fold_splits):
        fold_train = train_data.iloc[train_idx]
        fold_train_labels = train_labels[train_idx]

        binarizer = deepcopy(config.binarizer)
        fold_train = binarizer.fit_transform(fold_train, fold_train_labels)
        binarizers.append(binarizer)
        feature_names_per_binarizer.append(binarizer.get_feature_names_out())
        fold_train = fold_train.to_numpy(dtype=bool)

        fold_val = binarizer.transform(train_data.iloc[val_idx]).to_numpy(dtype=bool)
        fold_val_labels = train_labels[val_idx]

        fold_gp_config = replace(
            gp_template,
            train_data=fold_train,
            train_labels=fold_train_labels,
        )
        fold_trainer_config = replace(
            trainer_template,
            gp_config=fold_gp_config,
            val_data=fold_val,
            val_labels=fold_val_labels,
            progress_bar=show_epochs,
            progress_callback=epoch_callback,
        )

        trainer = GPTrainer(fold_trainer_config)
        history = trainer.fit()

        # Prefer validation score; fall back to training score
        fold_score = (
            history.best_val_score
            if history.best_val_score is not None
            else history.best_train_score
        )
        if fold_score is not None and fold_score > best_fold_score:
            best_fold_score = fold_score
            best_fold_idx = fold_idx

        folds.append(history)

        reporter.fold()

    del train_data, train_labels

    best_rule = folds[best_fold_idx].global_best_rule

    # Transform test data using best fold's binarizer
    best_binarizer = binarizers[best_fold_idx]
    feature_names = feature_names_per_binarizer[best_fold_idx]
    test_data = best_binarizer.transform(test_data).to_numpy(dtype=bool)

    if gp_template.score_fn is None:
        gp_template.score_fn = fast_f1_score
    if gp_template.optimize_scorer:
        test_score_fn, test_cm, test_data, test_labels = optimize_scorers_for_data(
            gp_template.score_fn, confusion_matrix, data=test_data, labels=test_labels
        )
    else:
        test_score_fn = gp_template.score_fn
        test_cm = confusion_matrix

    test_pred = best_rule.evaluate(test_data)

    test_score = float(test_score_fn(test_labels, test_pred))

    tp, fp, fn, tn = test_cm(test_labels, test_pred)

    reporter.run()

    return RunResult(
        run_id=run_id,
        seed=seed,
        best_fold_idx=best_fold_idx,
        folds=folds,
        test_score=test_score,
        test_tp=tp,
        test_fp=fp,
        test_fn=fn,
        test_tn=tn,
        feature_names=feature_names,
        binarizer=best_binarizer,
    )

hgp_lib.benchmarkers.progress.ProgressListener

Listener thread for aggregating progress updates from worker processes.

Runs in the main process and listens to a multiprocessing queue for progress updates from worker processes. Updates three tqdm progress bars: - Experiments (runs) - Folds - Epochs

The listener uses timeouts on queue.get() to periodically check for stop signals, preventing indefinite hangs if workers crash.

Parameters:

Name	Type	Description	Default
progress_queue	Queue	Multiprocessing queue for receiving progress updates.	required
config	ProgressConfig	Progress configuration with totals and display settings.	required

Examples:

>>> from multiprocessing import Queue
>>> from hgp_lib.benchmarkers.progress import ProgressConfig, ProgressListener
>>> q = Queue()
>>> cfg = ProgressConfig(
...     total_runs=1, total_folds=1, total_epochs=1,
...     show_run_progress=False, show_fold_progress=False,
...     show_epoch_progress=False,
... )
>>> listener = ProgressListener(q, cfg)
>>> listener.start()
>>> q.put(("epoch", 1))
>>> q.put(("fold", 1))
>>> q.put(("run", 1))
>>> listener.join()

Source code in src/hgp_lib/benchmarkers/progress.py

class ProgressListener:
    """
    Listener thread for aggregating progress updates from worker processes.

    Runs in the main process and listens to a multiprocessing queue for
    progress updates from worker processes. Updates three tqdm progress bars:
    - Experiments (runs)
    - Folds
    - Epochs

    The listener uses timeouts on queue.get() to periodically check for stop
    signals, preventing indefinite hangs if workers crash.

    Args:
        progress_queue: Multiprocessing queue for receiving progress updates.
        config: Progress configuration with totals and display settings.

    Examples:
        >>> from multiprocessing import Queue
        >>> from hgp_lib.benchmarkers.progress import ProgressConfig, ProgressListener
        >>> q = Queue()
        >>> cfg = ProgressConfig(
        ...     total_runs=1, total_folds=1, total_epochs=1,
        ...     show_run_progress=False, show_fold_progress=False,
        ...     show_epoch_progress=False,
        ... )
        >>> listener = ProgressListener(q, cfg)
        >>> listener.start()
        >>> q.put(("epoch", 1))
        >>> q.put(("fold", 1))
        >>> q.put(("run", 1))
        >>> listener.join()
    """

    def __init__(self, progress_queue: Queue, config: ProgressConfig):
        self.queue = progress_queue
        self.config = config
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._pbar_exp: tqdm | None = None
        self._pbar_fold: tqdm | None = None
        self._pbar_epoch: tqdm | None = None

    def start(self) -> None:
        """Start the listener thread."""
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._listen, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        """Signal the listener to stop and wait for it to finish."""
        self._stop_event.set()
        # Send sentinel to unblock queue.get() if waiting
        try:
            self.queue.put(_SHUTDOWN_SENTINEL)
        except (BrokenPipeError, EOFError):
            pass  # Queue already closed
        if self._thread is not None:
            self._thread.join(timeout=10.0)

    def join(self) -> None:
        """Wait for the listener thread to finish naturally (all runs completed)."""
        if self._thread is not None:
            self._thread.join()

    def _listen(self) -> None:
        """
        Main listener loop. Uses timeout to allow periodic stop checks.

        Expected message format: ``(channel, count)``, where ``channel`` is a
        :class:`ProgressChannel` value. The matching bar advances by ``count``.
        Unknown channels are ignored.
        """
        # Initialize progress bars (position=0 is bottom, position=2 is top)
        self._pbar_exp = tqdm(
            total=self.config.total_runs,
            position=0,
            desc="Runs",
            leave=True,
            disable=not self.config.show_run_progress,
        )
        self._pbar_fold = tqdm(
            total=self.config.total_folds,
            position=1,
            desc="Folds",
            leave=True,
            disable=not self.config.show_fold_progress,
        )
        self._pbar_epoch = tqdm(
            total=self.config.total_epochs,
            position=2,
            desc="Epochs",
            leave=True,
            disable=not self.config.show_epoch_progress,
        )

        bars = {
            ProgressChannel.EPOCH.value: self._pbar_epoch,
            ProgressChannel.FOLD.value: self._pbar_fold,
            ProgressChannel.RUN.value: self._pbar_exp,
        }

        finished_runs = 0

        try:
            while finished_runs < self.config.total_runs:
                # Check if stop was requested
                if self._stop_event.is_set():
                    break

                try:
                    msg, count = self.queue.get(timeout=_QUEUE_TIMEOUT_SECONDS)
                except Empty:
                    # Timeout - loop back to check stop_event and continue waiting
                    continue

                # Check for shutdown sentinel
                if msg == _SHUTDOWN_SENTINEL[0]:
                    break

                pbar = bars.get(msg)
                if pbar is None:
                    continue  # Unknown channel

                pbar.update(count)
                if msg == ProgressChannel.RUN.value:
                    finished_runs += count
        finally:
            # Ensure bars are closed properly
            self._pbar_epoch.close()
            self._pbar_fold.close()
            self._pbar_exp.close()

start()

Start the listener thread.

Source code in src/hgp_lib/benchmarkers/progress.py

def start(self) -> None:
    """Start the listener thread."""
    self._stop_event.clear()
    self._thread = threading.Thread(target=self._listen, daemon=True)
    self._thread.start()

stop()

Signal the listener to stop and wait for it to finish.

Source code in src/hgp_lib/benchmarkers/progress.py

def stop(self) -> None:
    """Signal the listener to stop and wait for it to finish."""
    self._stop_event.set()
    # Send sentinel to unblock queue.get() if waiting
    try:
        self.queue.put(_SHUTDOWN_SENTINEL)
    except (BrokenPipeError, EOFError):
        pass  # Queue already closed
    if self._thread is not None:
        self._thread.join(timeout=10.0)

join()

Wait for the listener thread to finish naturally (all runs completed).

Source code in src/hgp_lib/benchmarkers/progress.py

def join(self) -> None:
    """Wait for the listener thread to finish naturally (all runs completed)."""
    if self._thread is not None:
        self._thread.join()

hgp_lib.benchmarkers.progress.ProgressReporter

Producer-side handle for reporting progress, one method per channel.

Wraps the queue that :class:ProgressListener consumes. A reporter built without a queue is inert and discards every update.

Every method has the shape Callable[[int], None], so a bound method such as reporter.epoch can be passed as TrainerConfig.progress_callback.

Parameters:

Name	Type	Description	Default
progress_queue	Queue | None	Queue to publish updates on, or None for an inert reporter. Default: None.	None

Examples:

>>> from multiprocessing import Queue
>>> from hgp_lib.benchmarkers.progress import ProgressReporter
>>> reporter = ProgressReporter(Queue())
>>> reporter.enabled
True
>>> reporter.fold()
>>> reporter.queue.get()
('fold', 1)
>>> reporter.epoch(10)
>>> reporter.queue.get()
('epoch', 10)

An inert reporter accepts the same calls and discards them:

>>> inert = ProgressReporter()
>>> inert.enabled
False
>>> inert.run(5)

Source code in src/hgp_lib/benchmarkers/progress.py

class ProgressReporter:
    """
    Producer-side handle for reporting progress, one method per channel.

    Wraps the queue that :class:`ProgressListener` consumes. A reporter built
    without a queue is inert and discards every update.

    Every method has the shape ``Callable[[int], None]``, so a bound method such
    as ``reporter.epoch`` can be passed as ``TrainerConfig.progress_callback``.

    Args:
        progress_queue (Queue | None):
            Queue to publish updates on, or ``None`` for an inert reporter.
            Default: `None`.

    Examples:
        >>> from multiprocessing import Queue
        >>> from hgp_lib.benchmarkers.progress import ProgressReporter
        >>> reporter = ProgressReporter(Queue())
        >>> reporter.enabled
        True
        >>> reporter.fold()
        >>> reporter.queue.get()
        ('fold', 1)
        >>> reporter.epoch(10)
        >>> reporter.queue.get()
        ('epoch', 10)

        An inert reporter accepts the same calls and discards them:

        >>> inert = ProgressReporter()
        >>> inert.enabled
        False
        >>> inert.run(5)
    """

    # "Queue | None" because type[Queue] is <method>..
    def __init__(self, progress_queue: "Queue | None" = None):
        self.queue = progress_queue

    @property
    def enabled(self) -> bool:
        """bool: Whether updates are published rather than discarded."""
        return self.queue is not None

    def epoch(self, count: int = 1) -> None:
        """Report ``count`` completed epochs. Default: `1`."""
        self._send(ProgressChannel.EPOCH, count)

    def fold(self, count: int = 1) -> None:
        """Report ``count`` completed folds. Default: `1`."""
        self._send(ProgressChannel.FOLD, count)

    def run(self, count: int = 1) -> None:
        """Report ``count`` completed runs. Default: `1`."""
        self._send(ProgressChannel.RUN, count)

    def _send(self, channel: ProgressChannel, count: int) -> None:
        """Publish ``(channel, count)`` unless this reporter is inert."""
        if self.queue is not None:
            self.queue.put((channel.value, count))

enabled property

bool: Whether updates are published rather than discarded.

epoch(count=1)

Report count completed epochs. Default: 1.

Source code in src/hgp_lib/benchmarkers/progress.py

def epoch(self, count: int = 1) -> None:
    """Report ``count`` completed epochs. Default: `1`."""
    self._send(ProgressChannel.EPOCH, count)

fold(count=1)

Report count completed folds. Default: 1.

Source code in src/hgp_lib/benchmarkers/progress.py

def fold(self, count: int = 1) -> None:
    """Report ``count`` completed folds. Default: `1`."""
    self._send(ProgressChannel.FOLD, count)

run(count=1)

Report count completed runs. Default: 1.

Source code in src/hgp_lib/benchmarkers/progress.py

def run(self, count: int = 1) -> None:
    """Report ``count`` completed runs. Default: `1`."""
    self._send(ProgressChannel.RUN, count)

hgp_lib.benchmarkers.progress.ProgressChannel

Bases: str, Enum

The nesting levels a worker can report progress on.

Each member's value is the literal that travels over the queue.

Examples:

>>> from hgp_lib.benchmarkers.progress import ProgressChannel
>>> ProgressChannel.FOLD.value
'fold'

Source code in src/hgp_lib/benchmarkers/progress.py

class ProgressChannel(str, Enum):
    """
    The nesting levels a worker can report progress on.

    Each member's value is the literal that travels over the queue.

    Examples:
        >>> from hgp_lib.benchmarkers.progress import ProgressChannel
        >>> ProgressChannel.FOLD.value
        'fold'
    """

    EPOCH = "epoch"
    FOLD = "fold"
    RUN = "run"