271 lines
9.4 KiB
Python
271 lines
9.4 KiB
Python
"""Orchestration — one call in, the whole costed picture out.
|
||
|
||
The notebook assembles :class:`CalculatorInputs` from its ``engagement-data``
|
||
cell and its widgets, calls :func:`calculate`, and renders the result. The
|
||
engine never invents a volume and never reads a widget.
|
||
|
||
:attr:`CalculatorResult.warnings` is deliberate: when a published rule makes
|
||
a number smaller — Copilot covering summaries, the allowance absorbing a
|
||
month, deflection zeroing TTS — the result says so. A total that shrank for
|
||
a reason nobody can see is indistinguishable from a bug.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field, replace
|
||
from typing import Any
|
||
|
||
from .billing import CostLine, CostTotals, cost_lines, total_cost
|
||
from .meters import LicenceModel, MeterBasis, TokenPrice
|
||
from .ratecard import meter
|
||
from .tts import TtsLine, TtsUsage, tts_cost
|
||
from .usage import (
|
||
DeflectionMix,
|
||
VoiceBotUsage,
|
||
Volumes,
|
||
apply_copilot_covers_summary,
|
||
subscription_tokens,
|
||
virtual_agent_tokens,
|
||
voice_bot_billable_minutes,
|
||
voice_bot_roundup_uplift,
|
||
voice_bot_tokens,
|
||
)
|
||
|
||
#: Meter keys whose units come straight off a channel volume. The notebook's
|
||
#: feature catalogue decides which are switched on; this maps each to the
|
||
#: volume that drives it.
|
||
_VOLUME_DRIVEN: dict[str, str] = {
|
||
"bots_digital": "digital_sessions_monthly",
|
||
"ai_summary_and_insights": "voice_total_monthly",
|
||
"ai_scoring": "evaluations_monthly",
|
||
"ai_translate": "translations_monthly",
|
||
"predictive_routing": "routed_interactions_monthly",
|
||
"genesys_cloud_copilot": "ai_actions_monthly",
|
||
"apple_messages_for_business": "messaging_monthly",
|
||
"facebook_messenger": "messaging_monthly",
|
||
"instagram_direct_messaging": "messaging_monthly",
|
||
"whatsapp_messaging": "messaging_monthly",
|
||
"x_direct_messaging": "messaging_monthly",
|
||
"genesys_cloud_social": "social_posts_monthly",
|
||
"social_post_responses": "social_responses_monthly",
|
||
"predictive_engagement": "all_interactions_monthly",
|
||
"knowledge_queries": "all_interactions_monthly",
|
||
}
|
||
|
||
#: The per-user subscription meters, by licence-model face.
|
||
_PER_USER_KEYS: frozenset[str] = frozenset(
|
||
{
|
||
"agent_copilot_named",
|
||
"agent_copilot_concurrent",
|
||
"speech_and_text_analytics_named",
|
||
"speech_and_text_analytics_concurrent",
|
||
}
|
||
)
|
||
|
||
#: Deflection-driven meters — priced from the highest-tier partition, not a
|
||
#: raw volume.
|
||
_TIER_KEYS: frozenset[str] = frozenset({"virtual_agent", "agentic_virtual_agent"})
|
||
|
||
#: The Copilot family, in either licence face.
|
||
_COPILOT_KEYS: frozenset[str] = frozenset(
|
||
{"agent_copilot_named", "agent_copilot_concurrent"}
|
||
)
|
||
|
||
|
||
def _volume_for(volumes: Volumes, attr: str) -> float:
|
||
return float(getattr(volumes, attr))
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CalculatorInputs:
|
||
"""Everything the notebook supplies. Client data plus widget state."""
|
||
|
||
volumes: Volumes
|
||
users: int
|
||
licence: LicenceModel = LicenceModel.NAMED
|
||
enabled: frozenset[str] = frozenset()
|
||
mix: DeflectionMix = field(default_factory=DeflectionMix)
|
||
price: TokenPrice = field(default_factory=TokenPrice)
|
||
voice_bot: VoiceBotUsage | None = None
|
||
tts: TtsUsage | None = None
|
||
apply_allowance: bool = True
|
||
|
||
def __post_init__(self) -> None:
|
||
if self.users < 0:
|
||
raise ValueError("users must not be negative")
|
||
unknown = sorted(k for k in self.enabled if k not in _known_keys())
|
||
if unknown:
|
||
raise ValueError(
|
||
f"enabled contains keys that are not published meters: {unknown}"
|
||
)
|
||
|
||
@property
|
||
def copilot_enabled(self) -> bool:
|
||
return bool(self.enabled & _COPILOT_KEYS)
|
||
|
||
def with_(self, **changes: Any) -> CalculatorInputs:
|
||
"""A copy with fields replaced — for scenarios and sensitivity."""
|
||
return replace(self, **changes)
|
||
|
||
|
||
def _known_keys() -> frozenset[str]:
|
||
from .ratecard import METERS
|
||
|
||
return frozenset(METERS)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CalculatorResult:
|
||
"""The costed picture, plus why any number shrank."""
|
||
|
||
lines: tuple[CostLine, ...]
|
||
totals: CostTotals
|
||
tts: TtsLine | None
|
||
grand_total_annual: float
|
||
warnings: tuple[str, ...]
|
||
|
||
@property
|
||
def token_cost_annual(self) -> float:
|
||
return self.totals.token_cost_annual
|
||
|
||
@property
|
||
def tts_cost_annual(self) -> float:
|
||
return self.tts.cost_annual if self.tts is not None else 0.0
|
||
|
||
|
||
def metered_units(inputs: CalculatorInputs) -> dict[str, float]:
|
||
"""Monthly metered units per enabled meter key, before the Copilot rule."""
|
||
units: dict[str, float] = {}
|
||
|
||
for key in sorted(inputs.enabled):
|
||
if key in _PER_USER_KEYS:
|
||
units[key] = float(inputs.users)
|
||
elif key in _TIER_KEYS:
|
||
share = (
|
||
inputs.mix.virtual_agent_share
|
||
if key == "virtual_agent"
|
||
else inputs.mix.agentic_va_share
|
||
)
|
||
units[key] = inputs.volumes.voice_total_monthly * share
|
||
elif key == "bots_voice":
|
||
units[key] = (
|
||
voice_bot_billable_minutes(inputs.voice_bot)
|
||
if inputs.voice_bot is not None
|
||
else 0.0
|
||
)
|
||
elif key in _VOLUME_DRIVEN:
|
||
units[key] = _volume_for(inputs.volumes, _VOLUME_DRIVEN[key])
|
||
else: # pragma: no cover — every published key is classified above
|
||
raise ValueError(f"{key} has no unit driver")
|
||
|
||
return units
|
||
|
||
|
||
def calculate(inputs: CalculatorInputs) -> CalculatorResult:
|
||
"""Price a scenario end to end."""
|
||
raw_units = metered_units(inputs)
|
||
usage = apply_copilot_covers_summary(raw_units, inputs.copilot_enabled)
|
||
units = usage.units
|
||
warnings: list[str] = list(usage.notes)
|
||
|
||
tokens: dict[str, float] = {}
|
||
consumption = 0.0
|
||
|
||
# Per-user subscription meters — exact, never rounded.
|
||
per_user_enabled = frozenset(inputs.enabled & _PER_USER_KEYS)
|
||
subs = subscription_tokens(inputs.users, per_user_enabled, inputs.licence)
|
||
tokens.update(subs)
|
||
subscription = sum(subs.values())
|
||
|
||
# Voice bots — minutes with the published per-call round-up.
|
||
if "bots_voice" in inputs.enabled and inputs.voice_bot is not None:
|
||
bot_tokens = voice_bot_tokens(inputs.voice_bot)
|
||
tokens["bots_voice"] = bot_tokens
|
||
consumption += bot_tokens
|
||
uplift = voice_bot_roundup_uplift(inputs.voice_bot)
|
||
if uplift > 0:
|
||
warnings.append(
|
||
f"Voice bots: the published 15-second per-call round-up adds "
|
||
f"{uplift:.1%} to billable bot minutes at a "
|
||
f"{inputs.voice_bot.avg_bot_seconds_per_call:.0f}s average."
|
||
)
|
||
|
||
# Virtual agents — the highest-tier partition of voice volume.
|
||
tier_tokens = virtual_agent_tokens(
|
||
inputs.volumes.voice_total_monthly, inputs.mix
|
||
)
|
||
for key in sorted(_TIER_KEYS & inputs.enabled):
|
||
tokens[key] = tier_tokens.get(key, 0.0)
|
||
consumption += tokens[key]
|
||
|
||
# Everything else — units × published rate.
|
||
for key in sorted(inputs.enabled):
|
||
if key in _PER_USER_KEYS or key in _TIER_KEYS or key == "bots_voice":
|
||
continue
|
||
m = meter(key)
|
||
value = m.tokens_for(units.get(key, 0.0))
|
||
tokens[key] = value
|
||
consumption += value
|
||
if m.basis is MeterBasis.ZERO_RATED and units.get(key, 0.0) > 0:
|
||
warnings.append(
|
||
f"{m.feature}: {m.published_rate} — shown at $0 because "
|
||
"Genesys publishes it as included."
|
||
)
|
||
|
||
totals = total_cost(
|
||
consumption, subscription, inputs.licence, inputs.price, inputs.apply_allowance
|
||
)
|
||
if inputs.apply_allowance and totals.billable_tokens_monthly == 0:
|
||
warnings.append(
|
||
f"The free monthly allowance ({totals.allowance_tokens_monthly} "
|
||
f"tokens, {inputs.licence.label.lower()} org) covers this month's "
|
||
"consumption entirely."
|
||
)
|
||
|
||
tts_line = (
|
||
tts_cost(inputs.tts, inputs.mix, inputs.price.concession_pct)
|
||
if inputs.tts is not None
|
||
else None
|
||
)
|
||
if tts_line is not None:
|
||
warnings.extend(tts_line.warnings)
|
||
|
||
lines = cost_lines(tokens, units, inputs.price)
|
||
grand_total = totals.token_cost_annual + (
|
||
tts_line.cost_annual if tts_line is not None else 0.0
|
||
)
|
||
|
||
return CalculatorResult(
|
||
lines=lines,
|
||
totals=totals,
|
||
tts=tts_line,
|
||
grand_total_annual=grand_total,
|
||
warnings=tuple(warnings),
|
||
)
|
||
|
||
|
||
def cost_per_interaction(result: CalculatorResult, volumes: Volumes) -> float:
|
||
"""Annual grand total ÷ annual interactions — the number clients recall."""
|
||
annual = volumes.all_interactions_monthly * 12
|
||
if annual == 0:
|
||
return 0.0
|
||
return result.grand_total_annual / annual
|
||
|
||
|
||
def compare(scenarios: dict[str, CalculatorInputs]) -> list[dict[str, Any]]:
|
||
"""One row per scenario. Cost only — a calculator does not build a case."""
|
||
rows: list[dict[str, Any]] = []
|
||
for name, inputs in scenarios.items():
|
||
result = calculate(inputs)
|
||
rows.append(
|
||
{
|
||
"Scenario": name,
|
||
"Billable tokens / mo": result.totals.billable_tokens_monthly,
|
||
"Token cost / yr": result.totals.token_cost_annual,
|
||
"TTS cost / yr": result.tts_cost_annual,
|
||
"Total / yr": result.grand_total_annual,
|
||
"$ / interaction": cost_per_interaction(result, inputs.volumes),
|
||
}
|
||
)
|
||
return rows
|