Skip to content

module looptalker

solverpy_learn.report.talker.looptalker

LoopTalker — progress reporter for hyperparameter tuning

Combines ATP evaluation progress (tqdm bars or log lines) and LightGBM tuning progress into a single talker for AutoTuner.

Runs entirely in the parent process; cross-process forwarding is handled by RemoteTalker wrapping it.

LoopTalker

Bases: EvalTalker

Progress talker for the hyperparameter tuning pipeline.

autotune-looptalker

Outside a tuning phase, behaves identically to EvalTalker (per-job SolvingBar + cumulative RunningBar).

Inside a tuning phase (between tune_begin / tune_end), switches to two-bar tuning mode: an outer tune bar counting completed trials, and an inner [N/M] build or [N/M] eval bar for the current LightGBM training or ATP evaluation. All desc strings are padded to the same width for alignment.

Set headless=True for non-interactive use: all handlers use logger.info and no tqdm bars are created.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class LoopTalker(EvalTalker):
   """
   Progress talker for the hyperparameter tuning pipeline.

   ```plantuml name="autotune-looptalker"
   abstract class solverpy.report.talker.talker.Talker
   class solverpy.report.talker.logtalker.LogTalker extends solverpy.report.talker.talker.Talker
   class solverpy.report.talker.evaltalker.EvalTalker extends solverpy.report.talker.logtalker.LogTalker
   class solverpy_learn.report.talker.looptalker.LoopTalker extends solverpy.report.talker.evaltalker.EvalTalker {
      - _tune_bar: DefaultBar | None
      - _builder_bar: BuilderBar | None
      - _total_trials: int
      - _current_trial: int
      - _in_tuning: bool
      - _in_optuna_trial: bool
      --
      + eval_begin(jobs, *, refjob, sidnames, miniters, **kwargs)
      + eval_end(results, refjob, **kwargs)
      + eval_launch(tasks)
      + eval_done()
      + tune_begin(t_start, total)
      + tune_end(t_end)
      + tune_trial_begin(nick, it, values)
      + tune_trial_done(stats)
      + build_begin(f_mod, total)
      + build_step(n, total, metrics)
      + build_selected(iteration, metrics)
      + build_done(score)
   }
   ```

   Outside a tuning phase, behaves identically to
   [`EvalTalker`][solverpy.report.talker.evaltalker.EvalTalker] (per-job
   ``SolvingBar`` + cumulative ``RunningBar``).

   Inside a tuning phase (between ``tune_begin`` / ``tune_end``), switches to
   two-bar tuning mode: an outer ``tune`` bar counting completed trials, and
   an inner ``[N/M] build`` or ``[N/M] eval`` bar for the current LightGBM
   training or ATP evaluation.  All desc strings are padded to the same width
   for alignment.

   Set ``headless=True`` for non-interactive use: all handlers use
   ``logger.info`` and no tqdm bars are created.
   """

   def __init__(self, headless: bool = False) -> None:
      """
      Args:
          headless: if ``True``, use log lines instead of tqdm bars.
      """
      super().__init__(headless=headless)
      self._builder_bar: BuilderBar | None = None
      self._tune_bar: DefaultBar | None = None
      self._total_trials: int = 0
      self._current_trial: int = 0
      self._in_tuning: bool = False
      self._in_optuna_trial: bool = False
      self._in_tune_eval: bool = False

   # --- Helpers ---

   def _trial_desc(self, name: str) -> str:
      """Return '[N/M] name' aligned with the tune bar, or just 'name' if total unknown."""
      if not self._total_trials:
         return name
      dw = len(str(self._total_trials))
      n = self._current_trial + 1
      return f"[{n:>{dw}}/{self._total_trials}] {name}"

   def _inner_width(self) -> int:
      """Width of '[N/M] build' / '[N/M] eval ' for tune bar desc alignment."""
      if not self._total_trials:
         return 0
      dw = len(str(self._total_trials))
      return 2 * dw + 9  # len("[N/M] build") with dw-wide fields

   # --- ATP evaluation overrides ---

   def eval_begin(self,
                  jobs,
                  *,
                  refjob=None,
                  sidnames=True,
                  miniters=1,
                  **kwargs) -> None:
      """Outside tuning: full EvalTalker bars. Inside tuning: single RunningBar, skips report."""
      if not self._in_tuning:
         EvalTalker.eval_begin(self,
                               jobs,
                               refjob=refjob,
                               sidnames=sidnames,
                               miniters=miniters,
                               **kwargs)
         return
      LogTalker.eval_begin(self,
                           jobs,
                           refjob=refjob,
                           sidnames=sidnames,
                           report=False,
                           **kwargs)
      if not self._headless:
         max_job = max(len(_bids.problems(bid)) for (_, bid, _) in jobs)
         self._total_bar = RunningBar(
            total=self._total_count,
            desc=self._trial_desc("eval "),
            miniters=miniters,
            colour="blue",
            leave=True,
            postfix_width=_postfix_width(max_job),
         )

   def eval_end(self, results, refjob=None, **kwargs) -> None:
      """Outside tuning: full EvalTalker end. Inside tuning: close eval bar, update tune bar."""
      if not self._in_tuning:
         EvalTalker.eval_end(self, results, refjob=refjob, **kwargs)
         return
      if self._total_bar:
         self._total_errors = self._total_bar._errors
         self._total_bar.close()
         self._total_bar = None
      LogTalker.eval_end(self, results, refjob=refjob, report=False)
      if not self._in_optuna_trial:
         self._current_trial += 1
         if self._tune_bar:
            self._tune_bar.update(1)

   def eval_status(self, new: "bool | None", n: int = 1) -> None:
      """Update eval bar and refresh tune bar to keep it visible."""
      super().eval_status(new, n)
      if self._tune_bar:
         self._tune_bar.refresh()

   def eval_launch(self, tasks: Sequence["Task"]) -> None:
      """Outside tuning: per-job SolvingBar. Inside tuning: record time only."""
      if not self._in_tuning:
         EvalTalker.eval_launch(self, tasks)
      else:
         LogTalker.eval_launch(self, tasks)

   def eval_done(self) -> None:
      """Outside tuning: close job bar. Inside tuning: log without closing any bar."""
      if not self._in_tuning:
         EvalTalker.eval_done(self)
      else:
         LogTalker.eval_done(self)

   # --- Tuning lifecycle overrides ---

   def tune_begin(self, t_start: float, total: int = 0) -> None:
      """Enter tuning mode; create outer tune bar if not headless."""
      self._in_tuning = True
      self._total_trials = total
      self._current_trial = 0
      self._suspend_info = not self._headless
      if self._headless or not total:
         return
      self._tune_bar = DefaultBar(total, "tune".ljust(self._inner_width()))

   def tune_end(self, t_end: float) -> None:
      """Leave tuning mode; close tune bar and restore info logs."""
      self._in_tuning = False
      self._suspend_info = False
      self._total_trials = 0
      self._current_trial = 0
      if self._tune_bar:
         self._tune_bar.close()
         self._tune_bar = None

   def tune_trial_begin(self, nick: str, it: int, values: list) -> None:
      """Mark start of Optuna trial and delegate to LogTalker."""
      self._in_optuna_trial = True
      LogTalker.tune_trial_begin(self, nick, it, values)

   def tune_trial_done(self, stats: dict[str, Any]) -> None:
      """Increment tune bar at end of Optuna trial."""
      LogTalker.tune_trial_done(self, stats)
      self._in_optuna_trial = False
      self._current_trial += 1
      if self._tune_bar:
         self._tune_bar.update(1)

   def tune_eval_begin(self) -> None:
      """Suppress eval log messages while ATP evaluation runs inside a tuning trial."""
      self._in_tune_eval = True

   def tune_eval_end(self, results: dict) -> None:
      """Re-enable eval log messages after tuning ATP evaluation."""
      self._in_tune_eval = False

   def terminate(self) -> None:
      """Delegate to parent; _tune_bar is closed by tune_end()."""
      super().terminate()

   # --- Model build overrides ---

   def build_begin(self, f_mod: str, total: int) -> None:
      """Open a ``BuilderBar`` with [N/M] build label if not headless."""
      LogTalker.build_begin(self, f_mod, total)
      if not self._headless:
         self._builder_bar = BuilderBar(total,
                                        self._trial_desc("build"),
                                        colour="blue",
                                        leave=True)

   def build_step(
      self,
      n: int,
      total: int,
      metrics: dict[str, dict[str, float]],
   ) -> None:
      """Update the builder bar, refresh tune bar, and emit periodic log lines."""
      if self._builder_bar:
         self._builder_bar.status(metrics)
      if self._tune_bar:
         self._tune_bar.refresh()
      LogTalker.build_step(self, n, total, metrics)

   def build_selected(
      self,
      iteration: int,
      metrics: dict[str, dict[str, float]],
   ) -> None:
      """Report selected-model metrics through the log/reporting layer."""
      LogTalker.build_selected(self, iteration, metrics)

   def build_done(self, score: float) -> None:
      """Close the builder bar."""
      if self._builder_bar:
         self._builder_bar.close()
         self._builder_bar = None
      LogTalker.build_done(self, score)

__init__(headless: bool = False) -> None

Parameters:

Name Type Description Default
headless bool

if True, use log lines instead of tqdm bars.

 False
Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(self, headless: bool = False) -> None:
   """
   Args:
       headless: if ``True``, use log lines instead of tqdm bars.
   """
   super().__init__(headless=headless)
   self._builder_bar: BuilderBar | None = None
   self._tune_bar: DefaultBar | None = None
   self._total_trials: int = 0
   self._current_trial: int = 0
   self._in_tuning: bool = False
   self._in_optuna_trial: bool = False
   self._in_tune_eval: bool = False

build_begin(f_mod: str, total: int) -> None

Open a BuilderBar with [N/M] build label if not headless.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
221
222
223
224
225
226
227
228
def build_begin(self, f_mod: str, total: int) -> None:
   """Open a ``BuilderBar`` with [N/M] build label if not headless."""
   LogTalker.build_begin(self, f_mod, total)
   if not self._headless:
      self._builder_bar = BuilderBar(total,
                                     self._trial_desc("build"),
                                     colour="blue",
                                     leave=True)

build_done(score: float) -> None

Close the builder bar.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
251
252
253
254
255
256
def build_done(self, score: float) -> None:
   """Close the builder bar."""
   if self._builder_bar:
      self._builder_bar.close()
      self._builder_bar = None
   LogTalker.build_done(self, score)

build_selected(iteration: int, metrics: dict[str, dict[str, float]]) -> None

Report selected-model metrics through the log/reporting layer.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
243
244
245
246
247
248
249
def build_selected(
   self,
   iteration: int,
   metrics: dict[str, dict[str, float]],
) -> None:
   """Report selected-model metrics through the log/reporting layer."""
   LogTalker.build_selected(self, iteration, metrics)

build_step(n: int, total: int, metrics: dict[str, dict[str, float]]) -> None

Update the builder bar, refresh tune bar, and emit periodic log lines.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
230
231
232
233
234
235
236
237
238
239
240
241
def build_step(
   self,
   n: int,
   total: int,
   metrics: dict[str, dict[str, float]],
) -> None:
   """Update the builder bar, refresh tune bar, and emit periodic log lines."""
   if self._builder_bar:
      self._builder_bar.status(metrics)
   if self._tune_bar:
      self._tune_bar.refresh()
   LogTalker.build_step(self, n, total, metrics)

eval_begin(jobs, *, refjob=None, sidnames=True, miniters=1, **kwargs) -> None

Outside tuning: full EvalTalker bars. Inside tuning: single RunningBar, skips report.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def eval_begin(self,
               jobs,
               *,
               refjob=None,
               sidnames=True,
               miniters=1,
               **kwargs) -> None:
   """Outside tuning: full EvalTalker bars. Inside tuning: single RunningBar, skips report."""
   if not self._in_tuning:
      EvalTalker.eval_begin(self,
                            jobs,
                            refjob=refjob,
                            sidnames=sidnames,
                            miniters=miniters,
                            **kwargs)
      return
   LogTalker.eval_begin(self,
                        jobs,
                        refjob=refjob,
                        sidnames=sidnames,
                        report=False,
                        **kwargs)
   if not self._headless:
      max_job = max(len(_bids.problems(bid)) for (_, bid, _) in jobs)
      self._total_bar = RunningBar(
         total=self._total_count,
         desc=self._trial_desc("eval "),
         miniters=miniters,
         colour="blue",
         leave=True,
         postfix_width=_postfix_width(max_job),
      )

eval_done() -> None

Outside tuning: close job bar. Inside tuning: log without closing any bar.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
165
166
167
168
169
170
def eval_done(self) -> None:
   """Outside tuning: close job bar. Inside tuning: log without closing any bar."""
   if not self._in_tuning:
      EvalTalker.eval_done(self)
   else:
      LogTalker.eval_done(self)

eval_end(results, refjob=None, **kwargs) -> None

Outside tuning: full EvalTalker end. Inside tuning: close eval bar, update tune bar.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def eval_end(self, results, refjob=None, **kwargs) -> None:
   """Outside tuning: full EvalTalker end. Inside tuning: close eval bar, update tune bar."""
   if not self._in_tuning:
      EvalTalker.eval_end(self, results, refjob=refjob, **kwargs)
      return
   if self._total_bar:
      self._total_errors = self._total_bar._errors
      self._total_bar.close()
      self._total_bar = None
   LogTalker.eval_end(self, results, refjob=refjob, report=False)
   if not self._in_optuna_trial:
      self._current_trial += 1
      if self._tune_bar:
         self._tune_bar.update(1)

eval_launch(tasks: Sequence[Task]) -> None

Outside tuning: per-job SolvingBar. Inside tuning: record time only.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
158
159
160
161
162
163
def eval_launch(self, tasks: Sequence["Task"]) -> None:
   """Outside tuning: per-job SolvingBar. Inside tuning: record time only."""
   if not self._in_tuning:
      EvalTalker.eval_launch(self, tasks)
   else:
      LogTalker.eval_launch(self, tasks)

eval_status(new: bool | None, n: int = 1) -> None

Update eval bar and refresh tune bar to keep it visible.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
152
153
154
155
156
def eval_status(self, new: "bool | None", n: int = 1) -> None:
   """Update eval bar and refresh tune bar to keep it visible."""
   super().eval_status(new, n)
   if self._tune_bar:
      self._tune_bar.refresh()

terminate() -> None

Delegate to parent; _tune_bar is closed by tune_end().

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
215
216
217
def terminate(self) -> None:
   """Delegate to parent; _tune_bar is closed by tune_end()."""
   super().terminate()

tune_begin(t_start: float, total: int = 0) -> None

Enter tuning mode; create outer tune bar if not headless.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
174
175
176
177
178
179
180
181
182
def tune_begin(self, t_start: float, total: int = 0) -> None:
   """Enter tuning mode; create outer tune bar if not headless."""
   self._in_tuning = True
   self._total_trials = total
   self._current_trial = 0
   self._suspend_info = not self._headless
   if self._headless or not total:
      return
   self._tune_bar = DefaultBar(total, "tune".ljust(self._inner_width()))

tune_end(t_end: float) -> None

Leave tuning mode; close tune bar and restore info logs.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
184
185
186
187
188
189
190
191
192
def tune_end(self, t_end: float) -> None:
   """Leave tuning mode; close tune bar and restore info logs."""
   self._in_tuning = False
   self._suspend_info = False
   self._total_trials = 0
   self._current_trial = 0
   if self._tune_bar:
      self._tune_bar.close()
      self._tune_bar = None

tune_eval_begin() -> None

Suppress eval log messages while ATP evaluation runs inside a tuning trial.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
207
208
209
def tune_eval_begin(self) -> None:
   """Suppress eval log messages while ATP evaluation runs inside a tuning trial."""
   self._in_tune_eval = True

tune_eval_end(results: dict) -> None

Re-enable eval log messages after tuning ATP evaluation.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
211
212
213
def tune_eval_end(self, results: dict) -> None:
   """Re-enable eval log messages after tuning ATP evaluation."""
   self._in_tune_eval = False

tune_trial_begin(nick: str, it: int, values: list) -> None

Mark start of Optuna trial and delegate to LogTalker.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
194
195
196
197
def tune_trial_begin(self, nick: str, it: int, values: list) -> None:
   """Mark start of Optuna trial and delegate to LogTalker."""
   self._in_optuna_trial = True
   LogTalker.tune_trial_begin(self, nick, it, values)

tune_trial_done(stats: dict[str, Any]) -> None

Increment tune bar at end of Optuna trial.

Source code in packages/solverpy-learn/src/solverpy_learn/report/talker/looptalker.py 
199
200
201
202
203
204
205
def tune_trial_done(self, stats: dict[str, Any]) -> None:
   """Increment tune bar at end of Optuna trial."""
   LogTalker.tune_trial_done(self, stats)
   self._in_optuna_trial = False
   self._current_trial += 1
   if self._tune_bar:
      self._tune_bar.update(1)