<!-- GENERATED by .ai/scripts/sync_agent_assets.py from .ai/skills-src/design-scalable-systems/scripts/capacity_cost_model.py. DO NOT EDIT. -->

#!/usr/bin/env python3
"""Deterministic, dependency-free capacity and cost arithmetic."""

from __future__ import annotations

import argparse
import hashlib
import json
import math
from pathlib import Path
from typing import Any

MONTH_SECONDS = 30 * 24 * 60 * 60
DAY_SECONDS = 24 * 60 * 60


def number(data: dict[str, Any], key: str) -> float | None:
    value = data.get(key)
    if value is None:
        return None
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:
        raise ValueError(f"{key} must be a finite non-negative number")
    return float(value)


def rounded(value: float | None) -> float | None:
    return None if value is None else round(value, 6)


def model(source: dict[str, Any]) -> dict[str, Any]:
    if source.get("schema_version") != 1:
        raise ValueError("schema_version must be 1")
    workload = source.get("workload") or {}
    capacity = source.get("capacity") or {}
    average_rps = number(workload, "average_rps")
    peak_rps = number(workload, "peak_rps")
    latency_ms = number(workload, "average_service_time_ms")
    request_bytes = number(workload, "request_bytes")
    response_bytes = number(workload, "response_bytes")
    write_rps = number(workload, "write_rps")
    stored_bytes = number(workload, "stored_bytes_per_write")
    replication_value = number(workload, "storage_replication_factor")
    replication = 1.0 if replication_value is None else replication_value
    safe_rps = number(capacity, "tested_safe_rps_per_replica")
    headroom_value = number(capacity, "headroom_factor")
    headroom = 1.25 if headroom_value is None else headroom_value
    min_replicas_value = number(capacity, "minimum_replicas")
    min_replicas = 1 if min_replicas_value is None else int(min_replicas_value)
    if average_rps is not None and peak_rps is not None and average_rps > peak_rps:
        raise ValueError("average_rps cannot exceed peak_rps")
    if headroom < 1:
        raise ValueError("headroom_factor must be at least 1")
    if replication < 1:
        raise ValueError("storage_replication_factor must be at least 1")
    if min_replicas_value is not None and min_replicas_value != min_replicas:
        raise ValueError("minimum_replicas must be an integer")

    in_flight = None if peak_rps is None or latency_ms is None else peak_rps * latency_ms / 1000
    monthly_requests = None if average_rps is None else average_rps * MONTH_SECONDS
    ingress_bps = None if peak_rps is None or request_bytes is None else peak_rps * request_bytes
    egress_bps = None if peak_rps is None or response_bytes is None else peak_rps * response_bytes
    storage_day = None if write_rps is None or stored_bytes is None else write_rps * stored_bytes * DAY_SECONDS * replication
    replicas = None if peak_rps is None or safe_rps in (None, 0) else max(min_replicas, math.ceil(peak_rps * headroom / safe_rps))

    cost_items = source.get("cost_items") or []
    totals: dict[str, float] = {}
    priced = 0
    sourced = 0
    for item in cost_items:
        if not isinstance(item, dict):
            raise ValueError("cost_items must contain objects")
        quantity = number(item, "monthly_quantity")
        unit_price = number(item, "unit_price")
        currency = item.get("currency")
        if quantity is None or unit_price is None or not isinstance(currency, str) or not currency:
            continue
        totals[currency] = totals.get(currency, 0.0) + quantity * unit_price
        priced += 1
        required_evidence = ["provider", "service", "sku", "region", "unit", "source_url", "retrieved_at", "effective_date"]
        if all(isinstance(item.get(key), str) and item[key] for key in required_evidence):
            sourced += 1
    cost_status = "UNAVAILABLE" if not cost_items or priced == 0 else "LIVE_ESTIMATE" if priced == len(cost_items) and sourced == priced else "PARTIAL"

    result = {
        "schema_version": 1,
        "capacity": {
            "monthly_requests": rounded(monthly_requests),
            "peak_inflight_requests": rounded(in_flight),
            "peak_ingress_bytes_per_second": rounded(ingress_bps),
            "peak_egress_bytes_per_second": rounded(egress_bps),
            "stored_bytes_per_day_with_replication": rounded(storage_day),
            "required_replicas": replicas,
            "replica_status": "CALCULATED" if replicas is not None else "UNAVAILABLE",
        },
        "cost": {
            "status": cost_status,
            "monthly_totals": {key: round(value, 6) for key, value in sorted(totals.items())},
            "priced_items": priced,
            "sourced_items": sourced,
            "total_items": len(cost_items),
        },
        "missing_evidence": [
            label
            for condition, label in [
                (peak_rps is None, "peak_rps"),
                (latency_ms is None, "average_service_time_ms"),
                (safe_rps is None, "tested_safe_rps_per_replica"),
                (request_bytes is None, "request_bytes"),
                (response_bytes is None, "response_bytes"),
            ]
            if condition
        ],
    }
    result["model_hash"] = hashlib.sha256(json.dumps(result, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    return result


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("input", type=Path)
    args = parser.parse_args()
    root = Path.cwd().resolve()
    source_path = args.input.resolve()
    try:
        source_path.relative_to(root)
    except ValueError as exc:
        raise ValueError("input must remain inside the repository") from exc
    stat = source_path.lstat()
    if source_path.is_symlink() or not source_path.is_file() or stat.st_size > 1024 * 1024:
        raise ValueError("input must be a bounded regular repository file")
    source = json.loads(source_path.read_text(encoding="utf-8"))
    result = json.dumps(model(source), indent=2, sort_keys=True) + "\n"
    print(result, end="")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())