Migrated file_ops.py and scene_file_writer.py from os.path to Pathlib (#2642)

* Migrated file_ops.py and scene_file_writer.py from os.path to Pathlib (github issue #45)

* Updated scene_file_writer.py to handle WindowsPaths

* Updated scene_file_writer.py to handle WindowsPaths

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Updated file_ops.py to handle WindowsPaths

* Updated scene_file_writer.py to handle WindowsPaths

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fixed Merge Conflict: Added Type Hints in file_ops.py

* Changed test_file_ops to use Pathlib in accordance with file_ops.py usage

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Co-authored-by: Will Soltas <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: 3 people

manim/scene/scene_file_writer.py

@@ -111,9 +111,8 @@ def init_output_directories(self, scene_name):
                     "images_dir", module_name=module_name, scene_name=scene_name
                 ),
             )
-            self.image_file_path = os.path.join(
-                image_dir,
-                add_extension_if_not_present(self.output_name, ".png"),
+            self.image_file_path = image_dir / add_extension_if_not_present(
+                self.output_name, ".png"
             )
 
         if write_to_movie():
@@ -122,14 +121,10 @@ def init_output_directories(self, scene_name):
                     "video_dir", module_name=module_name, scene_name=scene_name
                 ),
             )
-
-            self.movie_file_path = os.path.join(
-                movie_dir,
-                add_extension_if_not_present(
-                    self.output_name,
-                    config["movie_file_extension"],
-                ),
+            self.movie_file_path = movie_dir / add_extension_if_not_present(
+                self.output_name, config["movie_file_extension"]
             )
+
             # TODO: /dev/null would be good in case sections_output_dir is used without bein set (doesn't work on Windows), everyone likes defensive programming, right?
             self.sections_output_dir = ""
             if config.save_sections:
@@ -149,7 +144,7 @@ def init_output_directories(self, scene_name):
                         self.gif_file_path
                     )
 
-                self.gif_file_path = os.path.join(movie_dir, self.gif_file_path)
+                self.gif_file_path = movie_dir / self.gif_file_path
 
             self.partial_movie_directory = guarantee_existence(
                 config.get_dir(
@@ -213,9 +208,9 @@ def add_partial_movie_file(self, hash_animation):
             self.partial_movie_files.append(None)
             self.sections[-1].partial_movie_files.append(None)
         else:
-            new_partial_movie_file = os.path.join(
-                self.partial_movie_directory,
-                f"{hash_animation}{config['movie_file_extension']}",
+            new_partial_movie_file = str(
+                self.partial_movie_directory
+                / f"{hash_animation}{config['movie_file_extension']}"
             )
             self.partial_movie_files.append(new_partial_movie_file)
             self.sections[-1].partial_movie_files.append(new_partial_movie_file)
@@ -380,7 +375,8 @@ def write_opengl_frame(self, renderer):
                 renderer.get_raw_frame_buffer_object_data(),
             )
         elif is_png_format() and not config["dry_run"]:
-            target_dir, extension = os.path.splitext(self.image_file_path)
+            target_dir = self.image_file_path.parent / self.image_file_path.stem
+            extension = self.image_file_path.suffix
             self.output_image(
                 renderer.get_image(),
                 target_dir,
@@ -389,7 +385,8 @@ def write_opengl_frame(self, renderer):
             )
 
     def output_image_from_array(self, frame_data):
-        target_dir, extension = os.path.splitext(self.image_file_path)
+        target_dir = self.image_file_path.parent / self.image_file_path.stem
+        extension = self.image_file_path.suffix
         self.output_image(
             Image.fromarray(frame_data),
             target_dir,
@@ -442,8 +439,8 @@ def finish(self):
             else:
                 self.clean_cache()
         elif is_png_format() and not config["dry_run"]:
-            target_dir, _ = os.path.splitext(self.image_file_path)
-            logger.info("\n%i images ready at %s\n", self.frame_count, target_dir)
+            target_dir = self.image_file_path.parent / self.image_file_path.stem
+            logger.info("\n%i images ready at %s\n", self.frame_count, str(target_dir))
         if self.subcaptions:
             self.write_subcaption_file()
 
@@ -524,11 +521,11 @@ def is_already_cached(self, hash_invocation):
         """
         if not hasattr(self, "partial_movie_directory") or not write_to_movie():
             return False
-        path = os.path.join(
-            self.partial_movie_directory,
-            f"{hash_invocation}{config['movie_file_extension']}",
+        path = (
+            self.partial_movie_directory
+            / f"{hash_invocation}{config['movie_file_extension']}"
         )
-        return os.path.exists(path)
+        return path.exists()
 
     def combine_files(
         self,
@@ -537,17 +534,15 @@ def combine_files(
         create_gif=False,
         includes_sound=False,
     ):
-        file_list = os.path.join(
-            self.partial_movie_directory,
-            "partial_movie_file_list.txt",
-        )
+        file_list = str(self.partial_movie_directory / "partial_movie_file_list.txt")
         logger.debug(
             f"Partial movie files to combine ({len(input_files)} files): %(p)s",
             {"p": input_files[:5]},
         )
         with open(file_list, "w", encoding="utf-8") as fp:
             fp.write("# This file is used internally by FFMPEG.\n")
             for pf_path in input_files:
+                pf_path = str(pf_path)
                 if os.name == "nt":
                     pf_path = pf_path.replace("\\", "/")
                 fp.write(f"file 'file:{pf_path}'\n")
@@ -578,7 +573,7 @@ def combine_files(
         if not includes_sound:
             commands += ["-an"]
 
-        commands += [output_file]
+        commands += [str(output_file)]
 
         combine_process = subprocess.Popen(commands)
         combine_process.wait()
@@ -598,6 +593,7 @@ def combine_to_movie(self):
         movie_file_path = self.movie_file_path
         if is_gif_format():
             movie_file_path = self.gif_file_path
+        movie_file_path = str(movie_file_path)
         logger.info("Combining to Movie file.")
         self.combine_files(
             partial_movie_files,
@@ -675,8 +671,8 @@ def combine_to_section_videos(self) -> None:
     def clean_cache(self):
         """Will clean the cache by removing the oldest partial_movie_files."""
         cached_partial_movies = [
-            os.path.join(self.partial_movie_directory, file_name)
-            for file_name in os.listdir(self.partial_movie_directory)
+            (self.partial_movie_directory / file_name)
+            for file_name in self.partial_movie_directory.iterdir()
             if file_name != "partial_movie_file_list.txt"
         ]
         if len(cached_partial_movies) > config["max_files_cached"]:
@@ -689,7 +685,7 @@ def clean_cache(self):
             )[:number_files_to_delete]
             # oldest_file_path = min(cached_partial_movies, key=os.path.getatime)
             for file_to_delete in oldest_files_to_delete:
-                os.remove(file_to_delete)
+                file_to_delete.unlink()
             logger.info(
                 f"The partial movie directory is full (> {config['max_files_cached']} files). Therefore, manim has removed the {number_files_to_delete} oldest file(s)."
                 " You can change this behaviour by changing max_files_cached in config.",
@@ -698,12 +694,12 @@ def clean_cache(self):
     def flush_cache_directory(self):
         """Delete all the cached partial movie files"""
         cached_partial_movies = [
-            os.path.join(self.partial_movie_directory, file_name)
-            for file_name in os.listdir(self.partial_movie_directory)
+            self.partial_movie_directory / file_name
+            for file_name in self.partial_movie_directory.iterdir()
             if file_name != "partial_movie_file_list.txt"
         ]
         for f in cached_partial_movies:
-            os.remove(f)
+            f.unlink()
         logger.info(
             f"Cache flushed. {len(cached_partial_movies)} file(s) deleted in %(par_dir)s.",
             {"par_dir": self.partial_movie_directory},

manim/utils/file_ops.py

@@ -135,27 +135,27 @@ def add_version_before_extension(file_name: Path) -> Path:
 
 
 def guarantee_existence(path: Path) -> Path:
-    if not os.path.exists(path):
-        os.makedirs(path)
-    return Path(os.path.abspath(path))
+    if not path.exists():
+        path.mkdir(parents=True)
+    return path.resolve(strict=True)
 
 
 def guarantee_empty_existence(path: Path) -> Path:
-    if os.path.exists(path):
-        shutil.rmtree(path)
-    os.makedirs(path)
-    return Path(os.path.abspath(path))
+    if path.exists():
+        shutil.rmtree(str(path))
+    path.mkdir(parents=True)
+    return path.resolve(strict=True)
 
 
 def seek_full_path_from_defaults(
-    file_name: Path, default_dir: str, extensions: str
+    file_name: str, default_dir: Path, extensions: list[str]
 ) -> Path:
-    possible_paths = [file_name]
+    possible_paths = [Path(file_name)]
     possible_paths += [
         Path(default_dir) / f"{file_name}{extension}" for extension in ["", *extensions]
     ]
     for path in possible_paths:
-        if os.path.exists(path):
+        if path.exists():
             return path
     error = f"From: {os.getcwd()}, could not find {file_name} at either of these locations: {possible_paths}"
     raise OSError(error)
@@ -175,14 +175,14 @@ def modify_atime(file_path) -> None:
 def open_file(file_path, in_browser=False):
     current_os = platform.system()
     if current_os == "Windows":
-        os.startfile(file_path if not in_browser else os.path.dirname(file_path))
+        os.startfile(file_path if not in_browser else file_path.parent)
     else:
         if current_os == "Linux":
             commands = ["xdg-open"]
-            file_path = file_path if not in_browser else os.path.dirname(file_path)
+            file_path = file_path if not in_browser else file_path.parent
         elif current_os.startswith("CYGWIN"):
             commands = ["cygstart"]
-            file_path = file_path if not in_browser else os.path.dirname(file_path)
+            file_path = file_path if not in_browser else file_path.parent
         elif current_os == "Darwin":
             if is_gif_format():
                 commands = ["ffplay", "-loglevel", config["ffmpeg_loglevel"].lower()]

tests/test_file_ops.py

@@ -1,32 +1,32 @@
 from __future__ import annotations
 
-import os
+from pathlib import Path
 
 from manim import *
 
 from .assert_utils import assert_dir_exists, assert_file_not_exists
 from .utils.video_tester import *
 
 
-def test_guarantee_existence(tmp_path):
-    test_dir = os.path.join(tmp_path, "test")
+def test_guarantee_existence(tmp_path: Path):
+    test_dir = tmp_path / "test"
     guarantee_existence(test_dir)
     # test if file dir got created
     assert_dir_exists(test_dir)
-    with open(os.path.join(test_dir, "test.txt"), "x") as f:
+    with open(test_dir / "test.txt", "x") as f:
         pass
     # test if file didn't get deleted
     guarantee_existence(test_dir)
 
 
-def test_guarantee_empty_existence(tmp_path):
-    test_dir = os.path.join(tmp_path, "test")
-    os.mkdir(test_dir)
-    with open(os.path.join(test_dir, "test.txt"), "x"):
+def test_guarantee_empty_existence(tmp_path: Path):
+    test_dir = tmp_path / "test"
+    test_dir.mkdir()
+    with open(test_dir / "test.txt", "x"):
         pass
 
     guarantee_empty_existence(test_dir)
     # test if dir got created
     assert_dir_exists(test_dir)
     # test if dir got cleaned
-    assert_file_not_exists(os.path.join(test_dir, "test.txt"))
+    assert_file_not_exists(test_dir / "test.txt")