Optimize inference

[project.entry-points."piloop.fact_sources"]
gpu_hours = "piloop_inference.gpu_hours:GpuHours"

"""``gpu_hours``: the GPU-hours that the campaign's evaluations held, and what they cost.

Each evaluation holds ``gpus`` GPUs for its wall time. ``hold_at = { gpu_usd = 50 }`` in ``[plugins.stop_rule]`` then
holds the campaign at a GPU budget.
"""

from datetime import datetime

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from piloop.capabilities import Fact, FactSource, FactValue
from piloop.db.models import Campaign
from piloop.plugins.campaign_type.kit.candidates.tables import Candidate
from piloop.plugins.campaign_type.kit.evaluation.tables import Measurement

GPU_HOURS = Fact[float]('gpu_hours', 'The GPU-hours that the evaluations of the campaign held.')
GPU_USD = Fact[float]('gpu_usd', 'What the GPU-hours of the evaluations cost, in USD.')


class GpuHours(FactSource):
    """The wall time of every evaluation of the campaign, times ``gpus``, and its price."""

    gpus: int = 1
    """The GPUs that one evaluation holds."""
    usd_per_gpu_hour: float
    """The price of one GPU for one hour."""

    def supplies(self) -> tuple[Fact[FactValue], ...]:
        return (GPU_HOURS, GPU_USD)

    def gather(
        self, s: Session, campaign: Campaign, wanted: frozenset[Fact[FactValue]], now: datetime
    ) -> dict[Fact[FactValue], FactValue]:
        mine = select(Candidate.id).where(Candidate.campaign_id == campaign.id)
        seconds = s.scalar(select(func.total(Measurement.seconds)).where(Measurement.candidate_id.in_(mine))) or 0.0
        hours = seconds * self.gpus / 3600
        found: dict[Fact[FactValue], FactValue] = {GPU_HOURS: hours, GPU_USD: hours * self.usd_per_gpu_hour}
        return {fact: value for fact, value in found.items() if fact in wanted}

[project.entry-points."piloop.verdict_rules"]
min_speedup = "piloop_inference.min_speedup:MinSpeedup"

"""``min_speedup``: the built-in verdict, but a keep must beat the bar by at least ``min_speedup`` of its value.

A throughput benchmark is noisy, and a gain of a few percent can be luck that three pairs do not rule out. Each keep
moves the head, and every later run must beat it, so a lucky keep costs the runs after it. The rule is ``bootstrap``
with one more test: its confidence, guardrails and reasons stay.
"""

from piloop.capabilities import Bootstrap, Case, Judgement, Outcome


class MinSpeedup(Bootstrap):
    """What ``bootstrap`` decides, but a keep that beats the bar by less than ``min_speedup`` is ``interesting``."""

    min_speedup: float
    """The smallest gain that keeps, as a fraction of the bar: ``0.05`` for 5%."""

    def decide(self, case: Case) -> Judgement:
        judged = super().decide(case)
        name = case.primary.name
        bar = case.bar.means.get(name)
        if judged.verdict is not Outcome.KEEP or not bar:
            return judged
        speedup = case.primary.direction.gain(case.candidate.means[name], bar) / abs(bar)
        if speedup >= self.min_speedup:
            return judged
        return Judgement(
            Outcome.INTERESTING, f'{judged.reason}; {speedup:.1%} is below min_speedup {self.min_speedup:.0%}'
        )

[project.entry-points."piloop.sections"]
latency = "piloop_inference.latency:Latency"

"""``latency``: a table of the latest candidates' latencies, in ``piloop status`` and in the implementers' briefs.

An implementer sees what its predecessors did to the tail latency, not only to the primary metric.
"""

from collections import defaultdict
from statistics import fmean

from sqlalchemy import select

from piloop.capabilities import Outcome, Rendered, Rendering, Section, Surface, Table
from piloop.plugins.campaign_type.kit.candidates.tables import Candidate
from piloop.plugins.campaign_type.kit.evaluation.tables import Measurement


class Latency(Section):
    """The mean of each of ``metrics`` over the full benchmark, for the latest ``rows`` candidates."""

    SHOWN_IN = frozenset({Surface.STATUS, Surface.RUN})
    ORDER = 175  # after the evaluations

    metrics: list[str] = ['p50_ms', 'p95_ms']
    """The metrics of the table, as the evaluate command reports them."""
    rows: int = 5
    """How many candidates, the latest first."""

    def render(self, r: Rendering) -> Rendered | None:
        full = (
            select(Candidate.id, Candidate.verdict, Measurement.metrics)
            .join(Measurement)
            .where(
                Candidate.campaign_id == r.campaign.id,
                Measurement.evaluator_id == r.campaign.evaluator_id,
                Measurement.rung.is_(None),
                Measurement.error.is_(None),
            )
        )
        samples: dict[tuple[int, Outcome | None], list[dict[str, float]]] = defaultdict(list)
        for candidate, verdict, metrics in r.s.execute(full.order_by(Candidate.id.desc())):
            samples[candidate, verdict].append(metrics)
        if not samples:
            return None
        table = Table(
            columns=('candidate', 'verdict', *self.metrics),
            rows=tuple(
                (f'#{candidate}', verdict.value if verdict else 'pending', *(_mean(each, m) for m in self.metrics))
                for (candidate, verdict), each in list(samples.items())[: self.rows]
            ),
        )
        return Rendered(title='Latency', blocks=(table,))


def _mean(samples: list[dict[str, float]], metric: str) -> float | None:
    values = [s[metric] for s in samples if metric in s]
    return round(fmean(values), 2) if values else None