utf-8 encoding

Author: KRRT7

codeflash/code_utils/checkpoint.py

@@ -11,13 +11,16 @@
 from rich.prompt import Confirm
 
 from codeflash.cli_cmds.console import console
+from codeflash.code_utils.compat import codeflash_temp_dir
 
 if TYPE_CHECKING:
     import argparse
 
 
 class CodeflashRunCheckpoint:
-    def __init__(self, module_root: Path, checkpoint_dir: Path = Path("/tmp")) -> None:  # noqa: S108
+    def __init__(self, module_root: Path, checkpoint_dir: Path | None = None) -> None:
+        if checkpoint_dir is None:
+            checkpoint_dir = codeflash_temp_dir
         self.module_root = module_root
         self.checkpoint_dir = Path(checkpoint_dir)
         # Create a unique checkpoint file name
@@ -37,7 +40,7 @@ def _initialize_checkpoint_file(self) -> None:
             "last_updated": time.time(),
         }
 
-        with self.checkpoint_path.open("w") as f:
+        with self.checkpoint_path.open("w", encoding="utf-8") as f:
             f.write(json.dumps(metadata) + "\n")
 
     def add_function_to_checkpoint(
@@ -66,7 +69,7 @@ def add_function_to_checkpoint(
             **additional_info,
         }
 
-        with self.checkpoint_path.open("a") as f:
+        with self.checkpoint_path.open("a", encoding="utf-8") as f:
             f.write(json.dumps(function_data) + "\n")
 
         # Update the metadata last_updated timestamp
@@ -75,7 +78,7 @@ def add_function_to_checkpoint(
     def _update_metadata_timestamp(self) -> None:
         """Update the last_updated timestamp in the metadata."""
         # Read the first line (metadata)
-        with self.checkpoint_path.open() as f:
+        with self.checkpoint_path.open(encoding="utf-8") as f:
             metadata = json.loads(f.readline())
             rest_content = f.read()
 
@@ -84,7 +87,7 @@ def _update_metadata_timestamp(self) -> None:
 
         # Write all lines to a temporary file
 
-        with self.checkpoint_path.open("w") as f:
+        with self.checkpoint_path.open("w", encoding="utf-8") as f:
             f.write(json.dumps(metadata) + "\n")
             f.write(rest_content)
 
@@ -94,7 +97,7 @@ def cleanup(self) -> None:
         self.checkpoint_path.unlink(missing_ok=True)
 
         for file in self.checkpoint_dir.glob("codeflash_checkpoint_*.jsonl"):
-            with file.open() as f:
+            with file.open(encoding="utf-8") as f:
                 # Skip the first line (metadata)
                 first_line = next(f)
                 metadata = json.loads(first_line)
@@ -116,7 +119,7 @@ def get_all_historical_functions(module_root: Path, checkpoint_dir: Path) -> dic
     to_delete = []
 
     for file in checkpoint_dir.glob("codeflash_checkpoint_*.jsonl"):
-        with file.open() as f:
+        with file.open(encoding="utf-8") as f:
             # Skip the first line (metadata)
             first_line = next(f)
             metadata = json.loads(first_line)
@@ -139,8 +142,8 @@ def get_all_historical_functions(module_root: Path, checkpoint_dir: Path) -> dic
 
 def ask_should_use_checkpoint_get_functions(args: argparse.Namespace) -> Optional[dict[str, dict[str, str]]]:
     previous_checkpoint_functions = None
-    if args.all and (sys.platform == "linux" or sys.platform == "darwin") and Path("/tmp").is_dir():  # noqa: S108 #TODO: use the temp dir from codeutils-compat.py
-        previous_checkpoint_functions = get_all_historical_functions(args.module_root, Path("/tmp"))  # noqa: S108
+    if args.all and codeflash_temp_dir.is_dir():
+        previous_checkpoint_functions = get_all_historical_functions(args.module_root, codeflash_temp_dir)
         if previous_checkpoint_functions and Confirm.ask(
             "Previous Checkpoint detected from an incomplete optimization run, shall I continue the optimization from that point?",
             default=True,

codeflash/code_utils/env_utils.py

@@ -120,7 +120,7 @@ def get_cached_gh_event_data() -> dict[str, Any]:
     event_path = os.getenv("GITHUB_EVENT_PATH")
     if not event_path:
         return {}
-    with Path(event_path).open() as f:
+    with Path(event_path).open(encoding='utf-8') as f:
         return json.load(f)  # type: ignore  # noqa
 
 

codeflash/verification/coverage_utils.py

@@ -46,13 +46,13 @@ def load_from_sqlite_database(
 
         reporter = JsonReporter(cov)
         temp_json_file = database_path.with_suffix(".report.json")
-        with temp_json_file.open("w") as f:
+        with temp_json_file.open("w", encoding="utf-8") as f:
             try:
                 reporter.report(morfs=[source_code_path.as_posix()], outfile=f)
             except NoDataError:
                 sentry_sdk.capture_message(f"No coverage data found for {function_name} in {source_code_path}")
                 return CoverageUtils.create_empty(source_code_path, function_name, code_context)
-        with temp_json_file.open() as f:
+        with temp_json_file.open(encoding="utf-8") as f:
             original_coverage_data = json.load(f)
 
         coverage_data, status = CoverageUtils._parse_coverage_file(temp_json_file, source_code_path)
@@ -92,7 +92,7 @@ def load_from_sqlite_database(
     def _parse_coverage_file(
         coverage_file_path: Path, source_code_path: Path
     ) -> tuple[dict[str, dict[str, Any]], CoverageStatus]:
-        with coverage_file_path.open() as f:
+        with coverage_file_path.open(encoding="utf-8") as f:
             coverage_data = json.load(f)
 
         candidates = generate_candidates(source_code_path)