93 lines
3.3 KiB
Python
Executable File
93 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import sys
|
|
import shlex
|
|
import subprocess
|
|
|
|
def run_command_ffmpeg(cmd_str: str, progress_prefix: str = "Progress") -> bool:
|
|
"""Runs a system command and streams its output live to the console."""
|
|
# shlex safely handles quotes and paths inside the command string
|
|
args = shlex.split(cmd_str)
|
|
|
|
# Redirect stderr to stdout because FFmpeg outputs status updates to stderr
|
|
process = subprocess.Popen(
|
|
args,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1
|
|
)
|
|
|
|
print(f"Executing: {cmd_str[:90]}...")
|
|
|
|
# Stream the output live to the terminal
|
|
while True:
|
|
line = process.stdout.readline()
|
|
if not line and process.poll() is not None:
|
|
break
|
|
if line:
|
|
clean_line = line.strip()
|
|
# Only print updates that show progress metrics to keep terminal clean
|
|
if any(metric in clean_line for metric in ["frame=", "time=", "fps=", "Rendering frame"]):
|
|
sys.stdout.write(f"\r[{progress_prefix}] {clean_line}")
|
|
sys.stdout.flush()
|
|
elif "Error" in clean_line or "failed" in clean_line:
|
|
print(f"\n[Alert] {clean_line}")
|
|
|
|
print("\n") # New line after process finishes
|
|
return process.returncode == 0
|
|
|
|
def run_command(command: list[str]) -> dict:
|
|
"""Executes a command, streams stdout only when the line changes, and detects OOM."""
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
shell=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1,
|
|
)
|
|
|
|
last_line = None # Track the previous line
|
|
|
|
if process.stdout:
|
|
for line in process.stdout:
|
|
if line != last_line: # Only print if the content changed
|
|
print(line, end="")
|
|
sys.stdout.flush()
|
|
last_line = line # Update the tracking variable
|
|
|
|
_, stderr = process.communicate()
|
|
|
|
if process.returncode != 0:
|
|
if process.returncode in [137, -9]:
|
|
return {
|
|
"success": False,
|
|
"error": "Process was KILLED by the Linux kernel (Out of Memory).",
|
|
"stderr": stderr,
|
|
}
|
|
return {
|
|
"success": False,
|
|
"error": f"Exit code {process.returncode}",
|
|
"stderr": stderr,
|
|
}
|
|
|
|
return {"success": True, "error": "", "stderr": stderr}
|
|
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e), "stderr": ""}
|
|
|
|
def run_command_old(command: str):
|
|
"""Executes a Linux command, waits for completion, and returns output."""
|
|
try:
|
|
# shell=True allows running full command strings with pipes/wildcards
|
|
# text=True returns strings instead of bytes
|
|
result = subprocess.run(
|
|
command, shell=True, check=True, capture_output=True, text=True
|
|
)
|
|
|
|
return {"success": True, "stdout": result.stdout, "stderr": result.stderr}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
# Handles errors if the Linux command returns a non-zero exit code
|
|
return {"success": False, "stdout": e.stdout, "stderr": e.stderr} |