(step_list: List[int], checkpoint_dir: str, converter)
| 51 | |
| 52 | |
| 53 | def _convert_multi_steps(step_list: List[int], checkpoint_dir: str, converter) -> None: |
| 54 | # When --step is provided, convert only the specified step checkpoints |
| 55 | # Resolve the root directory that contains global_step_* folders |
| 56 | root_dir = checkpoint_dir |
| 57 | if "global_step_" in root_dir: |
| 58 | while not os.path.basename(root_dir).startswith("global_step_"): |
| 59 | root_dir = os.path.dirname(root_dir) |
| 60 | # Go one level up to the parent that contains global_step_* dirs |
| 61 | root_dir = os.path.dirname(root_dir) |
| 62 | |
| 63 | succeeded: List[int] = [] |
| 64 | failed: List[tuple] = [] # (step, reason) |
| 65 | |
| 66 | for s in step_list: |
| 67 | step_dir = os.path.join(root_dir, f"global_step_{s}") |
| 68 | if not os.path.isdir(step_dir): |
| 69 | reason = f"Directory not found: {step_dir}" |
| 70 | typer.echo(f"[ERROR] Step {s}: {reason}", err=True) |
| 71 | failed.append((s, reason)) |
| 72 | continue |
| 73 | try: |
| 74 | converter.convert(step_dir) |
| 75 | succeeded.append(s) |
| 76 | except Exception as e: |
| 77 | typer.echo(f"[ERROR] Step {s}: {e}", err=True) |
| 78 | failed.append((s, str(e))) |
| 79 | |
| 80 | # Print summary report |
| 81 | typer.echo("\n" + "=" * 50) |
| 82 | typer.echo("Conversion Report") |
| 83 | typer.echo("=" * 50) |
| 84 | typer.echo(f"Total requested: {len(step_list)}") |
| 85 | typer.echo(f"Succeeded: {len(succeeded)}") |
| 86 | typer.echo(f"Failed: {len(failed)}") |
| 87 | if succeeded: |
| 88 | typer.echo(f"\nSucceeded steps: {', '.join(str(s) for s in succeeded)}") |
| 89 | if failed: |
| 90 | typer.echo("\nFailed steps:") |
| 91 | for s, reason in failed: |
| 92 | typer.echo(f" Step {s}: {reason}") |
| 93 | typer.echo("=" * 50) |
| 94 | |
| 95 | if failed: |
| 96 | raise typer.Exit(code=1) |
no test coverage detected