Skip to content

Tuner Timer

PhaseTimingRecord dataclass

Timing record for a single PolicyTuner pipeline phase.

Source code in src/autoslo/tuner/policy_tuner_timer.py 
16
17
18
19
20
21
22
23
@dataclass
class PhaseTimingRecord:
    """Timing record for a single PolicyTuner pipeline phase."""

    phase_key: str
    phase_name: str
    start_wall_utc: float  # epoch seconds from wall_clock_utc()
    elapsed_s: float

PolicyTunerTimer

Utility class for timing the phases of a PolicyTuner run and generating a report.

Source code in src/autoslo/tuner/policy_tuner_timer.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class PolicyTunerTimer:
    """
    Utility class for timing the phases of a PolicyTuner run and generating a
    report.
    """

    def __init__(self) -> None:
        self._phase_timings: list[PhaseTimingRecord] = []

    @staticmethod
    def _format_duration(elapsed_s: float) -> str:
        """Format a duration in seconds as a human-readable string."""
        total = int(elapsed_s)
        hours, remainder = divmod(total, 3600)
        minutes, seconds = divmod(remainder, 60)
        if hours > 0:
            return f"{hours}h {minutes}m {seconds}s"
        if minutes > 0:
            return f"{minutes}m {seconds}s"
        return f"{elapsed_s:.1f}s"

    @contextmanager
    def timed_phase(self, phase_key: str, phase_name: str):
        """Context manager that times a pipeline phase and appends a record."""
        start_wall = wall_clock_utc()
        t0 = time.perf_counter()
        try:
            yield
        finally:
            elapsed_s = time.perf_counter() - t0
            self._phase_timings.append(
                PhaseTimingRecord(
                    phase_key=phase_key,
                    phase_name=phase_name,
                    start_wall_utc=start_wall,
                    elapsed_s=elapsed_s,
                )
            )

    def finalize(self, out_dir: Path) -> None:
        """
        Write the timing report CSV and print a rich report to the console.
        """
        csv_path = out_dir / "timing_report.csv"
        total_s = sum(r.elapsed_s for r in self._phase_timings)
        fieldnames = [
            "phase_key",
            "phase_name",
            "start_wall_utc",
            "elapsed_s",
        ]
        with open(csv_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for r in self._phase_timings:
                writer.writerow(
                    {
                        "phase_key": r.phase_key,
                        "phase_name": r.phase_name,
                        "start_wall_utc": datetime.fromtimestamp(
                            r.start_wall_utc, tz=timezone.utc
                        ).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                        "elapsed_s": f"{r.elapsed_s:.3f}",
                    }
                )
            if self._phase_timings:
                writer.writerow(
                    {
                        "phase_key": "TOTAL",
                        "phase_name": "TOTAL",
                        "start_wall_utc": datetime.fromtimestamp(
                            self._phase_timings[0].start_wall_utc,
                            tz=timezone.utc,
                        ).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                        "elapsed_s": f"{total_s:.3f}",
                    }
                )

        # Print to console.
        total_s = sum(r.elapsed_s for r in self._phase_timings)
        table = Table(title="Tuning Pipeline — Timing Report", show_footer=True)
        table.add_column("Phase", footer="[bold]TOTAL[/]")
        table.add_column(
            "Duration",
            footer=f"[bold]{self._format_duration(total_s)}[/]",
            justify="right",
        )
        table.add_column("Wall start (UTC)", footer="", style="dim")
        table.add_column("% of total", footer="[bold]100%[/]", justify="right")

        for r in self._phase_timings:
            duration = self._format_duration(r.elapsed_s)
            wall_str = datetime.fromtimestamp(
                r.start_wall_utc, tz=timezone.utc
            ).strftime("%H:%M:%S")
            pct = f"{r.elapsed_s / total_s * 100:.0f}%" if total_s > 0 else "–"
            table.add_row(r.phase_name, duration, wall_str, pct)

        console.print()
        console.rule("[bold cyan]Timing Report")
        console.print(table)
        console.print(f"  Timing report saved to [bold]{csv_path}[/]")

_format_duration(elapsed_s) staticmethod

Format a duration in seconds as a human-readable string.

Source code in src/autoslo/tuner/policy_tuner_timer.py 
35
36
37
38
39
40
41
42
43
44
45
@staticmethod
def _format_duration(elapsed_s: float) -> str:
    """Format a duration in seconds as a human-readable string."""
    total = int(elapsed_s)
    hours, remainder = divmod(total, 3600)
    minutes, seconds = divmod(remainder, 60)
    if hours > 0:
        return f"{hours}h {minutes}m {seconds}s"
    if minutes > 0:
        return f"{minutes}m {seconds}s"
    return f"{elapsed_s:.1f}s"

timed_phase(phase_key, phase_name)

Context manager that times a pipeline phase and appends a record.

Source code in src/autoslo/tuner/policy_tuner_timer.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@contextmanager
def timed_phase(self, phase_key: str, phase_name: str):
    """Context manager that times a pipeline phase and appends a record."""
    start_wall = wall_clock_utc()
    t0 = time.perf_counter()
    try:
        yield
    finally:
        elapsed_s = time.perf_counter() - t0
        self._phase_timings.append(
            PhaseTimingRecord(
                phase_key=phase_key,
                phase_name=phase_name,
                start_wall_utc=start_wall,
                elapsed_s=elapsed_s,
            )
        )

finalize(out_dir)

Write the timing report CSV and print a rich report to the console.

Source code in src/autoslo/tuner/policy_tuner_timer.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def finalize(self, out_dir: Path) -> None:
    """
    Write the timing report CSV and print a rich report to the console.
    """
    csv_path = out_dir / "timing_report.csv"
    total_s = sum(r.elapsed_s for r in self._phase_timings)
    fieldnames = [
        "phase_key",
        "phase_name",
        "start_wall_utc",
        "elapsed_s",
    ]
    with open(csv_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for r in self._phase_timings:
            writer.writerow(
                {
                    "phase_key": r.phase_key,
                    "phase_name": r.phase_name,
                    "start_wall_utc": datetime.fromtimestamp(
                        r.start_wall_utc, tz=timezone.utc
                    ).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                    "elapsed_s": f"{r.elapsed_s:.3f}",
                }
            )
        if self._phase_timings:
            writer.writerow(
                {
                    "phase_key": "TOTAL",
                    "phase_name": "TOTAL",
                    "start_wall_utc": datetime.fromtimestamp(
                        self._phase_timings[0].start_wall_utc,
                        tz=timezone.utc,
                    ).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
                    "elapsed_s": f"{total_s:.3f}",
                }
            )

    # Print to console.
    total_s = sum(r.elapsed_s for r in self._phase_timings)
    table = Table(title="Tuning Pipeline — Timing Report", show_footer=True)
    table.add_column("Phase", footer="[bold]TOTAL[/]")
    table.add_column(
        "Duration",
        footer=f"[bold]{self._format_duration(total_s)}[/]",
        justify="right",
    )
    table.add_column("Wall start (UTC)", footer="", style="dim")
    table.add_column("% of total", footer="[bold]100%[/]", justify="right")

    for r in self._phase_timings:
        duration = self._format_duration(r.elapsed_s)
        wall_str = datetime.fromtimestamp(
            r.start_wall_utc, tz=timezone.utc
        ).strftime("%H:%M:%S")
        pct = f"{r.elapsed_s / total_s * 100:.0f}%" if total_s > 0 else "–"
        table.add_row(r.phase_name, duration, wall_str, pct)

    console.print()
    console.rule("[bold cyan]Timing Report")
    console.print(table)
    console.print(f"  Timing report saved to [bold]{csv_path}[/]")