mirror of
https://github.com/showlab/Code2Video.git
synced 2026-08-28 06:54:58 +00:00
Delete utils.py
This commit is contained in:
Anno Yanzhe Chen
GitHub
parent
45e62d0003
commit
cc8ce2e1ef
1 changed files with 0 additions and 209 deletions
209
utils.py
209
utils.py
|
|
@ -1,209 +0,0 @@
|
|||
import os
|
||||
import subprocess
|
||||
from typing import List
|
||||
from manim import *
|
||||
import multiprocessing
|
||||
import re
|
||||
import psutil
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def extract_json_from_markdown(text):
|
||||
# Match ```json ... ``` or ``` ... ```
|
||||
match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return text
|
||||
|
||||
|
||||
def extract_answer_from_response(response):
|
||||
try:
|
||||
content = response.candidates[0].content.parts[0].text
|
||||
except Exception:
|
||||
try:
|
||||
content = response.choices[0].message.content
|
||||
except Exception:
|
||||
content = str(response)
|
||||
content = extract_json_from_markdown(content)
|
||||
return content
|
||||
|
||||
|
||||
def fix_png_path(code_str: str, assets_dir: Path) -> str:
|
||||
assets_dir = Path(assets_dir).resolve()
|
||||
|
||||
def replacer(match):
|
||||
original_path = match.group(1) # matched XXX.png
|
||||
path_obj = Path(original_path)
|
||||
# not an absolute path and is not under assets_dir
|
||||
if not path_obj.is_absolute():
|
||||
# concat to absolute path
|
||||
return f'"{assets_dir / path_obj.name}"'
|
||||
# absolute path but not under assets_dir
|
||||
try:
|
||||
if assets_dir not in path_obj.parents:
|
||||
return f'"{assets_dir / path_obj.name}"'
|
||||
except RuntimeError:
|
||||
return f'"{assets_dir / path_obj.name}"'
|
||||
return match.group(0) # keep original
|
||||
|
||||
pattern = r'["\']([^"\']+\.png)["\']'
|
||||
return re.sub(pattern, replacer, code_str)
|
||||
|
||||
|
||||
def get_optimal_workers():
|
||||
"""Calculate the optimal number of parallel processes adaptively based on # CPU cores and load"""
|
||||
try:
|
||||
cpu_count = multiprocessing.cpu_count()
|
||||
except NotImplementedError:
|
||||
cpu_count = 6 # default
|
||||
|
||||
# Manim rendering is CPU-intensive; usually set workers to CPU cores or cores minus one
|
||||
# reserve 1 core for system/other processes
|
||||
optimal = max(1, cpu_count - 1)
|
||||
|
||||
# If the machine is high-performance multicore (>16 cores),
|
||||
# it's appropriate to limit the number of workers to avoid memory overflow
|
||||
if optimal > 16:
|
||||
optimal = 16
|
||||
|
||||
print(f"⚙️ Detected {cpu_count} cores, using {optimal} parallel processes")
|
||||
return optimal
|
||||
|
||||
|
||||
def monitor_system_resources():
|
||||
"""Monitor system resource usage"""
|
||||
try:
|
||||
cpu_percent = psutil.cpu_percent(interval=0.1)
|
||||
memory = psutil.virtual_memory()
|
||||
|
||||
print(f"📊 Resource usage: CPU {cpu_percent:.1f}% | Memory {memory.percent:.1f}%")
|
||||
|
||||
if cpu_percent > 95:
|
||||
print("⚠️ CPU usage is high")
|
||||
if memory.percent > 90:
|
||||
print("⚠️ Memory usage is high")
|
||||
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def replace_base_class(code: str, new_class_def: str) -> str:
|
||||
lines = code.splitlines(keepends=True)
|
||||
class_start = None
|
||||
class_end = None
|
||||
|
||||
# Find the start line of class TeachingScene(Scene):
|
||||
for i, line in enumerate(lines):
|
||||
if re.match(r"^\s*class\s+TeachingScene\s*\(Scene\)\s*:", line):
|
||||
class_start = i
|
||||
break
|
||||
|
||||
if class_start is not None:
|
||||
# Find the end line of the class definition
|
||||
# The class ends when a line with the same or less indentation is found
|
||||
base_indent = len(lines[class_start]) - len(lines[class_start].lstrip())
|
||||
class_end = class_start + 1
|
||||
while class_end < len(lines):
|
||||
line = lines[class_end]
|
||||
# If an empty line or a line with less indentation is found,
|
||||
# it means the class definition has ended
|
||||
if line.strip() != "" and (len(line) - len(line.lstrip()) <= base_indent):
|
||||
break
|
||||
class_end += 1
|
||||
|
||||
# Replace the original TeachingScene definition with the new one
|
||||
new_block = new_class_def.strip() + "\n\n"
|
||||
return "".join(lines[:class_start]) + new_block + "".join(lines[class_end:])
|
||||
else:
|
||||
# If TeachingScene does not exist, it should be inserted before the first class definition
|
||||
for i, line in enumerate(lines):
|
||||
if re.match(r"^\s*class\s+\w+", line):
|
||||
insert_pos = i
|
||||
break
|
||||
else:
|
||||
insert_pos = 0
|
||||
|
||||
new_block = new_class_def.strip() + "\n\n"
|
||||
return "".join(lines[:insert_pos]) + new_block + "".join(lines[insert_pos:])
|
||||
|
||||
|
||||
# Save the program to the.py file
|
||||
def save_code_to_file(code: str, filename: str = "scene.py"):
|
||||
with open(filename, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
print(f"Saved code to {filename}")
|
||||
|
||||
|
||||
# Run the manim code to generate a video
|
||||
def run_manim_script(filename: str, scene_name: str, output_dir: str = "videos") -> str:
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
output_path = os.path.join(output_dir, f"{scene_name}.mp4")
|
||||
|
||||
cmd = [
|
||||
"manim",
|
||||
"-pql", # play + low quality（can changed to -pqm or -pqh）
|
||||
str(filename), # script path
|
||||
scene_name, # class name
|
||||
"--output_file",
|
||||
f"{scene_name}.mp4",
|
||||
"--media_dir",
|
||||
str(output_dir), # media output directory
|
||||
]
|
||||
|
||||
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
if result.returncode != 0:
|
||||
print("Manim error:", result.stderr.decode())
|
||||
raise RuntimeError(f"Failed to render scene {scene_name}.")
|
||||
|
||||
print(f"Video saved to {output_path}")
|
||||
return output_path
|
||||
|
||||
|
||||
# Use ffmpeg to concatenate multiple mp4 files
|
||||
def stitch_videos(video_files: List[str], output_path: str = "final_output.mp4"):
|
||||
list_file = "video_list.txt"
|
||||
with open(list_file, "w") as f:
|
||||
for vf in video_files:
|
||||
f.write(f"file '{os.path.abspath(vf)}'\n")
|
||||
|
||||
cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file, "-c", "copy", output_path]
|
||||
print("Stitching videos:", cmd)
|
||||
subprocess.run(cmd, check=True)
|
||||
print(f"Final stitched video saved to {output_path}")
|
||||
|
||||
|
||||
def topic_to_safe_name(knowledge_point):
|
||||
# Allowed: alphanumeric Spaces _ - { } [ ] . , + & ' =
|
||||
SAFE_PATTERN = r"[^A-Za-z0-9 _\-\{\}\[\]\+&=\u03C0]"
|
||||
safe_name = re.sub(SAFE_PATTERN, "", knowledge_point)
|
||||
# Replace consecutive spaces with a single underscore
|
||||
safe_name = re.sub(r"\s+", "_", safe_name.strip())
|
||||
return safe_name
|
||||
|
||||
|
||||
def get_output_dir(idx, knowledge_point, base_dir, get_safe_name=False):
|
||||
safe_name = topic_to_safe_name(knowledge_point)
|
||||
# Prefix with idx-
|
||||
folder_name = f"{idx}-{safe_name}"
|
||||
if get_safe_name:
|
||||
return Path(base_dir) / folder_name, safe_name
|
||||
|
||||
return Path(base_dir) / folder_name
|
||||
|
||||
|
||||
def eva_video_list(knowledge_points, base_dir):
|
||||
|
||||
video_list = []
|
||||
for idx, kp in enumerate(knowledge_points):
|
||||
folder, safe_name = get_output_dir(idx, kp, base_dir, get_safe_name=True)
|
||||
|
||||
# mp4 filename must be safe, the same
|
||||
mp4_name = f"{safe_name}.mp4"
|
||||
mp4_path = folder / mp4_name
|
||||
video_list.append({"path": str(mp4_path), "knowledge_point": kp})
|
||||
return video_list
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(get_optimal_workers())
|
||||
Loading…
Add table
Reference in a new issue