refactor(logic): everything
This commit is contained in:
Björn Steinhagen
@@ -1,24 +1,124 @@
|
||||
from collections import defaultdict
|
||||
|
||||
from speckle_automate import (
|
||||
AutomateBase,
|
||||
AutomationContext,
|
||||
execute_automate_function,
|
||||
)
|
||||
|
||||
from src.applications.revit.revit_material_processor import RevitMaterialProcessor
|
||||
from src.applications.revit.revit_carbon_processor import RevitCarbonProcessor
|
||||
from src.applications.revit.revit_compliance import RevitCompliance
|
||||
from src.applications.revit.revit_model import RevitElementProcessor
|
||||
from src.applications.revit.revit_source_validator import RevitSourceValidator
|
||||
from src.carbon.aggregator import MassAggregator
|
||||
from src.applications.revit.revit_logger import RevitLogger
|
||||
from src.carbon import WoodSupplier
|
||||
from typing import Dict, Generator, Any, List
|
||||
|
||||
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
|
||||
|
||||
|
||||
# TODO: Function inputs
|
||||
class FunctionInputs(AutomateBase):
|
||||
"""User-defined function inputs."""
|
||||
|
||||
wood_supplier: WoodSupplier = WoodSupplier.INDUSTRY_AVERAGE
|
||||
# wood_supplier: WoodSupplier = WoodSupplier.INDUSTRY_AVERAGE
|
||||
|
||||
|
||||
class RevitCarbonAnalyzer:
|
||||
"""Main application for analyzing carbon in Revit models."""
|
||||
|
||||
def __init__(self):
|
||||
self.material_processor = MaterialProcessor()
|
||||
self.element_processor = ElementProcessor(
|
||||
material_processor=self.material_processor, logger=Logging()
|
||||
)
|
||||
self.carbon_calculator = CarbonCalculator()
|
||||
|
||||
def analyze_model(self, model_root) -> dict:
|
||||
"""Analyze a Revit model for carbon emissions."""
|
||||
results = {
|
||||
"processed_elements": [],
|
||||
"skipped_elements": [],
|
||||
"errors": [],
|
||||
"total_carbon": 0.0,
|
||||
}
|
||||
|
||||
# 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"]
|
||||
elif element_result["status"] == "skipped":
|
||||
results["skipped_elements"].append(element_result)
|
||||
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"),
|
||||
"error": str(e),
|
||||
"status": "error",
|
||||
}
|
||||
)
|
||||
|
||||
print(f"Total elements found: {element_count}") # Debug
|
||||
return results
|
||||
|
||||
def _process_single_element(self, element: Dict) -> Dict:
|
||||
"""Process a single element and return its results."""
|
||||
element_id = getattr(element, "id", "unknown")
|
||||
|
||||
# Process element
|
||||
processed_element = self.element_processor.process_element(element)
|
||||
if not processed_element:
|
||||
return {
|
||||
"id": element_id,
|
||||
"status": "skipped",
|
||||
"reason": "Invalid element structure",
|
||||
}
|
||||
|
||||
# Calculate carbon
|
||||
try:
|
||||
carbon_results = self.carbon_calculator.calculate_carbon(processed_element)
|
||||
return {
|
||||
"id": element_id,
|
||||
"status": "processed",
|
||||
"level": processed_element.level,
|
||||
"category": processed_element.category.value,
|
||||
"materials": [
|
||||
{
|
||||
"name": m.properties.name,
|
||||
"type": m.type.value,
|
||||
"volume": m.properties.volume,
|
||||
# Add other material properties as needed
|
||||
}
|
||||
for m in processed_element.materials
|
||||
],
|
||||
"carbon_results": carbon_results,
|
||||
"total_carbon": sum(r.total_carbon for r in carbon_results.values()),
|
||||
}
|
||||
except Exception as e:
|
||||
return {
|
||||
"id": element_id,
|
||||
"status": "error",
|
||||
"error": f"Carbon calculation failed: {str(e)}",
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _iterate_elements(model_data) -> Generator[Dict, None, None]:
|
||||
"""Iterate through all elements in the model."""
|
||||
for level in getattr(model_data, "elements", []):
|
||||
for type_group in getattr(level, "elements", []):
|
||||
for element_group in getattr(type_group, "elements", []):
|
||||
for element in getattr(element_group, "elements", []):
|
||||
yield element
|
||||
|
||||
|
||||
def automate_function(
|
||||
@@ -27,113 +127,88 @@ def automate_function(
|
||||
) -> None:
|
||||
"""Program entry point."""
|
||||
try:
|
||||
# Get version data
|
||||
# Initialize analyzer
|
||||
analyzer = RevitCarbonAnalyzer()
|
||||
|
||||
# Get commit root
|
||||
version_id = automate_context.automation_run_data.triggers[0].payload.version_id
|
||||
commit_root = automate_context.speckle_client.commit.get(
|
||||
automate_context.automation_run_data.project_id, version_id
|
||||
)
|
||||
|
||||
# Get model root
|
||||
model_root = automate_context.receive_version()
|
||||
|
||||
# Validate source application
|
||||
source_validator = RevitSourceValidator() # Built for revit, therefore check
|
||||
if not source_validator.validate_source_application(
|
||||
commit_root.sourceApplication
|
||||
):
|
||||
automate_context.mark_run_failed(
|
||||
f"Automation requires models from Revit. Received: {commit_root.sourceApplication}"
|
||||
)
|
||||
return
|
||||
if not source_validator.validate_connector_version(
|
||||
int(getattr(model_root, "version", 2))
|
||||
):
|
||||
automate_context.mark_run_failed(
|
||||
"Automation required Revit models using the v3 "
|
||||
"connector. Received: v2."
|
||||
)
|
||||
# Validate Revit source
|
||||
if not _validate_revit_source(commit_root):
|
||||
automate_context.mark_run_failed("Model must be from Revit")
|
||||
return
|
||||
|
||||
# Create processor chain and get logger for results
|
||||
processor = initialize_revit_processor()
|
||||
# Run analysis - convert Speckle model to dict for processing
|
||||
results = analyzer.analyze_model(model_root)
|
||||
|
||||
# Analyze elements
|
||||
processor.analyze_elements(model_root)
|
||||
# Process results
|
||||
_process_automation_results(automate_context, results)
|
||||
|
||||
# Logger information - successes
|
||||
logger_infos = processor.logger.get_info_summary()
|
||||
logger_successes = processor.logger.get_success_summary()
|
||||
logger_warnings = processor.logger.get_warnings_summary()
|
||||
logger_failures = processor.logger.get_errors_summary()
|
||||
|
||||
for category, object_ids in logger_successes.items():
|
||||
automate_context.attach_success_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Carbon calculations completed successfully for these elements!",
|
||||
)
|
||||
|
||||
for category, object_ids in logger_infos.items():
|
||||
automate_context.attach_info_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Elements deemed not applicable and skipped.",
|
||||
)
|
||||
|
||||
for category, object_ids in logger_warnings.items():
|
||||
automate_context.attach_warning_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Elements requiring careful review.",
|
||||
)
|
||||
|
||||
for category, object_ids in logger_failures.items():
|
||||
automate_context.attach_error_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Failure processing the following elements.",
|
||||
)
|
||||
|
||||
# TODO: Create new version
|
||||
# automate_context.create_new_version_in_project(model_root, "dev", "")
|
||||
|
||||
automate_context.mark_run_success("Processing completed successfully.")
|
||||
# Mark success
|
||||
automate_context.mark_run_success(
|
||||
f"Analysis complete. Processed {len(results['processed_elements'])} elements. "
|
||||
f"Total carbon: {results['total_carbon']:.2f} kgCO2e"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
automate_context.mark_run_failed(f"Processing failed: {str(e)}")
|
||||
raise # Re-raise for proper error tracking
|
||||
automate_context.mark_run_failed(f"Analysis failed: {str(e)}")
|
||||
raise
|
||||
|
||||
|
||||
# TODO instead of hard-coding revit, demo a factory method to inject implementations based on
|
||||
# function input
|
||||
def initialize_revit_processor() -> RevitElementProcessor:
|
||||
"""Configures and wires up processor components with dependencies.
|
||||
def _validate_revit_source(commit_root: Any) -> bool:
|
||||
"""Validate that the model is from Revit."""
|
||||
source_app = getattr(commit_root, "sourceApplication", "").lower()
|
||||
return source_app.startswith("revit")
|
||||
|
||||
Creates core system components (logger, aggregator) and configures processors
|
||||
with required dependencies injected.
|
||||
|
||||
Returns:
|
||||
tuple:
|
||||
- Model: Main processor configured with dependencies
|
||||
"""
|
||||
def _process_automation_results(
|
||||
automate_context: AutomationContext, results: dict
|
||||
) -> None:
|
||||
"""Process results and attach them to the automation context."""
|
||||
# Group results by category
|
||||
successes: Dict[str, List[str]] = defaultdict(list)
|
||||
warnings: Dict[str, List[str]] = defaultdict(list)
|
||||
errors: Dict[str, List[str]] = defaultdict(list)
|
||||
|
||||
# Core components
|
||||
logger = RevitLogger() # For tracking issues
|
||||
mass_aggregator = MassAggregator() # For collecting computed masses
|
||||
# TODO: results_aggregator = ResultAggregator and get rid of mass_aggregator
|
||||
# Group successful elements
|
||||
for element in results["processed_elements"]:
|
||||
successes["Carbon Analysis"].append(element["id"])
|
||||
|
||||
# Create processors
|
||||
material_processor = RevitMaterialProcessor(
|
||||
mass_aggregator, logger
|
||||
) # Material handler to "inject"
|
||||
carbon_processor = RevitCarbonProcessor()
|
||||
compliance_checker = RevitCompliance(logger) # Compliance checker to "inject"
|
||||
# Group skipped elements
|
||||
for element in results["skipped_elements"]:
|
||||
warnings["Skipped Elements"].append(element["id"])
|
||||
|
||||
# Create and return the main processor with dependencies "injected"
|
||||
return RevitElementProcessor(
|
||||
material_processor=material_processor,
|
||||
carbon_processor=carbon_processor,
|
||||
compliance_checker=compliance_checker,
|
||||
logger=logger,
|
||||
)
|
||||
# Group errors
|
||||
for element in results["errors"]:
|
||||
errors["Processing Errors"].append(element["id"])
|
||||
|
||||
# Attach grouped results
|
||||
for category, object_ids in successes.items():
|
||||
automate_context.attach_success_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Carbon calculations completed successfully for these elements!",
|
||||
)
|
||||
|
||||
for category, object_ids in warnings.items():
|
||||
automate_context.attach_warning_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Elements requiring careful review.",
|
||||
)
|
||||
|
||||
for category, object_ids in errors.items():
|
||||
automate_context.attach_error_to_objects(
|
||||
category=category,
|
||||
object_ids=object_ids,
|
||||
message="Failure processing the following elements.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -1,57 +0,0 @@
|
||||
from src.core.base import CarbonProcessor
|
||||
from src.core.types import MetalClass, WoodClass
|
||||
from src.carbon.types import CarbonData
|
||||
from src.carbon.data import wood_factors, metal_factors
|
||||
from src.applications.revit.utils.material_type_handler import (
|
||||
MaterialType,
|
||||
)
|
||||
from src.applications.revit.utils.constants import (
|
||||
PROPERTIES,
|
||||
)
|
||||
import json
|
||||
|
||||
|
||||
class RevitCarbonProcessor(CarbonProcessor):
|
||||
"""Implementation of CarbonProcessor for Revit context."""
|
||||
|
||||
def process(self, model_object) -> CarbonData:
|
||||
"""Compute embodied carbon per-element based previously asserted material properties.
|
||||
|
||||
Args:
|
||||
model_object (Any): Model object to process
|
||||
"""
|
||||
material_type = model_object[PROPERTIES]["Embodied Carbon Data"]["type"]
|
||||
|
||||
print(material_type)
|
||||
|
||||
match material_type:
|
||||
case MaterialType.CONCRETE.value:
|
||||
# TODO
|
||||
pass
|
||||
case MaterialType.WOOD.value:
|
||||
# TODO
|
||||
pass
|
||||
case MaterialType.METAL.value:
|
||||
material_quantities = model_object[PROPERTIES]["Material Quantities"][
|
||||
"FE_Steel"
|
||||
] # NOTE: This is dangerous.
|
||||
volume, density = (
|
||||
material_quantities.volume.value,
|
||||
material_quantities.density.value,
|
||||
)
|
||||
mass = volume * density
|
||||
|
||||
# Default to hot rolled
|
||||
factor = metal_factors[MetalClass.HOT_ROLLED]
|
||||
|
||||
embodied_carbon = mass * factor
|
||||
|
||||
model_object[PROPERTIES]["Embodied Carbon Data"][
|
||||
"category"
|
||||
] = MetalClass.HOT_ROLLED
|
||||
model_object[PROPERTIES]["Embodied Carbon Data"][
|
||||
"factor"
|
||||
] = factor # TODO: Append with units
|
||||
model_object[PROPERTIES]["Embodied Carbon Data"][
|
||||
"embodied_carbon"
|
||||
] = embodied_carbon # TODO: Append with units
|
||||
@@ -1,91 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
from src.core.base import Compliance, Logger
|
||||
from src.applications.revit.utils.constants import (
|
||||
ID,
|
||||
SPECKLE_TYPE,
|
||||
LINE,
|
||||
ARC,
|
||||
CIRCLE,
|
||||
PROPERTIES,
|
||||
MATERIAL_QUANTITIES,
|
||||
VOLUME,
|
||||
MATERIAL_CATEGORY,
|
||||
MATERIAL_CLASS,
|
||||
MATERIAL_NAME,
|
||||
)
|
||||
|
||||
|
||||
class RevitCompliance(Compliance):
|
||||
"""Implementation of the ComplianceChecker in the context of Revit.
|
||||
Checks if elements contain required properties for carbon calculations.
|
||||
"""
|
||||
|
||||
def __init__(self, logger: Logger):
|
||||
self._logger = logger
|
||||
|
||||
def check_compliance(self, element: Any) -> bool:
|
||||
"""
|
||||
Validates element and returns validation result with material data if valid.
|
||||
|
||||
Args:
|
||||
element: Element to validate
|
||||
|
||||
Returns:
|
||||
ValidationResult containing validation status and material data if valid
|
||||
"""
|
||||
|
||||
# Check ID
|
||||
object_id = getattr(element, ID, None)
|
||||
if not object_id:
|
||||
raise ValueError("Should have an id.")
|
||||
|
||||
# Skip geometry elements
|
||||
speckle_type = getattr(element, SPECKLE_TYPE, None)
|
||||
if speckle_type in [LINE, ARC, CIRCLE]:
|
||||
self._logger.log_info(
|
||||
object_id, "Skipped Geometry", "Skipped based on 'speckle_type'."
|
||||
)
|
||||
return False
|
||||
|
||||
# Check Properties
|
||||
properties = getattr(element, PROPERTIES, None)
|
||||
if not properties:
|
||||
self._logger.log_error(
|
||||
object_id,
|
||||
"Missing 'properties'",
|
||||
"Valid object without a 'properties' " "attribute shouldn't happen.",
|
||||
)
|
||||
return False
|
||||
|
||||
# Check Material Quantities
|
||||
material_quantities = properties.get(MATERIAL_QUANTITIES, None)
|
||||
if not material_quantities:
|
||||
self._logger.log_warning(
|
||||
object_id,
|
||||
"Missing 'Material Quantities'",
|
||||
"Absense of 'Material Quantities' " "indicates a non model-object.",
|
||||
)
|
||||
return False
|
||||
|
||||
# Validate material properties
|
||||
# After discussions 11.02.2025, we're being forgiving on missing "Physical" (aka
|
||||
# StructuralAsset)
|
||||
for material_name, material_data in material_quantities.items():
|
||||
for required_prop in [
|
||||
VOLUME,
|
||||
MATERIAL_CATEGORY,
|
||||
MATERIAL_CLASS,
|
||||
MATERIAL_NAME,
|
||||
]:
|
||||
if required_prop not in material_data:
|
||||
self._logger.log_error(
|
||||
object_id,
|
||||
f"Missing {required_prop}.",
|
||||
"Indicates changes to the Revit "
|
||||
"connector. Inspect commit and "
|
||||
"update accordingly.",
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
@@ -1,43 +0,0 @@
|
||||
from typing import Dict, Any
|
||||
from src.core.base.logger import Logger
|
||||
from src.applications.revit.utils.material_quality_strategy import (
|
||||
HighQualityStrategy,
|
||||
LowQualityStrategy,
|
||||
)
|
||||
from src.core.base import MaterialProcessor
|
||||
from src.core.types.material_data import (
|
||||
MaterialData,
|
||||
)
|
||||
from src.applications.revit.utils.constants import (
|
||||
VOLUME,
|
||||
VALUE,
|
||||
STRUCTURAL_ASSET,
|
||||
)
|
||||
|
||||
|
||||
class RevitMaterialProcessor(MaterialProcessor):
|
||||
"""Implementation of the MaterialProcessor for the Revit context."""
|
||||
|
||||
def __init__(self, mass_aggregator: "MassAggregator", logger: Logger):
|
||||
self._mass_aggregator = mass_aggregator
|
||||
self._logger = logger
|
||||
self._high_quality_strategy = HighQualityStrategy()
|
||||
self._low_quality_strategy = LowQualityStrategy()
|
||||
|
||||
def process(
|
||||
self, object_id: str, material_data: Dict[str, Any], level: str, type_name: str
|
||||
) -> MaterialData:
|
||||
# Volume has already been checked for
|
||||
volume = material_data[VOLUME][VALUE]
|
||||
|
||||
try:
|
||||
if STRUCTURAL_ASSET in material_data:
|
||||
return self._high_quality_strategy.process(
|
||||
object_id, material_data, volume, self._logger
|
||||
)
|
||||
else:
|
||||
return self._low_quality_strategy.process(
|
||||
object_id, material_data, volume, self._logger
|
||||
)
|
||||
except Exception as e:
|
||||
raise ValueError(str(e))
|
||||
@@ -1,132 +0,0 @@
|
||||
from typing import Generator, Any, Tuple
|
||||
from src.core.base import MaterialProcessor, Compliance, CarbonProcessor
|
||||
from src.core.base.logger import Logger
|
||||
from src.applications.revit.utils.constants import (
|
||||
ELEMENTS,
|
||||
NAME,
|
||||
ID,
|
||||
MATERIAL_QUANTITIES,
|
||||
PROPERTIES,
|
||||
)
|
||||
|
||||
|
||||
class RevitElementProcessor:
|
||||
"""Processes Revit model elements to extract and analyze material data."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
material_processor: MaterialProcessor,
|
||||
carbon_processor: CarbonProcessor,
|
||||
compliance_checker: Compliance,
|
||||
logger: Logger,
|
||||
):
|
||||
self.material_processor = material_processor
|
||||
self.carbon_processor = carbon_processor
|
||||
self.compliance_checker = compliance_checker
|
||||
self.logger = logger
|
||||
|
||||
def analyze_elements(self, model: Any) -> None:
|
||||
"""Processes all valid elements from the model."""
|
||||
for level, type_name, element in self._extract_valid_elements(model):
|
||||
try:
|
||||
self._process_materials(level, type_name, element)
|
||||
except Exception as e:
|
||||
self.logger.log_error(
|
||||
object_id=getattr(element, ID, "Unknown"),
|
||||
category="Processing Error",
|
||||
message=f"Error processing element: {str(e)}",
|
||||
)
|
||||
|
||||
def _extract_valid_elements(
|
||||
self, model: Any
|
||||
) -> Generator[Tuple[str, str, Any], None, None]:
|
||||
"""Yields valid elements (level, type_name, revit_object) after compliance checks."""
|
||||
for level, type_name, revit_object in self._get_element_hierarchy(model):
|
||||
if self._is_compliant(revit_object):
|
||||
yield level, type_name, revit_object
|
||||
else:
|
||||
self._log_skipped_element(revit_object)
|
||||
|
||||
def _get_element_hierarchy(
|
||||
self, model: Any
|
||||
) -> Generator[Tuple[str, str, Any], None, None]:
|
||||
"""Flattens nested elements to yield (level, type_name, element)."""
|
||||
for level in self._get_elements(model, "model"):
|
||||
level_name = getattr(level, NAME, "Unknown")
|
||||
|
||||
for type_group in self._get_elements(level, f"level {level_name}"):
|
||||
type_name = getattr(type_group, NAME, "Unknown")
|
||||
|
||||
for group in self._get_elements(type_group, f"type {type_name}"):
|
||||
yield from (
|
||||
(level_name, type_name, revit_object)
|
||||
for revit_object in self._get_elements(
|
||||
group, f"group {getattr(group, NAME, 'Unknown')}"
|
||||
)
|
||||
)
|
||||
|
||||
def _is_compliant(self, model_object: Any) -> bool:
|
||||
"""Checks if an element passes compliance checks."""
|
||||
return self.compliance_checker.check_compliance(model_object)
|
||||
|
||||
def _process_materials(self, level: str, type_name: str, model_object: Any) -> None:
|
||||
"""Extracts material data and processes carbon calculations."""
|
||||
object_id = getattr(model_object, ID, "Unknown")
|
||||
material_quantities = model_object[PROPERTIES].get(MATERIAL_QUANTITIES, {})
|
||||
|
||||
for material_name, material_data in material_quantities.items():
|
||||
processed_material = self.material_processor.process(
|
||||
object_id, material_data, level, type_name
|
||||
)
|
||||
if not processed_material:
|
||||
continue
|
||||
|
||||
model_object[PROPERTIES]["Embodied Carbon Data"] = vars(processed_material)
|
||||
|
||||
# Dictionary-based lookup instead of multiple if-elif checks
|
||||
category_map = {
|
||||
"floor": "Slabs",
|
||||
"stair": "Slabs",
|
||||
"slab edges": "Slabs",
|
||||
"wall": "Walls",
|
||||
"column": "Columns",
|
||||
"framing": "Beam",
|
||||
"beam": "Beam",
|
||||
"foundation": "Foundations",
|
||||
}
|
||||
category = next(
|
||||
(v for k, v in category_map.items() if k in type_name.lower()), None
|
||||
)
|
||||
|
||||
if category and getattr(processed_material, "type") == "Concrete":
|
||||
model_object[PROPERTIES]["Embodied Carbon Data"]["element"] = category
|
||||
|
||||
processed_carbon = self.carbon_processor.process(model_object)
|
||||
if processed_carbon:
|
||||
self.logger.log_success(
|
||||
object_id=object_id,
|
||||
category="Successfully Processed",
|
||||
message="Carbon calculations completed successfully.",
|
||||
)
|
||||
model_object[PROPERTIES]["Embodied Carbon Calculations"] = vars(
|
||||
processed_carbon
|
||||
)
|
||||
|
||||
def _log_skipped_element(self, model_object: Any) -> None:
|
||||
"""Logs elements that fail compliance checks."""
|
||||
self.logger.log_info(
|
||||
object_id=getattr(model_object, ID, "Unknown"),
|
||||
category="Skipped Elements",
|
||||
message="Element did not meet compliance criteria.",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _get_elements(node: Any, context: str) -> list:
|
||||
"""Get elements from a node, with consistent error handling."""
|
||||
elements = getattr(node, ELEMENTS, None)
|
||||
if not elements:
|
||||
name = getattr(node, NAME, "Unknown")
|
||||
raise ValueError(
|
||||
f"Invalid structure: missing elements in {context} '{name}'"
|
||||
)
|
||||
return elements
|
||||
@@ -1,18 +0,0 @@
|
||||
from src.core.base.source_validator import SourceApplicationValidator
|
||||
|
||||
|
||||
class RevitSourceValidator(SourceApplicationValidator):
|
||||
"""Validates that the source application is Revit"""
|
||||
|
||||
# ℹ️ sourceApplication value for v2: AppName + Version => Revit2024, Revit2023 etc.
|
||||
# ℹ️ sourceApplication value for v3: slug => revit
|
||||
# ⚠️ We're just working with v3 data - adapt commit_processor for v2 data structure if you want
|
||||
# ⚠️ Alternatively, write a model factory that injects the correct CommitProcessor()
|
||||
def validate_source_application(self, source_app: str) -> bool:
|
||||
return source_app.lower().startswith("revit")
|
||||
|
||||
def validate_connector_version(self, connector_version: int) -> bool:
|
||||
if connector_version == 2:
|
||||
return False # TODO: If you want to support v2, implement a factory method
|
||||
elif connector_version == 3:
|
||||
return True
|
||||
@@ -1,113 +0,0 @@
|
||||
from typing import Dict, Any, Protocol
|
||||
|
||||
from src.core.base import Logger
|
||||
from src.core.types.material_data import MaterialData
|
||||
from src.applications.revit.utils.material_type_handler import (
|
||||
MaterialType,
|
||||
ConcreteHandler,
|
||||
MetalHandler,
|
||||
WoodHandler,
|
||||
)
|
||||
from src.applications.revit.utils.constants import (
|
||||
MATERIAL_TYPE,
|
||||
MATERIAL_NAME,
|
||||
)
|
||||
|
||||
|
||||
class MaterialQualityStrategy(Protocol):
|
||||
"""Protocol defining how to process materials of different quality levels"""
|
||||
|
||||
def process(
|
||||
self,
|
||||
object_id: str,
|
||||
material_data: Dict[str, Any],
|
||||
volume: float,
|
||||
logger: Logger,
|
||||
) -> MaterialData:
|
||||
...
|
||||
|
||||
|
||||
class HighQualityStrategy(MaterialQualityStrategy):
|
||||
"""Strategy for processing high-quality materials (with structural asset)"""
|
||||
|
||||
def __init__(self):
|
||||
self._handlers = {
|
||||
MaterialType.CONCRETE.value: ConcreteHandler(),
|
||||
MaterialType.METAL.value: MetalHandler(),
|
||||
MaterialType.WOOD.value: WoodHandler(),
|
||||
}
|
||||
|
||||
def process(
|
||||
self,
|
||||
object_id: str,
|
||||
material_data: Dict[str, Any],
|
||||
volume: float,
|
||||
logger: Logger,
|
||||
) -> MaterialData:
|
||||
if MATERIAL_TYPE not in material_data:
|
||||
raise ValueError("Missing material type") # Rather safe than sorry
|
||||
|
||||
material_type = material_data[MATERIAL_TYPE]
|
||||
handler = self._handlers.get(material_type)
|
||||
if not handler:
|
||||
raise ValueError(f"Unsupported material type: {material_type}")
|
||||
|
||||
try:
|
||||
result = handler.create_material_data(material_data, volume)
|
||||
logger.log_success(
|
||||
object_id,
|
||||
"High-Quality Material Definitions",
|
||||
"Contains all expected attributes.",
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
raise Exception(f"Failed to process material: {str(e)}")
|
||||
|
||||
|
||||
class LowQualityStrategy(MaterialQualityStrategy):
|
||||
"""Strategy for processing low-quality materials (without structural asset)"""
|
||||
|
||||
DEFAULT_CONCRETE_GRADE = "35"
|
||||
DEFAULT_STEEL_DENSITY = 7851.81483993
|
||||
|
||||
def process(
|
||||
self,
|
||||
object_id: str,
|
||||
material_data: Dict[str, Any],
|
||||
volume: float,
|
||||
logger: Logger,
|
||||
) -> MaterialData:
|
||||
material_name = material_data[MATERIAL_NAME].lower()
|
||||
|
||||
if "clt" in material_name:
|
||||
logger.log_warning(
|
||||
object_id,
|
||||
"Low-Quality Wood Material Definitions",
|
||||
"Wood has no structural asset and found base on string search",
|
||||
)
|
||||
return MaterialData(MaterialType.WOOD.value, volume)
|
||||
elif "concrete" in material_name:
|
||||
logger.log_warning(
|
||||
object_id,
|
||||
"Low-Quality Concrete Material Definitions",
|
||||
"Concrete has no structural asset and found based on string search",
|
||||
)
|
||||
return MaterialData(
|
||||
MaterialType.CONCRETE.value, volume, grade=self.DEFAULT_CONCRETE_GRADE
|
||||
)
|
||||
elif "steel" in material_name:
|
||||
logger.log_warning(
|
||||
object_id,
|
||||
"Low-Quality Steel Material Definitions",
|
||||
"Steel has no structural asset and found based on string search",
|
||||
)
|
||||
return MaterialData(
|
||||
MaterialType.METAL.value,
|
||||
volume,
|
||||
density=self.DEFAULT_STEEL_DENSITY,
|
||||
mass=volume * self.DEFAULT_STEEL_DENSITY,
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unable to determine material type from name: {material_name}"
|
||||
)
|
||||
@@ -1,87 +0,0 @@
|
||||
from typing import Dict, Any
|
||||
from enum import Enum
|
||||
from abc import ABC, abstractmethod
|
||||
from src.core.types.material_data import MaterialData
|
||||
from src.applications.revit.utils.constants import (
|
||||
VALUE,
|
||||
DENSITY,
|
||||
STRUCTURAL_ASSET,
|
||||
UNITS,
|
||||
MATERIAL_NAME,
|
||||
COMPRESSIVE_STRENGTH,
|
||||
)
|
||||
|
||||
|
||||
class MaterialType(Enum):
|
||||
CONCRETE = "Concrete"
|
||||
METAL = "Metal"
|
||||
WOOD = "Wood"
|
||||
|
||||
|
||||
class MaterialTypeHandler(ABC):
|
||||
"""Abstract base class for handling different material types"""
|
||||
|
||||
@abstractmethod
|
||||
def create_material_data(
|
||||
self, material_data: Dict[str, Any], volume: float
|
||||
) -> MaterialData:
|
||||
pass
|
||||
|
||||
|
||||
class ConcreteHandler(MaterialTypeHandler):
|
||||
def create_material_data(
|
||||
self, material_data: Dict[str, Any], volume: float
|
||||
) -> MaterialData:
|
||||
if DENSITY not in material_data:
|
||||
raise ValueError("Missing density in concrete")
|
||||
density_dict = material_data[DENSITY]
|
||||
|
||||
if COMPRESSIVE_STRENGTH not in material_data:
|
||||
raise ValueError("Missing compression strength in concrete")
|
||||
compressive_strength_dict = material_data[COMPRESSIVE_STRENGTH]
|
||||
|
||||
density = density_dict[VALUE]
|
||||
comp_strength_units = compressive_strength_dict[UNITS]
|
||||
comp_strength_value = compressive_strength_dict[VALUE]
|
||||
|
||||
if comp_strength_units != "Kilonewtons per square meter":
|
||||
raise ValueError(f"Unsupported units: {comp_strength_units}")
|
||||
|
||||
grade = comp_strength_value * 0.001 # Convert to MPa
|
||||
|
||||
return MaterialData(
|
||||
type=MaterialType.CONCRETE.value,
|
||||
volume=volume,
|
||||
structural_asset=material_data[STRUCTURAL_ASSET],
|
||||
density=density,
|
||||
mass=volume * density,
|
||||
grade=str(grade),
|
||||
)
|
||||
|
||||
|
||||
class MetalHandler(MaterialTypeHandler):
|
||||
def create_material_data(
|
||||
self, material_data: Dict[str, Any], volume: float
|
||||
) -> MaterialData:
|
||||
if "steel" not in material_data[MATERIAL_NAME].lower():
|
||||
raise ValueError(
|
||||
f"Material name '{material_data[MATERIAL_NAME]}' does not contain 'steel'"
|
||||
)
|
||||
|
||||
density = material_data[DENSITY][VALUE]
|
||||
|
||||
return MaterialData(
|
||||
type=MaterialType.METAL.value,
|
||||
volume=volume,
|
||||
structural_asset=material_data[STRUCTURAL_ASSET],
|
||||
density=density,
|
||||
mass=density * volume,
|
||||
grade=material_data[STRUCTURAL_ASSET],
|
||||
)
|
||||
|
||||
|
||||
class WoodHandler(MaterialTypeHandler):
|
||||
def create_material_data(
|
||||
self, material_data: Dict[str, Any], volume: float
|
||||
) -> MaterialData:
|
||||
return MaterialData(type=MaterialType.WOOD.value, volume=volume)
|
||||
@@ -1,9 +0,0 @@
|
||||
from .data import ( metal_factors, wood_factors )
|
||||
from .types import ( CarbonData, WoodSupplier )
|
||||
|
||||
__all__ = [
|
||||
"metal_factors",
|
||||
"wood_factors",
|
||||
"CarbonData",
|
||||
"WoodSupplier"
|
||||
]
|
||||
@@ -1,53 +0,0 @@
|
||||
from typing import Dict
|
||||
from collections import defaultdict
|
||||
|
||||
|
||||
# TODO: Replace with carbon aggregator when ready
|
||||
class MassAggregator:
|
||||
"""Cumulative sum of the computed masses. Grouped by level and type."""
|
||||
|
||||
def __init__(self):
|
||||
self.totals = defaultdict(lambda: defaultdict(lambda: defaultdict(float)))
|
||||
|
||||
# TODO: Check! Changes for Revit model groups probably broke this
|
||||
def add_mass(
|
||||
self, mass: float, level: str, collection_type: str, material: str
|
||||
) -> None:
|
||||
"""Adds computed mass of a single object to the totals.
|
||||
Ignores computed masses < 1e-6.
|
||||
|
||||
Args:
|
||||
mass (float): computed mass
|
||||
level (str): string of the associated level of the object
|
||||
collection_type (str): object collection (e.g. "Columns", "Structural Foundations")
|
||||
material (str): name of the structural asset
|
||||
"""
|
||||
if mass <= 1e-6:
|
||||
return
|
||||
self.totals[level][collection_type][material] += mass
|
||||
|
||||
# TODO: Check! Changes for Revit model groups probably broke this
|
||||
def get_totals(self) -> Dict:
|
||||
"""Return computed totals in a structured dictionary.
|
||||
|
||||
Returns:
|
||||
Dict: nested "by_level" and then "by_type"
|
||||
"""
|
||||
return {
|
||||
"by_level": {
|
||||
level: {
|
||||
"total": sum(
|
||||
sum(material_masses.values())
|
||||
for material_masses in types.values()
|
||||
),
|
||||
"by_type": {
|
||||
type_name: {
|
||||
"total": sum(material_masses.values()),
|
||||
"by_material": material_masses,
|
||||
}
|
||||
for type_name, material_masses in types.items()
|
||||
},
|
||||
}
|
||||
for level, types in self.totals.items()
|
||||
},
|
||||
}
|
||||
@@ -1,98 +0,0 @@
|
||||
|
||||
from enum import Enum
|
||||
from .types import WoodSupplier
|
||||
from src.core.types import ( MetalClass, WoodClass )
|
||||
|
||||
"""kgCO2e/kg"""
|
||||
metal_factors = {
|
||||
MetalClass.HOT_ROLLED: 1.22,
|
||||
MetalClass.HSS: 1.99,
|
||||
MetalClass.PLATE: 1.73,
|
||||
MetalClass.REBAR: 0.854,
|
||||
MetalClass.OWSJ: 1.380,
|
||||
MetalClass.FASTENERS: 1.730,
|
||||
MetalClass.METAL_DECK: 2.370,
|
||||
}
|
||||
|
||||
"""kgCO2e/m3"""
|
||||
wood_factors = {
|
||||
WoodSupplier.INDUSTRY_AVERAGE: {
|
||||
WoodClass.GLULAM: 113,
|
||||
WoodClass.CLT: 135,
|
||||
WoodClass.LVL: 265,
|
||||
WoodClass.SOFTWOOD_LUMBER: 56,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: 142,
|
||||
WoodClass.WOOD_JOISTS: 2,
|
||||
WoodClass.REDWOOD_LUMBER: 38,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: 212,
|
||||
WoodClass.GLT_NLT_DLT: 123
|
||||
},
|
||||
WoodSupplier.ATHENA: {
|
||||
WoodClass.GLULAM: 107,
|
||||
WoodClass.CLT: 69,
|
||||
WoodClass.LVL: 169,
|
||||
WoodClass.SOFTWOOD_LUMBER: 48,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: 65,
|
||||
WoodClass.WOOD_JOISTS: None,
|
||||
WoodClass.REDWOOD_LUMBER: None,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: 182,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
},
|
||||
WoodSupplier.STRUCTURLAM: {
|
||||
WoodClass.GLULAM: 115,
|
||||
WoodClass.CLT: 124,
|
||||
WoodClass.LVL: None,
|
||||
WoodClass.SOFTWOOD_LUMBER: None,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: None,
|
||||
WoodClass.WOOD_JOISTS: None,
|
||||
WoodClass.REDWOOD_LUMBER: None,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: None,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
},
|
||||
WoodSupplier.AWC_CWC: {
|
||||
WoodClass.GLULAM: 137,
|
||||
WoodClass.CLT: 0,
|
||||
WoodClass.LVL: 361,
|
||||
WoodClass.SOFTWOOD_LUMBER: 63,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: 219,
|
||||
WoodClass.WOOD_JOISTS: 2,
|
||||
WoodClass.REDWOOD_LUMBER: 38,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: 243,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
},
|
||||
WoodSupplier.KATERRA: {
|
||||
WoodClass.GLULAM: None,
|
||||
WoodClass.CLT: 158,
|
||||
WoodClass.LVL: None,
|
||||
WoodClass.SOFTWOOD_LUMBER: None,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: None,
|
||||
WoodClass.WOOD_JOISTS: None,
|
||||
WoodClass.REDWOOD_LUMBER: None,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: None,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
},
|
||||
WoodSupplier.NORDIC_STRUCTURES: {
|
||||
WoodClass.GLULAM: 100,
|
||||
WoodClass.CLT: 122,
|
||||
WoodClass.LVL: None,
|
||||
WoodClass.SOFTWOOD_LUMBER: None,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: None,
|
||||
WoodClass.WOOD_JOISTS: None,
|
||||
WoodClass.REDWOOD_LUMBER: None,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: None,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
},
|
||||
WoodSupplier.BINDERHOLZ: {
|
||||
WoodClass.GLULAM: 118,
|
||||
WoodClass.CLT: 200,
|
||||
WoodClass.LVL: None,
|
||||
WoodClass.SOFTWOOD_LUMBER: None,
|
||||
WoodClass.SOFTWOOD_PLYWOOD: None,
|
||||
WoodClass.WOOD_JOISTS: None,
|
||||
WoodClass.REDWOOD_LUMBER: None,
|
||||
WoodClass.ORIENTED_STRAND_BOARD: None,
|
||||
WoodClass.GLT_NLT_DLT: 0
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
from enum import Enum
|
||||
from dataclasses import dataclass
|
||||
|
||||
class WoodSupplier(str, Enum):
|
||||
INDUSTRY_AVERAGE = "Industry Average"
|
||||
ATHENA = "Athena, 2021"
|
||||
STRUCTURLAM = "Structurlam, 2020"
|
||||
AWC_CWC = "AWC, CWC, 2018"
|
||||
KATERRA = "Katerra, 2020"
|
||||
NORDIC_STRUCTURES = "Nordic Structures, 2018"
|
||||
BINDERHOLZ = "Binderholz, 2019"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CarbonData:
|
||||
"""Data class for embodied carbon data"""
|
||||
|
||||
factor: float
|
||||
embodied_carbon: float
|
||||
@@ -1,14 +0,0 @@
|
||||
from .logger import Logger
|
||||
from .source_validator import SourceApplicationValidator
|
||||
from .material_processor import MaterialProcessor
|
||||
from .compliance import Compliance
|
||||
from .carbon_processor import CarbonProcessor
|
||||
|
||||
__all__ = [
|
||||
"Logger",
|
||||
"Model",
|
||||
"SourceApplicationValidator",
|
||||
"MaterialProcessor",
|
||||
"Compliance",
|
||||
"CarbonProcessor",
|
||||
]
|
||||
@@ -1,17 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any
|
||||
|
||||
|
||||
class CarbonProcessor(ABC):
|
||||
"""Interface for processing embodied carbon on an object."""
|
||||
|
||||
@abstractmethod
|
||||
def process(
|
||||
self, modeL_object: Any
|
||||
) -> None:
|
||||
"""Compute embodied carbon per-element based previously asserted material properties.
|
||||
|
||||
Args:
|
||||
model_object (Any): Model object to process
|
||||
"""
|
||||
pass
|
||||
@@ -1,13 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any
|
||||
|
||||
|
||||
class Compliance(ABC):
|
||||
"""Interface for compliance checks.
|
||||
Compliance are intended to be called for every object where an attribute is assumed.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def check_compliance(self, element: Any) -> bool:
|
||||
"""Check if element contains attribute(s)"""
|
||||
pass
|
||||
@@ -1,54 +0,0 @@
|
||||
from typing import Dict, Optional
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class Logger(ABC):
|
||||
"""Abstract base class for logging functionality"""
|
||||
|
||||
@abstractmethod
|
||||
def log_error(
|
||||
self, object_id: str, category: str, message: Optional[str] = None
|
||||
) -> None:
|
||||
"""Log an error for a specific object"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def log_warning(
|
||||
self, object_id: str, category: str, message: Optional[str] = None
|
||||
) -> None:
|
||||
"""Log a warning for a specific object"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def log_success(
|
||||
self, object_id: str, category: str, message: Optional[str] = None
|
||||
) -> None:
|
||||
"""Log a success for a specific object"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def log_info(
|
||||
self, object_id: str, category: str, message: Optional[str] = None
|
||||
) -> None:
|
||||
"""Log information for a specific object"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_warnings_summary(self) -> Dict[str, list]:
|
||||
"""Get summary of all warnings grouped by category"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_errors_summary(self) -> Dict[str, list]:
|
||||
"""Get summary of all errors grouped by category"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_success_summary(self) -> Dict[str, list]:
|
||||
"""Get summary of all successes grouped by category"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_info_summary(self) -> Dict[str, list]:
|
||||
"""Get summary of all info logs grouped by category"""
|
||||
pass
|
||||
@@ -1,20 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
class MaterialProcessor(ABC):
|
||||
"""Interface for processing a material."""
|
||||
|
||||
@abstractmethod
|
||||
def process(
|
||||
self, object_id: str, material_data: Dict[str, Any], level: str, type_name: str
|
||||
) -> None:
|
||||
"""Process material data and compute required properties
|
||||
|
||||
Args:
|
||||
object_id (str): Object ID
|
||||
material_data (Dict[str, Any]): material data
|
||||
level (str): associated level of object
|
||||
type_name (str): object type
|
||||
"""
|
||||
pass
|
||||
@@ -1,17 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class SourceApplicationValidator(ABC):
|
||||
"""Interface for source application validator.
|
||||
Host app should be supported by the automation.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def validate_source_application(self, source_app: str) -> bool:
|
||||
"""Assert that the source application is supported."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def validate_connector_version(self, connector_version: str) -> bool:
|
||||
"""Assert that the connector version is supported."""
|
||||
pass
|
||||
@@ -1,8 +0,0 @@
|
||||
from .material_class import ( MetalClass, WoodClass )
|
||||
from .material_data import MaterialData
|
||||
|
||||
__all__ = [
|
||||
"MaterialData",
|
||||
"MetalClass",
|
||||
"WoodClass"
|
||||
]
|
||||
@@ -1,21 +0,0 @@
|
||||
from enum import Enum
|
||||
|
||||
class MetalClass(Enum):
|
||||
HOT_ROLLED = "Hot Rolled"
|
||||
HSS = "HSS"
|
||||
PLATE = "Plate"
|
||||
REBAR = "Rebar"
|
||||
OWSJ = "OWSJ"
|
||||
FASTENERS = "Fasteners"
|
||||
METAL_DECK = "Metal Deck"
|
||||
|
||||
class WoodClass(Enum):
|
||||
GLULAM = "Glulam"
|
||||
CLT = "CLT"
|
||||
LVL = "LVL"
|
||||
SOFTWOOD_LUMBER = "Softwood Lumber"
|
||||
SOFTWOOD_PLYWOOD = "Softwood Plywood"
|
||||
WOOD_JOISTS = "Wood Joists"
|
||||
REDWOOD_LUMBER = "Redwood Lumber"
|
||||
ORIENTED_STRAND_BOARD = "Oriented Strand Board"
|
||||
GLT_NLT_DLT = "GLT/NLT/DLT"
|
||||
@@ -1,14 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class MaterialData:
|
||||
"""Data class for material properties"""
|
||||
|
||||
type: str # Concrete / Steel / Wood
|
||||
volume: float
|
||||
structural_asset: Optional[str] = None # Not ideal, but we're being forgiving
|
||||
density: Optional[float] = None # Only needed for steel
|
||||
mass: Optional[float] = None # Only needed for steel
|
||||
grade: Optional[str] = None # Needed for concrete
|
||||
@@ -0,0 +1,25 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional, Dict
|
||||
from src.domain.carbon.schema import EmissionFactor
|
||||
|
||||
|
||||
class EmissionFactorDatabase(ABC):
|
||||
"""Base class for emission factor databases"""
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
@@ -0,0 +1,40 @@
|
||||
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)
|
||||
@@ -0,0 +1,29 @@
|
||||
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())
|
||||
@@ -0,0 +1,36 @@
|
||||
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
|
||||
epd_number: Optional[str] = None
|
||||
publication_date: Optional[str] = None
|
||||
valid_until: Optional[str] = None
|
||||
manufacturer: Optional[str] = None
|
||||
plant_location: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CarbonResult:
|
||||
"""Result of a carbon calculation"""
|
||||
|
||||
material_name: str
|
||||
emission_factor: EmissionFactor
|
||||
quantity: float
|
||||
total_carbon: float
|
||||
category: str
|
||||
@@ -0,0 +1,44 @@
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
class MaterialType(Enum):
|
||||
CONCRETE = "Concrete"
|
||||
METAL = "Metal"
|
||||
WOOD = "Wood"
|
||||
|
||||
class ElementCategory(Enum):
|
||||
SLAB = "Slabs"
|
||||
WALL = "Walls"
|
||||
COLUMN = "Columns"
|
||||
BEAM = "Beams"
|
||||
FOUNDATION = "Foundations"
|
||||
|
||||
@dataclass
|
||||
class MaterialProperties:
|
||||
name: str
|
||||
volume: float
|
||||
density: Optional[float] = None
|
||||
structural_asset: Optional[str] = None
|
||||
compressive_strength: Optional[float] = None
|
||||
|
||||
@dataclass
|
||||
class Material:
|
||||
type: MaterialType
|
||||
properties: MaterialProperties
|
||||
grade: Optional[str] = None
|
||||
mass: Optional[float] = None
|
||||
|
||||
@dataclass
|
||||
class BuildingElement:
|
||||
id: str
|
||||
level: str
|
||||
category: ElementCategory
|
||||
materials: List[Material]
|
||||
carbon_data: Optional[Dict] = None
|
||||
|
||||
@dataclass
|
||||
class CarbonResult:
|
||||
factor: float # kgCO2e/kg for metals, kgCO2e/m3 for wood
|
||||
total_carbon: float # kgCO2e
|
||||
category: str
|
||||
@@ -2,10 +2,8 @@ import structlog
|
||||
from typing import Dict, Set, Optional
|
||||
from collections import defaultdict
|
||||
|
||||
from src.core.base.logger import Logger # Import the interface
|
||||
|
||||
|
||||
class RevitLogger(Logger):
|
||||
class Logging:
|
||||
"""Implements Logger interface with category-based logging"""
|
||||
|
||||
def __init__(self):
|
||||
@@ -0,0 +1,87 @@
|
||||
from typing import Dict
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
self.wood_factors = {
|
||||
"CLT": 135,
|
||||
"Glulam": 113,
|
||||
"default": 135, # Default to CLT
|
||||
}
|
||||
|
||||
def calculate_carbon(self, element: BuildingElement) -> Dict[str, CarbonResult]:
|
||||
"""Calculate carbon emissions for an element's materials."""
|
||||
results = {}
|
||||
|
||||
for material in element.materials:
|
||||
try:
|
||||
result = self._calculate_material_carbon(material)
|
||||
results[material.properties.name] = result
|
||||
except Exception as e:
|
||||
# Log error but continue with other materials
|
||||
print(
|
||||
f"Error calculating carbon for {material.properties.name}: {str(e)}"
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
def _calculate_material_carbon(self, material: Material) -> 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)
|
||||
else:
|
||||
raise ValueError(f"Unsupported material type: {material.type}")
|
||||
|
||||
def _calculate_metal_carbon(self, material: Material) -> CarbonResult:
|
||||
"""Calculate carbon emissions for metal."""
|
||||
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"])
|
||||
|
||||
return CarbonResult(
|
||||
factor=factor, total_carbon=material.mass * factor, 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"]
|
||||
)
|
||||
|
||||
return CarbonResult(
|
||||
factor=factor,
|
||||
total_carbon=material.properties.volume * factor,
|
||||
category="Wood",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _calculate_concrete_carbon(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
|
||||
return CarbonResult(
|
||||
factor=0.0, # Placeholder
|
||||
total_carbon=0.0, # Placeholder
|
||||
category="Concrete",
|
||||
)
|
||||
@@ -0,0 +1,121 @@
|
||||
from typing import Optional, List
|
||||
|
||||
from src.domain.types import BuildingElement, ElementCategory, Material
|
||||
from src.infrastructure.logging import Logging
|
||||
from src.services.material_processor import MaterialProcessor
|
||||
|
||||
|
||||
class ElementProcessor:
|
||||
"""Processes Revit building elements."""
|
||||
|
||||
SKIP_TYPES = [
|
||||
"Objects.Geometry.Line",
|
||||
"Objects.Geometry.Arc",
|
||||
"Objects.Geometry.Circle",
|
||||
]
|
||||
|
||||
def __init__(self, material_processor: MaterialProcessor, logger: Logging):
|
||||
self.material_processor = material_processor
|
||||
self.logger = logger
|
||||
|
||||
def process_element(self, element: dict) -> Optional[BuildingElement]:
|
||||
"""Process a single Revit element."""
|
||||
try:
|
||||
# Basic validation
|
||||
if not self._is_valid_element(element):
|
||||
return None
|
||||
|
||||
# Extract basic properties
|
||||
element_id = getattr(element, "id", "unknown")
|
||||
level = self._get_element_level(element)
|
||||
category = self._determine_category(element)
|
||||
|
||||
# Process materials
|
||||
materials = self._process_materials(element)
|
||||
|
||||
# Create building element
|
||||
return BuildingElement(
|
||||
id=element_id, level=level, category=category, materials=materials
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log_error(
|
||||
getattr(element, "id"),
|
||||
"Element Processing",
|
||||
f"Error processing element {getattr(element, "id")}: {str(e)}",
|
||||
)
|
||||
return None
|
||||
|
||||
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")
|
||||
return False
|
||||
|
||||
print(f"Material Quantities found: {properties['Material Quantities']}")
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _get_element_level(element) -> str:
|
||||
"""Extract element level."""
|
||||
return getattr(element, "level", "Unknown")
|
||||
|
||||
@staticmethod
|
||||
def _determine_category(element: dict) -> ElementCategory:
|
||||
"""Determine element category based on type name."""
|
||||
type_name = getattr(element, "name", "").lower()
|
||||
|
||||
category_mapping = {
|
||||
"floor": ElementCategory.SLAB,
|
||||
"stair": ElementCategory.SLAB,
|
||||
"slab": ElementCategory.SLAB,
|
||||
"wall": ElementCategory.WALL,
|
||||
"column": ElementCategory.COLUMN,
|
||||
"beam": ElementCategory.BEAM,
|
||||
"framing": ElementCategory.BEAM,
|
||||
"foundation": ElementCategory.FOUNDATION,
|
||||
}
|
||||
|
||||
for key, category in category_mapping.items():
|
||||
if key in type_name:
|
||||
return category
|
||||
|
||||
return ElementCategory.BEAM # Default category
|
||||
|
||||
def _process_materials(self, element) -> List[Material]:
|
||||
"""Process all materials in the element."""
|
||||
materials = []
|
||||
properties = getattr(element, "properties")
|
||||
material_quantities = properties["Material Quantities"]
|
||||
|
||||
for material_data in material_quantities.values(): # Added .values()
|
||||
try:
|
||||
material = self.material_processor.process_material(material_data)
|
||||
materials.append(material)
|
||||
except Exception as e:
|
||||
self.logger.log_warning(
|
||||
getattr(element, "id"),
|
||||
"Material Processing",
|
||||
f"Failed to process material in element {getattr(element, 'id')}: {str(e)}",
|
||||
)
|
||||
|
||||
return materials
|
||||
@@ -0,0 +1,91 @@
|
||||
from typing import Dict, Any
|
||||
|
||||
from src.domain.types import MaterialProperties, Material, MaterialType
|
||||
|
||||
|
||||
class MaterialProcessor:
|
||||
"""Processes Revit materials and calculates quantities."""
|
||||
|
||||
DEFAULT_CONCRETE_GRADE = "35"
|
||||
DEFAULT_STEEL_DENSITY = 7851.81483993 # kg/m3
|
||||
|
||||
def process_material(self, raw_material: Dict[str, Any]) -> Material:
|
||||
"""Process raw material data from Revit into domain model."""
|
||||
properties = MaterialProperties(
|
||||
name=raw_material["materialName"],
|
||||
volume=raw_material["volume"]["value"],
|
||||
density=raw_material.get("density", {}).get(
|
||||
"value"
|
||||
), # Using .get() for optional fields
|
||||
structural_asset=raw_material.get("structuralAsset"),
|
||||
compressive_strength=raw_material.get("compressiveStrength", {}).get(
|
||||
"value"
|
||||
),
|
||||
)
|
||||
|
||||
# Determine material type and create material
|
||||
if self._is_high_grade_material(raw_material):
|
||||
return self._process_high_grade_material(properties)
|
||||
else:
|
||||
return self._process_low_grade_material(properties)
|
||||
|
||||
@staticmethod
|
||||
def _is_high_grade_material(raw_material: Dict[str, Any]) -> bool:
|
||||
return "structuralAsset" in raw_material
|
||||
|
||||
def _process_high_grade_material(self, props: MaterialProperties) -> Material:
|
||||
"""Process materials with structural assets."""
|
||||
if "concrete" in props.name.lower():
|
||||
return self._process_concrete(props)
|
||||
elif "steel" in props.name.lower():
|
||||
return self._process_steel(props)
|
||||
elif "clt" in props.name.lower() or "timber" in props.name.lower():
|
||||
return Material(type=MaterialType.WOOD, properties=props)
|
||||
else:
|
||||
raise ValueError(f"Unknown high-grade material: {props.name}")
|
||||
|
||||
def _process_low_grade_material(self, props: MaterialProperties) -> Material:
|
||||
"""Process materials without structural assets."""
|
||||
name = props.name.lower()
|
||||
|
||||
if "concrete" in name:
|
||||
return Material(
|
||||
type=MaterialType.CONCRETE,
|
||||
properties=props,
|
||||
grade=self.DEFAULT_CONCRETE_GRADE,
|
||||
)
|
||||
elif "steel" in name:
|
||||
mass = props.volume * self.DEFAULT_STEEL_DENSITY
|
||||
return Material(
|
||||
type=MaterialType.METAL,
|
||||
properties=props,
|
||||
mass=mass,
|
||||
grade="default_steel",
|
||||
)
|
||||
elif "clt" in name or "timber" in name:
|
||||
return Material(type=MaterialType.WOOD, properties=props)
|
||||
else:
|
||||
raise ValueError(f"Unknown material type: {props.name}")
|
||||
|
||||
@staticmethod
|
||||
def _process_concrete(props: MaterialProperties) -> Material:
|
||||
"""Process concrete-specific properties."""
|
||||
if not props.compressive_strength:
|
||||
raise ValueError("Missing compressive strength for concrete")
|
||||
|
||||
grade = str(props.compressive_strength * 0.001) # Convert to MPa
|
||||
return Material(type=MaterialType.CONCRETE, properties=props, grade=grade)
|
||||
|
||||
@staticmethod
|
||||
def _process_steel(props: MaterialProperties) -> Material:
|
||||
"""Process steel-specific properties."""
|
||||
if not props.density:
|
||||
raise ValueError("Missing density for steel")
|
||||
|
||||
mass = props.volume * props.density
|
||||
return Material(
|
||||
type=MaterialType.METAL,
|
||||
properties=props,
|
||||
mass=mass,
|
||||
grade=props.structural_asset,
|
||||
)
|
||||
@@ -1,12 +1,11 @@
|
||||
"""Run integration tests with a speckle server."""
|
||||
|
||||
from pydantic import SecretStr
|
||||
|
||||
from speckle_automate import (
|
||||
AutomationContext,
|
||||
AutomationRunData,
|
||||
AutomationStatus,
|
||||
run_function
|
||||
run_function,
|
||||
)
|
||||
|
||||
from main import FunctionInputs, automate_function
|
||||
@@ -14,7 +13,9 @@ from main import FunctionInputs, automate_function
|
||||
from speckle_automate.fixtures import *
|
||||
|
||||
|
||||
def test_function_run(test_automation_run_data: AutomationRunData, test_automation_token: str):
|
||||
def test_function_run(
|
||||
test_automation_run_data: AutomationRunData, test_automation_token: str
|
||||
):
|
||||
"""Run an integration test for the automate function."""
|
||||
automation_context = AutomationContext.initialize(
|
||||
test_automation_run_data, test_automation_token
|
||||
|
||||
Reference in New Issue
Block a user