File size: 2,883 Bytes
bb2c701
 
 
 
 
 
 
 
 
c7f896e
bb2c701
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
"""
Cost Estimation Engine – Deterministic pricing with Bayesian uncertainty.
"""

import os
from functools import lru_cache
from typing import Dict, Optional, Union, Any
import yaml

from .intents import ProvisionResourceIntent, ResourceType

class CostEstimator:
    DEFAULT_PRICING = {
        ResourceType.VM: {
            "Standard_D2s_v3": 70.0,
            "Standard_D4s_v3": 140.0,
            "Standard_D8s_v3": 280.0,
            "Standard_D16s_v3": 560.0,
        },
        ResourceType.STORAGE_ACCOUNT: {
            "50GB": 5.0,
            "100GB": 10.0,
            "1TB": 100.0,
            "10TB": 900.0,
        },
        ResourceType.DATABASE: {
            "Basic": 15.0,
            "Standard": 50.0,
            "Premium": 200.0,
        },
        ResourceType.KUBERNETES_CLUSTER: {
            "Small": 100.0,
            "Medium": 300.0,
            "Large": 600.0,
        },
        ResourceType.FUNCTION_APP: {
            "Consumption": 0.0,
            "Premium": 75.0,
        },
        ResourceType.VIRTUAL_NETWORK: {
            "default": 0.0,
        },
    }

    def __init__(self, pricing_file: Optional[str] = None):
        if pricing_file and os.path.exists(pricing_file):
            with open(pricing_file, 'r') as f:
                raw = yaml.safe_load(f)
            self._pricing = {}
            for res_str, sizes in raw.items():
                try:
                    res_type = ResourceType(res_str)
                except ValueError:
                    continue
                self._pricing[res_type] = sizes
        else:
            self._pricing = self.DEFAULT_PRICING.copy()

    @lru_cache(maxsize=256)
    def estimate_monthly_cost(self, intent: ProvisionResourceIntent) -> Optional[float]:
        resource_pricing = self._pricing.get(intent.resource_type)
        if not resource_pricing:
            return None
        return resource_pricing.get(intent.size)

    def cost_delta_vs_baseline(
        self,
        intent: ProvisionResourceIntent,
        baseline_intent: Optional[ProvisionResourceIntent] = None,
    ) -> Optional[float]:
        proposed = self.estimate_monthly_cost(intent)
        if proposed is None:
            return None

        if baseline_intent:
            baseline = self.estimate_monthly_cost(baseline_intent)
            if baseline is None:
                return None
            return proposed - baseline
        else:
            resource_pricing = self._pricing.get(intent.resource_type)
            if not resource_pricing:
                return None
            min_cost = min(resource_pricing.values())
            return proposed - min_cost

    def estimate_cost_distribution(self, intent: ProvisionResourceIntent) -> Dict[str, float]:
        cost = self.estimate_monthly_cost(intent)
        if cost is None:
            return {}
        return {str(cost): 1.0}