Cleaning up files and code for pugetbench

This commit is contained in:
J-Doiron
2026-01-08 14:38:16 -08:00
parent 1ccdc423be
commit f431fb6727
2 changed files with 79 additions and 103 deletions

View File

@@ -33,63 +33,76 @@ logging.getLogger('').addHandler(console)
EXECUTABLE_NAME = "PugetBench for Creators.exe"
APP_CONFIG = {
"premierepro": {
"label": "Adobe Premiere Pro",
"version_func": get_premierepro_version,
"suffix": None,
},
"photoshop": {
"label": "Adobe Photoshop",
"version_func": get_photoshop_version,
"suffix": None,
},
"aftereffects": {
"label": "Adobe After Effects",
"version_func": get_aftereffects_version,
"suffix": None,
},
"lightroom": {
"label": "Adobe Lightroom Classic",
"version_func": get_lightroom_version,
"suffix": None,
},
"resolve": {
"label": "Davinci Resolve Studio",
"version_func": get_davinci_version,
"suffix": "-studio",
},
}
def read_output(stream, log_func, error_func, error_in_output):
"""Read and log output in real-time from a stream (stdout or stderr)."""
while True:
line = stream.readline()
if not line:
break
for line in iter(stream.readline, ''):
line = line.strip()
log_func(line) # Log the output
# If line contains "Error!:", store RuntimeError to be raised later
log_func(line)
if line.startswith("Error!:"):
error_func(line)
error_in_output["exception"] = RuntimeError(
f"Benchmark failed with error: {line}")
error_in_output["exception"] = RuntimeError(f"Benchmark failed with error: {line}")
break
# If line contains "Benchmark failed:", store RuntimeError
if line.startswith("Benchmark failed:"):
error_in_output["exception"] = RuntimeError(
"Benchmark had an unknown failure.")
error_in_output["exception"] = RuntimeError("Benchmark had an unknown failure.")
break
sys.stdout.flush() # optional here, but fine to keep
sys.stdout.flush()
def run_benchmark(application: str, app_version: str, benchmark_version: str):
"""run benchmark"""
start_time = time.time()
executable_path = Path(
f"C:\\Program Files\\PugetBench for Creators\\{EXECUTABLE_NAME}")
executable_path = Path(f"C:\\Program Files\\PugetBench for Creators\\{EXECUTABLE_NAME}")
if not executable_path.exists():
logging.error(f"PugetBench executable not found at {executable_path}")
logging.error("PugetBench executable not found at %s", executable_path)
sys.exit(1)
command_args = ["--run_count", "1", "--rerun_count", "1",
"--benchmark_version", f"{benchmark_version}", "--preset",
"Standard", "--app_version", f"{app_version}"]
command = None
if application == "premierepro":
command = [executable_path] + command_args + ["--app", "premierepro"]
elif application == "photoshop":
command = [executable_path] + command_args + ["--app", "photoshop"]
elif application == "aftereffects":
command = [executable_path] + command_args + ["--app", "aftereffects"]
elif application == "lightroom":
command = [executable_path] + command_args + ["--app", "lightroom"]
elif application == "resolve":
command = [executable_path] + command_args + ["--app", "resolve"]
command = [
executable_path,
"--run_count", "1",
"--rerun_count", "1",
"--benchmark_version", benchmark_version,
"--preset", "Standard",
"--app_version", app_version,
"--app", application
]
logging.info("Running benchmark command: %s", command)
logging.info(command)
error_in_output = {"exception": None} # Shared state for error reporting
with Popen(command, stdout=PIPE, stderr=PIPE, text=True) as process:
stdout_thread = threading.Thread(target=read_output, args=(
process.stdout, logging.info, logging.error, error_in_output))
stderr_thread = threading.Thread(target=read_output, args=(
process.stderr, logging.error, logging.error, error_in_output))
stdout_thread = threading.Thread(target=read_output, args=(process.stdout, logging.info, logging.error, error_in_output))
stderr_thread = threading.Thread(target=read_output, args=(process.stderr, logging.error, logging.error, error_in_output))
stdout_thread.start()
stderr_thread.start()
@@ -113,77 +126,38 @@ def main():
start_time = time.time()
parser = ArgumentParser()
parser.add_argument("--app", dest="app",
help="Application name to test", required=True)
parser.add_argument(
"--app_version", dest="app_version",
help="Application version to test", required=False)
parser.add_argument(
"--benchmark_version", dest="benchmark_version",
help="Puget Bench Benchmark version to use", required=False)
parser.add_argument("--app", choices=APP_CONFIG.keys(), dest="app", help="Application name to test", required=True)
parser.add_argument("--app_version", dest="app_version",help="Application version to test", required=False)
parser.add_argument("--benchmark_version", dest="benchmark_version", help="Puget Bench Benchmark version to use", required=False)
args = parser.parse_args()
apps = [
"premierepro",
"photoshop",
"aftereffects",
"lightroom",
"resolve"
]
if args.app is None or args.app not in apps:
logging.info("unrecognized option for program")
sys.exit(1)
config = APP_CONFIG[args.app]
test_label = config["label"]
# if args.app is None or args.app not in apps:
# logging.info("unrecognized option for program")
# sys.exit(1)
# Determine app version
if args.app_version is None:
full_version, trimmed_version = config["version_func"]()
if not full_version or not trimmed_version:
logging.error("Could not determine %s version. Is it installed?", test_label)
sys.exit(1)
else:
full_version = args.app_version
trimmed_version = trim_to_major_minor(full_version)
# Apply optional suffix (e.g., Resolve)
if config["suffix"]:
full_version += config["suffix"]
trimmed_version += config["suffix"]
if args.benchmark_version is None:
args.benchmark_version = get_latest_benchmark_by_version(args.app)
version = args.app_version
score = 0
full_version = version
trimmed_version = trim_to_major_minor(version)
test = ""
if args.app == "premierepro":
test = "Adobe Premiere Pro"
if version is None:
full_version, trimmed_version = get_premierepro_version()
if full_version is None or trimmed_version is None:
logging.error("Could not determine Premiere Pro version. Is it installed?")
sys.exit(1)
elif args.app == "photoshop":
test = "Adobe Photoshop"
if version is None:
full_version, trimmed_version = get_photoshop_version()
if full_version is None or trimmed_version is None:
logging.error("Could not determine Photoshop version. Is it installed?")
sys.exit(1)
elif args.app == "aftereffects":
test = "Adobe After Effects"
if version is None:
full_version, trimmed_version = get_aftereffects_version()
if full_version is None or trimmed_version is None:
logging.error("Could not determine After Effects version. Is it installed?")
sys.exit(1)
elif args.app == "lightroom":
test = "Adobe Lightroom Classic"
if version is None:
full_version, trimmed_version = get_lightroom_version()
if full_version is None or trimmed_version is None:
logging.error("Could not determine Lightroom version. Is it installed?")
sys.exit(1)
elif args.app == "resolve":
test = "Davinci Resolve Studio"
if version is None:
full_version, trimmed_version = get_davinci_version()
if full_version is None or trimmed_version is None:
logging.error("Could not determine Resolve Studio version. Is it installed?")
sys.exit(1)
full_version += "-studio"
trimmed_version += "-studio"
try:
start_time, end_time = run_benchmark(
args.app, trimmed_version, args.benchmark_version)
start_time, end_time = run_benchmark(args.app, trimmed_version, args.benchmark_version)
log_file = find_latest_log()
score = find_score_in_log(log_file)
destination = Path(script_dir) / "run" / os.path.split(log_file)[1]
@@ -193,7 +167,7 @@ def main():
"start_time": seconds_to_milliseconds(start_time),
"end_time": seconds_to_milliseconds(end_time),
"test": "PugetBench",
"test_parameter": test,
"test_parameter": test_label,
"app_version": full_version,
"benchmark_version": args.benchmark_version,
"pugetbench_version": get_pugetbench_version(),
@@ -202,6 +176,7 @@ def main():
}
write_report_json(log_dir, "report.json", report)
except Exception as e:
logging.error("Something went wrong running the benchmark!")
logging.exception(e)

View File

@@ -6,6 +6,7 @@ import win32api
import csv
def trim_to_major_minor(version: str | None) -> str | None:
"""Trims the versioning into major.minor."""
if version is None:
return None
# Match major.minor at the start
@@ -59,11 +60,11 @@ def get_latest_benchmark_by_version(benchmark_name: str):
# Sort numerically, with releases before beta
def version_key(v: str):
main, *suffix = v.split("-")
main = v.split("-")[0] # take numeric part only
nums = tuple(int(x) for x in main.split("."))
beta_flag = 1 if "-beta" in v else 0
return nums, -beta_flag # release first
versions.sort(key=version_key, reverse=True)
# Return the latest version