bex/tag_preprocessor/analyze.py

@@ -0,0 +1,138 @@
+"""Orchestrator: directory scan → preprocess → frequency filter → ensemble infer.
+
+Usage:
+    python -m bex.tag_preprocessor.analyze <directory>
+
+Runs the full Phase 1.0 pipeline over a directory of source files.
+"""
+
+import os
+import sys
+from pathlib import Path
+from collections import Counter
+
+from .code import preprocess
+from bex.ensemble import infer_ensemble
+
+SUPPORTED_EXTENSIONS = {
+    ".py", ".js", ".ts", ".kt", ".rb", ".go", ".rs", ".java", ".c", ".cpp",
+}
+
+
+def scan_directory(dir_path):
+    """Walk dir_path, return dict mapping extension → [file paths].
+
+    Only includes supported extensions. Walks recursively.
+    """
+    result = {}
+    for root, _, files in os.walk(dir_path):
+        for f in files:
+            ext = os.path.splitext(f)[1].lower()
+            if ext in SUPPORTED_EXTENSIONS:
+                result.setdefault(ext, []).append(os.path.join(root, f))
+    return result
+
+
+def frequency_filter(sequences, min_coverage=0.2):
+    """Remove symbols appearing in fewer than min_coverage fraction of files.
+
+    Args:
+        sequences: list of lists of (capture_name, text, line_number) tuples.
+        min_coverage: minimum fraction of files a symbol must appear in.
+
+    Returns:
+        Filtered sequences with rare symbols removed.
+    """
+    if not sequences:
+        return sequences
+
+    n_files = len(sequences)
+    threshold = max(1, int(n_files * min_coverage))
+
+    symbol_file_count = Counter()
+    for seq in sequences:
+        seen = set()
+        for _, text, _ in seq:
+            if text not in seen:
+                symbol_file_count[text] += 1
+                seen.add(text)
+
+    keep = {text for text, count in symbol_file_count.items()
+            if count >= threshold}
+
+    filtered = []
+    for seq in sequences:
+        new_seq = [(cap, text, line) for cap, text, line in seq if text in keep]
+        filtered.append(new_seq)
+
+    return filtered
+
+
+def infer(file_paths, extension, min_coverage=0.2, prefer=None, kmax=2, N=3):
+    """Run full pipeline: preprocess → frequency filter → ensemble infer.
+
+    Args:
+        file_paths: list of source file paths (same language).
+        extension: language extension (e.g. '.py').
+        min_coverage: minimum file fraction for a symbol to be kept.
+        prefer: inference algorithm preference ('crx', 'idregex', or None).
+        kmax: max k for k-ORE algorithms.
+        N: number of random trials.
+
+    Returns:
+        Ensemble result dict from infer_ensemble.
+    """
+    sequences = []
+    for fp in file_paths:
+        with open(fp) as f:
+            code = f.read()
+        seq = preprocess(fp, code)
+        if seq:
+            sequences.append(seq)
+
+    sequences = frequency_filter(sequences, min_coverage)
+
+    symbol_seqs = [[text for _, text, _ in seq] for seq in sequences]
+
+    return infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer)
+
+
+def analyze_directory(dir_path, min_coverage=0.2, prefer=None):
+    """Scan a directory and run infer for each language found.
+
+    Args:
+        dir_path: directory to scan.
+        min_coverage: minimum file fraction for symbol to be kept.
+        prefer: algorithm preference.
+
+    Returns:
+        dict mapping extension → ensemble result dict.
+    """
+    groups = scan_directory(dir_path)
+    results = {}
+    for ext, files in groups.items():
+        if len(files) < 1:
+            continue
+        results[ext] = infer(files, ext, min_coverage=min_coverage, prefer=prefer)
+    return results
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: python -m bex.tag_preprocessor.analyze <directory>", file=sys.stderr)
+        sys.exit(1)
+    dir_path = sys.argv[1]
+    results = analyze_directory(dir_path)
+    for ext, result in results.items():
+        best = result.get("best")
+        if best:
+            print(f"\n{ext}:")
+            print(f"  Algorithm: {best['algorithm']}")
+            print(f"  Grammar:   {best['grammar']}")
+            print(f"  MDL:       {best['mdl_score']}")
+        else:
+            print(f"\n{ext}: no grammar inferred")
+
+
+if __name__ == "__main__":
+    main()

bex/tag_preprocessor/code.py

@@ -1,7 +1,7 @@
 """Universal tree-sitter tag preprocessor.
 
 Usage:
-    python -m bex.tag-preprocessor.code <file>
+    python -m bex.tag_preprocessor.code <file>
 
 Emits an ordered sequence of behavioral tokens using community highlights.scm
 queries from nvim-treesitter. One code path for all languages.

tests/test_analyze.py

@@ -0,0 +1,146 @@
+"""Tests for tag-preprocessor orchestrator (analyze.py)."""
+
+from pathlib import Path
+import tempfile
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from bex.tag_preprocessor.analyze import scan_directory, frequency_filter, infer
+
+
+def test_scan_directory_empty():
+    with tempfile.TemporaryDirectory() as td:
+        result = scan_directory(td)
+        assert result == {}
+    print("  PASS test_scan_directory_empty")
+
+
+def test_scan_directory_groups_by_extension():
+    with tempfile.TemporaryDirectory() as td:
+        d = Path(td)
+        (d / "a.py").write_text("x = 1")
+        (d / "b.py").write_text("y = 2")
+        (d / "c.js").write_text("let x = 1;")
+        (d / "d.rs").write_text("fn main() {}")
+        result = scan_directory(td)
+        assert ".py" in result
+        assert ".js" in result
+        assert ".rs" in result
+        assert len(result[".py"]) == 2
+        assert len(result[".js"]) == 1
+        assert len(result[".rs"]) == 1
+    print("  PASS test_scan_directory_groups_by_extension")
+
+
+def test_scan_directory_skips_unsupported():
+    with tempfile.TemporaryDirectory() as td:
+        d = Path(td)
+        (d / "f.py").write_text("x = 1")
+        (d / "f.txt").write_text("hello")
+        (d / "f.md").write_text("# doc")
+        result = scan_directory(td)
+        assert ".py" in result
+        assert ".txt" not in result
+        assert ".md" not in result
+    print("  PASS test_scan_directory_skips_unsupported")
+
+
+def test_scan_directory_nested():
+    with tempfile.TemporaryDirectory() as td:
+        d = Path(td)
+        (d / "sub").mkdir()
+        (d / "sub" / "a.py").write_text("x = 1")
+        (d / "b.py").write_text("y = 2")
+        result = scan_directory(td)
+        assert len(result[".py"]) == 2
+    print("  PASS test_scan_directory_nested")
+
+
+def test_frequency_filter_nothing_to_filter():
+    sequences = [
+        [("function", "foo", 1), ("keyword.return", "return", 2)],
+        [("function", "bar", 1), ("keyword.return", "return", 2)],
+    ]
+    filtered = frequency_filter(sequences, min_coverage=0.5)
+    assert len(filtered) == 2
+    assert filtered == sequences
+    print("  PASS test_frequency_filter_nothing_to_filter")
+
+
+def test_frequency_filter_removes_infrequent_symbol():
+    sequences = [
+        [("keyword.return", "return", 1)],
+        [("keyword.return", "return", 1)],
+        [("function", "rare_fn", 1)],
+    ]
+    filtered = frequency_filter(sequences, min_coverage=0.67)
+    assert len(filtered) == 3
+    assert len(filtered[0]) == 1  # "return" kept
+    assert len(filtered[1]) == 1  # "return" kept
+    assert len(filtered[2]) == 0  # "rare_fn" removed
+    print("  PASS test_frequency_filter_removes_infrequent_symbol")
+
+
+def test_frequency_filter_edge_empty_sequences():
+    assert frequency_filter([], min_coverage=0.5) == []
+    assert frequency_filter([[], []], min_coverage=0.5) == [[], []]
+    print("  PASS test_frequency_filter_edge_empty_sequences")
+
+
+def test_infer_returns_ensemble_dict():
+    with tempfile.TemporaryDirectory() as td:
+        d = Path(td)
+        (d / "a.py").write_text("def foo():\n    return 1")
+        (d / "b.py").write_text("def bar():\n    return 2")
+        result = infer([str(d / "a.py"), str(d / "b.py")], ".py", min_coverage=0.5)
+        assert isinstance(result, dict)
+        assert "best" in result
+        assert "all" in result
+        assert "why" in result
+    print("  PASS test_infer_returns_ensemble_dict")
+
+
+def test_infer_low_coverage_filters_noise():
+    with tempfile.TemporaryDirectory() as td:
+        d = Path(td)
+        (d / "common.py").write_text(
+            "def setup():\n    pass\ndef teardown():\n    pass"
+        )
+        (d / "rare.py").write_text(
+            "def setup():\n    pass\ndef one_off():\n    raise Exception('boom')"
+        )
+        result = infer(
+            [str(d / "common.py"), str(d / "rare.py")], ".py", min_coverage=0.6
+        )
+        assert result["best"] is not None
+        assert result["best"]["grammar"] is not None
+    print("  PASS test_infer_low_coverage_filters_noise")
+
+
+def run_all():
+    tests = [
+        test_scan_directory_empty,
+        test_scan_directory_groups_by_extension,
+        test_scan_directory_skips_unsupported,
+        test_scan_directory_nested,
+        test_frequency_filter_nothing_to_filter,
+        test_frequency_filter_removes_infrequent_symbol,
+        test_frequency_filter_edge_empty_sequences,
+        test_infer_returns_ensemble_dict,
+        test_infer_low_coverage_filters_noise,
+    ]
+    passed = 0
+    failed = 0
+    for t in tests:
+        try:
+            t()
+            passed += 1
+        except Exception as e:
+            print(f"  FAIL {t.__name__}: {e}")
+            failed += 1
+    print(f"\n{passed} passed, {failed} failed")
+
+
+if __name__ == "__main__":
+    run_all()