from __future__ import annotations import difflib import re from pathlib import Path from .ingest import iter_text_files def generate_patch_diff(root: Path) -> str: diff_parts: list[str] = [] for relative_path, original in iter_text_files(root): transformed = transform_text(relative_path, original) if transformed == original: continue diff = difflib.unified_diff( original.splitlines(keepends=True), transformed.splitlines(keepends=True), fromfile=f"a/{relative_path}", tofile=f"b/{relative_path}", ) diff_parts.extend(diff) if not diff_parts: return "# No deterministic patch was generated. Review manual findings in the migration report.\n" return "".join(diff_parts) def transform_text(relative_path: str, text: str) -> str: path = Path(relative_path) lower_name = path.name.lower() suffix = path.suffix.lower() if suffix == ".py": return _transform_python(text) if lower_name.startswith("dockerfile"): return _transform_dockerfile(text) if suffix in {".sh", ".bash", ".zsh", ".yaml", ".yml", ".txt", ".md"}: return _transform_shellish(text) return text def _transform_python(text: str) -> str: changed = text needs_device = bool( re.search(r"\.cuda\s*\(\s*\)", changed) or re.search(r"\.to\(\s*['\"]cuda['\"]\s*\)", changed) or re.search(r"torch\.device\(\s*['\"]cuda['\"]\s*\)", changed) ) if needs_device and "import torch" in changed and "_rocmport_device" not in changed: changed = _insert_device_helper(changed) changed = re.sub(r"\.cuda\s*\(\s*\)", ".to(_rocmport_device)", changed) changed = re.sub(r"\.to\(\s*['\"]cuda['\"]\s*\)", ".to(_rocmport_device)", changed) changed = re.sub(r"torch\.device\(\s*['\"]cuda['\"]\s*\)", "_rocmport_device", changed) return changed def _insert_device_helper(text: str) -> str: lines = text.splitlines() insert_at = 0 for index, line in enumerate(lines): if line.startswith("import ") or line.startswith("from "): insert_at = index + 1 helper = [ "", "# ROCm PyTorch exposes AMD GPUs through the torch.cuda namespace.", '_rocmport_device = torch.device("cuda" if torch.cuda.is_available() else "cpu")', ] return "\n".join(lines[:insert_at] + helper + lines[insert_at:]) + ("\n" if text.endswith("\n") else "") def _transform_dockerfile(text: str) -> str: changed = re.sub( r"(?im)^\s*FROM\s+nvidia/cuda:[^\n]+", "FROM vllm/vllm-openai-rocm:latest", text, ) changed = _transform_shellish(changed) return changed def _transform_shellish(text: str) -> str: changed = text.replace("nvidia-smi", "rocm-smi") changed = changed.replace("NVIDIA_VISIBLE_DEVICES", "HIP_VISIBLE_DEVICES") changed = changed.replace("CUDA_VISIBLE_DEVICES", "HIP_VISIBLE_DEVICES") changed = changed.replace("NVIDIA_DRIVER_CAPABILITIES", "ROCM_VISIBLE_DEVICES") return changed