diff --git a/bex/tag_preprocessor/analyze.py b/bex/tag_preprocessor/analyze.py index d539183..cec0dc5 100644 --- a/bex/tag_preprocessor/analyze.py +++ b/bex/tag_preprocessor/analyze.py @@ -14,7 +14,7 @@ from collections import Counter import pathspec -from .code import preprocess_by_method +from .code import preprocess_by_method, _extract_call_tokens from bex.ensemble import infer_ensemble SUPPORTED_EXTENSIONS = { @@ -117,6 +117,79 @@ def frequency_filter(sequences, min_coverage=0.2): return filtered +def cluster_methods(sequences, min_cluster_size=3, ngram_size=3): + """Group method sequences by shared n-gram call patterns. + + Extracts call tokens from each sequence, builds an n-gram index, + and assigns methods to the largest matching clusters first. + Remaining methods go to an '(other)' cluster. + + Args: + sequences: list of (capture, text, line) lists. + min_cluster_size: minimum methods to form a cluster. + ngram_size: length of n-grams to match (default 3). + + Returns: + list of (label, [sequences]) tuples. + """ + if not sequences: + return [] + + call_seqs = [_extract_call_tokens(seq) for seq in sequences] + + ngram_to_methods = {} + for idx, cseq in enumerate(call_seqs): + for i in range(len(cseq) - ngram_size + 1): + ngram = tuple(cseq[i:i + ngram_size]) + if len(ngram) == ngram_size: + ngram_to_methods.setdefault(ngram, set()).add(idx) + + used = set() + clusters = [] + for ngram, indices in sorted(ngram_to_methods.items(), key=lambda x: -len(x[1])): + indices = indices - used + if len(indices) >= min_cluster_size: + label = " → ".join(ngram) + cluster_seqs = [sequences[i] for i in indices] + clusters.append((label, cluster_seqs)) + used.update(indices) + + remaining = [i for i in range(len(sequences)) if i not in used] + if remaining: + clusters.append(("(other)", [sequences[i] for i in remaining])) + + return clusters + + +def analyze_clusters(file_paths, extension, min_coverage=0.2, prefer=None, kmax=2, N=3): + """Run full pipeline with clustering: preprocess → cluster → per-cluster infer. + + Returns: + list of (label, ensemble_result_dict, method_count) tuples. + """ + sequences = [] + for fp in file_paths: + with open(fp) as f: + code = f.read() + for method_seq in preprocess_by_method(fp, code): + if method_seq: + sequences.append(method_seq) + + if not sequences: + return [] + + sequences = frequency_filter(sequences, min_coverage) + clusters = cluster_methods(sequences) + + results = [] + for label, cluster_seqs in clusters: + symbol_seqs = [[text for _, text, _ in seq] for seq in cluster_seqs] + result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer) + results.append((label, result, len(cluster_seqs))) + + return results + + def infer(file_paths, extension, min_coverage=0.2, prefer=None, kmax=2, N=3): """Run full pipeline: preprocess → frequency filter → ensemble infer. @@ -152,8 +225,9 @@ def analyze_directory( prefer=None, kmax=2, include=None, + cluster=True, ): - """Scan a directory and run infer for each language found. + """Scan a directory and run cluster analysis for each language found. Args: dir_path: directory to scan. @@ -161,9 +235,10 @@ def analyze_directory( prefer: algorithm preference. kmax: max k for k-ORE algorithms. include: optional glob pattern to filter files (e.g. '**/src/main/**'). + cluster: if True, run method-level clustering (default). Returns: - dict mapping extension → ensemble result dict. + dict mapping extension → list of (label, result_dict, count) tuples. """ groups = scan_directory(dir_path) results = {} @@ -174,12 +249,16 @@ def analyze_directory( files = [f for f in files if _match_glob(f, include)] if not files: continue - results[ext] = infer( - files, ext, - min_coverage=min_coverage, - prefer=prefer, - kmax=kmax, - ) + if cluster: + results[ext] = analyze_clusters( + files, ext, + min_coverage=min_coverage, + prefer=prefer, + kmax=kmax, + ) + else: + r = infer(files, ext, min_coverage=min_coverage, prefer=prefer, kmax=kmax) + results[ext] = [("(all methods)", r, 0)] return results @@ -205,6 +284,14 @@ def _parse_args(argv=None): "--include", help="Glob pattern to filter files (e.g. '**/src/main/**')", ) + parser.add_argument( + "--min-cluster-size", type=int, default=3, + help="Minimum methods to form a cluster (default: 3)", + ) + parser.add_argument( + "--ngram-size", type=int, default=3, + help="N-gram length for clustering (default: 3)", + ) return parser.parse_args(argv) @@ -217,15 +304,17 @@ def main(): kmax=args.kmax, include=args.include, ) - 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") + for ext, clusters in results.items(): + print(f"\n{ext}:") + for label, result, count in clusters: + if result and result.get("best"): + best = result["best"] + print(f" ╰─ {label} ({count} methods)") + print(f" Algorithm: {best['algorithm']}") + print(f" Grammar: {best['grammar']}") + print(f" MDL: {best['mdl_score']}") + else: + print(f" ╰─ {label} ({count} methods) — no grammar") if __name__ == "__main__": diff --git a/bex/tag_preprocessor/code.py b/bex/tag_preprocessor/code.py index 3afa89b..206b575 100644 --- a/bex/tag_preprocessor/code.py +++ b/bex/tag_preprocessor/code.py @@ -51,6 +51,24 @@ BEHAVIORAL_PREFIXES = ( "module", ) +CALL_PREFIXES = ("function", "reference.call", "reference.class") + +def _extract_call_tokens(seq): + """Extract ordered call-like tokens from a method sequence. + + Filters to captures representing function calls, constructors, + or references — the 'what happens in what order'. + Returns list of text values. + """ + result = [] + for capname, text, _ in seq: + for prefix in CALL_PREFIXES: + if capname.startswith(prefix): + result.append(text) + break + return result + + _grammar_cache = {} _query_cache = {} diff --git a/tests/test_analyze.py b/tests/test_analyze.py index 50e2c51..b562285 100644 --- a/tests/test_analyze.py +++ b/tests/test_analyze.py @@ -133,11 +133,14 @@ def test_analyze_directory_include_glob(): d = Path(td) (d / "src" / "main").mkdir(parents=True) (d / "src" / "test").mkdir(parents=True) - (d / "src" / "main" / "prod.py").write_text("x = 1") - (d / "src" / "test" / "test_prod.py").write_text("y = 2") + (d / "src" / "main" / "prod.py").write_text("def setup():\n pass\ndef run():\n return x") + (d / "src" / "test" / "test_prod.py").write_text("def test_run():\n assert run() == x") results = analyze_directory(td, include="**/src/main/**") assert ".py" in results - assert len(results[".py"]["all"]) >= 1 + assert len(results[".py"]) >= 1 + for label, r, count in results[".py"]: + if r and r.get("best"): + assert r["best"]["grammar"] is not None print(" PASS test_analyze_directory_include_glob")