LibreMail/scripts/device-testing/report.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""report.py -- aggregate scenario samples and render ``timing-tables.md``.

Pure and testable: given the samples a scenario collected, it produces the same
per-scenario markdown tables + aggregates as the hand-written ``timing-tables.md`` from the
manual run. No device or I/O dependency beyond writing the final file.
"""

from __future__ import annotations

import statistics
from dataclasses import dataclass
from typing import List, Optional, Sequence

from breadcrumbs import OpenSample

# The floor below which adb uiautomator-dump timings cannot resolve an in-app transition
# (see perf_summary.md: nav/back were dominated by the ~2.5-3 s dump latency).
UIAUTOMATOR_LATENCY_FLOOR_MS = 3000


# --------------------------------------------------------------------------- #
# Stats + markdown helpers
# --------------------------------------------------------------------------- #
@dataclass
class Aggregate:
    n: int
    mean: Optional[float]
    median: Optional[float]
    minimum: Optional[int]
    maximum: Optional[int]


def aggregate(values: Sequence[float]) -> Aggregate:
    """Summarise a list of numbers; safe on an empty list."""
    vals = [v for v in values if v is not None]
    if not vals:
        return Aggregate(0, None, None, None, None)
    return Aggregate(
        n=len(vals),
        mean=statistics.fmean(vals),
        median=statistics.median(vals),
        minimum=min(vals),
        maximum=max(vals),
    )


def md_table(headers: Sequence[str], rows: Sequence[Sequence[object]]) -> str:
    """Render a GitHub-flavoured markdown table."""
    head = "| " + " | ".join(str(h) for h in headers) + " |"
    sep = "| " + " | ".join("---" for _ in headers) + " |"
    body = [
        "| " + " | ".join("" if c is None else str(c) for c in row) + " |" for row in rows
    ]
    return "\n".join([head, sep, *body])


def _ms(value: Optional[float]) -> str:
    return "" if value is None else f"{round(value)}"


# --------------------------------------------------------------------------- #
# Cold open
# --------------------------------------------------------------------------- #
@dataclass
class ColdOpenSample:
    run: int
    total_time_ms: Optional[int]
    wait_time_ms: Optional[int]
    note: str = ""


def render_cold_open(samples: Sequence[ColdOpenSample]) -> str:
    rows: List[Sequence[object]] = []
    for s in samples:
        rows.append([s.run, _ms(s.total_time_ms), _ms(s.wait_time_ms), s.note])
    total_agg = aggregate([s.total_time_ms for s in samples if s.total_time_ms is not None])
    wait_agg = aggregate([s.wait_time_ms for s in samples if s.wait_time_ms is not None])
    rows.append(
        [
            "**mean**",
            f"**{_ms(total_agg.mean)}**",
            f"**{_ms(wait_agg.mean)}**",
            f"median {_ms(total_agg.median)} / {_ms(wait_agg.median)}",
        ]
    )
    table = md_table(["Run", "TotalTime (ms)", "WaitTime (ms)", "note"], rows)
    return (
        "## Cold open (am start -W, cache cleared each run)\n\n"
        + table
        + "\n\nColdest run is the first post-clear launch (class-load/JIT); steady state is lower.\n"
    )


# --------------------------------------------------------------------------- #
# Message open (reader body-load)
# --------------------------------------------------------------------------- #
@dataclass
class ReaderOpenRow:
    """A flattened, render-ready reader-open row (mirrors a timing-tables.md line)."""

    index: int
    label: str
    account_ref: str = ""
    cached: bool = False
    took_ms: Optional[int] = None
    reader_ready_ms: Optional[int] = None
    connect_ms: Optional[int] = None
    work_ms: Optional[int] = None
    select_ms: Optional[int] = None
    body_ms: Optional[int] = None
    flag_ms: Optional[int] = None
    live: Optional[int] = None
    rfc822_bytes: Optional[int] = None
    body_kb_per_s: Optional[float] = None
    skipped: bool = False
    reason: str = ""

    @classmethod
    def from_open_sample(
        cls, sample: OpenSample, index: int, label: str
    ) -> "ReaderOpenRow":
        bf = sample.body_fetch
        op = sample.body_fetch_op
        return cls(
            index=index,
            label=label,
            account_ref=sample.account_ref,
            cached=sample.cached,
            took_ms=sample.took_ms,
            reader_ready_ms=sample.reader_ready.took_ms if sample.reader_ready else None,
            connect_ms=op.connect_ms if op else None,
            work_ms=op.work_ms if op else None,
            select_ms=bf.select_ms if bf else None,
            body_ms=bf.body_ms if bf else None,
            flag_ms=bf.flag_ms if bf else None,
            live=op.live if op else None,
            rfc822_bytes=bf.rfc822_bytes if bf else None,
            body_kb_per_s=sample.body_kb_per_s,
        )


_OPEN_HEADERS = [
    "#",
    "message",
    "cached",
    "rfc822 B",
    "openMessage took",
    "reader ready",
    "connect",
    "work",
    "select",
    "body-dl",
    "flag",
    "live",
    "body KB/s",
]


def _open_row_cells(row: ReaderOpenRow) -> Sequence[object]:
    if row.skipped:
        return [
            row.index,
            row.label,
            "-",
            f"SKIPPED: {row.reason}",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
        ]
    kbps = "" if row.body_kb_per_s is None else f"{row.body_kb_per_s:.1f}"
    return [
        row.index,
        row.label,
        "yes" if row.cached else "no",
        row.rfc822_bytes if row.rfc822_bytes is not None else "",
        _fmt_ms(row.took_ms),
        _fmt_ms(row.reader_ready_ms),
        _ms(row.connect_ms),
        _ms(row.work_ms),
        _ms(row.select_ms),
        _ms(row.body_ms),
        _ms(row.flag_ms),
        _ms(row.live),
        kbps,
    ]


def _fmt_ms(value: Optional[int]) -> str:
    return "" if value is None else f"{value} ms"


def render_message_open(title: str, rows: Sequence[ReaderOpenRow]) -> str:
    table = md_table(_OPEN_HEADERS, [_open_row_cells(r) for r in rows])
    uncached = [r for r in rows if not r.skipped and not r.cached and r.took_ms is not None]
    agg = aggregate([r.took_ms for r in uncached])
    lines = [f"## {title}", "", table, ""]
    if agg.n:
        lines.append(
            f"Uncached opens: n={agg.n}, openMessage median "
            f"{_ms(agg.median)} ms, mean {_ms(agg.mean)} ms, "
            f"range {agg.minimum}-{agg.maximum} ms."
        )
    skipped = [r for r in rows if r.skipped]
    if skipped:
        lines.append(f"Skipped samples: {len(skipped)} (see rows above).")
    return "\n".join(lines) + "\n"


# --------------------------------------------------------------------------- #
# Back navigation
# --------------------------------------------------------------------------- #
@dataclass
class BackNavSample:
    index: int
    back_ms: Optional[int]
    note: str = ""


def render_back_nav(samples: Sequence[BackNavSample]) -> str:
    rows = [[s.index, _fmt_ms(s.back_ms), s.note] for s in samples]
    agg = aggregate([s.back_ms for s in samples if s.back_ms is not None])
    table = md_table(["#", "back (reader->mailbox)", "note"], rows)
    caveat = (
        f"\n\n**Caveat:** these are dominated by the ~{UIAUTOMATOR_LATENCY_FLOOR_MS} ms "
        "uiautomator-dump latency floor; true in-app back is sub-second and not precisely "
        "measurable via adb UI polling under load (see perf_summary.md)."
    )
    summary = "" if not agg.n else f"\n\nBack: n={agg.n}, mean {_ms(agg.mean)} ms, median {_ms(agg.median)} ms."
    return "## Reader -> mailbox (back)\n\n" + table + summary + caveat + "\n"


# --------------------------------------------------------------------------- #
# Document assembly
# --------------------------------------------------------------------------- #
def render_header(metadata: dict) -> str:
    lines = ["# LibreMail device perf run", ""]
    for key, value in metadata.items():
        lines.append(f"- **{key}:** {value}")
    lines.append("")
    return "\n".join(lines)


def build_document(metadata: dict, sections: Sequence[str]) -> str:
    parts = [render_header(metadata), *sections]
    return "\n".join(parts).rstrip() + "\n"