Function parse_usage_metrics

uncommon_route/cache_support.py:204–313

(
    content: bytes,
    model: str,
    pricing: dict[str, ModelPricing],
)

def parse_usage_metrics(
    content: bytes,
    model: str,
    pricing: dict[str, ModelPricing],
) -> UsageMetrics | None:
    try:
        data = json.loads(content)
    except (json.JSONDecodeError, TypeError):
        return None

    usage = data.get("usage") or {}
    if not isinstance(usage, dict):
        return None

    prompt_details = usage.get("prompt_tokens_details") or usage.get("input_tokens_details") or {}
    output_details = usage.get("completion_tokens_details") or usage.get("output_tokens_details") or {}

    raw_prompt_tokens = _as_int(usage.get("prompt_tokens", usage.get("input_tokens", 0)))
    output_tokens = _as_int(usage.get("completion_tokens", usage.get("output_tokens", 0)))
    total_tokens = _as_int(usage.get("total_tokens", 0))

    prompt_cache_hit = _as_int(
        usage.get("prompt_cache_hit_tokens", prompt_details.get("prompt_cache_hit_tokens", 0)),
    )
    prompt_cache_miss = _as_int(
        usage.get("prompt_cache_miss_tokens", prompt_details.get("prompt_cache_miss_tokens", 0)),
    )
    cache_read_input_tokens = _first_positive_int(
        usage.get("cache_read_input_tokens"),
        prompt_details.get("cache_read_input_tokens"),
        usage.get("cached_tokens"),
        prompt_details.get("cached_tokens"),
        prompt_cache_hit,
    )
    cache_write_input_tokens = _first_positive_int(
        usage.get("cache_creation_input_tokens"),
        prompt_details.get("cache_creation_input_tokens"),
        usage.get("cache_write_tokens"),
        prompt_details.get("cache_write_tokens"),
    )

    if prompt_cache_hit or prompt_cache_miss:
        input_tokens_uncached = prompt_cache_miss
        input_tokens_total = prompt_cache_hit + prompt_cache_miss
    elif "input_tokens" in usage and "prompt_tokens" not in usage:
        input_tokens_uncached = raw_prompt_tokens
        input_tokens_total = raw_prompt_tokens + cache_read_input_tokens + cache_write_input_tokens
    else:
        input_tokens_total = max(raw_prompt_tokens, raw_prompt_tokens + cache_write_input_tokens)
        input_tokens_uncached = max(raw_prompt_tokens - cache_read_input_tokens, 0)

    if total_tokens <= 0:
        total_tokens = input_tokens_total + output_tokens

    if (
        input_tokens_total <= 0
        and input_tokens_uncached <= 0
        and output_tokens <= 0