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

docs: class and method docs

Browse Source
This commit is contained in:
Björn Steinhagen
2025-02-10 08:30:15 +01:00
parent 9ca9a1910b
commit 9d46562419
13 changed files with 155 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+3 -2
View File
@@ -12,9 +12,10 @@ from src.validators.revit import RevitSourceValidator
from src.aggregators.carbon_totals import MassAggregator
from src.logging.compliance_logger import ComplianceLogger
# TODO: Function inputs
class FunctionInputs(AutomateBase):
# An example of how to use secret values.
"""User-defined function inputs.
"""
whisper_message: SecretStr = Field(title="This is a secret message")
forbidden_speckle_type: str = Field(
title="Forbidden speckle type",
src/aggregators/carbon_totals.py
+21 -2
View File
@@ -2,16 +2,35 @@ 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)))
def add_mass(self, mass: float, level: str, type: str, material: str) -> None:
# 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
type (str): object collection (e.g. "Columns", "Structural Foundations")
material (str): name of the structural asset
"""
if mass <= 1e-6:
return
self.totals[level][type][material] += mass
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: {
src/interfaces/compliance_checker.py
+6 -1
View File
@@ -2,8 +2,13 @@ from abc import ABC, abstractmethod
from typing import Any, List
# 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
@abstractmethod
def check_compliance(self, element: Any, required_properties: List[str]) -> bool:
"""Check if element meets compliance requirements"""
"""Check if element contains attribute(s)"""
pass
src/interfaces/logger.py
+28 -6
View File
@@ -1,29 +1,51 @@
from abc import ABC, abstractmethod
from typing import Dict, List
# TODO: Implementations use **kwargs which is silly. Formalize.
class Logger(ABC):
"""Interface for logging.
"""
@abstractmethod
def log_error(self, message: str, **kwargs) -> None:
"""Log an error message"""
"""Log an error.
Args:
message (str): error message
"""
pass
@abstractmethod
def log_warning(self, message: str, **kwargs) -> None:
"""Log a warning message"""
"""Log a warning.
Args:
message (str): warning message
"""
pass
@abstractmethod
def log_success(self, object_id: str, **kwargs) -> None:
"""Log a success message"""
"""Log a success.
Args:
object_id (str): success message
"""
pass
@abstractmethod
def get_warnings_summary(self) -> Dict:
"""Get summary of logged warning messages"""
"""Returns a dictionary of warning messages. The dictionary groups the wanring types.
Returns:
Dict: {warning_type : [object_ids], ...}
"""
pass
@abstractmethod
def get_successful_summary(self) -> List:
"""Get list of successfully processed elements"""
"""Returns a list of all successfully processed elements.
Returns:
List: [object_ids]
"""
pass
src/interfaces/material_processor.py
+9 -1
View File
@@ -3,9 +3,17 @@ from typing import Dict, Any
class MaterialProcessor(ABC):
"""Interface for processing a material.
"""
@abstractmethod
def process_material(
self, material_data: Dict[str, Any], level: str, type_name: str
) -> None:
"""Process material data and compute required properties"""
"""Process material data and compute required properties
Args:
material_data (Dict[str, Any]): material data
level (str): associated level of object
type_name (str): object type
"""
pass
src/interfaces/model_processor.py
+16 -3
View File
@@ -3,12 +3,25 @@ from typing import Any
class ModelProcessor(ABC):
"""Interface for model processing.
"""
@abstractmethod
def process_elements(self, model: Any) -> None:
"""Process all elements in the model"""
"""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, element: Any) -> None:
"""Process a single element"""
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
src/interfaces/validator.py
+11 -1
View File
@@ -2,7 +2,17 @@ from abc import ABC, abstractmethod
class SourceApplicationValidator(ABC):
"""Interface for source application validator.
Host app should be supported by the automation.
"""
@abstractmethod
def validate(self, source_app: str) -> bool:
"""Validate if the source application is supported"""
"""Assert that the source application is supported.
Args:
source_app (str): sourceApplication from the commit root
Returns:
bool: True if supported, False if not
"""
pass
src/logging/compliance_logger.py
+6 -8
View File
@@ -4,21 +4,22 @@ from collections import defaultdict
from src.interfaces.logger import Logger # Import the interface
# NOTE: Only provide docstring if not covered by base class
class ComplianceLogger(Logger): # Explicitly implement Logger interface
class ComplianceLogger(Logger):
"""Implements Logger interface
"""
def __init__(self):
self.missing_properties: DefaultDict[str, set] = defaultdict(set)
self.successful_elements: set = set()
self._structlog = structlog.get_logger()
def log_error(self, message: str, **kwargs) -> None:
"""Log an error message"""
self._structlog.error(message, **kwargs)
def log_warning(self, message: str, **kwargs) -> None:
"""
Log a warning message. Also tracks missing properties if they are specified
in the kwargs.
""" Log a warning message.
Categorises and caches missing properties if 'missing_property' and 'object_id' specified in the kwargs.
Args:
message: Warning message to log
@@ -34,16 +35,13 @@ class ComplianceLogger(Logger): # Explicitly implement Logger interface
self.missing_properties[missing_property].add(object_id)
def log_success(self, object_id: str) -> None:
"""Log a successful element processing"""
self._structlog.info(f"Successfully processed element", object_id=object_id)
self.successful_elements.add(object_id)
def get_warnings_summary(self) -> Dict[str, list]:
"""Get summary of logged messages"""
return {
prop: list(elements) for prop, elements in self.missing_properties.items()
}
def get_successful_summary(self) -> list:
"""Get list of successfully processed elements"""
return list(self.successful_elements)
src/processors/compliance.py
+3
View File
@@ -3,8 +3,11 @@ from src.interfaces.compliance_checker import ComplianceChecker
from src.interfaces.logger import Logger
from src.utils.constants import ID
# NOTE: Only provide docstring if not covered by base class
class RevitComplianceChecker(ComplianceChecker):
"""Implementation of the ComplianceChecker in the context of Revit.
"""
def __init__(self, logger: Logger):
self._logger = logger
src/processors/material.py
+5 -2
View File
@@ -2,11 +2,14 @@ from typing import Dict, Any
from src.interfaces.material_processor import MaterialProcessor
from src.models.material import (
MaterialData,
) # Import from models instead of defining here
from src.utils.constants import *
)
from src.utils.constants import VOLUME, VALUE, DENSITY, STRUCTURAL_ASSET
# NOTE: Only provide docstring if not covered by base class
class RevitMaterialProcessor(MaterialProcessor):
"""Implementation of the MaterialProcessor for the Revit context.
"""
def __init__(self, mass_aggregator: "MassAggregator"):
self._mass_aggregator = mass_aggregator
src/processors/model.py
+45 -11
View File
@@ -18,8 +18,12 @@ from src.utils.constants import (
DENSITY,
)
# NOTE: Only provide docstring if not covered by base class
class RevitModelProcessor(ModelProcessor):
"""Implementation of the ModelProcessor in the Revit context.
"""
def __init__(
self,
material_processor: MaterialProcessor,
@@ -32,7 +36,14 @@ class RevitModelProcessor(ModelProcessor):
def process_elements(self, model: Any) -> None:
"""Process all elements in the model hierarchically.
Model → Levels → Type Groups → Elements"""
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.")
@@ -41,7 +52,14 @@ class RevitModelProcessor(ModelProcessor):
self._process_level(level)
def _process_level(self, level: Any) -> None:
"""Process a single level in the model"""
"""Process all groups contained on a level.
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")
@@ -54,7 +72,16 @@ class RevitModelProcessor(ModelProcessor):
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"""
"""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")
@@ -72,23 +99,30 @@ class RevitModelProcessor(ModelProcessor):
for revit_object in revit_objects:
self.process_element(level_name, type_name, revit_object)
def process_element(self, level: str, type_name: str, revit_object: Any) -> None:
"""Process a single element following original logic exactly"""
def process_element(self, level: str, type_name: str, model_object: Any) -> None:
"""Processing of actual model object after successful traversal of the commit.
Args:
level (str): associated level of the object
type_name (str): family / type of the object
revit_object (Any): speckle object
"""
# We can probably straight up skip Line and Arc. Logging it as a warning is dumb
speckle_type = getattr(revit_object, SPECKLE_TYPE, None)
# 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]:
return # TODO: Possibly add to logger for info? Not a warning though.
return
element_id = getattr(revit_object, ID, None)
element_id = getattr(model_object, ID, None)
if not element_id:
return
# Check Material Quantities
properties = getattr(revit_object, PROPERTIES, None)
properties = getattr(model_object, PROPERTIES, None)
if not properties:
self._logger.log_warning(
f"Missing Material Quantities",
"Missing Material Quantities",
object_id=element_id,
missing_property=PROPERTIES,
)
@@ -96,7 +130,7 @@ class RevitModelProcessor(ModelProcessor):
material_quantities = properties.get(MATERIAL_QUANTITIES, None)
if not material_quantities:
self._logger.log_warning(
f"Missing Material Quantities",
"Missing Material Quantities",
object_id=element_id,
missing_property=MATERIAL_QUANTITIES,
)
src/utils/constants.py
+2
View File
@@ -1,3 +1,5 @@
# TODO: Implementation specific constants.
REQUIRED_PROPERTIES = ["volume", "density", "structuralAsset"]
# Keys
src/validators/validators.py
View File