diff --git a/main.py b/main.py index 9a18ebc..fbcf749 100644 --- a/main.py +++ b/main.py @@ -1,25 +1,124 @@ -from pydantic import Field, SecretStr +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 RevitModel -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.core.base import Model -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( @@ -28,115 +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 = configure_components() + # Run analysis - convert Speckle model to dict for processing + results = analyzer.analyze_model(model_root) - # Process model - processor.process_elements(model_root) + # Process results + _process_automation_results(automate_context, results) - # Logger information - successes - ( - logger_successes, - logger_infos, - logger_warnings, - logger_failures, - ) = processor.get_processing_results() - - 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 configure_components() -> Model: - """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 RevitModel( - 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__": diff --git a/src/applications/revit/revit_carbon_processor.py b/src/applications/revit/revit_carbon_processor.py deleted file mode 100644 index 27a47ea..0000000 --- a/src/applications/revit/revit_carbon_processor.py +++ /dev/null @@ -1,48 +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"] - 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 - model_object[PROPERTIES]["Embodied Carbon Data"]["embodied_carbon"] = embodied_carbon - pass diff --git a/src/applications/revit/revit_compliance.py b/src/applications/revit/revit_compliance.py deleted file mode 100644 index 8658dea..0000000 --- a/src/applications/revit/revit_compliance.py +++ /dev/null @@ -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 diff --git a/src/applications/revit/revit_material_processor.py b/src/applications/revit/revit_material_processor.py deleted file mode 100644 index 855324b..0000000 --- a/src/applications/revit/revit_material_processor.py +++ /dev/null @@ -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)) diff --git a/src/applications/revit/revit_model.py b/src/applications/revit/revit_model.py deleted file mode 100644 index 22773f4..0000000 --- a/src/applications/revit/revit_model.py +++ /dev/null @@ -1,180 +0,0 @@ -from typing import Any, Tuple, Dict, List -from src.core.base import Model -from src.core.base import MaterialProcessor -from src.core.base import Compliance -from src.core.base import CarbonProcessor -from src.core.base.logger import Logger -from src.applications.revit.utils.constants import ( - ELEMENTS, - NAME, - ID, - MATERIAL_QUANTITIES, - PROPERTIES, -) - -# NOTE: Only provide docstring if not covered by base class - - -class RevitModel(Model): - """Implementation of the ModelProcessor in the Revit context.""" - - 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 process_elements(self, model: Any) -> None: - """Model traversal. - Model → Levels → Type Groups → Elements. - - Performance Notes: - - Threading? Could be cool, but: - - Profile code first. Do we need it? - - Not really I/O bound. Our traversal is just walking through an in-memory object structure - - TBH I'm scared of the hierarchy - - Thread safety? Shared loggers and aggregators. - - Flattening nested iterations? - - Code less readable (e.g. chain.from_iterable(..)) - - Obscures hierarchical nature of data - - Performance benefit(s) minimal? - - Harder to debug - """ - levels = self._get_elements(model, "model") - - for level in levels: - level_name = self._get_name(level) - type_groups = self._get_elements(level, f"level {level_name}") - - for type_group in type_groups: - type_name = self._get_name(type_group) - groups = self._get_elements(type_group, f"type {type_name}") - - for group in groups: - revit_objects = self._get_elements( - group, f"group {self._get_name(group)}" - ) - for revit_object in revit_objects: - self.process_element(level_name, type_name, revit_object) - - def process_element(self, level: str, type_name: str, model_object: Any) -> None: - """Process a single model element if it passes compliance checks. - - Args: - level (str): associated level of the object - type_name (str): family / type of the object - model_object (Any): speckle object containing properties for processing - """ - # First check compliance - this also handles logging any validation warnings - valid = self._compliance_checker.check_compliance(model_object) - if not valid: - return - - # These are now safe to do after compliance checking - material_quantities = model_object[PROPERTIES][MATERIAL_QUANTITIES] - object_id = getattr(model_object, ID) - - try: - # Process each material if element passed validation - for material_name, material_data in material_quantities.items(): - processed_material = self._material_processor.process( - object_id, material_data, level, type_name - ) - if processed_material: # If processing was successful - model_object[PROPERTIES]["Embodied Carbon Data"] = vars( - processed_material - ) - - if getattr(processed_material, "type") == "Concrete": - model_object[PROPERTIES]["Embodied Carbon Data"][ - "element" - ] = self._categorize(type_name) - - 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 for this element.", - ) - - model_object[PROPERTIES]["Embodied Carbon Calculations"] = vars(processed_carbon) - - except Exception as e: - # Log any processing errors that occur - self._logger.log_error(object_id, "Material Processing Error", str(e)) - - def get_processing_results( - self, - ) -> Tuple[ - Dict[str, List[str]], - Dict[str, List[str]], - Dict[str, List[str]], - Dict[str, List[str]], - ]: - """Get processing results in the format expected by main.py. - - Returns: - Tuple containing: - - Dict mapping success categories to lists of successfully processed object IDs - - Dict mapping information categories to lists of affected object IDs - - Dict mapping warning categories to lists of affected object IDs - - Dict mapping error categories to lists of affected object IDs - """ - return ( - self._logger.get_success_summary(), - self._logger.get_info_summary(), - self._logger.get_warnings_summary(), - self._logger.get_errors_summary(), - ) - - @staticmethod - def _get_elements(node: Any, context: str) -> list: - """Get elements from a node, with consistent error handling. - - Args: - node: Node to extract elements from - context: Context for error message if elements missing - - Raises: - ValueError: If elements are missing - """ - 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 - - @staticmethod - def _get_name(node: Any) -> str: - """Safely get name from node, with fallback""" - return getattr(node, NAME, "Unknown") - - @staticmethod - def _categorize(type_name: str) -> str: - searchable_string = type_name.lower() - if ( - "floor" in searchable_string - or "stair" in searchable_string - or "slab edges" in searchable_string - ): - return "Slabs" - elif "wall" in searchable_string: - return "Walls" - elif "column" in searchable_string: - return "Columns" - elif "framing" in searchable_string or "beam" in searchable_string: - return "Beam" - elif "foundation" in searchable_string: - return "Foundations" - else: - raise ValueError(f"{type_name} not accounted for.") diff --git a/src/applications/revit/revit_source_validator.py b/src/applications/revit/revit_source_validator.py deleted file mode 100644 index 77e867a..0000000 --- a/src/applications/revit/revit_source_validator.py +++ /dev/null @@ -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 diff --git a/src/applications/revit/utils/material_quality_strategy.py b/src/applications/revit/utils/material_quality_strategy.py deleted file mode 100644 index 17bb690..0000000 --- a/src/applications/revit/utils/material_quality_strategy.py +++ /dev/null @@ -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}" - ) diff --git a/src/applications/revit/utils/material_type_handler.py b/src/applications/revit/utils/material_type_handler.py deleted file mode 100644 index 18e5bbf..0000000 --- a/src/applications/revit/utils/material_type_handler.py +++ /dev/null @@ -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) diff --git a/src/carbon/__init__.py b/src/carbon/__init__.py deleted file mode 100644 index ab892a4..0000000 --- a/src/carbon/__init__.py +++ /dev/null @@ -1,9 +0,0 @@ -from .data import ( metal_factors, wood_factors ) -from .types import ( CarbonData, WoodSupplier ) - -__all__ = [ - "metal_factors", - "wood_factors", - "CarbonData", - "WoodSupplier" - ] \ No newline at end of file diff --git a/src/carbon/aggregator.py b/src/carbon/aggregator.py deleted file mode 100644 index d0a2b53..0000000 --- a/src/carbon/aggregator.py +++ /dev/null @@ -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() - }, - } diff --git a/src/carbon/data.py b/src/carbon/data.py deleted file mode 100644 index 9bb8b25..0000000 --- a/src/carbon/data.py +++ /dev/null @@ -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 - } -} - - diff --git a/src/carbon/types.py b/src/carbon/types.py deleted file mode 100644 index 21cb30b..0000000 --- a/src/carbon/types.py +++ /dev/null @@ -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 diff --git a/src/core/base/__init__.py b/src/core/base/__init__.py deleted file mode 100644 index 63fe5ef..0000000 --- a/src/core/base/__init__.py +++ /dev/null @@ -1,15 +0,0 @@ -from .logger import Logger -from .model import Model -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" -] diff --git a/src/core/base/carbon_processor.py b/src/core/base/carbon_processor.py deleted file mode 100644 index e2ab08a..0000000 --- a/src/core/base/carbon_processor.py +++ /dev/null @@ -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 diff --git a/src/core/base/compliance.py b/src/core/base/compliance.py deleted file mode 100644 index 65cb988..0000000 --- a/src/core/base/compliance.py +++ /dev/null @@ -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 diff --git a/src/core/base/logger.py b/src/core/base/logger.py deleted file mode 100644 index 49f64f8..0000000 --- a/src/core/base/logger.py +++ /dev/null @@ -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 diff --git a/src/core/base/material_processor.py b/src/core/base/material_processor.py deleted file mode 100644 index 5eb2236..0000000 --- a/src/core/base/material_processor.py +++ /dev/null @@ -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 diff --git a/src/core/base/model.py b/src/core/base/model.py deleted file mode 100644 index 616f0b3..0000000 --- a/src/core/base/model.py +++ /dev/null @@ -1,40 +0,0 @@ -from abc import ABC, abstractmethod -from typing import Any, Dict, Tuple, List - - -class Model(ABC): - """Interface for model processing.""" - - @abstractmethod - def process_elements(self, model: Any) -> None: - """Process all elements in the model. - - Args: - model (Any): root commit - """ - pass - - # TODO: element should be Base? - @abstractmethod - def process_element(self, level: str, type_name: str, model_object: Any) -> None: - """Process a single element. - - Args: - level (str): associated level of object - type_name (str): object type - model_object (Any): speckle object - """ - pass - - # TODO: This is gross - @abstractmethod - def get_processing_results( - self, - ) -> Tuple[ - Dict[str, List[str]], - Dict[str, List[str]], - Dict[str, List[str]], - Dict[str, List[str]], - ]: - """Expose logging results.""" - pass diff --git a/src/core/base/source_validator.py b/src/core/base/source_validator.py deleted file mode 100644 index 00e6ec7..0000000 --- a/src/core/base/source_validator.py +++ /dev/null @@ -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 diff --git a/src/core/types/__init__.py b/src/core/types/__init__.py deleted file mode 100644 index 145b055..0000000 --- a/src/core/types/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -from .material_class import ( MetalClass, WoodClass ) -from .material_data import MaterialData - -__all__ = [ - "MaterialData", - "MetalClass", - "WoodClass" - ] diff --git a/src/core/types/material_class.py b/src/core/types/material_class.py deleted file mode 100644 index fab89b8..0000000 --- a/src/core/types/material_class.py +++ /dev/null @@ -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" \ No newline at end of file diff --git a/src/core/types/material_data.py b/src/core/types/material_data.py deleted file mode 100644 index b866dc6..0000000 --- a/src/core/types/material_data.py +++ /dev/null @@ -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 diff --git a/src/applications/__init__.py b/src/domain/__init__.py similarity index 100% rename from src/applications/__init__.py rename to src/domain/__init__.py diff --git a/src/applications/revit/__init__.py b/src/domain/carbon/__init__.py similarity index 100% rename from src/applications/revit/__init__.py rename to src/domain/carbon/__init__.py diff --git a/src/applications/revit/utils/__init__.py b/src/domain/carbon/databases/__init__.py similarity index 100% rename from src/applications/revit/utils/__init__.py rename to src/domain/carbon/databases/__init__.py diff --git a/src/domain/carbon/databases/base.py b/src/domain/carbon/databases/base.py new file mode 100644 index 0000000..3e0a889 --- /dev/null +++ b/src/domain/carbon/databases/base.py @@ -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 diff --git a/src/domain/carbon/databases/epd_global.py b/src/domain/carbon/databases/epd_global.py new file mode 100644 index 0000000..8511b2a --- /dev/null +++ b/src/domain/carbon/databases/epd_global.py @@ -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) diff --git a/src/domain/carbon/registry.py b/src/domain/carbon/registry.py new file mode 100644 index 0000000..f007d5c --- /dev/null +++ b/src/domain/carbon/registry.py @@ -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()) diff --git a/src/domain/carbon/schema.py b/src/domain/carbon/schema.py new file mode 100644 index 0000000..32b81d8 --- /dev/null +++ b/src/domain/carbon/schema.py @@ -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 diff --git a/src/applications/revit/utils/constants.py b/src/domain/constants.py similarity index 100% rename from src/applications/revit/utils/constants.py rename to src/domain/constants.py diff --git a/src/domain/types.py b/src/domain/types.py new file mode 100644 index 0000000..5c16499 --- /dev/null +++ b/src/domain/types.py @@ -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 \ No newline at end of file diff --git a/src/core/__init__.py b/src/infrastructure/__init__.py similarity index 100% rename from src/core/__init__.py rename to src/infrastructure/__init__.py diff --git a/src/applications/revit/revit_logger.py b/src/infrastructure/logging.py similarity index 97% rename from src/applications/revit/revit_logger.py rename to src/infrastructure/logging.py index da8399e..93a76f0 100644 --- a/src/applications/revit/revit_logger.py +++ b/src/infrastructure/logging.py @@ -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): diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/services/carbon_calculator.py b/src/services/carbon_calculator.py new file mode 100644 index 0000000..8e4776a --- /dev/null +++ b/src/services/carbon_calculator.py @@ -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", + ) diff --git a/src/services/element_processor.py b/src/services/element_processor.py new file mode 100644 index 0000000..b8a7226 --- /dev/null +++ b/src/services/element_processor.py @@ -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 diff --git a/src/services/material_processor.py b/src/services/material_processor.py new file mode 100644 index 0000000..49e7bd0 --- /dev/null +++ b/src/services/material_processor.py @@ -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, + ) diff --git a/tests/manual_test.py b/tests/manual_test.py deleted file mode 100644 index b87b4cd..0000000 --- a/tests/manual_test.py +++ /dev/null @@ -1,93 +0,0 @@ -# pytest: skip-file - -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 RevitModel -from src.applications.revit.revit_source_validator import RevitSourceValidator -from src.carbon.aggregator import MassAggregator -from src.applications.revit.revit_logger import RevitLogger - -# Import required libraries -from specklepy.api.client import SpeckleClient -from specklepy.core.api import operations -from specklepy.transports.server import ServerTransport - -# Define global variables -HOST = "https://app.speckle.systems/" -AUTHENTICATION_TOKEN = "840e5a18cda38ccc2a9ed8b52e9316530505c14181" -STREAM_ID = "99bdf924fb" -BRANCH_NAME = "2843" - -# Setting up SpeckleClient and authenticating -client = SpeckleClient(host=HOST) -client.authenticate_with_token(token=AUTHENTICATION_TOKEN) - -# Receiving commit -transport = ServerTransport(STREAM_ID, client) -branch = client.branch.get(stream_id=STREAM_ID, name=BRANCH_NAME) -model_data = operations.receive(branch.commits.items[0].referencedObject, transport) - - -def create_processor_chain() -> tuple[RevitModel, RevitLogger]: - """ - Creates and configures the processing chain with all necessary dependencies. - - Returns: - tuple[RevitModel, RevitLogger]: - - Configured processor ready to handle Revit types - - Logger instance for accessing compliance results - """ - # Create core components - logger = RevitLogger() - mass_aggregator = MassAggregator() - - # Create processors - material_processor = RevitMaterialProcessor(mass_aggregator, logger) - carbon_processor = RevitCarbonProcessor() - compliance_checker = RevitCompliance(logger) - - # Create and return the main processor with logger - return ( - RevitModel( - material_processor=material_processor, - carbon_processor=carbon_processor, - compliance_checker=compliance_checker, - logger=logger, - ), - logger, - ) - - -try: - # Get version data - commit_root = branch.commits.items[0] - model_root = model_data - - # Validate source application - source_validator = RevitSourceValidator() - if not source_validator.validate_source_application(commit_root.sourceApplication): - print( - f"Automation requires Revit v3 commits. Received: {commit_root.sourceApplication}" - ) - if not source_validator.validate_connector_version( - int(getattr(model_root, "version", 2)) - ): - print( - "Automation required Revit models using the v3 " "connector. Received: v2." - ) - - # Create processor chain and get logger for results - processor, logger = create_processor_chain() - - # Process model - processor.process_elements(model_root) - - # Report compliance issues - compliance_summary = logger.get_warnings_summary() - - print("Processing completed successfully.") - -except Exception as e: - print(f"Processing failed: {str(e)}") - raise # Re-raise for proper error tracking diff --git a/tests/test_function.py b/tests/test_function.py index 964b633..8d2ff8d 100644 --- a/tests/test_function.py +++ b/tests/test_function.py @@ -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