carboncue_sdk

carboncue_sdk

CarbonCue SDK - Core library for carbon-aware computing.

CarbonClient

CarbonClient(config: CarbonConfig | None = None)

Client for accessing carbon intensity data and calculating SCI scores.

Example

client = CarbonClient() intensity = await client.get_current_intensity(region="us-west-2") sci = client.calculate_sci( ... operations=100.0, ... materials=50.0, ... functional_unit=1000, ... region="us-west-2" ... )

Parameters:

Name	Type	Description	Default
config	CarbonConfig | None	Optional configuration. If not provided, loads from environment.	None

Source code in packages/sdk/src/carboncue_sdk/client.py

def __init__(self, config: CarbonConfig | None = None) -> None:
    """Initialize the carbon client.

    Args:
        config: Optional configuration. If not provided, loads from environment.
    """
    self.config = config or CarbonConfig()
    self._http_client: httpx.AsyncClient | None = None
    self._cache: dict[str, tuple[Any, datetime]] = {}

__aenter__ async

__aenter__() -> CarbonClient

Async context manager entry.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def __aenter__(self) -> "CarbonClient":
    """Async context manager entry."""
    self._http_client = httpx.AsyncClient(
        timeout=self.config.request_timeout,
        headers={"auth-token": self.config.electricity_maps_api_key or ""},
    )
    return self

__aexit__ async

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Async context manager exit."""
    if self._http_client:
        await self._http_client.aclose()
        self._http_client = None

get_current_intensity async

get_current_intensity(region: str, provider: str = 'aws') -> CarbonIntensity

Get current carbon intensity for a region.

Parameters:

Name	Type	Description	Default
region	str	Region code (e.g., us-west-2)	required
provider	str	Cloud provider (aws, azure, gcp, etc.)	'aws'

Returns:

Type	Description
CarbonIntensity	Current carbon intensity data

Raises:

Type	Description
InvalidRegionError	If region is not supported
InvalidProviderError	If provider is not supported
AuthenticationError	If API key is missing or invalid
RateLimitError	If API rate limit exceeded
DataNotAvailableError	If data not available for region
APIError	If API request fails

Source code in packages/sdk/src/carboncue_sdk/client.py

async def get_current_intensity(self, region: str, provider: str = "aws") -> CarbonIntensity:
    """Get current carbon intensity for a region.

    Args:
        region: Region code (e.g., us-west-2)
        provider: Cloud provider (aws, azure, gcp, etc.)

    Returns:
        Current carbon intensity data

    Raises:
        InvalidRegionError: If region is not supported
        InvalidProviderError: If provider is not supported
        AuthenticationError: If API key is missing or invalid
        RateLimitError: If API rate limit exceeded
        DataNotAvailableError: If data not available for region
        APIError: If API request fails
    """
    cache_key = f"intensity:{provider}:{region}"
    cached = self._get_from_cache(cache_key)
    if cached is not None:
        return cached  # type: ignore[no-any-return]

    # Map cloud region to Electricity Maps zone
    try:
        zone_id = RegionMapper.get_zone_id(region, provider)
    except ValueError as e:
        error_msg = str(e)
        if "cloud provider" in error_msg.lower():
            raise InvalidProviderError(error_msg) from e
        raise InvalidRegionError(error_msg) from e

    # Get data from Electricity Maps API
    intensity = await self._fetch_from_electricity_maps(zone_id, region)

    self._set_cache(cache_key, intensity)
    return intensity

calculate_sci

calculate_sci(operational_emissions: float, embodied_emissions: float, functional_unit: float, functional_unit_type: str = 'requests', region: str = 'us-west-2') -> SCIScore

Calculate Software Carbon Intensity (SCI) score.

Implements GSF SCI specification: SCI = (O + M) / R

Parameters:

Name	Type	Description	Default
operational_emissions	float	O - Operational emissions in gCO2eq	required
embodied_emissions	float	M - Embodied emissions in gCO2eq	required
functional_unit	float	R - Number of functional units	required
functional_unit_type	str	Type of functional unit (requests, users, etc.)	'requests'
region	str	Region where computation occurred	'us-west-2'

Returns:

Type	Description
SCIScore	Calculated SCI score

Raises:

Type	Description
ValueError	If functional_unit is <= 0

Source code in packages/sdk/src/carboncue_sdk/client.py

def calculate_sci(
    self,
    operational_emissions: float,
    embodied_emissions: float,
    functional_unit: float,
    functional_unit_type: str = "requests",
    region: str = "us-west-2",
) -> SCIScore:
    """Calculate Software Carbon Intensity (SCI) score.

    Implements GSF SCI specification: SCI = (O + M) / R

    Args:
        operational_emissions: O - Operational emissions in gCO2eq
        embodied_emissions: M - Embodied emissions in gCO2eq
        functional_unit: R - Number of functional units
        functional_unit_type: Type of functional unit (requests, users, etc.)
        region: Region where computation occurred

    Returns:
        Calculated SCI score

    Raises:
        ValueError: If functional_unit is <= 0
    """
    if functional_unit <= 0:
        raise ValueError("Functional unit must be greater than 0")

    score = (operational_emissions + embodied_emissions) / functional_unit

    return SCIScore(
        score=score,
        operational_emissions=operational_emissions,
        embodied_emissions=embodied_emissions,
        functional_unit=functional_unit,
        functional_unit_type=functional_unit_type,
        region=region,
    )

close async

close() -> None

Close the HTTP client and cleanup resources.

Source code in packages/sdk/src/carboncue_sdk/client.py

async def close(self) -> None:
    """Close the HTTP client and cleanup resources."""
    if self._http_client:
        await self._http_client.aclose()
        self._http_client = None

CarbonConfig

Bases: BaseSettings

Configuration for CarbonCue SDK.

All settings can be provided via environment variables with CARBONCUE_ prefix. Example: CARBONCUE_ELECTRICITY_MAPS_API_KEY=your_key

APIError

Bases: CarbonCueError

Error communicating with external API.

AuthenticationError

Bases: APIError

API authentication failed.

CarbonCueError

Bases: Exception

Base exception for all CarbonCue errors.

DataNotAvailableError

Bases: APIError

Carbon intensity data not available for requested region.

InvalidProviderError

Bases: CarbonCueError

Invalid cloud provider specified.

InvalidRegionError

Bases: CarbonCueError

Invalid cloud region specified.

RateLimitError

Bases: APIError

API rate limit exceeded.

CarbonIntensity

Bases: BaseModel

Carbon intensity data for a specific region and time.

Region

Bases: BaseModel

Cloud region information.

SCIScore

Bases: BaseModel

Software Carbon Intensity (SCI) score per GSF specification.

SCI = (O + M) / R Where: O = Operational emissions (energy × carbon intensity) M = Embodied emissions (manufacturing impact) R = Functional unit (requests, users, etc.)

RegionMapper

Maps cloud provider regions to Electricity Maps zone identifiers.

get_zone_id staticmethod

get_zone_id(region: str, provider: str = 'aws') -> str

Get Electricity Maps zone ID for a cloud region.

Parameters:

Name	Type	Description	Default
region	str	Cloud provider region code (e.g., us-west-2, eastus)	required
provider	str	Cloud provider name (aws, azure, gcp, digitalocean)	'aws'

Returns:

Type	Description
str	Electricity Maps zone ID (e.g., US-CAL-CISO)

Raises:

Type	Description
ValueError	If region or provider is not supported

Source code in packages/sdk/src/carboncue_sdk/region_mapper.py

@staticmethod
def get_zone_id(region: str, provider: str = "aws") -> str:
    """Get Electricity Maps zone ID for a cloud region.

    Args:
        region: Cloud provider region code (e.g., us-west-2, eastus)
        provider: Cloud provider name (aws, azure, gcp, digitalocean)

    Returns:
        Electricity Maps zone ID (e.g., US-CAL-CISO)

    Raises:
        ValueError: If region or provider is not supported
    """
    provider_lower = provider.lower()

    if provider_lower not in REGION_TO_ZONE_MAP:
        raise ValueError(
            f"Unsupported cloud provider: {provider}. "
            f"Supported: {', '.join(REGION_TO_ZONE_MAP.keys())}"
        )

    region_map = REGION_TO_ZONE_MAP[provider_lower]

    if region not in region_map:
        raise ValueError(
            f"Unsupported region '{region}' for provider '{provider}'. "
            f"Supported regions: {', '.join(region_map.keys())}"
        )

    return region_map[region]

get_supported_regions staticmethod

get_supported_regions(provider: str = 'aws') -> list[str]

Get list of supported regions for a cloud provider.

Parameters:

Name	Type	Description	Default
provider	str	Cloud provider name	'aws'

Returns:

Type	Description
list[str]	List of supported region codes

Raises:

Type	Description
ValueError	If provider is not supported

Source code in packages/sdk/src/carboncue_sdk/region_mapper.py

@staticmethod
def get_supported_regions(provider: str = "aws") -> list[str]:
    """Get list of supported regions for a cloud provider.

    Args:
        provider: Cloud provider name

    Returns:
        List of supported region codes

    Raises:
        ValueError: If provider is not supported
    """
    provider_lower = provider.lower()

    if provider_lower not in REGION_TO_ZONE_MAP:
        raise ValueError(f"Unsupported cloud provider: {provider}")

    return list(REGION_TO_ZONE_MAP[provider_lower].keys())

get_supported_providers staticmethod

get_supported_providers() -> list[str]

Get list of all supported cloud providers.

Returns:

Type	Description
list[str]	List of supported provider names

Source code in packages/sdk/src/carboncue_sdk/region_mapper.py

@staticmethod
def get_supported_providers() -> list[str]:
    """Get list of all supported cloud providers.

    Returns:
        List of supported provider names
    """
    return list(REGION_TO_ZONE_MAP.keys())