palladium/calculators/Genesys_Token_Calculator/genesyscalc/usage.py

"""Volumes → metered units → tokens.

Three published rules are modelled here that the obvious arithmetic gets
wrong. Each one is traceable to a sentence in the tokens-model article:

1. **The 15-second per-call round-up.** "Genesys rounds up each call to the
   next 15-second increment." Applied per call and *then* summed — averaging
   first and rounding once understates the bill, badly, on short bot touches.

2. **The highest-tier rule.** "Genesys bases charges on the highest tier
   (price) resource that Genesys Cloud uses during the interaction." So the
   AI-resource shares of a volume are a **partition**, not overlapping slices:
   an interaction is charged once, at the highest tier it touched.

3. **Copilot covers Supervisor summaries.** "if you enable Agent Copilot
   simultaneously, then Supervisor Copilot summaries and insights do not
   consume tokens." Billing AI Summary while Copilot is on double-charges.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field

from .meters import LicenceModel, MeterBasis, Tier
from .ratecard import VOICE_BOT_ROUNDUP_SECONDS, meter

#: Meters covered by the Copilot exclusion when Agent Copilot is enabled.
COPILOT_COVERED_METERS: frozenset[str] = frozenset({"ai_summary_and_insights"})

#: Which per-interaction meter prices each tier of the partition.
TIER_METER: dict[Tier, str] = {
    Tier.VIRTUAL_AGENT: "virtual_agent",
    Tier.AGENTIC_VIRTUAL_AGENT: "agentic_virtual_agent",
}


@dataclass(frozen=True)
class VoiceBotUsage:
    """Voice-bot activity, in the shape the round-up needs.

    The round-up is per *call*, so a call count is required — an aggregate
    minute total cannot be billed correctly.
    """

    calls_per_month: float
    avg_bot_seconds_per_call: float

    def __post_init__(self) -> None:
        if self.calls_per_month < 0:
            raise ValueError("calls_per_month must not be negative")
        if self.avg_bot_seconds_per_call < 0:
            raise ValueError("avg_bot_seconds_per_call must not be negative")


@dataclass(frozen=True)
class DeflectionMix:
    """Shares of interactions by the **highest tier** each one touched.

    This is a partition, not a set of overlapping rates. ``bot_only_share``
    is the share whose highest resource was a bot flow; an interaction that
    escalated bot → virtual agent counts once, in ``virtual_agent_share``.
    The shares therefore sum to at most 1.0 — the remainder went straight to
    an agent and consumed no AI-resource token.

    Modelling these as independent percentages (and summing the tokens) is
    the over-billing bug the published rule exists to prevent.
    """

    bot_only_share: float = 0.0
    virtual_agent_share: float = 0.0
    agentic_va_share: float = 0.0

    def __post_init__(self) -> None:
        for name, value in (
            ("bot_only_share", self.bot_only_share),
            ("virtual_agent_share", self.virtual_agent_share),
            ("agentic_va_share", self.agentic_va_share),
        ):
            if not 0.0 <= value <= 1.0:
                raise ValueError(f"{name} must be in [0, 1], got {value}")
        if self.total_deflected_share > 1.0 + 1e-9:
            raise ValueError(
                "tier shares must sum to at most 1.0 — they partition the "
                "volume by highest tier touched, they do not overlap "
                f"(got {self.total_deflected_share})"
            )

    @property
    def total_deflected_share(self) -> float:
        """Share of interactions that touched any AI resource."""
        return self.bot_only_share + self.virtual_agent_share + self.agentic_va_share

    @property
    def agent_handled_share(self) -> float:
        """Share that reached a human without touching an AI resource."""
        return 1.0 - self.total_deflected_share


@dataclass(frozen=True)
class Volumes:
    """Monthly interaction volumes by channel. Client data — from the
    notebook's ``engagement-data`` cell, never invented by the engine."""

    voice_inbound_monthly: float = 0.0
    voice_outbound_monthly: float = 0.0
    digital_sessions_monthly: float = 0.0
    email_monthly: float = 0.0
    messaging_monthly: float = 0.0
    social_posts_monthly: float = 0.0
    social_responses_monthly: float = 0.0
    translations_monthly: float = 0.0
    evaluations_monthly: float = 0.0
    ai_actions_monthly: float = 0.0
    routed_interactions_monthly: float = 0.0

    def __post_init__(self) -> None:
        for name, value in vars(self).items():
            if value < 0:
                raise ValueError(f"{name} must not be negative")

    @property
    def voice_total_monthly(self) -> float:
        return self.voice_inbound_monthly + self.voice_outbound_monthly

    @property
    def all_interactions_monthly(self) -> float:
        return (
            self.voice_total_monthly
            + self.digital_sessions_monthly
            + self.email_monthly
            + self.messaging_monthly
        )


def voice_bot_billable_minutes(
    usage: VoiceBotUsage, roundup_seconds: int = VOICE_BOT_ROUNDUP_SECONDS
) -> float:
    """Billable bot minutes per month, with the published per-call round-up.

    "Genesys rounds up each call to the next 15-second increment." The
    round-up lands on every call, so it is applied first and summed after::

        per_call = ceil(seconds / 15) * 15 / 60
        total    = per_call * calls

    A 40-second average bills as 45 seconds — 0.75 min/call, not 0.667. The
    penalty is worst on short touches: a 5-second bot greeting bills as 15
    seconds, three times its duration.
    """
    if roundup_seconds <= 0:
        raise ValueError("roundup_seconds must be positive")
    increments = math.ceil(usage.avg_bot_seconds_per_call / roundup_seconds)
    billable_seconds_per_call = increments * roundup_seconds
    return billable_seconds_per_call / 60.0 * usage.calls_per_month


def voice_bot_roundup_uplift(
    usage: VoiceBotUsage, roundup_seconds: int = VOICE_BOT_ROUNDUP_SECONDS
) -> float:
    """How much the round-up adds, as a fraction of the raw duration.

    The number worth saying out loud on stage: at a 40-second average it is
    0.125 (12.5% more bot cost than the raw minutes suggest).
    """
    raw_minutes = usage.avg_bot_seconds_per_call / 60.0 * usage.calls_per_month
    if raw_minutes == 0:
        return 0.0
    return voice_bot_billable_minutes(usage, roundup_seconds) / raw_minutes - 1.0


def voice_bot_tokens(
    usage: VoiceBotUsage, roundup_seconds: int = VOICE_BOT_ROUNDUP_SECONDS
) -> float:
    """Voice-bot tokens per month: billable minutes ÷ 17."""
    minutes = voice_bot_billable_minutes(usage, roundup_seconds)
    return minutes * meter("bots_voice").rate


def tier_interactions(volume: float, mix: DeflectionMix) -> dict[Tier, float]:
    """Split ``volume`` across tiers by highest resource touched.

    A partition — the returned values sum to at most ``volume``, so no
    interaction can be billed twice.
    """
    if volume < 0:
        raise ValueError("volume must not be negative")
    return {
        Tier.BOT: volume * mix.bot_only_share,
        Tier.VIRTUAL_AGENT: volume * mix.virtual_agent_share,
        Tier.AGENTIC_VIRTUAL_AGENT: volume * mix.agentic_va_share,
    }


def virtual_agent_tokens(volume: float, mix: DeflectionMix) -> dict[str, float]:
    """Tokens for the VA and Agentic VA tiers of a deflection partition.

    The bot tier is deliberately absent: voice bots are metered in minutes
    with a per-call round-up, so :func:`voice_bot_tokens` owns that line.
    """
    split = tier_interactions(volume, mix)
    return {
        TIER_METER[tier]: meter(TIER_METER[tier]).tokens_for(interactions)
        for tier, interactions in split.items()
        if tier in TIER_METER
    }


@dataclass(frozen=True)
class FeatureUsage:
    """Metered units per meter key, plus the notes explaining any zeroing."""

    units: dict[str, float] = field(default_factory=dict)
    notes: tuple[str, ...] = ()

    def get(self, key: str) -> float:
        return self.units.get(key, 0.0)


def apply_copilot_covers_summary(
    units: dict[str, float], copilot_enabled: bool
) -> FeatureUsage:
    """Zero the summary meters when Agent Copilot is enabled.

    "if you enable Agent Copilot simultaneously, then Supervisor Copilot
    summaries and insights do not consume tokens." The zeroing is reported,
    not silent — a number that shrank for a published reason should say so.
    """
    if not copilot_enabled:
        return FeatureUsage(units=dict(units))

    adjusted = dict(units)
    notes: list[str] = []
    for key in sorted(COPILOT_COVERED_METERS):
        covered = adjusted.get(key, 0.0)
        if covered > 0:
            adjusted[key] = 0.0
            notes.append(
                f"{meter(key).feature}: {covered:,.0f} units not billed — "
                "Agent Copilot is enabled, and Genesys does not charge "
                "Supervisor summaries and insights alongside it."
            )
    return FeatureUsage(units=adjusted, notes=tuple(notes))


def subscription_tokens(
    users: int, enabled_per_user_keys: frozenset[str], licence: LicenceModel
) -> dict[str, float]:
    """Per-user subscription tokens per month, by meter key.

    Exact — per-user totals are not rounded; only consumption is (see
    :mod:`genesyscalc.billing`).
    """
    if users < 0:
        raise ValueError("users must not be negative")
    out: dict[str, float] = {}
    for key in sorted(enabled_per_user_keys):
        m = meter(key)
        if m.basis is not MeterBasis.PER_USER_PER_MONTH:
            raise ValueError(f"{key} is not a per-user meter")
        out[key] = users * m.tokens_per_user_month(licence)
    return out