mirror of
https://github.com/Xahau/xahaud.git
synced 2026-08-21 07:20:53 +00:00
Revert "chore: use improved levelization script with threading and argparse"
This reverts commit 5c1d7d9ae9.
This commit is contained in:
Nicholas Dudfield
@@ -1,273 +1,283 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Levelization generator.
|
||||
|
||||
Produces the same result artifacts as levelization.sh, but much faster by
|
||||
doing parsing/counting in-process instead of spawning external tools in
|
||||
tight loops.
|
||||
"""
|
||||
Usage: levelization.py
|
||||
This script takes no parameters, and can be called from any directory in the file system.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import concurrent.futures
|
||||
import os
|
||||
import posixpath
|
||||
import re
|
||||
import shutil
|
||||
import time
|
||||
from collections import Counter, defaultdict
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
# Compile regex patterns once at module level
|
||||
INCLUDE_PATTERN = re.compile(r"^\s*#include.*/.*\.h")
|
||||
INCLUDE_TARGET_PATTERN = re.compile(r'.*["<]([^">]+)[">].*')
|
||||
PATHS_LINE_PATTERN = re.compile(r"^\s*(\d+)\s+(\S+)\s+(\S+)\s*$")
|
||||
INCLUDE_PATH_PATTERN = re.compile(r'[<"]([^>"]+)[>"]')
|
||||
|
||||
|
||||
def dictionary_sort_key(value: str) -> str:
|
||||
"""Approximate `sort -d` behavior used by the shell script."""
|
||||
return "".join(ch for ch in value if ch.isalnum() or ch.isspace())
|
||||
def dictionary_sort_key(s):
|
||||
"""
|
||||
Create a sort key that mimics 'sort -d' (dictionary order).
|
||||
Dictionary order only considers blanks and alphanumeric characters.
|
||||
"""
|
||||
return "".join(c for c in s if c.isalnum() or c.isspace())
|
||||
|
||||
|
||||
def normalize_level(value: str) -> str:
|
||||
# Match shell behavior: if level includes a file component (contains "."),
|
||||
# replace with dirname + "/toplevel".
|
||||
if "." in value:
|
||||
parent = posixpath.dirname(value) or "."
|
||||
value = f"{parent}/toplevel"
|
||||
return value.replace("/", ".")
|
||||
def get_level(file_path):
|
||||
"""
|
||||
Extract the level from a file path (second and third directory components).
|
||||
Equivalent to bash: cut -d/ -f 2,3
|
||||
|
||||
Examples:
|
||||
src/ripple/app/main.cpp -> ripple.app
|
||||
src/test/app/Import_test.cpp -> test.app
|
||||
"""
|
||||
parts = file_path.split("/")
|
||||
|
||||
if len(parts) >= 3:
|
||||
level = f"{parts[1]}/{parts[2]}"
|
||||
elif len(parts) >= 2:
|
||||
level = f"{parts[1]}/toplevel"
|
||||
else:
|
||||
level = file_path
|
||||
|
||||
# If the "level" indicates a file, cut off the filename
|
||||
if "." in level.split("/")[-1]:
|
||||
# Use the "toplevel" label as a workaround for `sort`
|
||||
# inconsistencies between different utility versions
|
||||
level = level.rsplit("/", 1)[0] + "/toplevel"
|
||||
|
||||
return level.replace("/", ".")
|
||||
|
||||
|
||||
def source_level(rel_path: str) -> str:
|
||||
parts = rel_path.split("/")
|
||||
return normalize_level("/".join(parts[1:3]))
|
||||
def extract_include_level(include_line):
|
||||
"""
|
||||
Extract the include path from an #include directive.
|
||||
Gets the first two directory components from the include path.
|
||||
Equivalent to bash: cut -d/ -f 1,2
|
||||
|
||||
|
||||
def include_level(include_line: str) -> str | None:
|
||||
match = INCLUDE_TARGET_PATTERN.match(include_line)
|
||||
Examples:
|
||||
#include <ripple/basics/base_uint.h> -> ripple.basics
|
||||
#include "ripple/app/main/Application.h" -> ripple.app
|
||||
"""
|
||||
match = INCLUDE_PATH_PATTERN.search(include_line)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
include_path = match.group(1)
|
||||
parts = include_path.split("/")
|
||||
return normalize_level("/".join(parts[:2]))
|
||||
|
||||
if len(parts) >= 2:
|
||||
include_level = f"{parts[0]}/{parts[1]}"
|
||||
else:
|
||||
include_level = include_path
|
||||
|
||||
# If the "includelevel" indicates a file, cut off the filename
|
||||
if "." in include_level.split("/")[-1]:
|
||||
include_level = include_level.rsplit("/", 1)[0] + "/toplevel"
|
||||
|
||||
return include_level.replace("/", ".")
|
||||
|
||||
|
||||
def scan_file(path: Path, repo_root: Path) -> tuple[list[str], list[tuple[str, str]]]:
|
||||
rel = path.relative_to(repo_root).as_posix()
|
||||
src_level = source_level(rel)
|
||||
def find_repository_directories(start_path, depth_limit=10):
|
||||
"""
|
||||
Find the repository root by looking for src or include folders.
|
||||
Walks up the directory tree from the start path.
|
||||
"""
|
||||
current = start_path.resolve()
|
||||
|
||||
raw_lines: list[str] = []
|
||||
paths: list[tuple[str, str]] = []
|
||||
for _ in range(depth_limit):
|
||||
src_path = current / "src"
|
||||
include_path = current / "include"
|
||||
has_src = src_path.exists()
|
||||
has_include = include_path.exists()
|
||||
|
||||
with path.open("r", encoding="utf-8", errors="ignore") as handle:
|
||||
for line in handle:
|
||||
if "boost" in line:
|
||||
continue
|
||||
if not INCLUDE_PATTERN.match(line):
|
||||
continue
|
||||
if has_src or has_include:
|
||||
dirs = []
|
||||
if has_src:
|
||||
dirs.append(src_path)
|
||||
if has_include:
|
||||
dirs.append(include_path)
|
||||
return current, dirs
|
||||
|
||||
line = line.rstrip("\n")
|
||||
raw_lines.append(f"{rel}:{line}")
|
||||
parent = current.parent
|
||||
if parent == current:
|
||||
break
|
||||
current = parent
|
||||
|
||||
dst_level = include_level(line)
|
||||
if dst_level is None:
|
||||
continue
|
||||
if src_level != dst_level:
|
||||
paths.append((src_level, dst_level))
|
||||
|
||||
return raw_lines, paths
|
||||
raise RuntimeError(
|
||||
"Could not find repository root. "
|
||||
"Expected to find a directory containing 'src' and/or 'include' folders."
|
||||
)
|
||||
|
||||
|
||||
def iter_source_files(repo_root: Path) -> list[Path]:
|
||||
files: list[Path] = []
|
||||
for top in ("include", "src"):
|
||||
root = repo_root / top
|
||||
if root.exists():
|
||||
files.extend(path for path in root.rglob("*") if path.is_file())
|
||||
files.sort(key=lambda p: p.relative_to(repo_root).as_posix())
|
||||
return files
|
||||
def main():
|
||||
script_dir = Path(__file__).parent.resolve()
|
||||
os.chdir(script_dir)
|
||||
|
||||
# Clean up and create results directory.
|
||||
results_dir = script_dir / "results"
|
||||
if results_dir.exists():
|
||||
import shutil
|
||||
|
||||
def write_relation_db(
|
||||
results_dir: Path,
|
||||
edge_counts: list[tuple[tuple[str, str], int]],
|
||||
) -> tuple[dict[str, list[tuple[str, int]]], dict[str, list[tuple[str, int]]]]:
|
||||
shutil.rmtree(results_dir)
|
||||
results_dir.mkdir()
|
||||
|
||||
# Find the repository root.
|
||||
try:
|
||||
repo_root, scan_dirs = find_repository_directories(script_dir)
|
||||
print(f"Found repository root: {repo_root}")
|
||||
for scan_dir in scan_dirs:
|
||||
print(f" Scanning: {scan_dir.relative_to(repo_root)}")
|
||||
except RuntimeError as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Find all #include directives.
|
||||
print("\nScanning for raw includes...")
|
||||
raw_includes = []
|
||||
rawincludes_file = results_dir / "rawincludes.txt"
|
||||
|
||||
with open(rawincludes_file, "w", buffering=8192) as raw_f:
|
||||
for dir_path in scan_dirs:
|
||||
for file_path in dir_path.rglob("*"):
|
||||
if not file_path.is_file():
|
||||
continue
|
||||
try:
|
||||
rel_path_str = str(file_path.relative_to(repo_root))
|
||||
with open(
|
||||
file_path, "r", encoding="utf-8", errors="ignore", buffering=8192
|
||||
) as f:
|
||||
for line in f:
|
||||
if "#include" not in line or "boost" in line:
|
||||
continue
|
||||
if INCLUDE_PATTERN.match(line):
|
||||
line_stripped = line.strip()
|
||||
entry = f"{rel_path_str}:{line_stripped}\n"
|
||||
print(entry, end="")
|
||||
raw_f.write(entry)
|
||||
raw_includes.append((rel_path_str, line_stripped))
|
||||
except Exception as e:
|
||||
print(f"Error reading {file_path}: {e}", file=sys.stderr)
|
||||
|
||||
# Build levelization paths and count directly.
|
||||
print("Build levelization paths")
|
||||
path_counts = defaultdict(int)
|
||||
|
||||
for file_path, include_line in raw_includes:
|
||||
include_level = extract_include_level(include_line)
|
||||
if not include_level:
|
||||
continue
|
||||
level = get_level(file_path)
|
||||
if level != include_level:
|
||||
path_counts[(level, include_level)] += 1
|
||||
|
||||
# Sort and deduplicate paths.
|
||||
print("Sort and deduplicate paths")
|
||||
sorted_items = sorted(
|
||||
path_counts.items(),
|
||||
key=lambda x: (dictionary_sort_key(x[0][0]), dictionary_sort_key(x[0][1])),
|
||||
)
|
||||
|
||||
paths_file = results_dir / "paths.txt"
|
||||
with open(paths_file, "w") as f:
|
||||
for (level, include_level), count in sorted_items:
|
||||
line = f"{count:7} {level} {include_level}\n"
|
||||
print(line.rstrip())
|
||||
f.write(line)
|
||||
|
||||
# Split into flat-file database.
|
||||
print("Split into flat-file database")
|
||||
includes_dir = results_dir / "includes"
|
||||
includedby_dir = results_dir / "includedby"
|
||||
includes_dir.mkdir(parents=True, exist_ok=True)
|
||||
includedby_dir.mkdir(parents=True, exist_ok=True)
|
||||
includes_dir.mkdir()
|
||||
includedby_dir.mkdir()
|
||||
|
||||
includes: dict[str, list[tuple[str, int]]] = defaultdict(list)
|
||||
includedby: dict[str, list[tuple[str, int]]] = defaultdict(list)
|
||||
includes_data = defaultdict(list)
|
||||
includedby_data = defaultdict(list)
|
||||
|
||||
with (results_dir / "paths.txt").open("w", encoding="utf-8") as out:
|
||||
for (src, dst), count in edge_counts:
|
||||
out.write(f"{count:7d} {src} {dst}\n")
|
||||
includes[src].append((dst, count))
|
||||
includedby[dst].append((src, count))
|
||||
for (level, include_level), count in sorted_items:
|
||||
includes_data[level].append((include_level, count))
|
||||
includedby_data[include_level].append((level, count))
|
||||
|
||||
for src, entries in includes.items():
|
||||
with (includes_dir / src).open("w", encoding="utf-8") as out:
|
||||
for dst, count in entries:
|
||||
out.write(f"{dst} {count}\n")
|
||||
for level in sorted(includes_data.keys(), key=dictionary_sort_key):
|
||||
with open(includes_dir / level, "w") as f:
|
||||
for include_level, count in includes_data[level]:
|
||||
line = f"{include_level} {count}\n"
|
||||
print(line.rstrip())
|
||||
f.write(line)
|
||||
|
||||
for dst, entries in includedby.items():
|
||||
with (includedby_dir / dst).open("w", encoding="utf-8") as out:
|
||||
for src, count in entries:
|
||||
out.write(f"{src} {count}\n")
|
||||
for include_level in sorted(includedby_data.keys(), key=dictionary_sort_key):
|
||||
with open(includedby_dir / include_level, "w") as f:
|
||||
for level, count in includedby_data[include_level]:
|
||||
line = f"{level} {count}\n"
|
||||
print(line.rstrip())
|
||||
f.write(line)
|
||||
|
||||
return includes, includedby
|
||||
# Search for loops.
|
||||
print("Search for loops")
|
||||
loops_file = results_dir / "loops.txt"
|
||||
ordering_file = results_dir / "ordering.txt"
|
||||
|
||||
# Pre-load all include files into memory for fast lookup.
|
||||
includes_cache = {}
|
||||
includes_lookup = {}
|
||||
|
||||
def build_loops_and_ordering(
|
||||
includes: dict[str, list[tuple[str, int]]],
|
||||
) -> tuple[list[str], list[str]]:
|
||||
include_map = {
|
||||
src: {dst: count for dst, count in entries}
|
||||
for src, entries in includes.items()
|
||||
}
|
||||
for include_file in sorted(includes_dir.iterdir(), key=lambda p: p.name):
|
||||
if not include_file.is_file():
|
||||
continue
|
||||
includes_cache[include_file.name] = []
|
||||
includes_lookup[include_file.name] = {}
|
||||
with open(include_file, "r") as f:
|
||||
for line in f:
|
||||
parts = line.strip().split()
|
||||
if len(parts) >= 2:
|
||||
name, count = parts[0], int(parts[1])
|
||||
includes_cache[include_file.name].append((name, count))
|
||||
includes_lookup[include_file.name][name] = count
|
||||
|
||||
ordering_lines: list[str] = []
|
||||
loops_lines: list[str] = []
|
||||
loops_found = set()
|
||||
|
||||
seen_pairs: set[tuple[str, str]] = set()
|
||||
with open(loops_file, "w", buffering=8192) as loops_f, open(
|
||||
ordering_file, "w", buffering=8192
|
||||
) as ordering_f:
|
||||
for source in sorted(includes_cache.keys()):
|
||||
for include, include_freq in includes_cache[source]:
|
||||
if include not in includes_lookup:
|
||||
continue
|
||||
|
||||
for source in sorted(includes.keys()):
|
||||
for include, includefreq in includes[source]:
|
||||
if include not in include_map:
|
||||
continue
|
||||
source_freq = includes_lookup[include].get(source)
|
||||
|
||||
sourcefreq = include_map[include].get(source)
|
||||
if sourcefreq is None:
|
||||
ordering_lines.append(f"{source} > {include}\n")
|
||||
continue
|
||||
if source_freq is not None:
|
||||
loop_key = tuple(sorted([source, include]))
|
||||
if loop_key in loops_found:
|
||||
continue
|
||||
loops_found.add(loop_key)
|
||||
|
||||
if (include, source) in seen_pairs:
|
||||
continue
|
||||
seen_pairs.add((source, include))
|
||||
loops_f.write(f"Loop: {source} {include}\n")
|
||||
|
||||
loops_lines.append(f"Loop: {source} {include}\n")
|
||||
if includefreq - sourcefreq > 3:
|
||||
loops_lines.append(f" {source} > {include}\n\n")
|
||||
elif sourcefreq - includefreq > 3:
|
||||
loops_lines.append(f" {include} > {source}\n\n")
|
||||
elif sourcefreq == includefreq:
|
||||
loops_lines.append(f" {include} == {source}\n\n")
|
||||
else:
|
||||
loops_lines.append(f" {include} ~= {source}\n\n")
|
||||
diff = include_freq - source_freq
|
||||
if diff > 3:
|
||||
loops_f.write(f" {source} > {include}\n\n")
|
||||
elif diff < -3:
|
||||
loops_f.write(f" {include} > {source}\n\n")
|
||||
elif source_freq == include_freq:
|
||||
loops_f.write(f" {include} == {source}\n\n")
|
||||
else:
|
||||
loops_f.write(f" {include} ~= {source}\n\n")
|
||||
else:
|
||||
ordering_f.write(f"{source} > {include}\n")
|
||||
|
||||
return ordering_lines, loops_lines
|
||||
# Print results.
|
||||
print("\nOrdering:")
|
||||
with open(ordering_file, "r") as f:
|
||||
print(f.read(), end="")
|
||||
|
||||
|
||||
def generate(results_dir: Path, repo_root: Path, workers: int) -> None:
|
||||
if results_dir.exists():
|
||||
shutil.rmtree(results_dir)
|
||||
results_dir.mkdir(parents=True)
|
||||
|
||||
files = iter_source_files(repo_root)
|
||||
|
||||
raw_by_file: dict[str, list[str]] = {}
|
||||
paths_by_file: dict[str, list[tuple[str, str]]] = {}
|
||||
|
||||
start = time.perf_counter()
|
||||
if workers <= 1:
|
||||
for file in files:
|
||||
rel = file.relative_to(repo_root).as_posix()
|
||||
raw, paths = scan_file(file, repo_root)
|
||||
raw_by_file[rel] = raw
|
||||
paths_by_file[rel] = paths
|
||||
else:
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
|
||||
futures = {
|
||||
file.relative_to(repo_root).as_posix(): pool.submit(
|
||||
scan_file, file, repo_root
|
||||
)
|
||||
for file in files
|
||||
}
|
||||
for rel in sorted(futures.keys()):
|
||||
raw, paths = futures[rel].result()
|
||||
raw_by_file[rel] = raw
|
||||
paths_by_file[rel] = paths
|
||||
|
||||
raw_lines: list[str] = []
|
||||
raw_lines.extend(
|
||||
line
|
||||
for rel in sorted(raw_by_file.keys())
|
||||
for line in raw_by_file[rel]
|
||||
)
|
||||
with (results_dir / "rawincludes.txt").open("w", encoding="utf-8") as out:
|
||||
out.write("\n".join(raw_lines))
|
||||
if raw_lines:
|
||||
out.write("\n")
|
||||
|
||||
path_pairs: list[tuple[str, str]] = []
|
||||
path_pairs.extend(
|
||||
pair
|
||||
for rel in sorted(paths_by_file.keys())
|
||||
for pair in paths_by_file[rel]
|
||||
)
|
||||
counts = Counter(path_pairs)
|
||||
|
||||
edge_counts = sorted(
|
||||
counts.items(),
|
||||
key=lambda item: (
|
||||
dictionary_sort_key(item[0][0]),
|
||||
dictionary_sort_key(item[0][1]),
|
||||
),
|
||||
)
|
||||
|
||||
includes, _ = write_relation_db(results_dir, edge_counts)
|
||||
ordering, loops = build_loops_and_ordering(includes)
|
||||
|
||||
with (results_dir / "ordering.txt").open("w", encoding="utf-8") as out:
|
||||
out.writelines(ordering)
|
||||
with (results_dir / "loops.txt").open("w", encoding="utf-8") as out:
|
||||
out.writelines(loops)
|
||||
|
||||
elapsed = time.perf_counter() - start
|
||||
print(
|
||||
f"levelization.py: scanned {len(files)} files, "
|
||||
f"{len(raw_lines)} includes, {len(edge_counts)} unique paths in "
|
||||
f"{elapsed:.2f}s"
|
||||
)
|
||||
print((results_dir / "ordering.txt").read_text(encoding="utf-8"), end="")
|
||||
print((results_dir / "loops.txt").read_text(encoding="utf-8"), end="")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
script_dir = Path(__file__).resolve().parent
|
||||
repo_root = script_dir.parents[1]
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument(
|
||||
"--repo-root",
|
||||
type=Path,
|
||||
default=repo_root,
|
||||
help="Repository root (defaults based on script location).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--results-dir",
|
||||
type=Path,
|
||||
default=script_dir / "results",
|
||||
help="Output results directory.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--workers",
|
||||
type=int,
|
||||
default=min(32, (os.cpu_count() or 1)),
|
||||
help="Thread count for source scanning (default: CPU count, max 32).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
generated_dir = args.results_dir.resolve()
|
||||
generate(
|
||||
results_dir=generated_dir,
|
||||
repo_root=args.repo_root.resolve(),
|
||||
workers=max(1, args.workers),
|
||||
)
|
||||
|
||||
return 0
|
||||
print("\nLoops:")
|
||||
with open(loops_file, "r") as f:
|
||||
print(f.read(), end="")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user