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

feat: add databases

Browse Source
This commit is contained in:
Björn Steinhagen
2025-02-24 15:11:54 +01:00
parent 973accd2be
commit 2020048dca
20 changed files with 700 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+5 -23
View File
@@ -1,5 +1,4 @@
from collections import defaultdict
from enum import Enum
from pydantic import Field
from speckle_automate import (
@@ -10,33 +9,13 @@ 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
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"
# TODO
class ConcreteDatabase(Enum):
pass
def create_one_of_enum(enum_cls):
"""
Helper function to create a JSON schema from an Enum class.
@@ -173,7 +152,10 @@ def automate_function(
"""Program entry point."""
try:
# Initialize analyzer
analyzer = RevitCarbonAnalyzer()
analyzer = RevitCarbonAnalyzer(
steel_database=function_inputs.steel_database,
timber_database=function_inputs.timber_database,
)
# Get commit root
version_id = automate_context.automation_run_data.triggers[0].payload.version_id
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
+181
View File
@@ -0,0 +1,181 @@
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",
"glue laminated timber",
"nail laminated timber",
"dowel laminated timber",
],
}
self._steel_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"],
}
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"""
name = name.lower()
# Check if name contains any alias
for standard_name, variations in aliases.items():
if name == standard_name:
return standard_name
for variation in variations:
if variation in name:
return standard_name
# If no match found, return original
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
+34 -42
View File
@@ -1,7 +1,6 @@
from typing import Dict
from typing import Dict, Optional
from src.domain.carbon.registry import EmissionFactorRegistry
from src.domain.carbon.schema import EmissionDatabase, EmissionFactor
from src.domain.carbon.emission_factor_registry import EmissionFactorRegistry
from src.domain.types import BuildingElement, CarbonResult, Material, MaterialType
@@ -10,39 +9,22 @@ class CarbonCalculator:
def __init__(
self,
metal_database: EmissionDatabase,
wood_database: EmissionDatabase,
registry: EmissionFactorRegistry,
steel_database: str,
timber_database: str,
concrete_database: Optional[str] = None,
):
self._registry = registry
self._metal_factors: dict[str, EmissionFactor] = {}
self._wood_factors: dict[str, EmissionFactor] = {}
# Store database selections
self._steel_database = steel_database
self._timber_database = timber_database
self._concrete_database = concrete_database
# Cache all metal factors
for grade in [
"Hot Rolled",
"HSS",
"Plate",
"Rebar",
"OWSJ",
"Fasteners",
"Metal Deck",
]:
factor = self._registry.get_factor(grade, metal_database)
if factor:
self._metal_factors[grade] = factor
# Initialize registry
self._registry = EmissionFactorRegistry()
# Cache all wood factors
for wood_type in [
"CLT",
"Glulam",
"LVL",
"Softwood Lumber",
"Softwood Plywood",
]:
factor = self._registry.get_factor(wood_type, wood_database)
if factor:
self._wood_factors[wood_type] = factor
# Cache common material factors to avoid repeated lookups
self._steel_factors_cache = {}
self._timber_factors_cache = {}
self._concrete_factors_cache = {}
def calculate_carbon(self, element: BuildingElement) -> Dict[str, CarbonResult]:
"""Calculate carbon emissions for an element's materials."""
@@ -76,12 +58,18 @@ class CarbonCalculator:
if not material.mass:
raise ValueError("Mass required for metal carbon calculation")
factor = self._metal_factors.get(material.grade)
# 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.value,
total_carbon=material.mass * factor.value,
@@ -90,25 +78,29 @@ class CarbonCalculator:
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)
structural_asset = material.properties.structural_asset
# Get factor from cache or registry
if structural_asset not in self._timber_factors_cache:
factor = self._registry.get_timber_factor(
structural_asset, self._timber_database
)
if not factor:
raise ValueError(
f"No emission factor found for wood type: {material.properties.structural_asset}"
f"No emission factor found for wood type: {structural_asset}"
)
self._timber_factors_cache[structural_asset] = factor
factor = self._timber_factors_cache[structural_asset]
return CarbonResult(
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