name: "carbon-calculator" description: "Calculate embodied carbon in construction materials. Track CO2 emissions, compare alternatives, and generate sustainability reports." homepage: "https://datadrivenconstruction.io" metadata: {"openclaw": {"emoji": "🌱", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}

Carbon Calculator

Business Case

Problem Statement

Sustainability requirements demand:

• Tracking embodied carbon
• Comparing material options
• Meeting carbon targets
• Reporting emissions

Solution

Calculate and track embodied carbon for construction materials using standard emission factors.

Technical Implementation

import pandas as pd
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum


class MaterialCategory(Enum):
    CONCRETE = "concrete"
    STEEL = "steel"
    ALUMINUM = "aluminum"
    TIMBER = "timber"
    BRICK = "brick"
    GLASS = "glass"
    INSULATION = "insulation"
    PLASTIC = "plastic"
    COPPER = "copper"
    OTHER = "other"


@dataclass
class CarbonFactor:
    material: str
    category: MaterialCategory
    ec_factor: float  # kgCO2e per unit
    unit: str
    source: str


@dataclass
class MaterialInput:
    material_code: str
    material_name: str
    quantity: float
    unit: str
    category: MaterialCategory


@dataclass
class CarbonResult:
    material_code: str
    material_name: str
    quantity: float
    unit: str
    ec_factor: float
    embodied_carbon: float  # kgCO2e
    category: str


# Embodied carbon factors (kgCO2e per unit)
CARBON_FACTORS = {
    # Concrete
    'concrete_c20': CarbonFactor('Concrete C20', MaterialCategory.CONCRETE, 240, 'm3', 'ICE Database'),
    'concrete_c30': CarbonFactor('Concrete C30', MaterialCategory.CONCRETE, 290, 'm3', 'ICE Database'),
    'concrete_c40': CarbonFactor('Concrete C40', MaterialCategory.CONCRETE, 350, 'm3', 'ICE Database'),
    'concrete_c50': CarbonFactor('Concrete C50', MaterialCategory.CONCRETE, 410, 'm3', 'ICE Database'),

    # Steel
    'steel_rebar': CarbonFactor('Rebar', MaterialCategory.STEEL, 1.99, 'kg', 'ICE Database'),
    'steel_section': CarbonFactor('Steel Section', MaterialCategory.STEEL, 1.55, 'kg', 'ICE Database'),
    'steel_sheet': CarbonFactor('Steel Sheet', MaterialCategory.STEEL, 2.03, 'kg', 'ICE Database'),
    'steel_stainless': CarbonFactor('Stainless Steel', MaterialCategory.STEEL, 6.15, 'kg', 'ICE Database'),

    # Aluminum
    'aluminum_general': CarbonFactor('Aluminum General', MaterialCategory.ALUMINUM, 9.16, 'kg', 'ICE Database'),
    'aluminum_recycled': CarbonFactor('Aluminum Recycled', MaterialCategory.ALUMINUM, 1.81, 'kg', 'ICE Database'),

    # Timber
    'timber_softwood': CarbonFactor('Softwood Timber', MaterialCategory.TIMBER, 0.31, 'kg', 'ICE Database'),
    'timber_hardwood': CarbonFactor('Hardwood Timber', MaterialCategory.TIMBER, 0.46, 'kg', 'ICE Database'),
    'timber_glulam': CarbonFactor('Glulam', MaterialCategory.TIMBER, 0.51, 'kg', 'ICE Database'),
    'timber_clt': CarbonFactor('CLT', MaterialCategory.TIMBER, 0.44, 'kg', 'ICE Database'),
    'timber_plywood': CarbonFactor('Plywood', MaterialCategory.TIMBER, 0.65, 'kg', 'ICE Database'),

    # Masonry
    'brick_common': CarbonFactor('Common Brick', MaterialCategory.BRICK, 0.24, 'kg', 'ICE Database'),
    'block_concrete': CarbonFactor('Concrete Block', MaterialCategory.BRICK, 0.10, 'kg', 'ICE Database'),

    # Glass
    'glass_float': CarbonFactor('Float Glass', MaterialCategory.GLASS, 1.44, 'kg', 'ICE Database'),
    'glass_double': CarbonFactor('Double Glazing', MaterialCategory.GLASS, 35.0, 'm2', 'ICE Database'),

    # Insulation
    'insul_mineral': CarbonFactor('Mineral Wool', MaterialCategory.INSULATION, 1.28, 'kg', 'ICE Database'),
    'insul_eps': CarbonFactor('EPS', MaterialCategory.INSULATION, 3.29, 'kg', 'ICE Database'),
    'insul_xps': CarbonFactor('XPS', MaterialCategory.INSULATION, 3.29, 'kg', 'ICE Database'),

    # Other
    'copper_pipe': CarbonFactor('Copper Pipe', MaterialCategory.COPPER, 2.71, 'kg', 'ICE Database'),
    'pvc_pipe': CarbonFactor('PVC Pipe', MaterialCategory.PLASTIC, 3.10, 'kg', 'ICE Database'),
}


class CarbonCalculator:
    """Calculate embodied carbon for construction."""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.materials: List[MaterialInput] = []
        self.results: List[CarbonResult] = []
        self.custom_factors: Dict[str, CarbonFactor] = {}

    def add_custom_factor(self,
                          code: str,
                          name: str,
                          category: MaterialCategory,
                          ec_factor: float,
                          unit: str,
                          source: str = "Custom"):
        """Add custom carbon factor."""

        self.custom_factors[code] = CarbonFactor(
            material=name,
            category=category,
            ec_factor=ec_factor,
            unit=unit,
            source=source
        )

    def get_factor(self, material_code: str) -> Optional[CarbonFactor]:
        """Get carbon factor for material."""

        # Check custom first
        if material_code in self.custom_factors:
            return self.custom_factors[material_code]

        # Check standard factors
        code_lower = material_code.lower().replace('-', '_').replace(' ', '_')
        return CARBON_FACTORS.get(code_lower)

    def add_material(self,
                     material_code: str,
                     material_name: str,
                     quantity: float,
                     unit: str,
                     category: MaterialCategory = MaterialCategory.OTHER):
        """Add material to calculation."""

        self.materials.append(MaterialInput(
            material_code=material_code,
            material_name=material_name,
            quantity=quantity,
            unit=unit,
            category=category
        ))

    def calculate(self) -> List[CarbonResult]:
        """Calculate embodied carbon for all materials."""

        self.results = []

        for mat in self.materials:
            factor = self.get_factor(mat.material_code)

            if factor:
                # Check unit compatibility
                if factor.unit == mat.unit:
                    ec = mat.quantity * factor.ec_factor
                else:
                    # Assume conversion needed - simplified
                    ec = mat.quantity * factor.ec_factor
            else:
                # Use default factor based on category
                default_factors = {
                    MaterialCategory.CONCRETE: 300,
                    MaterialCategory.STEEL: 1.8,
                    MaterialCategory.ALUMINUM: 9.0,
                    MaterialCategory.TIMBER: 0.4,
                    MaterialCategory.BRICK: 0.2,
                    MaterialCategory.GLASS: 1.5,
                    MaterialCategory.INSULATION: 2.0,
                    MaterialCategory.OTHER: 1.0
                }
                ec_factor = default_factors.get(mat.category, 1.0)
                ec = mat.quantity * ec_factor

            self.results.append(CarbonResult(
                material_code=mat.material_code,
                material_name=mat.material_name,
                quantity=mat.quantity,
                unit=mat.unit,
                ec_factor=factor.ec_factor if factor else 0,
                embodied_carbon=round(ec, 2),
                category=mat.category.value
            ))

        return self.results

    def get_total_carbon(self) -> float:
        """Get total embodied carbon (kgCO2e)."""
        return sum(r.embodied_carbon for r in self.results)

    def get_carbon_by_category(self) -> Dict[str, float]:
        """Get carbon breakdown by category."""

        by_category = {}
        for r in self.results:
            if r.category not in by_category:
                by_category[r.category] = 0
            by_category[r.category] += r.embodied_carbon

        return {k: round(v, 2) for k, v in by_category.items()}

    def compare_alternatives(self,
                              original_code: str,
                              original_qty: float,
                              alternative_code: str,
                              alternative_qty: float) -> Dict[str, Any]:
        """Compare carbon impact of material alternatives."""

        original_factor = self.get_factor(original_code)
        alt_factor = self.get_factor(alternative_code)

        if not original_factor or not alt_factor:
            return {}

        original_carbon = original_qty * original_factor.ec_factor
        alt_carbon = alternative_qty * alt_factor.ec_factor
        savings = original_carbon - alt_carbon

        return {
            'original_material': original_factor.material,
            'original_carbon': round(original_carbon, 2),
            'alternative_material': alt_factor.material,
            'alternative_carbon': round(alt_carbon, 2),
            'carbon_savings': round(savings, 2),
            'savings_percent': round(savings / original_carbon * 100, 1) if original_carbon > 0 else 0
        }

    def generate_report(self) -> Dict[str, Any]:
        """Generate carbon report."""

        if not self.results:
            self.calculate()

        total = self.get_total_carbon()
        by_category = self.get_carbon_by_category()

        # Find top contributors
        sorted_results = sorted(self.results, key=lambda x: x.embodied_carbon, reverse=True)
        top_5 = sorted_results[:5]

        # Convert to tonnes
        total_tonnes = total / 1000

        return {
            'project': self.project_name,
            'total_kgCO2e': round(total, 2),
            'total_tCO2e': round(total_tonnes, 2),
            'material_count': len(self.results),
            'by_category': by_category,
            'top_contributors': [
                {
                    'material': r.material_name,
                    'carbon': r.embodied_carbon,
                    'percentage': round(r.embodied_carbon / total * 100, 1) if total > 0 else 0
                }
                for r in top_5
            ]
        }

    def suggest_reductions(self) -> List[Dict[str, Any]]:
        """Suggest carbon reduction opportunities."""

        if not self.results:
            self.calculate()

        suggestions = []

        for r in self.results:
            # Steel -> Timber
            if r.category == 'steel' and r.embodied_carbon > 1000:
                suggestions.append({
                    'material': r.material_name,
                    'current_carbon': r.embodied_carbon,
                    'suggestion': 'Consider timber alternative where structurally feasible',
                    'potential_reduction': '60-80%'
                })

            # Standard concrete -> Low carbon
            if r.category == 'concrete' and r.embodied_carbon > 5000:
                suggestions.append({
                    'material': r.material_name,
                    'current_carbon': r.embodied_carbon,
                    'suggestion': 'Use low-carbon concrete mix with SCMs',
                    'potential_reduction': '20-40%'
                })

            # Virgin aluminum -> Recycled
            if r.category == 'aluminum':
                suggestions.append({
                    'material': r.material_name,
                    'current_carbon': r.embodied_carbon,
                    'suggestion': 'Specify recycled aluminum content',
                    'potential_reduction': '70-80%'
                })

        return suggestions

    def export_to_excel(self, output_path: str) -> str:
        """Export carbon calculation to Excel."""

        if not self.results:
            self.calculate()

        report = self.generate_report()

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer: