From dac33db616b41f6919bc329b5e7396612816f187 Mon Sep 17 00:00:00 2001 From: tobjend Date: Sun, 12 Jul 2026 23:46:05 +0200 Subject: [PATCH] feat: BEX_VERBOSE env var + real-time parallel progress logging _vprint now honors the BEX_VERBOSE env var in addition to the CLI --verbose flag, so non-CLI callers (MCP, dervish integration) and worker processes (which start with a fresh module) emit logs too. --verbose exports BEX_VERBOSE=1 so ProcessPoolExecutor workers inherit it. analyze_by_package now logs real-time per-group completion with [i/N] label (count methods) done (Xs), making it obvious the parallel pipeline is progressing and not hung. --- bex/tag_preprocessor/analyze.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/bex/tag_preprocessor/analyze.py b/bex/tag_preprocessor/analyze.py index 8613616..e8e01d5 100644 --- a/bex/tag_preprocessor/analyze.py +++ b/bex/tag_preprocessor/analyze.py @@ -178,14 +178,16 @@ def scan_directory(dir_path, gitignore_spec=None): DEFAULT_COVERAGE = 0.05 -VERBOSE = False -_vstart = 0.0 +VERBOSE = os.environ.get("BEX_VERBOSE", "").lower() in ("1", "true", "yes", "on") +_vstart = time.time() def _vprint(*args, **kwargs): - if VERBOSE: + # Emit if enabled via CLI flag (VERBOSE global) or BEX_VERBOSE env var. + # Env var also propagates into worker processes, which start with a fresh module. + if VERBOSE or os.environ.get("BEX_VERBOSE", "").lower() in ("1", "true", "yes", "on"): elapsed = time.time() - _vstart - print(f"[{elapsed:6.1f}s]", *args, file=sys.stderr, **kwargs) + print(f"[{elapsed:6.1f}s]", *args, file=sys.stderr, flush=True, **kwargs) def frequency_filter(sequences, min_coverage=0.2): @@ -239,6 +241,7 @@ def _preprocess_files(file_paths): sequences = [] seq_files = [] n_workers = os.cpu_count() + _vprint(f"Preprocessing {len(file_paths)} files across {n_workers} workers ...") with ProcessPoolExecutor(max_workers=n_workers) as ex: futures = {ex.submit(_preprocess_file, fp): fp for fp in file_paths} for f in as_completed(futures): @@ -516,6 +519,7 @@ def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFA results = [] n_workers = os.cpu_count() _vprint(f"Inferring {len(groups)} groups across {n_workers} workers ...") + t_infer = time.time() with ProcessPoolExecutor(max_workers=n_workers) as ex: futures = {} for label, indices in groups: @@ -525,10 +529,12 @@ def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFA min_coverage, prefer, kmax, N, include_kore, include_idregex, method, min_methods, crx_method, min_structure, split_mixed, idregex_refine, cluster_method, decompose, max_seq_length) futures[f] = label + done = 0 for f in as_completed(futures): label, result, count, meta = f.result() results.append((label, result, count, meta)) - _vprint(f"Infer {label} ({count} methods) done") + done += 1 + _vprint(f" [{done}/{len(futures)}] {label} ({count} methods) done ({time.time()-t_infer:.1f}s)") results.sort(key=lambda x: x[0]) @@ -1052,6 +1058,8 @@ def main(): args = _parse_args() global VERBOSE, _vstart VERBOSE = args.verbose + if args.verbose: + os.environ["BEX_VERBOSE"] = "1" # propagate to worker processes _vstart = time.time() _vprint(f"Scanning {args.directory} ...") results = analyze_directory(