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

refactor(revit_model): cleanup

Browse Source 
first of many
This commit is contained in:
Björn Steinhagen
2025-02-23 22:27:11 +01:00
parent d130815f20
commit 3e8d80dd21
6 changed files with 126 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+11 -14
View File
@@ -1,4 +1,3 @@
from pydantic import Field, SecretStr
from speckle_automate import (
AutomateBase,
AutomationContext,
@@ -8,13 +7,13 @@ from speckle_automate import (
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_model import RevitElementProcessor
from src.applications.revit.revit_source_validator import RevitSourceValidator
from src.carbon.aggregator import MassAggregator
from src.applications.revit.revit_logger import RevitLogger
from src.core.base import Model
from src.carbon import WoodSupplier
# TODO: Function inputs
class FunctionInputs(AutomateBase):
"""User-defined function inputs."""
@@ -54,18 +53,16 @@ def automate_function(
return
# Create processor chain and get logger for results
processor = configure_components()
processor = initialize_revit_processor()
# Process model
processor.process_elements(model_root)
# Analyze elements
processor.analyze_elements(model_root)
# Logger information - successes
(
logger_successes,
logger_infos,
logger_warnings,
logger_failures,
) = processor.get_processing_results()
logger_infos = processor.logger.get_info_summary()
logger_successes = processor.logger.get_success_summary()
logger_warnings = processor.logger.get_warnings_summary()
logger_failures = processor.logger.get_errors_summary()
for category, object_ids in logger_successes.items():
automate_context.attach_success_to_objects(
@@ -107,7 +104,7 @@ def automate_function(
# TODO instead of hard-coding revit, demo a factory method to inject implementations based on
# function input
def configure_components() -> Model:
def initialize_revit_processor() -> RevitElementProcessor:
"""Configures and wires up processor components with dependencies.
Creates core system components (logger, aggregator) and configures processors
@@ -131,7 +128,7 @@ def configure_components() -> Model:
compliance_checker = RevitCompliance(logger) # Compliance checker to "inject"
# Create and return the main processor with dependencies "injected"
return RevitModel(
return RevitElementProcessor(
material_processor=material_processor,
carbon_processor=carbon_processor,
compliance_checker=compliance_checker,
src/applications/revit/revit_carbon_processor.py
+21 -12
View File
@@ -1,7 +1,7 @@
from src.core.base import CarbonProcessor
from src.core.types import ( MetalClass, WoodClass )
from src.core.types import MetalClass, WoodClass
from src.carbon.types import CarbonData
from src.carbon.data import ( wood_factors, metal_factors )
from src.carbon.data import wood_factors, metal_factors
from src.applications.revit.utils.material_type_handler import (
MaterialType,
)
@@ -10,12 +10,11 @@ from src.applications.revit.utils.constants import (
)
import json
class RevitCarbonProcessor(CarbonProcessor):
"""Implementation of CarbonProcessor for Revit context."""
def process(
self, model_object
) -> CarbonData:
def process(self, model_object) -> CarbonData:
"""Compute embodied carbon per-element based previously asserted material properties.
Args:
@@ -33,16 +32,26 @@ class RevitCarbonProcessor(CarbonProcessor):
# 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
material_quantities = model_object[PROPERTIES]["Material Quantities"][
"FE_Steel"
] # NOTE: This is dangerous.
volume, density = (
material_quantities.volume.value,
material_quantities.density.value,
)
mass = volume * density
# Default to hot rolled
factor = metal_factors[MetalClass.HOT_ROLLED]
embodied_carbon = mass * factor
model_object[PROPERTIES]["Embodied Carbon Data"]["category"] = MetalClass.HOT_ROLLED
model_object[PROPERTIES]["Embodied Carbon Data"]["factor"] = factor
model_object[PROPERTIES]["Embodied Carbon Data"]["embodied_carbon"] = embodied_carbon
pass
model_object[PROPERTIES]["Embodied Carbon Data"][
"category"
] = MetalClass.HOT_ROLLED
model_object[PROPERTIES]["Embodied Carbon Data"][
"factor"
] = factor # TODO: Append with units
model_object[PROPERTIES]["Embodied Carbon Data"][
"embodied_carbon"
] = embodied_carbon # TODO: Append with units
src/applications/revit/revit_model.py
+93 -141
View File
@@ -1,8 +1,5 @@
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 typing import Generator, Any, Tuple
from src.core.base import MaterialProcessor, Compliance, CarbonProcessor
from src.core.base.logger import Logger
from src.applications.revit.utils.constants import (
ELEMENTS,
@@ -12,11 +9,9 @@ from src.applications.revit.utils.constants import (
PROPERTIES,
)
# NOTE: Only provide docstring if not covered by base class
class RevitModel(Model):
"""Implementation of the ModelProcessor in the Revit context."""
class RevitElementProcessor:
"""Processes Revit model elements to extract and analyze material data."""
def __init__(
self,
@@ -25,127 +20,109 @@ class RevitModel(Model):
compliance_checker: Compliance,
logger: Logger,
):
self._material_processor = material_processor
self._carbon_processor = carbon_processor
self._compliance_checker = compliance_checker
self._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
def analyze_elements(self, model: Any) -> None:
"""Processes all valid elements from the model."""
for level, type_name, element in self._extract_valid_elements(model):
try:
self._process_materials(level, type_name, element)
except Exception as e:
self.logger.log_error(
object_id=getattr(element, ID, "Unknown"),
category="Processing Error",
message=f"Error processing element: {str(e)}",
)
if processed_material: # If processing was successful
model_object[PROPERTIES]["Embodied Carbon Data"] = vars(
processed_material
def _extract_valid_elements(
self, model: Any
) -> Generator[Tuple[str, str, Any], None, None]:
"""Yields valid elements (level, type_name, revit_object) after compliance checks."""
for level, type_name, revit_object in self._get_element_hierarchy(model):
if self._is_compliant(revit_object):
yield level, type_name, revit_object
else:
self._log_skipped_element(revit_object)
def _get_element_hierarchy(
self, model: Any
) -> Generator[Tuple[str, str, Any], None, None]:
"""Flattens nested elements to yield (level, type_name, element)."""
for level in self._get_elements(model, "model"):
level_name = getattr(level, NAME, "Unknown")
for type_group in self._get_elements(level, f"level {level_name}"):
type_name = getattr(type_group, NAME, "Unknown")
for group in self._get_elements(type_group, f"type {type_name}"):
yield from (
(level_name, type_name, revit_object)
for revit_object in self._get_elements(
group, f"group {getattr(group, NAME, 'Unknown')}"
)
)
if getattr(processed_material, "type") == "Concrete":
model_object[PROPERTIES]["Embodied Carbon Data"][
"element"
] = self._categorize(type_name)
def _is_compliant(self, model_object: Any) -> bool:
"""Checks if an element passes compliance checks."""
return self.compliance_checker.check_compliance(model_object)
processed_carbon = self._carbon_processor.process(model_object)
def _process_materials(self, level: str, type_name: str, model_object: Any) -> None:
"""Extracts material data and processes carbon calculations."""
object_id = getattr(model_object, ID, "Unknown")
material_quantities = model_object[PROPERTIES].get(MATERIAL_QUANTITIES, {})
if processed_carbon:
self._logger.log_success(
object_id=object_id,
category="Successfully Processed",
message="Carbon calculations completed successfully for this element.",
)
for material_name, material_data in material_quantities.items():
processed_material = self.material_processor.process(
object_id, material_data, level, type_name
)
if not processed_material:
continue
model_object[PROPERTIES]["Embodied Carbon Calculations"] = vars(processed_carbon)
model_object[PROPERTIES]["Embodied Carbon Data"] = vars(processed_material)
except Exception as e:
# Log any processing errors that occur
self._logger.log_error(object_id, "Material Processing Error", str(e))
# Dictionary-based lookup instead of multiple if-elif checks
category_map = {
"floor": "Slabs",
"stair": "Slabs",
"slab edges": "Slabs",
"wall": "Walls",
"column": "Columns",
"framing": "Beam",
"beam": "Beam",
"foundation": "Foundations",
}
category = next(
(v for k, v in category_map.items() if k in type_name.lower()), None
)
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.
if category and getattr(processed_material, "type") == "Concrete":
model_object[PROPERTIES]["Embodied Carbon Data"]["element"] = category
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(),
processed_carbon = self.carbon_processor.process(model_object)
if processed_carbon:
self.logger.log_success(
object_id=object_id,
category="Successfully Processed",
message="Carbon calculations completed successfully.",
)
model_object[PROPERTIES]["Embodied Carbon Calculations"] = vars(
processed_carbon
)
def _log_skipped_element(self, model_object: Any) -> None:
"""Logs elements that fail compliance checks."""
self.logger.log_info(
object_id=getattr(model_object, ID, "Unknown"),
category="Skipped Elements",
message="Element did not meet compliance criteria.",
)
@staticmethod
def _get_elements(node: Any, context: str) -> list:
"""Get elements from a node, with consistent error handling.
Args:
node: Node to extract elements from
context: Context for error message if elements missing
Raises:
ValueError: If elements are missing
"""
"""Get elements from a node, with consistent error handling."""
elements = getattr(node, ELEMENTS, None)
if not elements:
name = getattr(node, NAME, "Unknown")
@@ -153,28 +130,3 @@ class RevitModel(Model):
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.")
src/core/base/__init__.py
+1 -2
View File
@@ -1,5 +1,4 @@
from .logger import Logger
from .model import Model
from .source_validator import SourceApplicationValidator
from .material_processor import MaterialProcessor
from .compliance import Compliance
@@ -11,5 +10,5 @@ __all__ = [
"SourceApplicationValidator",
"MaterialProcessor",
"Compliance",
"CarbonProcessor"
"CarbonProcessor",
]
src/core/base/model.py
-40
View File
@@ -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
tests/manual_test.py
-93
View File
@@ -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