Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 5 additions & 56 deletions codeflash/code_utils/git_worktree_utils.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from __future__ import annotations

import json
import subprocess
import tempfile
import time
Expand All @@ -9,15 +8,12 @@
from typing import TYPE_CHECKING, Optional

import git
from filelock import FileLock

from codeflash.cli_cmds.console import logger
from codeflash.code_utils.compat import codeflash_cache_dir
from codeflash.code_utils.git_utils import check_running_in_git_repo, git_root_dir

if TYPE_CHECKING:
from typing import Any

from git import Repo


Expand Down Expand Up @@ -100,71 +96,24 @@ def get_patches_dir_for_project() -> Path:
return Path(patches_dir / project_id)


def get_patches_metadata() -> dict[str, Any]:
project_patches_dir = get_patches_dir_for_project()
meta_file = project_patches_dir / "metadata.json"
if meta_file.exists():
with meta_file.open("r", encoding="utf-8") as f:
return json.load(f)
return {"id": get_git_project_id() or "", "patches": []}


def save_patches_metadata(patch_metadata: dict) -> dict:
project_patches_dir = get_patches_dir_for_project()
meta_file = project_patches_dir / "metadata.json"
lock_file = project_patches_dir / "metadata.json.lock"

# we are not supporting multiple concurrent optimizations within the same process, but keep that in case we decide to do so in the future.
with FileLock(lock_file, timeout=10):
metadata = get_patches_metadata()

patch_metadata["id"] = time.strftime("%Y%m%d-%H%M%S")
metadata["patches"].append(patch_metadata)

meta_file.write_text(json.dumps(metadata, indent=2))

return patch_metadata


def overwrite_patch_metadata(patches: list[dict]) -> bool:
project_patches_dir = get_patches_dir_for_project()
meta_file = project_patches_dir / "metadata.json"
lock_file = project_patches_dir / "metadata.json.lock"

with FileLock(lock_file, timeout=10):
metadata = get_patches_metadata()
metadata["patches"] = patches
meta_file.write_text(json.dumps(metadata, indent=2))
return True


def create_diff_patch_from_worktree(
worktree_dir: Path,
files: list[str],
fto_name: Optional[str] = None,
metadata_input: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
worktree_dir: Path, files: list[str], fto_name: Optional[str] = None
) -> Optional[Path]:
repository = git.Repo(worktree_dir, search_parent_directories=True)
uni_diff_text = repository.git.diff(None, "HEAD", *files, ignore_blank_lines=True, ignore_space_at_eol=True)

if not uni_diff_text:
logger.warning("No changes found in worktree.")
return {}
return None

if not uni_diff_text.endswith("\n"):
uni_diff_text += "\n"

project_patches_dir = get_patches_dir_for_project()
project_patches_dir.mkdir(parents=True, exist_ok=True)

final_function_name = fto_name or metadata_input.get("fto_name", "unknown")
patch_path = project_patches_dir / f"{worktree_dir.name}.{final_function_name}.patch"
patch_path = project_patches_dir / f"{worktree_dir.name}.{fto_name}.patch"
with patch_path.open("w", encoding="utf8") as f:
f.write(uni_diff_text)

final_metadata = {"patch_path": str(patch_path)}
if metadata_input:
final_metadata.update(metadata_input)
final_metadata = save_patches_metadata(final_metadata)

return final_metadata
return patch_path
Loading
Loading