#!/usr/bin/env python3 """ pa_analyzer.py — PA Usage Analyzer for MC1710 Codebase Analyzes which MC types in mc1710/ use which Platform-Abstraction (PA) members. Usage: python tools/migration/pa_analyzer.py # dry-run python tools/migration/pa_analyzer.py --execute # run analysis python tools/migration/pa_analyzer.py --execute --pa INbt --pa IPlayer # filter python tools/migration/pa_analyzer.py --execute --json # machine output python tools/migration/pa_analyzer.py --execute --verbose # detailed """ import argparse import json import os import sys from datetime import datetime, timezone DEFAULT_CLASSIFICATIONS_PATH = "tools/migration/classification_results.json" DEFAULT_INDEX_PATH = "tools/migration/mc_usage_index.json" DEFAULT_OUTPUT_PATH = "tools/migration/pa_usage_analysis.json" NON_SUPPRESSED_CLASSIFICATIONS = {"EXISTING", "INTERFACE", "SERVICE", "AUTO"} CONFIDENCE_HIGH = 90 CONFIDENCE_MEDIUM = 50 def detect_project_root(start=None): cur = os.path.abspath(start or os.path.dirname(os.path.abspath(__file__))) for _ in range(20): if os.path.isfile(os.path.join(cur, "settings.gradle")): return cur parent = os.path.dirname(cur) if parent == cur: break cur = parent return os.path.abspath(start or os.path.dirname(os.path.abspath(__file__))) def load_json_file(project_root, rel_path, description): full_path = os.path.join(project_root, rel_path) if not os.path.isfile(full_path): raise FileNotFoundError(f"{description} not found: {full_path}") with open(full_path, "r", encoding="utf-8") as f: return json.load(f) def extract_non_suppressed_pas( classifications_data, classification_filter=None, pa_filter=None ): types_dict = classifications_data.get("types", {}) pa_map = {} for fqn, info in types_dict.items(): classification = info.get("classification", "") pa_name = info.get("pa_name") or info.get("service_name") if not pa_name: continue if classification not in NON_SUPPRESSED_CLASSIFICATIONS: continue if classification_filter and classification not in classification_filter: continue if pa_filter and pa_name not in pa_filter: continue if pa_name not in pa_map: pa_map[pa_name] = {"mc_types": [], "classifications": set()} pa_map[pa_name]["mc_types"].append( { "fqn": fqn, "shortName": info.get("shortName", ""), "classification": classification, "import_count": info.get("import_count", 0), } ) pa_map[pa_name]["classifications"].add(classification) for pa_name in pa_map: pa_map[pa_name]["classifications"] = sorted(pa_map[pa_name]["classifications"]) pa_map[pa_name]["mc_types"].sort(key=lambda x: -x["import_count"]) return pa_map def _ensure_member(member_dict, name): if name not in member_dict: member_dict[name] = {"count": 0, "files": set()} return member_dict[name] def _accumulate_members(member_dict, data_source, all_files, total_usage_list): for name, info in data_source.items(): count = info.get("count", 0) files = info.get("files", []) entry = _ensure_member(member_dict, name) entry["count"] += count entry["files"].update(files) all_files.update(files) total_usage_list[0] += count def _accumulate_usage( target_count_files, usage_data, usage_key, all_files, total_usage_list ): kind_entries = usage_data.get(usage_key, []) if isinstance(kind_entries, list): # Index format: array of {file, line} objects (constructors, casts, instanceof) for entry in kind_entries: f = entry.get("file", "") if isinstance(entry, dict) else "" if f: target_count_files["files"].add(f) all_files.add(f) target_count_files["count"] += len(kind_entries) total_usage_list[0] += len(kind_entries) elif isinstance(kind_entries, dict): # Fallback: {count, files} dict format (methodCalls, staticCalls, fieldAccesses) kind_count = kind_entries.get("count", 0) kind_files = kind_entries.get("files", []) target_count_files["count"] += kind_count target_count_files["files"].update(kind_files) all_files.update(kind_files) total_usage_list[0] += kind_count def analyze_pa_usage(pa_map, index_data): index_types = index_data.get("types", {}) results = {} for pa_name, pa_info in pa_map.items(): methods = {} static_methods = {} fields = {} all_files = set() total_usage = [0] total_import_count = 0 constructors = {"count": 0, "files": set()} casts = {"count": 0, "files": set()} instanceof_usage = {"count": 0, "files": set()} for mc_type in pa_info["mc_types"]: fqn = mc_type["fqn"] total_import_count += mc_type["import_count"] type_data = index_types.get(fqn, {}) _accumulate_members( methods, type_data.get("methodCalls", {}), all_files, total_usage ) _accumulate_members( static_methods, type_data.get("staticCalls", {}), all_files, total_usage ) _accumulate_members( fields, type_data.get("fieldAccesses", {}), all_files, total_usage ) usage = type_data.get("usage", {}) _accumulate_usage(constructors, usage, "new", all_files, total_usage) _accumulate_usage(casts, usage, "cast", all_files, total_usage) _accumulate_usage( instanceof_usage, usage, "instanceof", all_files, total_usage ) unique_members = len(methods) + len(static_methods) + len(fields) confidence = _compute_confidence( total_usage=total_usage[0], unique_members=unique_members, file_count=len(all_files), total_import_count=total_import_count, ) results[pa_name] = { "mc_types": pa_info["mc_types"], "classifications": pa_info["classifications"], "members": { "methods": _finalize_member_dict(methods), "static_methods": _finalize_member_dict(static_methods), "fields": _finalize_member_dict(fields), }, "source_files": sorted(all_files), "usage_count": total_usage[0], "total_import_count": total_import_count, "unique_members": unique_members, "file_count": len(all_files), "constructors": _finalize_usage_entry(constructors), "casts": _finalize_usage_entry(casts), "instanceof": _finalize_usage_entry(instanceof_usage), "confidence": round(confidence, 1), } return results def _finalize_member_dict(member_dict): result = {} for name, info in sorted(member_dict.items(), key=lambda x: -x[1]["count"]): result[name] = { "count": info["count"], "file_count": len(info["files"]), "files": sorted(info["files"]), } return result def _finalize_usage_entry(entry): return { "count": entry["count"], "file_count": len(entry["files"]), "files": sorted(entry["files"]), } def _compute_confidence(total_usage, unique_members, file_count, total_import_count): """Weighted score from usage density, member coverage, and file spread.""" if total_import_count == 0: return 0.0 usage_density = min(total_usage / max(total_import_count, 1), 5.0) / 5.0 member_score = min(unique_members / 20.0, 1.0) file_score = min(file_count / 50.0, 1.0) return min(usage_density * 40 + member_score * 35 + file_score * 25, 100.0) def build_output(results, args): all_files = set() for r in results.values(): all_files.update(r["source_files"]) metadata = { "analysis_date": datetime.now(timezone.utc).isoformat(), "tool_version": "1.0.0", "classifications_source": args.classifications, "index_source": args.index, "total_PAs_scanned": len(results), "total_mc_types_covered": sum(len(r["mc_types"]) for r in results.values()), "total_source_files_touched": len(all_files), "confidence_thresholds": { "high": f">={CONFIDENCE_HIGH}%", "medium": f">={CONFIDENCE_MEDIUM}%", "low": f"<{CONFIDENCE_MEDIUM}%", }, "filters_applied": { "classifications": args.classification or "all non-suppressed", "pa_names": args.pa or "all", }, } sorted_results = dict(sorted(results.items())) high_conf = [] medium_conf = [] low_conf = [] by_classification = {} unused_pas = [] for pa_name, data in sorted_results.items(): conf = data["confidence"] if conf >= CONFIDENCE_HIGH: high_conf.append(pa_name) elif conf >= CONFIDENCE_MEDIUM: medium_conf.append(pa_name) else: low_conf.append(pa_name) for cls in data["classifications"]: if cls not in by_classification: by_classification[cls] = [] by_classification[cls].append(pa_name) if data["usage_count"] == 0: unused_pas.append(pa_name) top_by_usage = [ { "pa_name": name, "usage_count": data["usage_count"], "file_count": data["file_count"], } for name, data in sorted(results.items(), key=lambda x: -x[1]["usage_count"])[ :20 ] ] summary = { "by_confidence": {"high": high_conf, "medium": medium_conf, "low": low_conf}, "by_classification": by_classification, "top_by_usage": top_by_usage, "unused_pas": unused_pas, } return {"metadata": metadata, "summary": summary, "pa_usage": sorted_results} def format_human_readable(output_data, verbose=False): lines = [] meta = output_data["metadata"] summary = output_data["summary"] pa_usage = output_data["pa_usage"] lines.append("") lines.append("=" * 72) lines.append("PA USAGE ANALYSIS — MC1710 Member Usage by Platform-Abstraction") lines.append("=" * 72) lines.append(f" Date: {meta['analysis_date']}") lines.append(f" PAs scanned: {meta['total_PAs_scanned']}") lines.append(f" MC types covered: {meta['total_mc_types_covered']}") lines.append(f" Source files touched: {meta['total_source_files_touched']}") lines.append("") high = summary["by_confidence"]["high"] med = summary["by_confidence"]["medium"] low = summary["by_confidence"]["low"] lines.append(" CONFIDENCE DISTRIBUTION:") lines.append(f" High (>={CONFIDENCE_HIGH}%): {len(high)} PAs") lines.append(f" Medium (>={CONFIDENCE_MEDIUM}%): {len(med)} PAs") lines.append(f" Low (<{CONFIDENCE_MEDIUM}%): {len(low)} PAs") lines.append(f" Unused (0 usage): {len(summary['unused_pas'])} PAs") lines.append("") lines.append(" TOP 20 PAs BY USAGE:") for entry in summary["top_by_usage"]: lines.append( f" {entry['pa_name']:35s} usage={entry['usage_count']:6d} files={entry['file_count']:4d}" ) lines.append("") if verbose: lines.append("-" * 72) lines.append("DETAILED PA BREAKDOWN:") lines.append("-" * 72) for pa_name, data in pa_usage.items(): lines.append("") lines.append(f" PA: {pa_name}") lines.append(f" Classifications: {', '.join(data['classifications'])}") lines.append( f" MC Types: {', '.join(t['shortName'] for t in data['mc_types'])}" ) lines.append(f" Total Usage: {data['usage_count']}") lines.append(f" Unique Members: {data['unique_members']}") lines.append(f" Files: {data['file_count']}") lines.append(f" Confidence: {data['confidence']}%") methods = data["members"]["methods"] if methods: lines.append(f" Instance Methods ({len(methods)}):") for name, info in list(methods.items())[:10]: lines.append( f" {name:40s} {info['count']:5d} calls {info['file_count']:3d} files" ) if len(methods) > 10: lines.append(f" ... and {len(methods) - 10} more") statics = data["members"]["static_methods"] if statics: lines.append(f" Static Methods ({len(statics)}):") for name, info in list(statics.items())[:5]: lines.append( f" {name:40s} {info['count']:5d} calls {info['file_count']:3d} files" ) if len(statics) > 5: lines.append(f" ... and {len(statics) - 5} more") fields_data = data["members"]["fields"] if fields_data: lines.append(f" Fields ({len(fields_data)}):") for name, info in list(fields_data.items())[:10]: lines.append( f" {name:40s} {info['count']:5d} accesses {info['file_count']:3d} files" ) if len(fields_data) > 10: lines.append(f" ... and {len(fields_data) - 10} more") if data["constructors"]["count"] > 0: lines.append( f" Constructors: {data['constructors']['count']} in {data['constructors']['file_count']} files" ) if data["casts"]["count"] > 0: lines.append( f" Casts: {data['casts']['count']} in {data['casts']['file_count']} files" ) if data["instanceof"]["count"] > 0: lines.append( f" instanceof: {data['instanceof']['count']} in {data['instanceof']['file_count']} files" ) if summary["unused_pas"]: lines.append("") lines.append(" UNUSED PAs (0 member usage detected in index):") for i in range(0, len(summary["unused_pas"]), 6): chunk = summary["unused_pas"][i : i + 6] lines.append(f" {', '.join(chunk)}") lines.append("") return "\n".join(lines) def _print_dry_run(pa_map, args): print("") print("=" * 72) print("PA USAGE ANALYZER — DRY RUN (add --execute to run)") print("=" * 72) print(f" Classifications source: {args.classifications}") print(f" Index source: {args.index}") print(f" Output target: {args.output}") print(f" PA filter: {args.pa or 'all'}") print(f" Classification filter: {args.classification or 'all non-suppressed'}") print(f"") print(f" Non-suppressed PAs found: {len(pa_map)}") total_mc = sum(len(v["mc_types"]) for v in pa_map.values()) print(f" Total MC types to scan: {total_mc}") print(f"") by_class = {} for pa_name, info in sorted(pa_map.items()): for cls in info["classifications"]: if cls not in by_class: by_class[cls] = [] by_class[cls].append(pa_name) for cls in sorted(by_class): pas = by_class[cls] print(f" {cls} ({len(pas)} PAs):") for pa_name in pas[:10]: mc_types = pa_map[pa_name]["mc_types"] mc_str = ", ".join(t["shortName"] for t in mc_types[:3]) if len(mc_types) > 3: mc_str += f" +{len(mc_types) - 3} more" total_imports = sum(t["import_count"] for t in mc_types) print(f" {pa_name:35s} mc=[{mc_str}] imports={total_imports}") if len(pas) > 10: print(f" ... and {len(pas) - 10} more") print("") print(" Run with --execute to perform the full analysis.") print("") def export_all_methods( index_data, output_path, classifications_data=None, classification_filter=None, as_json=False, ): index_types = index_data.get("types", {}) allowed_fqns = None fqn_to_classification = {} if classifications_data: types_dict = classifications_data.get("types", {}) allowed_fqns = set() for fqn, info in types_dict.items(): cls = info.get("classification", "") if classification_filter and cls not in classification_filter: continue allowed_fqns.add(fqn) fqn_to_classification[fqn] = cls def _sort_members(members): return sorted(members, key=lambda x: (-x["count"], x["name"])) types_output = [] total_method_usage = 0 total_field_usage = 0 total_static_usage = 0 total_methods = 0 total_fields = 0 total_statics = 0 per_classification_stats = {} for fqn, type_data in sorted(index_types.items()): if allowed_fqns is not None and fqn not in allowed_fqns: continue short_name = type_data.get("shortName", "") fqn_cls = fqn_to_classification.get(fqn, "UNKNOWN") methods = [] for name, info in type_data.get("methodCalls", {}).items(): count = info.get("count", 0) files = info.get("files", []) methods.append( {"name": name, "count": count, "file_count": len(files), "files": files} ) total_method_usage += count fields = [] for name, info in type_data.get("fieldAccesses", {}).items(): count = info.get("count", 0) files = info.get("files", []) fields.append( {"name": name, "count": count, "file_count": len(files), "files": files} ) total_field_usage += count statics = [] for name, info in type_data.get("staticCalls", {}).items(): count = info.get("count", 0) files = info.get("files", []) statics.append( {"name": name, "count": count, "file_count": len(files), "files": files} ) total_static_usage += count methods = _sort_members(methods) fields = _sort_members(fields) statics = _sort_members(statics) total_methods += len(methods) total_fields += len(fields) total_statics += len(statics) type_usage = ( sum(m["count"] for m in methods) + sum(f["count"] for f in fields) + sum(s["count"] for s in statics) ) type_entry = { "fqn": fqn, "shortName": short_name, "classification": fqn_cls, "import_count": type_data.get("importCount", 0), "total_usage": type_usage, "methods": methods, "fields": fields, "static_methods": statics, } types_output.append(type_entry) if fqn_cls not in per_classification_stats: per_classification_stats[fqn_cls] = { "types": 0, "methods": 0, "fields": 0, "statics": 0, "usage": 0, } per_classification_stats[fqn_cls]["types"] += 1 per_classification_stats[fqn_cls]["methods"] += len(methods) per_classification_stats[fqn_cls]["fields"] += len(fields) per_classification_stats[fqn_cls]["statics"] += len(statics) per_classification_stats[fqn_cls]["usage"] += type_usage filter_label = "ALL (unfiltered)" if classification_filter: filter_label = ", ".join(sorted(classification_filter)) output = { "metadata": { "export_date": datetime.now(timezone.utc).isoformat(), "classification_filter": filter_label, "total_mc_types": len(types_output), "total_methods": total_methods, "total_fields": total_fields, "total_statics": total_statics, "total_unique_members": total_methods + total_fields + total_statics, "total_method_usage": total_method_usage, "total_field_usage": total_field_usage, "total_static_usage": total_static_usage, "total_usage": total_method_usage + total_field_usage + total_static_usage, "per_classification": { cls: stats for cls, stats in sorted(per_classification_stats.items()) }, }, "types": types_output, } os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) with open(output_path, "w", encoding="utf-8", newline="\n") as f: json.dump(output, f, indent=2, ensure_ascii=False) meta = output["metadata"] if as_json: print(json.dumps(meta, indent=2)) else: print("") print("=" * 72) print(f"ALL METHODS EXPORT — Classifications: {filter_label}") print("=" * 72) print(f" MC types scanned: {meta['total_mc_types']}") print( f" Unique methods: {meta['total_methods']:,} ({meta['total_method_usage']:,} total calls)" ) print( f" Unique fields: {meta['total_fields']:,} ({meta['total_field_usage']:,} total accesses)" ) print( f" Unique statics: {meta['total_statics']:,} ({meta['total_static_usage']:,} total calls)" ) print(f" Total unique members:{meta['total_unique_members']:,}") print(f" Total usage: {meta['total_usage']:,}") if per_classification_stats: print("") print(" PER-CLASSIFICATION BREAKDOWN:") for cls in sorted(per_classification_stats): s = per_classification_stats[cls] print( f" {cls:12s} {s['types']:3d} types {s['methods']:4d} methods {s['fields']:4d} fields {s['statics']:4d} statics {s['usage']:6d} usage" ) top_types = sorted(types_output, key=lambda t: -t["total_usage"])[:20] print("") print(" TOP 20 TYPES BY USAGE:") for t in top_types: print( f" {t['shortName']:30s} [{t['classification']:9s}] {t['total_usage']:6d} usage {len(t['methods']):3d}m {len(t['fields']):3d}f {len(t['static_methods']):3d}s" ) print("") print(f" Output written to: {output_path}") print("") return output def main(): parser = argparse.ArgumentParser( description="PA Usage Analyzer — Maps PA member usage across mc1710/ source files.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python tools/migration/pa_analyzer.py # dry-run python tools/migration/pa_analyzer.py --execute # run analysis python tools/migration/pa_analyzer.py --execute --pa INbt --pa IPlayer python tools/migration/pa_analyzer.py --execute --classification EXISTING INTERFACE python tools/migration/pa_analyzer.py --execute --classification SERVICE python tools/migration/pa_analyzer.py --execute --output /tmp/pa_usage.json python tools/migration/pa_analyzer.py --execute --verbose python tools/migration/pa_analyzer.py --execute --json python tools/migration/pa_analyzer.py --all-methods # export all MC members python tools/migration/pa_analyzer.py --all-methods --json # JSON metadata only """, ) parser.add_argument( "--execute", action="store_true", help="Run analysis and write output (default: dry-run)", ) parser.add_argument( "--output", default=None, help=f"Output JSON path (default: {DEFAULT_OUTPUT_PATH})", ) parser.add_argument( "--classifications", default=DEFAULT_CLASSIFICATIONS_PATH, help=f"Classification results JSON (default: {DEFAULT_CLASSIFICATIONS_PATH})", ) parser.add_argument( "--index", default=DEFAULT_INDEX_PATH, help=f"MC usage index JSON (default: {DEFAULT_INDEX_PATH})", ) parser.add_argument( "--pa", action="append", default=None, help="Filter to specific PA name(s). Repeatable.", ) parser.add_argument( "--classification", nargs="*", default=None, help="Filter by classification(s): EXISTING, INTERFACE, SERVICE, AUTO", ) parser.add_argument( "--json", action="store_true", help="Machine-readable JSON summary to stdout" ) parser.add_argument( "--verbose", action="store_true", help="Show detailed per-PA breakdown" ) parser.add_argument( "--all-methods", action="store_true", help="Export MC method/field/static usage from the index, filtered by classification", ) parser.add_argument( "--all-methods-classification", nargs="*", default=None, help="Classifications for --all-methods: EXISTING INTERFACE SERVICE AUTO TRANSFORM (default: all). Use ALL explicitly to skip loading classifications file.", ) parser.add_argument( "--project-root", default=None, help="Project root override (auto-detected by default)", ) args = parser.parse_args() project_root = ( os.path.abspath(args.project_root) if args.project_root else detect_project_root() ) if args.output is None: args.output = DEFAULT_OUTPUT_PATH try: if args.all_methods: index_data = load_json_file(project_root, args.index, "Index") all_methods_output = args.output.replace(".json", "_all_methods.json") if args.output == DEFAULT_OUTPUT_PATH: all_methods_output = "tools/migration/all_methods_export.json" full_output_path = os.path.join(project_root, all_methods_output) cls_data = None cls_filter = None if args.all_methods_classification and "ALL" in [ c.upper() for c in args.all_methods_classification ]: pass else: cls_data = load_json_file( project_root, args.classifications, "Classifications" ) if args.all_methods_classification: cls_filter = set(c.upper() for c in args.all_methods_classification) export_all_methods( index_data, full_output_path, classifications_data=cls_data, classification_filter=cls_filter, as_json=args.json, ) return classifications_data = load_json_file( project_root, args.classifications, "Classifications" ) classification_filter = ( set(args.classification) if args.classification else None ) pa_filter = set(args.pa) if args.pa else None pa_map = extract_non_suppressed_pas( classifications_data, classification_filter=classification_filter, pa_filter=pa_filter, ) if not args.execute: _print_dry_run(pa_map, args) return index_data = load_json_file(project_root, args.index, "Index") results = analyze_pa_usage(pa_map, index_data) output_data = build_output(results, args) output_path = os.path.join(project_root, args.output) os.makedirs(os.path.dirname(output_path), exist_ok=True) with open(output_path, "w", encoding="utf-8", newline="\n") as f: json.dump(output_data, f, indent=2, ensure_ascii=False) if args.json: conf_data = output_data["summary"]["by_confidence"] summary_output = { "status": "success", "output_file": args.output, "total_pas": len(results), "total_mc_types": sum(len(r["mc_types"]) for r in results.values()), "total_files": output_data["metadata"]["total_source_files_touched"], "confidence_distribution": { "high": len(conf_data["high"]), "medium": len(conf_data["medium"]), "low": len(conf_data["low"]), }, "unused_pas": len(output_data["summary"]["unused_pas"]), } print(json.dumps(summary_output, indent=2)) else: print(format_human_readable(output_data, verbose=args.verbose)) print(f" Output written to: {args.output}") print("") except FileNotFoundError as e: print(f"ERROR: {e}", file=sys.stderr) sys.exit(1) except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON — {e}", file=sys.stderr) sys.exit(1) except Exception as e: print(f"ERROR: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()