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

refactor: RevitModelProcessor

Browse Source
This commit is contained in:
Björn Steinhagen
2025-02-10 08:53:42 +01:00
parent 9d46562419
commit 809bfa4102
5 changed files with 167 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/interfaces/compliance_checker.py
+15 -3
View File
@@ -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
src/interfaces/logger.py
+4 -3
View File
@@ -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], ...}
src/processors/compliance.py
+90 -11
View File
@@ -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
)
src/processors/model.py
+57 -117
View File
@@ -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")
tests/manual_test.py
+1 -1
View File
@@ -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)