huanld/speckle-embodied-carbon-calculator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #9 from bjoernsteinhagen/bjorn/web-2678-compute-concrete-embodied-carbon

Browse Source 
feature: Implement embodied carbon calculation for concrete materials
This commit is contained in:
Björn Steinhagen
2025-02-25 10:38:38 +01:00
committed by GitHub
parent e7c73d0681 c31ab5a303
commit f1ebf4ce39
7 changed files with 629 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+134 -14
View File
@@ -9,7 +9,11 @@ from speckle_automate import (
from typing import Dict, Generator, Any, List
from src.domain.carbon.databases.enums import SteelDatabase, TimberDatabase
from src.domain.carbon.databases.enums import (
SteelDatabase,
TimberDatabase,
ConcreteDatabase,
)
from src.infrastructure.logging import Logging
from src.services.carbon_calculator import CarbonCalculator
from src.services.element_processor import ElementProcessor
@@ -24,7 +28,6 @@ def create_one_of_enum(enum_cls):
return [{"const": item.value, "title": item.name} for item in enum_cls]
# TODO: Function inputs
class FunctionInputs(AutomateBase):
"""User-defined function inputs."""
@@ -42,22 +45,108 @@ class FunctionInputs(AutomateBase):
json_schema_extra={"oneOf": create_one_of_enum(TimberDatabase)},
)
concrete_database: str = Field(
default=ConcreteDatabase.GulLowAir.value,
title="Concrete Database",
description="Database used for the GWP of concrete objects",
json_schema_extra={"oneOf": create_one_of_enum(ConcreteDatabase)},
)
country: str = Field(
default="CAN",
title="Country",
description="Country for concrete strength units (CAN: MPa, USA: PSI)",
json_schema_extra={
"oneOf": [
{"const": "CAN", "title": "Canada (MPa)"},
{"const": "USA", "title": "United States (PSI)"},
]
},
)
# Add reinforcement rates based on Image 1
reinforcement_grade_beam: float = Field(
default=100.0,
title="Grade Beam Reinforcement (kg/m³)",
)
reinforcement_slab_on_grade: float = Field(
default=85.0,
title="Slab on Grade Reinforcement (kg/m³)",
)
reinforcement_pad_footing: float = Field(
default=100.0,
title="Pad Footing Reinforcement (kg/m³)",
)
reinforcement_pile: float = Field(
default=100.0,
title="Pile Reinforcement (kg/m³)",
)
reinforcement_strip_footing: float = Field(
default=100.0,
title="Strip Footing Reinforcement (kg/m³)",
)
reinforcement_pile_cap: float = Field(
default=100.0,
title="Pile Cap Reinforcement (kg/m³)",
)
reinforcement_gravity_wall: float = Field(
default=150.0,
title="Gravity Wall Reinforcement (kg/m³)",
)
reinforcement_column: float = Field(
default=450.0,
title="Column Reinforcement (kg/m³)",
)
reinforcement_shear_wall: float = Field(
default=150.0,
title="Shear Wall Reinforcement (kg/m³)",
)
reinforcement_concrete_slab: float = Field(
default=120.0,
title="Concrete Slab Reinforcement (kg/m³)",
)
reinforcement_beam: float = Field(
default=220.0,
title="Beam Reinforcement (kg/m³)",
)
reinforcement_topping_slab: float = Field(
default=85.0,
title="Topping Slab Reinforcement (kg/m³)",
)
class RevitCarbonAnalyzer:
"""Main application for analyzing carbon in Revit models."""
def __init__(self, steel_database: str, timber_database: str):
def __init__(
self,
steel_database: str,
timber_database: str,
concrete_database: str,
country: str,
reinforcement_rates: Dict[str, float],
):
self.material_processor = MaterialProcessor()
self.element_processor = ElementProcessor(
material_processor=self.material_processor, logger=Logging()
)
self.carbon_calculator = CarbonCalculator(
steel_database=steel_database.value
if isinstance(steel_database, SteelDatabase)
else steel_database,
timber_database=timber_database.value
if isinstance(timber_database, SteelDatabase)
else timber_database,
steel_database=steel_database,
timber_database=timber_database,
concrete_database=concrete_database,
country=country,
custom_reinforcement_rates=reinforcement_rates,
)
def analyze_model(self, model_root) -> dict:
@@ -67,7 +156,7 @@ class RevitCarbonAnalyzer:
"skipped_elements": [],
"errors": [],
"total_carbon": 0.0,
"missing_factors": {"timber": [], "steel": []},
"missing_factors": {"timber": [], "steel": [], "concrete": []},
}
# Process each element
@@ -91,9 +180,14 @@ class RevitCarbonAnalyzer:
)
# Get missing factors
missing_timber, missing_steel = self.carbon_calculator.get_missing_factors()
(
missing_timber,
missing_steel,
missing_concrete,
) = self.carbon_calculator.get_missing_factors()
results["missing_factors"]["timber"] = missing_timber
results["missing_factors"]["steel"] = missing_steel
results["missing_factors"]["concrete"] = missing_concrete
# Log missing factors
if missing_timber:
@@ -106,6 +200,11 @@ class RevitCarbonAnalyzer:
for item in missing_steel:
print(f" - {item}")
if missing_concrete:
print(f"Missing concrete factors ({len(missing_concrete)}):")
for item in missing_concrete:
print(f" - {item}")
return results
def _process_single_element(self, element: Dict) -> Dict:
@@ -167,17 +266,39 @@ def automate_function(
# Get string values from enums if needed
steel_db = function_inputs.steel_database
timber_db = function_inputs.timber_database
concrete_db = function_inputs.concrete_database
country = function_inputs.country
# Ensure we're working with string values
# Ensure we're working with string values, not enum objects
if hasattr(steel_db, "value"):
steel_db = steel_db.value
if hasattr(timber_db, "value"):
timber_db = timber_db.value
if hasattr(concrete_db, "value"):
concrete_db = concrete_db.value
# Create custom reinforcement rates dictionary
custom_reinforcement_rates = {
"Grade Beam": function_inputs.reinforcement_grade_beam,
"Slab on Grade": function_inputs.reinforcement_slab_on_grade,
"Pad Footing": function_inputs.reinforcement_pad_footing,
"Pile": function_inputs.reinforcement_pile,
"Strip Footing": function_inputs.reinforcement_strip_footing,
"Pile Cap": function_inputs.reinforcement_pile_cap,
"Walls - wind/gravity": function_inputs.reinforcement_gravity_wall,
"Column": function_inputs.reinforcement_column,
"Shear Walls": function_inputs.reinforcement_shear_wall,
"Concrete Slabs": function_inputs.reinforcement_concrete_slab,
"Beams": function_inputs.reinforcement_beam,
"Topping Slabs": function_inputs.reinforcement_topping_slab,
}
# Initialize analyzer
analyzer = RevitCarbonAnalyzer(
steel_database=steel_db,
timber_database=timber_db,
concrete_database=concrete_db,
country=country,
reinforcement_rates=custom_reinforcement_rates,
)
# Get commit root
@@ -202,7 +323,7 @@ def automate_function(
# Prepare detailed success message
success_message = (
f"🚀 Analysis complete.\n\n\tProcessed:\t\t{len(results['processed_elements'])} elements.\n\t"
f"🚀 Analysis complete.\n\n\tProcessed:\t\t{len(results['processed_elements'])} elements\n\t"
f"Total carbon:\t{results['total_carbon']:.2f} kgCO₂e\n"
)
@@ -234,7 +355,6 @@ def automate_function(
else:
success_message += (
"\nNOTE: All materials successfully matched with emission factors."
"complete."
)
# Mark success with detailed message
src/domain/carbon/concrete_reinforcement.py
+57
View File
@@ -0,0 +1,57 @@
from enum import Enum
from typing import Dict
class ConcreteElementType(Enum):
GRADE_BEAM = "Grade Beam"
SLAB_ON_GRADE = "Slab on Grade"
PAD_FOOTING = "Pad Footing"
PILE = "Pile"
STRIP_FOOTING = "Strip Footing"
PILE_CAP = "Pile Cap"
GRAVITY_WALL = "Walls - wind/gravity"
COLUMN = "Column"
SHEAR_WALL = "Shear Walls"
CONCRETE_SLAB = "Concrete Slabs"
BEAM = "Beams"
TOPPING_SLAB = "Topping Slabs"
class ReinforcementRates:
"""Reinforcement rates for concrete elements by element type."""
def __init__(self, rates_dict: Dict[str, float]):
"""Initialize with rates provided from function inputs."""
self._rates = {}
# Convert string keys to enum values
for type_str, rate in rates_dict.items():
for enum_type in ConcreteElementType:
if enum_type.value.lower() == type_str.lower():
self._rates[enum_type] = rate
break
# If no match found, store with string key
if type_str.lower() not in [e.value.lower() for e in ConcreteElementType]:
self._rates[type_str] = rate
def get_rate(self, element_type: str) -> float:
"""Get reinforcement rate for a concrete element type."""
# Try direct match with enum values
for enum_type in ConcreteElementType:
if enum_type.value.lower() == element_type.lower():
return self._rates.get(enum_type, 100.0) # Default if missing
# Try fuzzy matching with string keys
element_lower = element_type.lower()
if "beam" in element_lower and "grade" in element_lower:
return self.get_rate_by_type(ConcreteElementType.GRADE_BEAM)
elif "slab" in element_lower and "grade" in element_lower:
return self.get_rate_by_type(ConcreteElementType.SLAB_ON_GRADE)
# ... other fuzzy matching logic ...
# Default value if no match found
return 100.0
def get_rate_by_type(self, element_type: ConcreteElementType) -> float:
"""Get rate by concrete element type enum."""
return self._rates.get(element_type, 100.0) # Default if missing
src/domain/carbon/databases/concrete/__init__.py
View File
src/domain/carbon/databases/concrete/metric.py
+270
View File
@@ -0,0 +1,270 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
from src.domain.carbon.databases.enums import ConcreteDatabase
UNIT = "kgCO₂e/m³"
class ConcreteEmissionDatabase(EmissionFactorDatabase):
"""Database implementation for concrete emission factors based on cement type and strength."""
def __init__(self, database_name: str):
super().__init__()
self._database_name = database_name
self._factors = {}
self._init_factors()
def _init_factors(self):
"""Initialize factors based on the specific database."""
# Define mappings from database type to strength values
database_strength_values = {
ConcreteDatabase.GulLowAir.value: {
"25": {
"Beam": 188,
"Slab": 188,
"Slab on Grade": 188,
"Foundation": 151,
"Column": 151,
"Wall": 151,
"Wall Foundation": 151,
},
"30": {
"Beam": 220,
"Slab": 220,
"Slab on Grade": 220,
"Foundation": 176,
"Column": 176,
"Wall": 176,
"Wall Foundation": 176,
},
"35": {
"Beam": 250,
"Slab": 250,
"Slab on Grade": 250,
"Foundation": 200,
"Column": 200,
"Wall": 200,
"Wall Foundation": 200,
},
"40": {
"Beam": 280,
"Slab": 280,
"Slab on Grade": 280,
"Foundation": 224,
"Column": 224,
"Wall": 224,
"Wall Foundation": 224,
},
"45": {
"Beam": 298,
"Slab": 298,
"Slab on Grade": 298,
"Foundation": 238,
"Column": 238,
"Wall": 238,
"Wall Foundation": 238,
},
"50": {
"Beam": 320,
"Slab": 320,
"Slab on Grade": 320,
"Foundation": 256,
"Column": 256,
"Wall": 256,
"Wall Foundation": 256,
},
},
ConcreteDatabase.GulHighAir.value: {
"25": {
"Beam": 201,
"Slab": 197,
"Slab on Grade": 197,
"Foundation": 157,
"Column": 157,
"Wall": 157,
"Wall Foundation": 157,
},
"30": {
"Beam": 236,
"Slab": 230,
"Slab on Grade": 230,
"Foundation": 184,
"Column": 184,
"Wall": 184,
"Wall Foundation": 184,
},
"35": {
"Beam": 268,
"Slab": 264,
"Slab on Grade": 264,
"Foundation": 211,
"Column": 211,
"Wall": 211,
"Wall Foundation": 211,
},
"40": {
"Beam": 292,
"Slab": 292,
"Slab on Grade": 292,
"Foundation": 234,
"Column": 234,
"Wall": 234,
"Wall Foundation": 234,
},
"45": {
"Beam": 316,
"Slab": 316,
"Slab on Grade": 316,
"Foundation": 254,
"Column": 254,
"Wall": 254,
"Wall Foundation": 254,
},
"50": {
"Beam": 343,
"Slab": 322,
"Slab on Grade": 322,
"Foundation": 257,
"Column": 257,
"Wall": 257,
"Wall Foundation": 257,
},
},
ConcreteDatabase.GuLowAir.value: {
"25": {
"Beam": 201,
"Slab": 201,
"Slab on Grade": 201,
"Foundation": 161,
"Column": 161,
"Wall": 161,
"Wall Foundation": 161,
},
"30": {
"Beam": 236,
"Slab": 236,
"Slab on Grade": 236,
"Foundation": 189,
"Column": 189,
"Wall": 189,
"Wall Foundation": 189,
},
"35": {
"Beam": 268,
"Slab": 268,
"Slab on Grade": 268,
"Foundation": 214,
"Column": 214,
"Wall": 214,
"Wall Foundation": 214,
},
"40": {
"Beam": 300,
"Slab": 300,
"Slab on Grade": 300,
"Foundation": 240,
"Column": 240,
"Wall": 240,
"Wall Foundation": 240,
},
"45": {
"Beam": 319,
"Slab": 319,
"Slab on Grade": 319,
"Foundation": 256,
"Column": 256,
"Wall": 256,
"Wall Foundation": 256,
},
"50": {
"Beam": 343,
"Slab": 343,
"Slab on Grade": 343,
"Foundation": 274,
"Column": 274,
"Wall": 274,
"Wall Foundation": 274,
},
},
ConcreteDatabase.GuHighAir.value: {
"25": {
"Beam": 210,
"Slab": 210,
"Slab on Grade": 210,
"Foundation": 168,
"Column": 168,
"Wall": 168,
"Wall Foundation": 168,
},
"30": {
"Beam": 246,
"Slab": 246,
"Slab on Grade": 246,
"Foundation": 197,
"Column": 197,
"Wall": 197,
"Wall Foundation": 197,
},
"35": {
"Beam": 283,
"Slab": 283,
"Slab on Grade": 283,
"Foundation": 227,
"Column": 227,
"Wall": 227,
"Wall Foundation": 227,
},
"40": {
"Beam": 313,
"Slab": 313,
"Slab on Grade": 313,
"Foundation": 251,
"Column": 251,
"Wall": 251,
"Wall Foundation": 251,
},
"45": {
"Beam": 339,
"Slab": 339,
"Slab on Grade": 339,
"Foundation": 271,
"Column": 271,
"Wall": 271,
"Wall Foundation": 271,
},
"50": {
"Beam": 345,
"Slab": 345,
"Slab on Grade": 345,
"Foundation": 276,
"Column": 276,
"Wall": 276,
"Wall Foundation": 276,
},
},
}
# Get the strength values for the selected database
strength_values = database_strength_values.get(self._database_name)
if not strength_values:
raise ValueError(f"Unknown concrete database: {self._database_name}")
# Create factors from the strength values
for strength, elements in strength_values.items():
for element, value in elements.items():
factor_key = f"{strength}_{element}"
self._factors[factor_key] = EmissionFactor(
value=value,
unit=UNIT,
database=self._database_name,
epd_number=f"CONCRETE-{self._database_name}-{strength}-{element}",
publication_date="2024-01-01",
valid_until="2029-01-01",
)
def get_factor_by_strength_and_element(
self, strength: str, element_type: str
) -> EmissionFactor:
"""Get emission factor based on concrete strength and element type."""
factor_key = f"{strength}_{element_type}"
return self._factors.get(factor_key)
src/domain/carbon/databases/enums.py
+4 -1
View File
@@ -18,4 +18,7 @@ class SteelDatabase(Enum):
class ConcreteDatabase(Enum):
pass
GulLowAir = "GUL Cement, Low Air"
GulHighAir = "GUL Cement, High Air"
GuLowAir = "GU Cement, Low Air"
GuHighAir = "GU Cement, High Air"
src/domain/carbon/emission_factor_registry.py
+28 -18
View File
@@ -1,4 +1,6 @@
from typing import Optional, Dict
from src.domain.carbon.databases.concrete.metric import ConcreteEmissionDatabase
from src.domain.carbon.schema import EmissionFactor
from src.domain.carbon.databases.enums import (
TimberDatabase,
@@ -32,6 +34,7 @@ class EmissionFactorRegistry:
self._concrete_databases = {}
# Material aliases for normalization
# NOTE: Purely demonstrative → add aliases as needed.
self._timber_aliases = {
"clt": ["cross laminated timber", "cross-laminated timber"],
"glulam": [
@@ -57,9 +60,9 @@ class EmissionFactorRegistry:
"hot-rolled",
"hot_rolled",
"hotrolled",
"345 MPa",
"350W",
"350W(1)",
"345 MPa", # NOTE: Needed!
"350W", # NOTE: Needed!
"350W(1)", # NOTE: Needed!
],
"hss": ["hollow structural section", "hollow section", "tube"],
"plate": ["flat plate"],
@@ -76,7 +79,7 @@ class EmissionFactorRegistry:
# Initialize all database instances
self._init_timber_databases()
self._init_steel_databases()
# self._init_concrete_databases() - empty for now
self._init_concrete_databases()
def _init_timber_databases(self) -> None:
"""Initialize timber database implementations"""
@@ -98,6 +101,23 @@ class EmissionFactorRegistry:
SteelDatabase.Type350MPa.value: Steel350MPa(),
}
def _init_concrete_databases(self) -> None:
"""Initialize concrete database implementations"""
self._concrete_databases = {
ConcreteDatabase.GulLowAir.value: ConcreteEmissionDatabase(
ConcreteDatabase.GulLowAir.value
),
ConcreteDatabase.GulHighAir.value: ConcreteEmissionDatabase(
ConcreteDatabase.GulHighAir.value
),
ConcreteDatabase.GuLowAir.value: ConcreteEmissionDatabase(
ConcreteDatabase.GuLowAir.value
),
ConcreteDatabase.GuHighAir.value: ConcreteEmissionDatabase(
ConcreteDatabase.GuHighAir.value
),
}
@staticmethod
def _normalize_material_name(name: str, aliases: Dict[str, list]) -> str:
"""Normalize material name using centralized aliases with enhanced matching.
@@ -176,25 +196,15 @@ class EmissionFactorRegistry:
return db.get_factor(normalized_name)
def get_concrete_factor(
self, material_name: str, database: str
self, strength: str, element_type: str, database: str
) -> Optional[EmissionFactor]:
"""Get emission factor for concrete from specified database"""
"""Get emission factor for concrete from specified database based on strength and element type."""
db = self._concrete_databases.get(database)
if not db:
raise ValueError(f"Unknown concrete database: {database}")
# Try direct lookup first
factor = db.get_factor(material_name)
if factor:
return factor
# If not found, try normalized name when concrete aliases are added
if self._concrete_aliases:
normalized_name = self._normalize_material_name(
material_name, self._concrete_aliases
)
return db.get_factor(normalized_name)
return None
# Concrete database requires both strength and element type
return db.get_factor_by_strength_and_element(strength, element_type)
def list_timber_databases(self) -> list[str]:
"""List all registered timber databases"""
src/services/carbon_calculator.py
+136 -14
View File
@@ -1,7 +1,14 @@
from typing import Dict, Optional, Tuple, List
from src.domain.carbon.concrete_reinforcement import ReinforcementRates
from src.domain.carbon.emission_factor_registry import EmissionFactorRegistry
from src.domain.types import BuildingElement, CarbonResult, Material, MaterialType
from src.domain.types import (
BuildingElement,
CarbonResult,
Material,
MaterialType,
ElementCategory,
)
class CarbonCalculator:
@@ -11,25 +18,32 @@ class CarbonCalculator:
self,
steel_database: str,
timber_database: str,
concrete_database: Optional[str] = None,
concrete_database: str,
country: str,
custom_reinforcement_rates: Dict[str, float],
):
# Store database selections
self._steel_database = steel_database
self._timber_database = timber_database
self._concrete_database = concrete_database
self._country = country
# Initialize registry
self._registry = EmissionFactorRegistry()
# Initialize reinforcement rates with the provided dictionary
# TODO: Validate inputs (e.g. C# int.TryParse()? )
self._reinforcement_rates = ReinforcementRates(custom_reinforcement_rates)
# Cache common material factors to avoid repeated lookups
self._steel_factors_cache = {}
self._timber_factors_cache = {}
self._concrete_factors_cache = {}
# TODO: Remove
# Track missing factors
self._missing_timber_factors = set()
self._missing_steel_factors = set()
self._missing_concrete_factors = set()
def calculate_carbon(self, element: BuildingElement) -> Dict[str, CarbonResult]:
"""Calculate carbon emissions for an element's materials."""
@@ -37,7 +51,10 @@ class CarbonCalculator:
for material in element.materials:
try:
result = self._calculate_material_carbon(material)
if material.type == MaterialType.CONCRETE:
result = self._calculate_concrete_carbon(material, element.category)
else:
result = self._calculate_material_carbon(material)
results[material.properties.name] = result
except Exception as e:
# Track missing factors
@@ -52,6 +69,13 @@ class CarbonCalculator:
self._missing_steel_factors.add(
material.grade or material.properties.name
)
elif material.type == MaterialType.CONCRETE:
# Track missing concrete factors
strength = str(int(material.properties.compressive_strength))
element_type = self._map_element_category_to_concrete_type(
element.category
)
self._missing_concrete_factors.add(f"{strength}_{element_type}")
print(
f"Error calculating carbon for {material.properties.name}: {str(e)}"
@@ -59,14 +83,20 @@ class CarbonCalculator:
return results
def _calculate_material_carbon(self, material: Material) -> CarbonResult:
def _calculate_material_carbon(
self, material: Material, element_category: Optional[ElementCategory] = None
) -> CarbonResult:
"""Calculate carbon emissions for a single material."""
if material.type == MaterialType.METAL:
return self._calculate_metal_carbon(material)
elif material.type == MaterialType.WOOD:
return self._calculate_wood_carbon(material)
elif material.type == MaterialType.CONCRETE:
return self._calculate_concrete_carbon(material)
if element_category is None:
raise ValueError(
"Element category is required for concrete carbon calculation"
)
return self._calculate_concrete_carbon(material, element_category)
else:
raise ValueError(f"Unsupported material type: {material.type}")
@@ -120,19 +150,111 @@ class CarbonCalculator:
category="Wood",
)
def _calculate_concrete_carbon(self, material: Material) -> CarbonResult:
"""Calculate carbon emissions for concrete."""
# TODO: Implement concrete-specific calculation when concrete database is added
return CarbonResult(
factor=0.0, # Placeholder
total_carbon=0.0, # Placeholder
def _calculate_concrete_carbon(
self, material: Material, element_category: ElementCategory
) -> CarbonResult:
"""Calculate carbon emissions for concrete, including reinforcement."""
if not material.properties.compressive_strength:
raise ValueError(
"Compressive strength required for concrete carbon calculation"
)
# Handle unit conversion based on country
# For US, convert PSI to MPa if needed
strength_value = material.properties.compressive_strength
if self._country == "USA":
# Check if value is in PSI (typically large numbers)
if strength_value > 100: # Assume PSI
strength_value = strength_value / 145.038 # Convert PSI to MPa
# Round to nearest valid strength category (25, 30, 35, 40, 45, 50)
valid_strengths = [25, 30, 35, 40, 45, 50]
strength_mpa = min(valid_strengths, key=lambda x: abs(x - strength_value))
strength = str(strength_mpa)
# Map element category to concrete element type for the database
element_type = self._map_element_category_to_concrete_type(element_category)
# Create cache key for concrete factors
concrete_cache_key = f"{strength}_{element_type}"
# Get concrete factor
if concrete_cache_key not in self._concrete_factors_cache:
try:
factor = self._registry.get_concrete_factor(
strength, element_type, self._concrete_database
)
if not factor:
self._missing_concrete_factors.add(concrete_cache_key)
raise ValueError(
f"No emission factor found for concrete: strength={strength}, element={element_type}"
)
self._concrete_factors_cache[concrete_cache_key] = factor
except Exception as e:
self._missing_concrete_factors.add(concrete_cache_key)
raise ValueError(f"Error getting concrete factor: {str(e)}")
concrete_factor = self._concrete_factors_cache[concrete_cache_key]
concrete_volume = material.properties.volume
concrete_carbon = concrete_volume * concrete_factor.value
# Calculate reinforcement carbon
reinforcement_rate = self._reinforcement_rates.get_rate(element_type)
reinforcement_mass = (
concrete_volume * reinforcement_rate / 1000
) # Convert kg to tons if needed
# Get rebar factor from steel database
if "Rebar" not in self._steel_factors_cache:
rebar_factor = self._registry.get_steel_factor(
"Rebar", self._steel_database
)
if not rebar_factor:
raise ValueError("No emission factor found for rebar")
self._steel_factors_cache["Rebar"] = rebar_factor
rebar_factor = self._steel_factors_cache["Rebar"]
reinforcement_carbon = reinforcement_mass * rebar_factor.value
# Total carbon is concrete + reinforcement
total_carbon = concrete_carbon + reinforcement_carbon
# Create result with additional metadata
result = CarbonResult(
factor=concrete_factor.value,
total_carbon=total_carbon,
category="Concrete",
)
# TODO: Remove
def get_missing_factors(self) -> Tuple[List[str], List[str]]:
# TODO: Add extra metadata (not in original CarbonResult class, would need extension)
# TODO: result.concrete_carbon = concrete_carbon
# TODO: result.reinforcement_carbon = reinforcement_carbon
# TODO: result.reinforcement_rate = reinforcement_rate
return result
@staticmethod
def _map_element_category_to_concrete_type(
element_category: ElementCategory,
) -> str:
"""Map BuildingElement category to concrete element type for database lookup."""
# Default mappings
category_mapping = {
ElementCategory.SLAB: "Slab",
ElementCategory.WALL: "Wall",
ElementCategory.COLUMN: "Column",
ElementCategory.BEAM: "Beam",
ElementCategory.FOUNDATION: "Foundation",
}
# Return the mapped type or default to "Beam" if unknown
return category_mapping.get(element_category, "Beam")
def get_missing_factors(self) -> Tuple[List[str], List[str], List[str]]:
"""Return lists of materials that had no emission factor."""
return (
sorted(list(self._missing_timber_factors)),
sorted(list(self._missing_steel_factors)),
sorted(list(self._missing_concrete_factors)),
)