From 809bfa4102fd85952c5e8bb68d0e2636dbb8a1a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Steinhagen?= Date: Mon, 10 Feb 2025 08:53:42 +0100 Subject: [PATCH] refactor: RevitModelProcessor --- src/interfaces/compliance_checker.py | 18 ++- src/interfaces/logger.py | 7 +- src/processors/compliance.py | 101 ++++++++++++++-- src/processors/model.py | 174 +++++++++------------------ tests/manual_test.py | 2 +- 5 files changed, 167 insertions(+), 135 deletions(-) diff --git a/src/interfaces/compliance_checker.py b/src/interfaces/compliance_checker.py index 3e9476c..02a4bec 100644 --- a/src/interfaces/compliance_checker.py +++ b/src/interfaces/compliance_checker.py @@ -1,14 +1,26 @@ from abc import ABC, abstractmethod from typing import Any, List +from dataclasses import dataclass +from typing import Optional -# TODO: Passed around but never used class ComplianceChecker(ABC): """Interface for compliance checks. Compliance are intended to be called for every object where an attribute is assumed. """ - # TODO: Passed around but never used + + @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]) -> bool: + def check_compliance( + self, element: Any, required_properties: List[str] + ) -> ValidationResult: """Check if element contains attribute(s)""" pass diff --git a/src/interfaces/logger.py b/src/interfaces/logger.py index 849e0be..50aa844 100644 --- a/src/interfaces/logger.py +++ b/src/interfaces/logger.py @@ -1,10 +1,11 @@ from abc import ABC, abstractmethod from typing import Dict, List + # TODO: Implementations use **kwargs which is silly. Formalize. class Logger(ABC): - """Interface for logging. - """ + """Interface for logging.""" + @abstractmethod def log_error(self, message: str, **kwargs) -> None: """Log an error. @@ -34,7 +35,7 @@ class Logger(ABC): @abstractmethod def get_warnings_summary(self) -> Dict: - """Returns a dictionary of warning messages. The dictionary groups the wanring types. + """Returns a dictionary of warning messages. The dictionary groups the warning types. Returns: Dict: {warning_type : [object_ids], ...} diff --git a/src/processors/compliance.py b/src/processors/compliance.py index a864648..a8817d1 100644 --- a/src/processors/compliance.py +++ b/src/processors/compliance.py @@ -1,25 +1,104 @@ -from typing import Any, List -from src.interfaces.compliance_checker import ComplianceChecker -from src.interfaces.logger import Logger -from src.utils.constants import ID +from typing import Any, List, Optional +from dataclasses import dataclass + +from src.interfaces import ComplianceChecker, Logger +from src.utils.constants import ( + ID, + SPECKLE_TYPE, + LINE, + ARC, + CIRCLE, + PROPERTIES, + MATERIAL_QUANTITIES, + STRUCTURAL_ASSET, + VOLUME, + DENSITY, +) -# NOTE: Only provide docstring if not covered by base class class RevitComplianceChecker(ComplianceChecker): """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, required_properties: List[str]) -> bool: + def check_compliance( + self, element: Any, required_properties: List[str] + ) -> ComplianceChecker.ValidationResult: """ - Checks basic element compliance (presence of ID and any custom required properties) + 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, + ) + + return validation + + def _validate_element(self, element: Any) -> ComplianceChecker.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: - self._logger.log_warning( - "Element missing ID", object_id="unknown", missing_property=ID + return self.ValidationResult( + is_valid=False, error_property=ID, error_message="Missing element ID" ) - return False - return True # Basic compliance only requires ID + # Check Properties + properties = getattr(element, PROPERTIES, None) + if not properties: + return self.ValidationResult( + is_valid=False, + error_property=PROPERTIES, + error_message="Missing Properties", + ) + + # 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", + ) + + # Validate material properties + for material_name, material_data in material_quantities.items(): + for required_prop in [VOLUME, STRUCTURAL_ASSET, DENSITY]: + if required_prop not in material_data: + return self.ValidationResult( + is_valid=False, + error_property=required_prop, + error_message=f"Missing {required_prop}", + ) + + return self.ValidationResult( + is_valid=True, material_quantities=material_quantities + ) diff --git a/src/processors/model.py b/src/processors/model.py index 0421414..7129712 100644 --- a/src/processors/model.py +++ b/src/processors/model.py @@ -4,26 +4,17 @@ from src.interfaces.material_processor import MaterialProcessor from src.interfaces.compliance_checker import ComplianceChecker from src.interfaces.logger import Logger from src.utils.constants import ( - LINE, - ARC, - CIRCLE, ELEMENTS, NAME, - PROPERTIES, - MATERIAL_QUANTITIES, - SPECKLE_TYPE, ID, - VOLUME, - STRUCTURAL_ASSET, - DENSITY, ) # NOTE: Only provide docstring if not covered by base class class RevitModelProcessor(ModelProcessor): - """Implementation of the ModelProcessor in the Revit context. - """ + """Implementation of the ModelProcessor in the Revit context.""" + def __init__( self, material_processor: MaterialProcessor, @@ -35,127 +26,76 @@ class RevitModelProcessor(ModelProcessor): self._logger = logger def process_elements(self, model: Any) -> None: - """Process all elements in the model hierarchically. + """Model traversal. Model → Levels → Type Groups → Elements. - - Args: - model (Any): commit root - - Raises: - ValueError: root["elements"] must exist """ - levels = getattr(model, ELEMENTS, None) - if not levels: - raise ValueError("Invalid model: missing elements at root.") + levels = self._get_elements(model, "model") for level in levels: - self._process_level(level) + level_name = self._get_name(level) + type_groups = self._get_elements(level, f"level {level_name}") - def _process_level(self, level: Any) -> None: - """Process all groups contained on a level. + for type_group in type_groups: + type_name = self._get_name(type_group) + groups = self._get_elements(type_group, f"type {type_name}") - Args: - level (Any): root["elements"][i] → level collection - - Raises: - ValueError: root["elements"][i]["elements"] must exist - """ - type_groups = getattr(level, ELEMENTS, None) - if not type_groups: - level_name = getattr(level, NAME, "Unknown") - raise ValueError( - f"Invalid level structure: missing elements in {level_name}" - ) - - level_name = getattr(level, NAME) - for type_group in type_groups: - self._process_type_group(type_group, level_name) - - def _process_type_group(self, type_group: Any, level_name: str) -> None: - """Process a group of elements of the same type - - Args: - type_group (Any): group to process - level_name (str): associated level of the groups - - Raises: - ValueError: root["elements"][i]["elements"][i]["elements"] must exist - ValueError: root["elements"][i]["elements"][i]["elements"][i]["elements"] must exist - """ - groups = getattr(type_group, ELEMENTS, None) - if not groups: - type_name = getattr(type_group, NAME, "Unknown") - raise ValueError(f"Invalid type structure: missing elements in {type_name}") - - type_name = getattr(type_group, NAME) - - for group in groups: - revit_objects = getattr(group, ELEMENTS, None) - if not revit_objects: - raise ValueError( - f"Invalid type structure: missing elements in " - f"{getattr(group, NAME, None)}" - ) - for revit_object in revit_objects: - self.process_element(level_name, type_name, revit_object) + 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: - """Processing of actual model object after successful traversal of the commit. + """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 - revit_object (Any): speckle object + model_object (Any): speckle object containing properties for processing """ - - # We can probably straight up skip Line and Arc. Logging it as a warning is dumb - # TODO: Possibly add to logger for info? Not a warning though. - speckle_type = getattr(model_object, SPECKLE_TYPE, None) - if speckle_type in [LINE, ARC, CIRCLE]: + # First check compliance - this also handles logging any validation warnings + validation = self._compliance_checker.check_compliance(model_object, []) + if not validation.is_valid: return - element_id = getattr(model_object, ID, None) - if not element_id: - return - - # Check Material Quantities - properties = getattr(model_object, PROPERTIES, None) - if not properties: - self._logger.log_warning( - "Missing Material Quantities", - object_id=element_id, - missing_property=PROPERTIES, - ) - return - material_quantities = properties.get(MATERIAL_QUANTITIES, None) - if not material_quantities: - self._logger.log_warning( - "Missing Material Quantities", - object_id=element_id, - missing_property=MATERIAL_QUANTITIES, - ) - return - - # Process each material - # 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 material_quantities.items(): - # Check required material properties - for required_prop in [VOLUME, STRUCTURAL_ASSET, DENSITY]: - if required_prop not in material_data: - self._logger.log_warning( - f"Missing {required_prop}", - object_id=element_id, - missing_property=required_prop, - ) - return - - try: + 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 ) - self._logger.log_success(element_id) - except Exception as e: - self._logger.log_error( - f"Failed to process element {element_id}", error=str(e) - ) + # Log success only after all material processing complete + self._logger.log_success(getattr(model_object, ID)) + + 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) + ) + + @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") diff --git a/tests/manual_test.py b/tests/manual_test.py index ff11b41..b1f3694 100644 --- a/tests/manual_test.py +++ b/tests/manual_test.py @@ -22,7 +22,7 @@ BRANCH_NAME = "2391" client = SpeckleClient(host=HOST) client.authenticate_with_token(token=AUTHENTICATION_TOKEN) -# Receving commit +# 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)