tobi/grammar-inference-engine
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat: BEX_VERBOSE env var + real-time parallel progress logging

Browse source 
_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.
This commit is contained in:
tobjend 2026-07-12 23:46:05 +02:00
parent ea6cac53e3
commit dac33db616
1 changed files with 13 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
bex/tag_preprocessor/analyze.py
View file

@ -178,14 +178,16 @@ def scan_directory(dir_path, gitignore_spec=None):
DEFAULT_COVERAGE = 0.05 DEFAULT_COVERAGE = 0.05
VERBOSE = False VERBOSE = os.environ.get("BEX_VERBOSE", "").lower() in ("1", "true", "yes", "on")
_vstart = 0.0 _vstart = time.time()
def _vprint(*args, **kwargs): 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 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): def frequency_filter(sequences, min_coverage=0.2):
@ -239,6 +241,7 @@ def _preprocess_files(file_paths):
sequences = [] sequences = []
seq_files = [] seq_files = []
n_workers = os.cpu_count() n_workers = os.cpu_count()
_vprint(f"Preprocessing {len(file_paths)} files across {n_workers} workers ...")
with ProcessPoolExecutor(max_workers=n_workers) as ex: with ProcessPoolExecutor(max_workers=n_workers) as ex:
futures = {ex.submit(_preprocess_file, fp): fp for fp in file_paths} futures = {ex.submit(_preprocess_file, fp): fp for fp in file_paths}
for f in as_completed(futures): for f in as_completed(futures):
@ -516,6 +519,7 @@ def analyze_by_package(file_paths, extension, project_root="", min_coverage=DEFA
results = [] results = []
n_workers = os.cpu_count() n_workers = os.cpu_count()
_vprint(f"Inferring {len(groups)} groups across {n_workers} workers ...") _vprint(f"Inferring {len(groups)} groups across {n_workers} workers ...")
t_infer = time.time()
with ProcessPoolExecutor(max_workers=n_workers) as ex: with ProcessPoolExecutor(max_workers=n_workers) as ex:
futures = {} futures = {}
for label, indices in groups: 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) 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 futures[f] = label
done = 0
for f in as_completed(futures): for f in as_completed(futures):
label, result, count, meta = f.result() label, result, count, meta = f.result()
results.append((label, result, count, meta)) 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]) results.sort(key=lambda x: x[0])
@ -1052,6 +1058,8 @@ def main():
args = _parse_args() args = _parse_args()
global VERBOSE, _vstart global VERBOSE, _vstart
VERBOSE = args.verbose VERBOSE = args.verbose
if args.verbose:
os.environ["BEX_VERBOSE"] = "1" # propagate to worker processes
_vstart = time.time() _vstart = time.time()
_vprint(f"Scanning {args.directory} ...") _vprint(f"Scanning {args.directory} ...")
results = analyze_directory( results = analyze_directory(