diff --git a/main.py b/main.py index 2a36d0e..b2127e2 100644 --- a/main.py +++ b/main.py @@ -5,13 +5,13 @@ from speckle_automate import ( execute_automate_function, ) -from src.applications.revit.revit_material import RevitMaterial +from src.applications.revit.revit_material_processor import RevitMaterialProcessor 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, Logger +from src.core.base import Model # TODO: Function inputs @@ -66,26 +66,40 @@ def automate_function( processor.process_elements(model_root) # Logger information - successes - logger_successes, logger_warnings = processor.get_processing_results() - if logger_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="Successfully Processed", - object_ids=logger_successes, - message="Carbon calculations completed successfully for this element.", + category=category, + object_ids=object_ids, + message="Carbon calculations completed successfully for these elements.", ) - # Logger information - warnings - if logger_warnings: - for missing_property, elements in logger_warnings.items(): - automate_context.attach_warning_to_objects( - category="Missing Required Revit Properties", - object_ids=elements, - message=( - f"Property '{missing_property}' is missing, which prevents carbon " - f"calculations. If this element is critical to your analysis, please " - f"update its Revit properties." - ), - ) + 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", "") @@ -116,7 +130,9 @@ def configure_components() -> Model: # TODO: results_aggregator = ResultAggregator and get rid of mass_aggregator # Create processors - material_processor = RevitMaterial(mass_aggregator) # Material handler to "inject" + material_processor = RevitMaterialProcessor( + mass_aggregator, logger + ) # Material handler to "inject" compliance_checker = RevitCompliance(logger) # Compliance checker to "inject" # Create and return the main processor with dependencies "injected" diff --git a/src/applications/revit/revit_compliance.py b/src/applications/revit/revit_compliance.py index f0e2d92..8658dea 100644 --- a/src/applications/revit/revit_compliance.py +++ b/src/applications/revit/revit_compliance.py @@ -1,4 +1,4 @@ -from typing import Any, List +from typing import Any from src.core.base import Compliance, Logger from src.applications.revit.utils.constants import ( @@ -9,9 +9,10 @@ from src.applications.revit.utils.constants import ( CIRCLE, PROPERTIES, MATERIAL_QUANTITIES, - STRUCTURAL_ASSET, VOLUME, - DENSITY, + MATERIAL_CATEGORY, + MATERIAL_CLASS, + MATERIAL_NAME, ) @@ -23,81 +24,68 @@ class RevitCompliance(Compliance): def __init__(self, logger: Logger): self._logger = logger - def check_compliance( - self, element: Any, required_properties: List[str] - ) -> Compliance.ValidationResult: + def check_compliance(self, element: Any) -> bool: """ Validates element and returns validation result with material data if valid. Args: element: Element to validate - required_properties: List of required properties (unused but kept for interface) Returns: ValidationResult containing validation status and material data if valid """ - validation = self._validate_element(element) - if not validation.is_valid: - self._logger.log_warning( - validation.error_message, - object_id=getattr(element, ID, "unknown"), - missing_property=validation.error_property, - ) + # Check ID + object_id = getattr(element, ID, None) + if not object_id: + raise ValueError("Should have an id.") - return validation - - def _validate_element(self, element: Any) -> Compliance.ValidationResult: - """Internal validation logic for a single element. - - Args: - element: Element to validate - - Returns: - ValidationResult with validation status and error details or material data - """ # Skip geometry elements speckle_type = getattr(element, SPECKLE_TYPE, None) if speckle_type in [LINE, ARC, CIRCLE]: - return self.ValidationResult( - is_valid=False, error_message="Geometry element - skipping" - ) - - # Check ID - element_id = getattr(element, ID, None) - if not element_id: - return self.ValidationResult( - is_valid=False, error_property=ID, error_message="Missing element ID" + 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: - return self.ValidationResult( - is_valid=False, - error_property=PROPERTIES, - error_message="Missing 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: - return self.ValidationResult( - is_valid=False, - error_property=MATERIAL_QUANTITIES, - error_message="Missing 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, STRUCTURAL_ASSET, DENSITY]: + for required_prop in [ + VOLUME, + MATERIAL_CATEGORY, + MATERIAL_CLASS, + MATERIAL_NAME, + ]: if required_prop not in material_data: - return self.ValidationResult( - is_valid=False, - error_property=required_prop, - error_message=f"Missing {required_prop}", + self._logger.log_error( + object_id, + f"Missing {required_prop}.", + "Indicates changes to the Revit " + "connector. Inspect commit and " + "update accordingly.", ) + return False - return self.ValidationResult( - is_valid=True, material_quantities=material_quantities - ) + return True diff --git a/src/applications/revit/revit_logger.py b/src/applications/revit/revit_logger.py index 2d45935..da8399e 100644 --- a/src/applications/revit/revit_logger.py +++ b/src/applications/revit/revit_logger.py @@ -1,48 +1,85 @@ import structlog -from typing import Dict, DefaultDict +from typing import Dict, Set, Optional from collections import defaultdict from src.core.base.logger import Logger # Import the interface -# NOTE: Only provide docstring if not covered by base class - class RevitLogger(Logger): - """Implements Logger interface""" + """Implements Logger interface with category-based logging""" def __init__(self): - self.missing_properties: DefaultDict[str, set] = defaultdict(set) - self.successful_elements: set = set() self._structlog = structlog.get_logger() + self._errors: Dict[str, Set[str]] = defaultdict(set) + self._warnings: Dict[str, Set[str]] = defaultdict(set) + self._successes: Dict[str, Set[str]] = defaultdict(set) + self._info: Dict[str, Set[str]] = defaultdict(set) - def log_error(self, message: str, **kwargs) -> None: - self._structlog.error(message, **kwargs) + def log_error( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log an error for a specific object under a category""" + self._errors[category].add(object_id) + if message: + self._structlog.error(message, object_id=object_id, category=category) + else: + self._structlog.error( + "Error logged", object_id=object_id, category=category + ) - def log_warning(self, message: str, **kwargs) -> None: - """Log a warning message. - Categorises and caches missing properties if 'missing_property' and 'object_id' specified in the kwargs. + def log_warning( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log a warning for a specific object under a category""" + self._warnings[category].add(object_id) + if message: + self._structlog.warning(message, object_id=object_id, category=category) + else: + self._structlog.warning( + "Warning logged", object_id=object_id, category=category + ) - Args: - message: Warning message to log - **kwargs: Additional context. If 'object_id' and 'missing_property' are - provided, they will be tracked for compliance reporting. - """ - self._structlog.warn(message, **kwargs) + def log_success( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log a success for a specific object under a category""" + self._successes[category].add(object_id) + if message: + self._structlog.info(message, object_id=object_id, category=category) + else: + self._structlog.info( + "Success logged", object_id=object_id, category=category + ) - # Track missing properties if relevant kwargs are provided - object_id = kwargs.get("object_id") - missing_property = kwargs.get("missing_property") - if object_id and missing_property: - self.missing_properties[missing_property].add(object_id) + def log_info( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log information for a specific object under a category""" + self._info[category].add(object_id) + if message: + self._structlog.info(message, object_id=object_id, category=category) + else: + self._structlog.info( + "Information logged", object_id=object_id, category=category + ) - def log_success(self, object_id: str, **kwargs) -> None: - self._structlog.info(f"Successfully processed element", object_id=object_id) - self.successful_elements.add(object_id) + @staticmethod + def _convert_sets_to_lists(data: Dict[str, Set[str]]) -> Dict[str, list]: + """Convert set values to lists in dictionary""" + return {category: list(objects) for category, objects in data.items()} def get_warnings_summary(self) -> Dict[str, list]: - return { - prop: list(elements) for prop, elements in self.missing_properties.items() - } + """Get all warnings grouped by category""" + return self._convert_sets_to_lists(self._warnings) - def get_successful_summary(self) -> list: - return list(self.successful_elements) + def get_errors_summary(self) -> Dict[str, list]: + """Get all errors grouped by category""" + return self._convert_sets_to_lists(self._errors) + + def get_success_summary(self) -> Dict[str, list]: + """Get all successes grouped by category""" + return self._convert_sets_to_lists(self._successes) + + def get_info_summary(self) -> Dict[str, list]: + """Get all info logs grouped by category""" + return self._convert_sets_to_lists(self._info) diff --git a/src/applications/revit/revit_material.py b/src/applications/revit/revit_material.py deleted file mode 100644 index cb5576a..0000000 --- a/src/applications/revit/revit_material.py +++ /dev/null @@ -1,35 +0,0 @@ -from typing import Dict, Any -from src.core.base import Material -from src.core.types.material import ( - MaterialData, -) -from src.applications.revit.utils.constants import ( - VOLUME, - VALUE, - DENSITY, - STRUCTURAL_ASSET, -) - -# NOTE: Only provide docstring if not covered by base class - - -class RevitMaterial(Material): - """Implementation of the MaterialProcessor for the Revit context.""" - - def __init__(self, mass_aggregator: "MassAggregator"): - self._mass_aggregator = mass_aggregator - - def process_material( - self, material_data: Dict[str, Any], level: str, type_name: str - ) -> MaterialData: - volume = material_data[VOLUME][VALUE] - density = material_data[DENSITY][VALUE] - mass = volume * density - structural_asset = material_data[STRUCTURAL_ASSET] - - mat_data = MaterialData( - volume=volume, density=density, structural_asset=structural_asset, mass=mass - ) - - self._mass_aggregator.add_mass(mass, level, type_name, structural_asset) - return mat_data diff --git a/src/applications/revit/revit_material_processor.py b/src/applications/revit/revit_material_processor.py new file mode 100644 index 0000000..855324b --- /dev/null +++ b/src/applications/revit/revit_material_processor.py @@ -0,0 +1,43 @@ +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 index 85c7d0c..1bef4ff 100644 --- a/src/applications/revit/revit_model.py +++ b/src/applications/revit/revit_model.py @@ -1,12 +1,14 @@ -from typing import Any +from typing import Any, Tuple, Dict, List from src.core.base import Model -from src.core.base import Material +from src.core.base import MaterialProcessor from src.core.base import Compliance 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 @@ -17,7 +19,7 @@ class RevitModel(Model): def __init__( self, - material_processor: Material, + material_processor: MaterialProcessor, compliance_checker: Compliance, logger: Logger, ): @@ -67,32 +69,62 @@ class RevitModel(Model): model_object (Any): speckle object containing properties for processing """ # First check compliance - this also handles logging any validation warnings - validation = self._compliance_checker.check_compliance(model_object, []) - if not validation.is_valid: + 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 - # TODO: Project 2427 is an interesting one with compound materials - # TODO: Checkout object fefcc95c2f0ecd28a49ecdd7764e2d79. Worth skipping if volume = 0? - for material_name, material_data in validation.material_quantities.items(): - self._material_processor.process_material( - material_data, level, type_name + for material_name, material_data in material_quantities.items(): + processed_material = self._material_processor.process( + object_id, material_data, level, type_name ) - # Log success only after all material processing complete - self._logger.log_success(getattr(model_object, ID)) + if processed_material: # If processing was successful + self._logger.log_success( + object_id=object_id, + category="Successfully Processed", + message="Carbon calculations completed successfully for this element.", + ) + + 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) except Exception as e: # Log any processing errors that occur - self._logger.log_error( - f"Failed to process element {getattr(model_object, ID)}", error=str(e) - ) + self._logger.log_error(object_id, "Material Processing Error", str(e)) - # TODO: This is gross. - def get_processing_results(self) -> tuple[list, dict]: + 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_successful_summary(), + self._logger.get_success_summary(), + self._logger.get_info_summary(), self._logger.get_warnings_summary(), + self._logger.get_errors_summary(), ) @staticmethod @@ -118,3 +150,23 @@ class RevitModel(Model): 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/utils/constants.py b/src/applications/revit/utils/constants.py index 0943789..a73c9f3 100644 --- a/src/applications/revit/utils/constants.py +++ b/src/applications/revit/utils/constants.py @@ -6,12 +6,17 @@ REQUIRED_PROPERTIES = ["volume", "density", "structuralAsset"] APPLICATION_ID = "applicationId" ARC = "Objects.Geometry.Arc" CIRCLE = "Objects.Geometry.Circle" +COMPRESSIVE_STRENGTH = "compressiveStrength" DENSITY = "density" ELEMENTS = "elements" ID = "id" LINE = "Objects.Geometry.Line" MASS = "mass" +MATERIAL_CATEGORY = "materialCategory" +MATERIAL_CLASS = "materialClass" +MATERIAL_NAME = "materialName" MATERIAL_QUANTITIES = "Material Quantities" +MATERIAL_TYPE = "materialType" NAME = "name" PROPERTIES = "properties" SOURCE_APPLICATION = "sourceApplication" diff --git a/src/applications/revit/utils/material_quality_strategy.py b/src/applications/revit/utils/material_quality_strategy.py new file mode 100644 index 0000000..17bb690 --- /dev/null +++ b/src/applications/revit/utils/material_quality_strategy.py @@ -0,0 +1,113 @@ +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 new file mode 100644 index 0000000..18e5bbf --- /dev/null +++ b/src/applications/revit/utils/material_type_handler.py @@ -0,0 +1,87 @@ +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/core/base/__init__.py b/src/core/base/__init__.py index 5d22cf3..f91cbec 100644 --- a/src/core/base/__init__.py +++ b/src/core/base/__init__.py @@ -1,13 +1,13 @@ from .logger import Logger from .model import Model from .source_validator import SourceApplicationValidator -from .material import Material +from .material_processor import MaterialProcessor from .compliance import Compliance __all__ = [ "Logger", "Model", "SourceApplicationValidator", - "Material", + "MaterialProcessor", "Compliance", ] diff --git a/src/core/base/compliance.py b/src/core/base/compliance.py index b71e4ae..65cb988 100644 --- a/src/core/base/compliance.py +++ b/src/core/base/compliance.py @@ -1,7 +1,5 @@ from abc import ABC, abstractmethod -from typing import Any, List -from dataclasses import dataclass -from typing import Optional +from typing import Any class Compliance(ABC): @@ -9,18 +7,7 @@ class Compliance(ABC): Compliance are intended to be called for every object where an attribute is assumed. """ - @dataclass - class ValidationResult: - """Results of element validation including material data if valid""" - - is_valid: bool - material_quantities: Optional[dict] = None - error_property: Optional[str] = None - error_message: Optional[str] = None - @abstractmethod - def check_compliance( - self, element: Any, required_properties: List[str] - ) -> ValidationResult: + 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 index 50aa844..49f64f8 100644 --- a/src/core/base/logger.py +++ b/src/core/base/logger.py @@ -1,52 +1,54 @@ +from typing import Dict, Optional from abc import ABC, abstractmethod -from typing import Dict, List -# TODO: Implementations use **kwargs which is silly. Formalize. class Logger(ABC): - """Interface for logging.""" + """Abstract base class for logging functionality""" @abstractmethod - def log_error(self, message: str, **kwargs) -> None: - """Log an error. - - Args: - message (str): error message - """ + 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, message: str, **kwargs) -> None: - """Log a warning. - - Args: - message (str): warning message - """ + 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, **kwargs) -> None: - """Log a success. - - Args: - object_id (str): success message - """ + def log_success( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log a success for a specific object""" pass @abstractmethod - def get_warnings_summary(self) -> Dict: - """Returns a dictionary of warning messages. The dictionary groups the warning types. - - Returns: - Dict: {warning_type : [object_ids], ...} - """ + def log_info( + self, object_id: str, category: str, message: Optional[str] = None + ) -> None: + """Log information for a specific object""" pass @abstractmethod - def get_successful_summary(self) -> List: - """Returns a list of all successfully processed elements. - - Returns: - List: [object_ids] - """ + 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.py b/src/core/base/material_processor.py similarity index 70% rename from src/core/base/material.py rename to src/core/base/material_processor.py index 7ecb0bf..5eb2236 100644 --- a/src/core/base/material.py +++ b/src/core/base/material_processor.py @@ -2,16 +2,17 @@ from abc import ABC, abstractmethod from typing import Dict, Any -class Material(ABC): +class MaterialProcessor(ABC): """Interface for processing a material.""" @abstractmethod - def process_material( - self, material_data: Dict[str, Any], level: str, type_name: str + 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 diff --git a/src/core/base/model.py b/src/core/base/model.py index efc13c7..616f0b3 100644 --- a/src/core/base/model.py +++ b/src/core/base/model.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from typing import Any +from typing import Any, Dict, Tuple, List class Model(ABC): @@ -28,6 +28,13 @@ class Model(ABC): # TODO: This is gross @abstractmethod - def get_processing_results(self) -> tuple[list, dict]: + 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/types/__init__.py b/src/core/types/__init__.py index 62f0216..5d7c233 100644 --- a/src/core/types/__init__.py +++ b/src/core/types/__init__.py @@ -1,3 +1,3 @@ -from .material import MaterialData +from .material_data import MaterialData __all__ = ["MaterialData"] diff --git a/src/core/types/material.py b/src/core/types/material.py deleted file mode 100644 index a34fbfe..0000000 --- a/src/core/types/material.py +++ /dev/null @@ -1,12 +0,0 @@ -from dataclasses import dataclass -from typing import Optional - - -@dataclass -class MaterialData: - """Data class for material properties""" - - volume: float - density: float - structural_asset: str - mass: Optional[float] = None diff --git a/src/core/types/material_data.py b/src/core/types/material_data.py new file mode 100644 index 0000000..b866dc6 --- /dev/null +++ b/src/core/types/material_data.py @@ -0,0 +1,14 @@ +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/tests/manual_test.py b/tests/manual_test.py index 70fa66c..3257325 100644 --- a/tests/manual_test.py +++ b/tests/manual_test.py @@ -1,6 +1,6 @@ # pytest: skip-file -from src.applications.revit.revit_material import RevitMaterial +from src.applications.revit.revit_material_processor import RevitMaterialProcessor 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 @@ -16,7 +16,7 @@ from specklepy.transports.server import ServerTransport HOST = "https://app.speckle.systems/" AUTHENTICATION_TOKEN = "840e5a18cda38ccc2a9ed8b52e9316530505c14181" STREAM_ID = "99bdf924fb" -BRANCH_NAME = "2391" +BRANCH_NAME = "2843" # Setting up SpeckleClient and authenticating client = SpeckleClient(host=HOST) @@ -42,7 +42,7 @@ def create_processor_chain() -> tuple[RevitModel, RevitLogger]: mass_aggregator = MassAggregator() # Create processors - material_processor = RevitMaterial(mass_aggregator) + material_processor = RevitMaterialProcessor(mass_aggregator, logger) compliance_checker = RevitCompliance(logger) # Create and return the main processor with logger