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

Merge pull request #8 from bjoernsteinhagen/bjorn/web-2680-setup-databases

Browse Source 
feat: database-driven carbon factors with material normalization and improved logging
This commit is contained in:
Björn Steinhagen
2025-02-25 08:15:20 +01:00
committed by GitHub
parent 3ab35a6d31 e92066a8de
commit e7c73d0681
23 changed files with 908 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+101 -17
View File
@@ -1,5 +1,6 @@
from collections import defaultdict
from pydantic import Field
from speckle_automate import (
AutomateBase,
AutomationContext,
@@ -8,28 +9,56 @@ from speckle_automate import (
from typing import Dict, Generator, Any, List
from src.domain.carbon.databases.enums import SteelDatabase, TimberDatabase
from src.infrastructure.logging import Logging
from src.services.carbon_calculator import CarbonCalculator
from src.services.element_processor import ElementProcessor
from src.services.material_processor import MaterialProcessor
def create_one_of_enum(enum_cls):
"""
Helper function to create a JSON schema from an Enum class.
This is used for generating user input forms in the UI.
"""
return [{"const": item.value, "title": item.name} for item in enum_cls]
# TODO: Function inputs
class FunctionInputs(AutomateBase):
"""User-defined function inputs."""
# wood_supplier: WoodSupplier = WoodSupplier.INDUSTRY_AVERAGE
steel_database: str = Field(
default=SteelDatabase.Type350MPa,
title="Steel Database",
description="Database used for the GWP of steel objects",
json_schema_extra={"oneOf": create_one_of_enum(SteelDatabase)},
)
timber_database: str = Field(
default=TimberDatabase.Binderholz2019,
title="Timber Database",
description="Database used for the GWP of timber objects",
json_schema_extra={"oneOf": create_one_of_enum(TimberDatabase)},
)
class RevitCarbonAnalyzer:
"""Main application for analyzing carbon in Revit models."""
def __init__(self):
def __init__(self, steel_database: str, timber_database: str):
self.material_processor = MaterialProcessor()
self.element_processor = ElementProcessor(
material_processor=self.material_processor, logger=Logging()
)
self.carbon_calculator = CarbonCalculator()
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,
)
def analyze_model(self, model_root) -> dict:
"""Analyze a Revit model for carbon emissions."""
@@ -38,19 +67,13 @@ class RevitCarbonAnalyzer:
"skipped_elements": [],
"errors": [],
"total_carbon": 0.0,
"missing_factors": {"timber": [], "steel": []},
}
# Debug: Print number of elements found
element_count = 0
# Process each element
for element in self._iterate_elements(model_root):
element_count += 1
try:
print(
f"Processing element {getattr(element, 'id', 'unknown')}"
) # Debug
element_result = self._process_single_element(element)
print(f"Result status: {element_result['status']}") # Debug
if element_result["status"] == "processed":
results["processed_elements"].append(element_result)
results["total_carbon"] += element_result["total_carbon"]
@@ -59,7 +82,6 @@ class RevitCarbonAnalyzer:
else:
results["errors"].append(element_result)
except Exception as e:
print(f"Error processing element: {str(e)}") # Debug
results["errors"].append(
{
"id": getattr(element, "id", "unknown"),
@@ -68,7 +90,22 @@ class RevitCarbonAnalyzer:
}
)
print(f"Total elements found: {element_count}") # Debug
# Get missing factors
missing_timber, missing_steel = self.carbon_calculator.get_missing_factors()
results["missing_factors"]["timber"] = missing_timber
results["missing_factors"]["steel"] = missing_steel
# Log missing factors
if missing_timber:
print(f"Missing timber factors ({len(missing_timber)}):")
for item in missing_timber:
print(f" - {item}")
if missing_steel:
print(f"Missing steel factors ({len(missing_steel)}):")
for item in missing_steel:
print(f" - {item}")
return results
def _process_single_element(self, element: Dict) -> Dict:
@@ -127,8 +164,21 @@ def automate_function(
) -> None:
"""Program entry point."""
try:
# Get string values from enums if needed
steel_db = function_inputs.steel_database
timber_db = function_inputs.timber_database
# Ensure we're working with string values
if hasattr(steel_db, "value"):
steel_db = steel_db.value
if hasattr(timber_db, "value"):
timber_db = timber_db.value
# Initialize analyzer
analyzer = RevitCarbonAnalyzer()
analyzer = RevitCarbonAnalyzer(
steel_database=steel_db,
timber_database=timber_db,
)
# Get commit root
version_id = automate_context.automation_run_data.triggers[0].payload.version_id
@@ -150,12 +200,46 @@ def automate_function(
# Process results
_process_automation_results(automate_context, results)
# Mark success
automate_context.mark_run_success(
f"Analysis complete. Processed {len(results['processed_elements'])} elements. "
f"Total carbon: {results['total_carbon']:.2f} kgCO2e"
# Prepare detailed success message
success_message = (
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"
)
# Add missing factors to message if any
missing_timber = results["missing_factors"]["timber"]
missing_steel = results["missing_factors"]["steel"]
if missing_timber or missing_steel:
success_message += "\nMissing emission factors detected:\n"
if missing_timber:
success_message += (
f"- Timber ({len(missing_timber)}): {', '.join(missing_timber[:5])}"
)
if len(missing_timber) > 5:
success_message += f" and {len(missing_timber) - 5} more"
success_message += "\n"
if missing_steel:
success_message += (
f"- Steel ({len(missing_steel)}): {', '.join(missing_steel[:5])}"
)
if len(missing_steel) > 5:
success_message += f" and {len(missing_steel) - 5} more"
success_message += "\n"
success_message += "\nThese materials were assigned zero carbon. Consider updating the database."
else:
success_message += (
"\nNOTE: All materials successfully matched with emission factors."
"complete."
)
# Mark success with detailed message
automate_context.mark_run_success(success_message)
except Exception as e:
automate_context.mark_run_failed(f"Analysis failed: {str(e)}")
raise
src/domain/carbon/databases/base.py
+7 -12
View File
@@ -1,4 +1,4 @@
from abc import ABC, abstractmethod
from abc import ABC
from typing import Optional, Dict
from src.domain.carbon.schema import EmissionFactor
@@ -8,18 +8,13 @@ class EmissionFactorDatabase(ABC):
def __init__(self):
self._factors: Dict[str, EmissionFactor] = {}
self._material_aliases: Dict[str, list[str]] = {}
@abstractmethod
def get_factor(self, material_name: str) -> Optional[EmissionFactor]:
"""Get emission factor for a material name"""
pass
material_name = material_name.lower()
for name, factor in self._factors.items():
if name.lower() == material_name:
return factor
def _normalize_material_name(self, name: str) -> str:
"""Normalize material name using aliases"""
normalized = name.lower()
for standard, variations in self._material_aliases.items():
for variation in variations:
if variation in normalized:
normalized = normalized.replace(variation, standard)
return normalized
# If no direct match, return None
return None
src/domain/carbon/databases/enums.py
+21
View File
@@ -0,0 +1,21 @@
from enum import Enum
class TimberDatabase(Enum):
Athena2021 = "ATHENA 2021"
Structurlam2020 = "Structurlam, 2020"
AwcCwc2018 = "AWC, CWC, 2018"
Katerra2020 = "Katerra, 2020"
NordicStructures2018 = "Nordic Structures, 2018"
Binderholz2019 = "Binderholz, 2019"
StructuralamAbbotsford = "Structuralam Abbotsford"
CLFBaselineDocument = "CLF Baseline Document"
IndustryAverage = "INDUSTRY AVERAGE"
class SteelDatabase(Enum):
Type350MPa = "Type 350 MPa"
class ConcreteDatabase(Enum):
pass
src/domain/carbon/databases/epd_global.py
-40
View File
@@ -1,40 +0,0 @@
from typing import Optional
from src.domain.carbon.schema import EmissionDatabase, EmissionFactor
from src.domain.carbon.databases.base import EmissionFactorDatabase
class EPDGlobalDatabase(EmissionFactorDatabase):
"""EPD Global emission factor database implementation"""
def __init__(self):
super().__init__()
self._factors = {
"hot rolled structural steel": EmissionFactor(
value=1.22,
unit="kgCO2e/kg",
database=EmissionDatabase.EPD_GLOBAL,
epd_number="EPD-123-2024",
publication_date="2024-01-01",
valid_until="2029-01-01",
manufacturer="SteelCo",
plant_location="Sheffield, UK",
),
# Add other factors...
}
self._material_aliases = {
"hot rolled": ["hot-rolled", "hot_rolled", "hotrolled"],
"structural steel": ["structural_steel", "struct steel"],
# Add other aliases...
}
def get_factor(self, material_name: str) -> Optional[EmissionFactor]:
"""Get emission factor for a material name, checking variations"""
# Try direct match first
material_name = material_name.lower()
if material_name in self._factors:
return self._factors[material_name]
# Try aliases
normalized_name = self._normalize_material_name(material_name)
return self._factors.get(normalized_name)
src/domain/carbon/databases/steel/__init__.py
View File
src/domain/carbon/databases/steel/steel_350_mpa.py
+81
View File
@@ -0,0 +1,81 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
from src.domain.carbon.databases.enums import SteelDatabase
UNIT = "kgCO₂e/kg"
class Steel350MPa(EmissionFactorDatabase):
"""Database implementation for Type 350 MPa steel emission factors."""
def __init__(self):
super().__init__()
self._factors = {
"Hot Rolled": EmissionFactor(
value=1.22,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-HR",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"HSS": EmissionFactor(
value=1.99,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-HSS",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"Plate": EmissionFactor(
value=1.73,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-PL",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"Rebar": EmissionFactor(
value=0.854,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-RB",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"OWSJ": EmissionFactor(
value=1.380,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-OWSJ",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"Fasteners": EmissionFactor(
value=1.730,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-FST",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
"Metal Deck": EmissionFactor(
value=2.370,
unit=UNIT,
database=SteelDatabase.Type350MPa.value,
epd_number="STEEL-350-MD",
publication_date="2024-01-01",
valid_until="2029-01-01",
),
}
# Set up common aliases for steel types
self._material_aliases = {
"hot rolled": ["hot-rolled", "hot_rolled", "hotrolled"],
"hss": ["hollow structural section", "hollow section", "tube steel"],
"plate": ["steel plate", "flat plate"],
"rebar": ["reinforcing bar", "reinforcement"],
"owsj": ["open web steel joist", "steel joist"],
"fasteners": ["bolts", "screws", "nails", "rivets"],
"metal deck": ["steel deck", "decking"],
}
src/domain/carbon/databases/timber/__init__.py
View File
src/domain/carbon/databases/timber/athena.py
+60
View File
@@ -0,0 +1,60 @@
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class Athena(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=107,
unit=UNIT,
database=TimberDatabase.Athena2021.value,
epd_number="ATHENA-2021-GL",
publication_date="2021-01-01",
valid_until="2026-01-01",
),
"CLT": EmissionFactor(
value=69,
unit="kgCO2e/m3",
database=TimberDatabase.Athena2021.value,
epd_number=UNIT,
publication_date="2021-01-01",
valid_until="2026-01-01",
),
"LVL": EmissionFactor(
value=169,
unit=UNIT,
database=TimberDatabase.Athena2021.value,
epd_number="ATHENA-2021-LVL",
publication_date="2021-01-01",
valid_until="2026-01-01",
),
"Softwood Lumber": EmissionFactor(
value=48,
unit=UNIT,
database=TimberDatabase.Athena2021.value,
epd_number="ATHENA-2021-SWL",
publication_date="2021-01-01",
valid_until="2026-01-01",
),
"Softwood Plywood": EmissionFactor(
value=65,
unit=UNIT,
database=TimberDatabase.Athena2021.value,
epd_number="ATHENA-2021-SWP",
publication_date="2021-01-01",
valid_until="2026-01-01",
),
"Oriented Strand Board": EmissionFactor(
value=182,
unit=UNIT,
database=TimberDatabase.Athena2021.value,
epd_number="ATHENA-2021-OSB",
publication_date="2021-01-01",
valid_until="2026-01-01",
),
}
src/domain/carbon/databases/timber/awc_cwc.py
+68
View File
@@ -0,0 +1,68 @@
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class AwcCwc(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=137,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-GL",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"LVL": EmissionFactor(
value=361,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-LVL",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"Softwood Lumber": EmissionFactor(
value=63,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-SWL",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"Softwood Plywood": EmissionFactor(
value=219,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-SWP",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"Wood Joists": EmissionFactor(
value=2,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-WJ",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"Redwood Lumber": EmissionFactor(
value=38,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-RWL",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"Oriented Strand Board": EmissionFactor(
value=243,
unit=UNIT,
database=TimberDatabase.AwcCwc2018.value,
epd_number="AWC-2018-OSB",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
}
src/domain/carbon/databases/timber/binderholz.py
+28
View File
@@ -0,0 +1,28 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class Binderholz(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=118,
unit=UNIT,
database=TimberDatabase.Binderholz2019.value,
epd_number="BH-2019-GL",
publication_date="2019-01-01",
valid_until="2024-01-01",
),
"CLT": EmissionFactor(
value=200,
unit=UNIT,
database=TimberDatabase.Binderholz2019.value,
epd_number="BH-2019-CLT",
publication_date="2019-01-01",
valid_until="2024-01-01",
),
}
src/domain/carbon/databases/timber/clf_baseline_document.py
+28
View File
@@ -0,0 +1,28 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class CLFBaselineDocument(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"CLT": EmissionFactor(
value=137,
unit=UNIT,
database=TimberDatabase.CLFBaselineDocument.value,
epd_number="CLF-CLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"GLT/NLT/DLT": EmissionFactor(
value=109,
unit=UNIT,
database=TimberDatabase.CLFBaselineDocument.value,
epd_number="CLF-GLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
}
src/domain/carbon/databases/timber/industry_average.py
+84
View File
@@ -0,0 +1,84 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class IndustryAverage(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=113,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-GL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"CLT": EmissionFactor(
value=135,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-CLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"LVL": EmissionFactor(
value=265,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-LVL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"Softwood Lumber": EmissionFactor(
value=56,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-SWL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"Softwood Plywood": EmissionFactor(
value=142,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-SWP",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"Wood Joists": EmissionFactor(
value=2,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-WJ",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"Redwood Lumber": EmissionFactor(
value=38,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-RWL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"Oriented Strand Board": EmissionFactor(
value=212,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-OSB",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"GLT/NLT/DLT": EmissionFactor(
value=123,
unit=UNIT,
database=TimberDatabase.IndustryAverage.value,
epd_number="IA-GLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
}
src/domain/carbon/databases/timber/katerra.py
+20
View File
@@ -0,0 +1,20 @@
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class Katerra(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"CLT": EmissionFactor(
value=158,
unit=UNIT,
database=TimberDatabase.Katerra2020.value,
epd_number="KAT-2020-CLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
}
src/domain/carbon/databases/timber/nordic_structures.py
+28
View File
@@ -0,0 +1,28 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class NordicStructures(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=100,
unit=UNIT,
database=TimberDatabase.NordicStructures2018.value,
epd_number="NS-2018-GL",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
"CLT": EmissionFactor(
value=122,
unit=UNIT,
database=TimberDatabase.NordicStructures2018.value,
epd_number="NS-2018-CLT",
publication_date="2018-01-01",
valid_until="2023-01-01",
),
}
src/domain/carbon/databases/timber/structuralam_abbotsford.py
+20
View File
@@ -0,0 +1,20 @@
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class StructuralamAbbotsford(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=103,
unit=UNIT,
database=TimberDatabase.StructuralamAbbotsford.value,
epd_number="SA-GL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
}
src/domain/carbon/databases/timber/structurlam.py
+28
View File
@@ -0,0 +1,28 @@
from src.domain.carbon.databases.enums import TimberDatabase
from src.domain.carbon.databases.base import EmissionFactorDatabase
from src.domain.carbon.schema import EmissionFactor
UNIT = "kgCO₂e/m³"
class Structurlam(EmissionFactorDatabase):
def __init__(self):
super().__init__()
self._factors = {
"Glulam": EmissionFactor(
value=115,
unit=UNIT,
database=TimberDatabase.Structurlam2020.value,
epd_number="STR-2020-GL",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
"CLT": EmissionFactor(
value=124,
unit=UNIT,
database=TimberDatabase.Structurlam2020.value,
epd_number="STR-2020-CLT",
publication_date="2020-01-01",
valid_until="2025-01-01",
),
}
src/domain/carbon/emission_factor_registry.py
+209
View File
@@ -0,0 +1,209 @@
from typing import Optional, Dict
from src.domain.carbon.schema import EmissionFactor
from src.domain.carbon.databases.enums import (
TimberDatabase,
SteelDatabase,
ConcreteDatabase,
)
# Import timber databases
from src.domain.carbon.databases.timber.athena import Athena
from src.domain.carbon.databases.timber.structurlam import Structurlam
from src.domain.carbon.databases.timber.awc_cwc import AwcCwc
from src.domain.carbon.databases.timber.katerra import Katerra
from src.domain.carbon.databases.timber.nordic_structures import NordicStructures
from src.domain.carbon.databases.timber.binderholz import Binderholz
from src.domain.carbon.databases.timber.structuralam_abbotsford import (
StructuralamAbbotsford,
)
from src.domain.carbon.databases.timber.clf_baseline_document import CLFBaselineDocument
from src.domain.carbon.databases.timber.industry_average import IndustryAverage
# Import steel databases
from src.domain.carbon.databases.steel.steel_350_mpa import Steel350MPa
class EmissionFactorRegistry:
"""Registry of available emission factor databases"""
def __init__(self):
self._timber_databases = {}
self._steel_databases = {}
self._concrete_databases = {}
# Material aliases for normalization
self._timber_aliases = {
"clt": ["cross laminated timber", "cross-laminated timber"],
"glulam": [
"glue laminated timber",
"glued laminated timber",
"glulam beam",
],
"lvl": ["laminated veneer lumber"],
"softwood lumber": ["dimensional lumber", "sawn lumber", "softwood"],
"softwood plywood": ["plywood", "softwood ply"],
"oriented strand board": ["osb", "osb board"],
"glt/nlt/dlt": [
"glt",
"nlt",
"dlt",
"nail laminated timber",
"dowel laminated timber",
],
}
self._steel_aliases = {
"hot rolled": [
"hot-rolled",
"hot_rolled",
"hotrolled",
"345 MPa",
"350W",
"350W(1)",
],
"hss": ["hollow structural section", "hollow section", "tube"],
"plate": ["flat plate"],
"rebar": ["reinforcing bar", "reinforcement"],
"owsj": ["open web steel joist", "steel joist"],
"fasteners": ["bolts", "screws", "nails", "rivets"],
"metal deck": ["deck", "decking"],
}
self._concrete_aliases = {
# To be added when concrete implementation is needed
}
# Initialize all database instances
self._init_timber_databases()
self._init_steel_databases()
# self._init_concrete_databases() - empty for now
def _init_timber_databases(self) -> None:
"""Initialize timber database implementations"""
self._timber_databases = {
TimberDatabase.Athena2021.value: Athena(),
TimberDatabase.Structurlam2020.value: Structurlam(),
TimberDatabase.AwcCwc2018.value: AwcCwc(),
TimberDatabase.Katerra2020.value: Katerra(),
TimberDatabase.NordicStructures2018.value: NordicStructures(),
TimberDatabase.Binderholz2019.value: Binderholz(),
TimberDatabase.StructuralamAbbotsford.value: StructuralamAbbotsford(),
TimberDatabase.CLFBaselineDocument.value: CLFBaselineDocument(),
TimberDatabase.IndustryAverage.value: IndustryAverage(),
}
def _init_steel_databases(self) -> None:
"""Initialize steel database implementations"""
self._steel_databases = {
SteelDatabase.Type350MPa.value: Steel350MPa(),
}
@staticmethod
def _normalize_material_name(name: str, aliases: Dict[str, list]) -> str:
"""Normalize material name using centralized aliases with enhanced matching.
This improved version handles:
- Case insensitivity
- Direct matches (exact)
- Substring matches (contains)
- Special known cases
"""
# Convert to lowercase for case-insensitive comparison
name = name.lower().strip()
# Special case handling
if any(
steel_name in name
for steel_name in ["345 mpa", "350w", "steel 345", "default_steel"]
):
return "Hot Rolled" # Map all these variants to Hot Rolled steel
# Check for direct match with standard names
for standard_name in aliases.keys():
if standard_name.lower() == name:
return standard_name
# Check for standard name appearing as substring
for standard_name in aliases.keys():
if standard_name.lower() in name:
return standard_name
# Check aliases
for standard_name, variations in aliases.items():
for variation in variations:
if variation.lower() == name or variation.lower() in name:
return standard_name
return name
def get_timber_factor(
self, material_name: str, database: str
) -> Optional[EmissionFactor]:
"""Get emission factor for timber from specified database with name normalization"""
db = self._timber_databases.get(database)
if not db:
raise ValueError(f"Unknown timber database: {database}")
# Try direct lookup first
factor = db.get_factor(material_name)
if factor:
return factor
# If not found, try normalized name
normalized_name = self._normalize_material_name(
material_name, self._timber_aliases
)
return db.get_factor(normalized_name)
def get_steel_factor(
self, material_name: str, database: str
) -> Optional[EmissionFactor]:
"""Get emission factor for steel from specified database with name normalization"""
db = self._steel_databases.get(database)
if not db:
raise ValueError(f"Unknown steel database: {database}")
# Try direct lookup first
factor = db.get_factor(material_name)
if factor:
return factor
# If not found, try normalized name
normalized_name = self._normalize_material_name(
material_name, self._steel_aliases
)
return db.get_factor(normalized_name)
def get_concrete_factor(
self, material_name: str, database: str
) -> Optional[EmissionFactor]:
"""Get emission factor for concrete from specified database"""
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
def list_timber_databases(self) -> list[str]:
"""List all registered timber databases"""
return list(self._timber_databases.keys())
def list_steel_databases(self) -> list[str]:
"""List all registered steel databases"""
return list(self._steel_databases.keys())
def list_concrete_databases(self) -> list[str]:
"""List all registered concrete databases"""
return list(self._concrete_databases.keys())
src/domain/carbon/registry.py
-29
View File
@@ -1,29 +0,0 @@
from typing import Optional
from src.domain.carbon.schema import EmissionDatabase, EmissionFactor
from src.domain.carbon.databases.base import EmissionFactorDatabase
class EmissionFactorRegistry:
"""Registry of available emission factor databases"""
def __init__(self):
self._databases: dict[EmissionDatabase, EmissionFactorDatabase] = {}
def register_database(
self, database_type: EmissionDatabase, implementation: EmissionFactorDatabase
) -> None:
"""Register a new database implementation"""
self._databases[database_type] = implementation
def get_factor(
self, material_name: str, database: EmissionDatabase
) -> Optional[EmissionFactor]:
"""Get emission factor from specified database"""
db = self._databases.get(database)
if not db:
raise ValueError(f"Unknown database: {database}")
return db.get_factor(material_name)
def list_databases(self) -> list[EmissionDatabase]:
"""List all registered databases"""
return list(self._databases.keys())
src/domain/carbon/schema.py
+1 -10
View File
@@ -1,23 +1,14 @@
from dataclasses import dataclass
from enum import Enum
from typing import Optional
class EmissionDatabase(str, Enum):
"""Available emission factor databases"""
EPD_GLOBAL = "EPD Global"
ICE = "Inventory of Carbon and Energy"
EC3 = "EC3 Database"
@dataclass
class EmissionFactor:
"""Emission factor with metadata"""
value: float
unit: str # e.g., "kgCO2e/kg" or "kgCO2e/m3"
database: EmissionDatabase
database: str
epd_number: Optional[str] = None
publication_date: Optional[str] = None
valid_until: Optional[str] = None
src/services/carbon_calculator.py
+81 -30
View File
@@ -1,26 +1,35 @@
from typing import Dict
from typing import Dict, Optional, Tuple, List
from src.domain.carbon.emission_factor_registry import EmissionFactorRegistry
from src.domain.types import BuildingElement, CarbonResult, Material, MaterialType
class CarbonCalculator:
"""Calculates embodied carbon for building elements."""
def __init__(self):
# Carbon factors (kgCO2e/kg for metals, kgCO2e/m3 for wood)
self.metal_factors = {
"Hot Rolled": 1.22,
"HSS": 1.99,
"Plate": 1.73,
"Rebar": 0.854,
"default": 1.22, # Default to hot rolled
}
def __init__(
self,
steel_database: str,
timber_database: str,
concrete_database: Optional[str] = None,
):
# Store database selections
self._steel_database = steel_database
self._timber_database = timber_database
self._concrete_database = concrete_database
self.wood_factors = {
"CLT": 135,
"Glulam": 113,
"default": 135, # Default to CLT
}
# Initialize registry
self._registry = EmissionFactorRegistry()
# 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()
def calculate_carbon(self, element: BuildingElement) -> Dict[str, CarbonResult]:
"""Calculate carbon emissions for an element's materials."""
@@ -31,7 +40,19 @@ class CarbonCalculator:
result = self._calculate_material_carbon(material)
results[material.properties.name] = result
except Exception as e:
# Log error but continue with other materials
# Track missing factors
if "No emission factor found" in str(e):
if material.type == MaterialType.WOOD:
material_key = (
material.properties.structural_asset
or material.properties.name
)
self._missing_timber_factors.add(material_key)
elif material.type == MaterialType.METAL:
self._missing_steel_factors.add(
material.grade or material.properties.name
)
print(
f"Error calculating carbon for {material.properties.name}: {str(e)}"
)
@@ -54,34 +75,64 @@ class CarbonCalculator:
if not material.mass:
raise ValueError("Mass required for metal carbon calculation")
# Determine factor based on grade or use default
factor = self.metal_factors.get(material.grade, self.metal_factors["default"])
# Get factor from cache or registry
if material.grade not in self._steel_factors_cache:
factor = self._registry.get_steel_factor(
material.grade, self._steel_database
)
if not factor:
raise ValueError(
f"No emission factor found for metal grade: {material.grade}"
)
self._steel_factors_cache[material.grade] = factor
factor = self._steel_factors_cache[material.grade]
return CarbonResult(
factor=factor, total_carbon=material.mass * factor, category="Metal"
factor=factor.value,
total_carbon=material.mass * factor.value,
category="Metal",
)
def _calculate_wood_carbon(self, material: Material) -> CarbonResult:
"""Calculate carbon emissions for wood."""
# Determine factor based on structural asset or use default
factor = self.wood_factors.get(
material.properties.structural_asset, self.wood_factors["default"]
)
material_name = material.properties.structural_asset
# Use name as a fallback if structural_asset is None
if material_name is None:
# Extract material type from name
material_name = material.properties.name
# Get factor from cache or registry
if material_name not in self._timber_factors_cache:
factor = self._registry.get_timber_factor(
material_name, self._timber_database
)
if not factor:
raise ValueError(
f"No emission factor found for wood type: {material_name}"
)
self._timber_factors_cache[material_name] = factor
factor = self._timber_factors_cache[material_name]
return CarbonResult(
factor=factor,
total_carbon=material.properties.volume * factor,
factor=factor.value,
total_carbon=material.properties.volume * factor.value,
category="Wood",
)
@staticmethod
def _calculate_concrete_carbon(material: Material) -> CarbonResult:
def _calculate_concrete_carbon(self, material: Material) -> CarbonResult:
"""Calculate carbon emissions for concrete."""
# TODO: Implement concrete-specific carbon calculation
# This would involve looking up factors based on concrete grade
# and calculating based on volume or mass depending on the data source
# TODO: Implement concrete-specific calculation when concrete database is added
return CarbonResult(
factor=0.0, # Placeholder
total_carbon=0.0, # Placeholder
category="Concrete",
)
# TODO: Remove
def get_missing_factors(self) -> Tuple[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)),
)
src/services/element_processor.py
+1 -11
View File
@@ -48,30 +48,20 @@ class ElementProcessor:
def _is_valid_element(self, element) -> bool:
"""Validate if element should be processed."""
element_id = getattr(element, "id", "unknown")
# Debug logs
print(f"\nValidating element {element_id}")
print(f"speckle_type: {getattr(element, 'speckle_type', None)}")
print(f"has properties: {hasattr(element, 'properties')}")
# Skip geometry elements
if getattr(element, "speckle_type", None) in self.SKIP_TYPES:
print("Skipped: geometry element")
return False
# Must have properties
if not hasattr(element, "properties"):
print("Skipped: no properties")
return False
# Must have material quantities
properties = getattr(element, "properties")
if "Material Quantities" not in properties: # Changed from hasattr to dictionary access
print("Skipped: no Material Quantities")
if "Material Quantities" not in properties:
return False
print(f"Material Quantities found: {properties['Material Quantities']}")
return True
@staticmethod
src/services/material_processor.py
+1 -3
View File
@@ -14,9 +14,7 @@ class MaterialProcessor:
properties = MaterialProperties(
name=raw_material["materialName"],
volume=raw_material["volume"]["value"],
density=raw_material.get("density", {}).get(
"value"
), # Using .get() for optional fields
density=raw_material.get("density", {}).get("value"),
structural_asset=raw_material.get("structuralAsset"),
compressive_strength=raw_material.get("compressiveStrength", {}).get(
"value"
tests/test_registry.py
+41
View File
@@ -0,0 +1,41 @@
import pytest
from src.domain.carbon.databases.enums import TimberDatabase, SteelDatabase
from src.domain.carbon.emission_factor_registry import EmissionFactorRegistry
class TestRegistry:
"""Test suite for the EmissionFactorRegistry"""
@pytest.fixture
def registry(self):
"""Create and return a registry instance"""
return EmissionFactorRegistry()
def test_timber_database_lookup(self, registry):
"""Test direct lookup of timber factors"""
# Test each database
factor = registry.get_timber_factor(
"FE_CLT Floor Panel (1)", TimberDatabase.Athena2021.value
)
assert factor is not None
assert factor.value == 69
factor = registry.get_timber_factor(
"FE_Glulam", TimberDatabase.Binderholz2019.value
)
assert factor is not None
assert factor.value == 118
def test_steel_database_lookup(self, registry):
"""Test direct lookup of steel factors"""
factor = registry.get_steel_factor(
"Metal - Steel CSA G40", SteelDatabase.Type350MPa.value
)
assert factor is not None
assert factor.value == 1.22
def test_invalid_database(self, registry):
"""Test error handling for invalid database"""
with pytest.raises(ValueError, match="Unknown timber database"):
registry.get_timber_factor("CLT", "NonExistentDatabase")