Feature: Pricing Services

src/aleph/sdk/client/http.py

@@ -37,6 +37,7 @@
 from aleph.sdk.client.services.dns import DNS
 from aleph.sdk.client.services.instance import Instance
 from aleph.sdk.client.services.port_forwarder import PortForwarder
+from aleph.sdk.client.services.pricing import Pricing
 from aleph.sdk.client.services.scheduler import Scheduler
 
 from ..conf import settings
@@ -135,7 +136,7 @@ async def __aenter__(self):
         self.crn = Crn(self)
         self.scheduler = Scheduler(self)
         self.instance = Instance(self)
-
+        self.pricing = Pricing(self)
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):

src/aleph/sdk/client/services/crn.py

@@ -1,8 +1,9 @@
-from typing import TYPE_CHECKING, Dict, Optional, Union
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
 
 import aiohttp
 from aiohttp.client_exceptions import ClientResponseError
 from aleph_message.models import ItemHash
+from pydantic import BaseModel
 
 from aleph.sdk.conf import settings
 from aleph.sdk.exceptions import MethodNotAvailableOnCRN, VmNotFoundOnHost
@@ -13,6 +14,22 @@
     from aleph.sdk.client.http import AlephHttpClient
 
 
+class GPU(BaseModel):
+    vendor: str
+    model: str
+    device_name: str
+    device_class: str
+    pci_host: str
+    compatible: bool
+
+
+class NetworkGPUS(BaseModel):
+    total_gpu_count: int
+    available_gpu_count: int
+    available_gpu_list: dict[str, List[GPU]]  # str = node_url
+    used_gpu_list: dict[str, List[GPU]]  # str = node_url
+
+
 class Crn:
     """
     This services allow interact with CRNS API
@@ -136,3 +153,48 @@ async def update_instance_config(self, crn_address: str, item_hash: ItemHash):
             async with session.post(full_url) as resp:
                 resp.raise_for_status()
                 return await resp.json()
+
+    # Gpu Functions Helper
+    async def fetch_gpu_on_network(
+        self,
+        crn_list: Optional[List[dict]] = None,
+    ) -> NetworkGPUS:
+        if not crn_list:
+            crn_list = (await self._client.crn.get_crns_list()).get("crns", [])
+
+        gpu_count: int = 0
+        available_gpu_count: int = 0
+
+        compatible_gpu: Dict[str, List[GPU]] = {}
+        available_compatible_gpu: Dict[str, List[GPU]] = {}
+
+        # Ensure crn_list is a list before iterating
+        if not isinstance(crn_list, list):
+            crn_list = []
+
+        for crn_ in crn_list:
+            if not crn_.get("gpu_support", False):
+                continue
+
+            # Only process CRNs with GPU support
+            crn_address = crn_["address"]
+
+            # Extracts used GPU
+            for gpu in crn_.get("compatible_gpus", []):
+                compatible_gpu[crn_address] = []
+                compatible_gpu[crn_address].append(GPU.model_validate(gpu))
+                gpu_count += 1
+
+            # Extracts available GPU
+            for gpu in crn_.get("compatible_available_gpus", []):
+                available_compatible_gpu[crn_address] = []
+                available_compatible_gpu[crn_address].append(GPU.model_validate(gpu))
+                gpu_count += 1
+                available_gpu_count += 1
+
+        return NetworkGPUS(
+            total_gpu_count=gpu_count,
+            available_gpu_count=available_gpu_count,
+            used_gpu_list=compatible_gpu,
+            available_gpu_list=available_compatible_gpu,
+        )

src/aleph/sdk/client/services/pricing.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import TYPE_CHECKING, Dict, List, Optional, Union
+
+from aleph.sdk.client.services.base import BaseService
+
+if TYPE_CHECKING:
+    pass
+
+from decimal import Decimal
+
+from pydantic import BaseModel, RootModel
+
+
+class PricingEntity(str, Enum):
+    STORAGE = "storage"
+    WEB3_HOSTING = "web3_hosting"
+    PROGRAM = "program"
+    PROGRAM_PERSISTENT = "program_persistent"
+    INSTANCE = "instance"
+    INSTANCE_CONFIDENTIAL = "instance_confidential"
+    INSTANCE_GPU_STANDARD = "instance_gpu_standard"
+    INSTANCE_GPU_PREMIUM = "instance_gpu_premium"
+
+
+class GroupEntity(str, Enum):
+    STORAGE = "storage"
+    WEBSITE = "website"
+    PROGRAM = "program"
+    INSTANCE = "instance"
+    CONFIDENTIAL = "confidential"
+    GPU = "gpu"
+    ALL = "all"
+
+
+class Price(BaseModel):
+    payg: Optional[Decimal] = None
+    holding: Optional[Decimal] = None
+    fixed: Optional[Decimal] = None
+
+
+class ComputeUnit(BaseModel):
+    vcpus: int
+    memory_mib: int
+    disk_mib: int
+
+
+class Tier(BaseModel):
+    id: str
+    compute_units: int
+    vram: Optional[int] = None
+    model: Optional[str] = None
+
+
+class PricingPerEntity(BaseModel):
+    price: Dict[str, Union[Price, Decimal]]
+    compute_unit: Optional[ComputeUnit] = None
+    tiers: Optional[List[Tier]] = None
+
+
+class PricingModel(RootModel[Dict[PricingEntity, PricingPerEntity]]):
+    def __iter__(self):
+        return iter(self.root)
+
+    def __getitem__(self, item):
+        return self.root[item]
+
+
+PRICING_GROUPS: dict[str, list[PricingEntity]] = {
+    GroupEntity.STORAGE: [PricingEntity.STORAGE],
+    GroupEntity.WEBSITE: [PricingEntity.WEB3_HOSTING],
+    GroupEntity.PROGRAM: [PricingEntity.PROGRAM, PricingEntity.PROGRAM_PERSISTENT],
+    GroupEntity.INSTANCE: [PricingEntity.INSTANCE],
+    GroupEntity.CONFIDENTIAL: [PricingEntity.INSTANCE_CONFIDENTIAL],
+    GroupEntity.GPU: [
+        PricingEntity.INSTANCE_GPU_STANDARD,
+        PricingEntity.INSTANCE_GPU_PREMIUM,
+    ],
+    GroupEntity.ALL: list(PricingEntity),
+}
+
+PAYG_GROUP: list[PricingEntity] = [
+    PricingEntity.INSTANCE,
+    PricingEntity.INSTANCE_CONFIDENTIAL,
+    PricingEntity.INSTANCE_GPU_STANDARD,
+    PricingEntity.INSTANCE_GPU_PREMIUM,
+]
+
+
+class Pricing(BaseService[PricingModel]):
+    """
+    This Service handle logic around Pricing
+    """
+
+    aggregate_key = "pricing"
+    model_cls = PricingModel
+
+    def __init__(self, client):
+        super().__init__(client=client)
+
+    # Config from aggregate
+    async def get_pricing_aggregate(
+        self,
+    ) -> PricingModel:
+        result = await self.get_config(
+            address="0xFba561a84A537fCaa567bb7A2257e7142701ae2A"
+        )
+        return result.data[0]
+
+    async def get_pricing_for_services(
+        self, services: List[PricingEntity], pricing_info: Optional[PricingModel] = None
+    ) -> Dict[PricingEntity, PricingPerEntity]:
+        """
+        Get pricing information for requested services
+
+        Args:
+            services: List of pricing entities to get information for
+            pricing_info: Optional pre-fetched pricing aggregate
+
+        Returns:
+            Dictionary with pricing information for requested services
+        """
+        if (
+            not pricing_info
+        ):  # Avoid reloading aggregate info if there is already fetched
+            pricing_info = await self.get_pricing_aggregate()
+
+        result = {}
+        for service in services:
+            if service in pricing_info:
+                result[service] = pricing_info[service]
+
+        return result

src/aleph/sdk/utils.py

@@ -5,6 +5,7 @@
 import hmac
 import json
 import logging
+import math
 import os
 import subprocess
 from datetime import date, datetime, time
@@ -66,6 +67,7 @@
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from jwcrypto.jwa import JWA
 
+from aleph.sdk.client.services.pricing import ComputeUnit
 from aleph.sdk.conf import settings
 from aleph.sdk.types import GenericMessage, SEVInfo, SEVMeasurement
 
@@ -613,3 +615,14 @@ def sanitize_url(url: str) -> str:
         url = f"https://{url}"
 
     return url
+
+
+def _get_nb_compute_units(
+    service_compute: ComputeUnit,
+    vcpus: int = 1,
+    memory_mib: int = 2048,
+):
+    memory = math.ceil(memory_mib / service_compute.memory_mib)
+
+    nb_compute = vcpus if vcpus >= memory else memory
+    return nb_compute