diff --git a/bex/tag_preprocessor/analyze.py b/bex/tag_preprocessor/analyze.py index bd63d62..231808d 100644 --- a/bex/tag_preprocessor/analyze.py +++ b/bex/tag_preprocessor/analyze.py @@ -16,7 +16,7 @@ from collections import Counter import pathspec -from .code import preprocess_by_method, _extract_call_tokens +from .code import preprocess_by_method, _extract_call_tokens, extract_arg_info, _summarize_arg_info from bex.ensemble import infer_ensemble IMPORT_PATTERNS = [ @@ -58,6 +58,21 @@ def _match_glob(filepath, pattern): return spec.match_file(filepath) +def _build_arg_patterns(file_paths): + """Extract merged argument patterns across a set of files.""" + merged = {} + for fp in file_paths: + try: + with open(fp) as f: + code = f.read() + except OSError: + continue + info = extract_arg_info(fp, code) + for call_name, observations in info.items(): + merged.setdefault(call_name, []).extend(observations) + return _summarize_arg_info(merged) + + def _extract_imports(file_paths): """Extract unique import lines from source files. @@ -227,9 +242,10 @@ def analyze_clusters(file_paths, extension, min_coverage=0.2, prefer=None, kmax= idx = next(i for i, s in enumerate(sequences) if s is seq) cluster_fps.add(seq_files[idx]) imports = _extract_imports(cluster_fps) + arg_patterns = _build_arg_patterns(cluster_fps) symbol_seqs = [[text for _, text, _ in seq] for seq in cluster_seqs] result = infer_ensemble(symbol_seqs, kmax=kmax, N=N, prefer=prefer) - meta = {"files": cluster_fps, "imports": imports} + meta = {"files": cluster_fps, "imports": imports, "arg_patterns": arg_patterns} results.append((label, result, len(cluster_seqs), meta)) return results @@ -365,6 +381,7 @@ def _build_json_output(results): entry["grammar"] = result["best"]["grammar"] entry["mdl_score"] = round(result["best"]["mdl_score"], 1) entry["imports"] = meta.get("imports", []) + entry["arg_patterns"] = meta.get("arg_patterns", {}) lang["conventions"].append(entry) lang["total_methods"] = total_methods output.append(lang) @@ -402,6 +419,15 @@ def main(): print(f" Imports: {joined}") if len(imps) > 6: print(f" ... and {len(imps) - 6} more") + argp = meta.get("arg_patterns", {}) + if argp: + top_calls = sorted(argp.items(), key=lambda x: -x[1]["occurrences"])[:4] + for call_name, asum in top_calls: + ac = asum["arg_count"] + pats = asum["patterns"][:2] + pat_strs = [f"{p['args']}:{','.join(p['types'])}" for p in pats] + print(f" Args({call_name}): n={ac['common']} " + f"[{'; '.join(pat_strs)}]") if __name__ == "__main__": diff --git a/bex/tag_preprocessor/code.py b/bex/tag_preprocessor/code.py index 206b575..d480e66 100644 --- a/bex/tag_preprocessor/code.py +++ b/bex/tag_preprocessor/code.py @@ -72,6 +72,135 @@ def _extract_call_tokens(seq): _grammar_cache = {} _query_cache = {} +ARG_LITERAL_TYPES = { + "string", "string_literal", "number", "integer", "float", + "decimal", "true", "false", "null", "nil", "none", +} + +LAMBDA_TYPES = { + "lambda", "do_block", "block", +} + + +def _classify_arg_node(node): + """Classify a tree-sitter argument node by structural role.""" + t = node.type + if t in ARG_LITERAL_TYPES: + return "lit" + if t in LAMBDA_TYPES or "ambda" in t or "block" in t: + return "lambda" + if t == "keyword_argument" or t.endswith("named_argument"): + return "kwarg" + if t.endswith("call_expression") or t in ("call", "method_invocation"): + return "call" + if t.endswith("identifier") or t.endswith("name"): + return "var" + if "subscript" in t: + return "subscript" + if "binary" in t or "unary" in t or "ternary" in t or "operator" in t: + return "expr" + if t in ("interpolation", "template_string"): + return "template" + return "other" + + +def _find_arglist_node(parent): + """Find the argument container node in a call expression, language-agnostic.""" + args = parent.child_by_field_name("arguments") + if args: + return args + for child in parent.children: + if child.type in ("argument_list", "arguments", "call_suffix"): + return child + if child.type == "template_string": + return child + return None + + +def _iterate_arg_nodes(arglist): + """Yield individual argument expression nodes from an arglist.""" + if arglist.type == "call_suffix": + for child in arglist.children: + if child.is_named and ("ambda" in child.type or "block" in child.type): + yield child + return + for child in arglist.children: + if child.type == "value_arguments": + for arg in child.children: + if arg.is_named and arg.type == "value_argument": + for expr in arg.children: + if expr.is_named: + yield expr + break + return + for child in arglist.children: + if child.is_named: + yield child + + +def extract_arg_info(file_path, code): + """For each behavioral call in the file, record argument structure. + + Returns dict: call_text -> [ (arg_count, (type1, type2, ...)), ... ] + where types are 'var','lit','call','lambda','kwarg','expr','other'. + """ + ext = os.path.splitext(file_path)[1].lower() + lang, query_name = _load_grammar(ext) + query_src = _load_query(query_name) + + parser = Parser(lang) + tree = parser.parse(code.encode()) + query = Query(lang, query_src) + + cursor = QueryCursor(query) + captures = cursor.captures(tree.root_node) + + info = {} + for capname, nodes in captures.items(): + if not capname.startswith(BEHAVIORAL_PREFIXES): + continue + for node in nodes: + text = code[node.start_byte:node.end_byte].strip() + parent = node.parent + if not parent: + continue + arglist = _find_arglist_node(parent) + if not arglist: + continue + types = [_classify_arg_node(c) for c in _iterate_arg_nodes(arglist)] + info.setdefault(text, []).append((len(types), tuple(types))) + return info + + +def _summarize_arg_info(info): + """Aggregate extract_arg_info across files into compact summaries. + + Returns dict: call_text -> { + occurrences: int, + arg_count: {min, max, common}, + patterns: [(count, (types..)), ...], # top-5 by frequency + } + """ + from collections import Counter + summary = {} + for call_name, observations in info.items(): + counts = [c for c, _ in observations] + pattern_counts = Counter(observations) + top_patterns = pattern_counts.most_common(5) + summary[call_name] = { + "occurrences": len(observations), + "arg_count": { + "min": min(counts), + "max": max(counts), + "common": max(set(counts), key=counts.count), + }, + "patterns": [ + {"count": c, "args": n, "types": list(t)} + for (n, t), c in top_patterns + ], + } + return summary + def _load_grammar(ext): entry = EXTENSION_MAP.get(ext)