(case: dict[str, Any], args: argparse.Namespace)
| 559 | |
| 560 | |
| 561 | def ace_step_loader(case: dict[str, Any], args: argparse.Namespace) -> tuple[Any, RunnerFn]: |
| 562 | set_threads(args.threads) |
| 563 | ref_root = REPO_ROOT / "reference" / "ACE-Step-1.5" |
| 564 | if str(REPO_ROOT) not in sys.path: |
| 565 | sys.path.insert(0, str(REPO_ROOT)) |
| 566 | if str(ref_root) not in sys.path: |
| 567 | sys.path.insert(0, str(ref_root)) |
| 568 | from acestep.handler import AceStepHandler |
| 569 | from acestep.inference import GenerationConfig, GenerationParams, generate_music |
| 570 | from acestep.llm_inference import LLMHandler |
| 571 | |
| 572 | backend = "cuda" if args.backend == "cuda" else "cpu" |
| 573 | checkpoint_dir = ensure_absolute_path(case["model"]) |
| 574 | os.environ["ACESTEP_CHECKPOINTS_DIR"] = str(checkpoint_dir) |
| 575 | dit_model_path = case.get("load_options", {}).get("ace_step.dit_model_path", "acestep-v15-turbo") |
| 576 | lm_model_path = case.get("load_options", {}).get("ace_step.lm_model_path", "acestep-5Hz-lm-1.7B") |
| 577 | |
| 578 | dit_handler = AceStepHandler() |
| 579 | init_message, ok = dit_handler.initialize_service( |
| 580 | project_root=str(checkpoint_dir), |
| 581 | config_path=dit_model_path, |
| 582 | device=backend, |
| 583 | force_dtype=None, |
| 584 | use_flash_attention=False, |
| 585 | compile_model=False, |
| 586 | offload_to_cpu=False, |
| 587 | offload_dit_to_cpu=False, |
| 588 | quantization=None, |
| 589 | ) |
| 590 | if not ok: |
| 591 | raise RuntimeError(f"ACE-Step DiT init failed: {init_message}") |
| 592 | |
| 593 | llm_handler = LLMHandler() |
| 594 | init_message, ok = llm_handler.initialize( |
| 595 | checkpoint_dir=str(checkpoint_dir), |
| 596 | lm_model_path=lm_model_path, |
| 597 | backend="pt", |
| 598 | device=backend, |
| 599 | offload_to_cpu=False, |
| 600 | dtype=None, |
| 601 | ) |
| 602 | if not ok: |
| 603 | raise RuntimeError(f"ACE-Step LM init failed: {init_message}") |
| 604 | |
| 605 | def run(request: dict[str, Any], _args: argparse.Namespace) -> tuple[float, float]: |
| 606 | options = dict(request.get("options", {})) |
| 607 | params_dict = { |
| 608 | "task_type": request.get("task_route", "text2music"), |
| 609 | "caption": request.get("text", ""), |
| 610 | "lyrics": request.get("lyrics", ""), |
| 611 | "vocal_language": request.get("language", "unknown"), |
| 612 | "duration": request.get("duration_seconds"), |
| 613 | "inference_steps": request.get("num_inference_steps"), |
| 614 | "guidance_scale": request.get("guidance_scale"), |
| 615 | "seed": request.get("seed"), |
| 616 | "src_audio": request.get("audio"), |
| 617 | "reference_audio": request.get("reference_audio"), |
| 618 | "instruction": request.get("instruction"), |
nothing calls this directly
no test coverage detected