ifc5d - continue refactor

src/ifc5d/README.md

@@ -23,6 +23,14 @@ Planned (would you like to contribute? Please reach out!):
 
 ## Usage CSV to IFC
 
+Simple example:
+```python
+import ifc5d.csv2ifc
+
+csv2ifc = ifc5d.csv2ifc.Csv2Ifc(csv_filepath, ifc_file)
+csv2ifc.execute()
+```
+
 See example files as a CSV file format reference:
 - `sample_cost_schedule_house_FR.csv` / `.ods`
 - `schedule.csv`, `rates.csv` (schedule of rates example)
@@ -31,21 +39,39 @@ Some notes on the format:
 - Empty lines are ignored.
 - Importing ods/xlsx is not currently supported, only csv.
 - 'Property', 'Query' columns are required only for non-schedule of rates cost schedules.
-- Some older exports may have 'Description' field instead of 'Name', it is safe to just rename it.
+- 'Index' was preferred for import hierarchy source over 'Hierarchy' as it's easier to edit from the table view.
+- In non-SoR if cost values are not provided in any way and cost item has subitems, then `SUM()` will added automatically as it's cost value.  
+Mixing `SUM()` cost items and items with their own value is not supported.
+
+## Columns Description
+
+**General Columns**
+- 'Index' - is a hierarchy depth that's used for building hierarchy during csv import. Starts from 1.  
+E.g. root items of the same level have index '1', their children have '2', etc.
+- 'Name' - IfcCostItem.Name.  
+Some older exports may have 'Description' field instead of 'Name', it is safe to just rename it.
+- 'Identification' - IfcCostItem.Identification.
+- 'Unit' - IfcCostValue unit, should be provided as a unit symbol.  
+E.g. 'm2', 'm3', 'kg', etc.
+- 'Value' - overall cost value to assign to IfcCostItem.
+- All other columns that are not mentioned in this list or the one below will be interpeted as cost value categories.  
+Note that if 'Value' is provided, it takes priority over subcategories.
+
+Non-schedule of rates columns:
+- 'Quantity' - total cost item quantity.
+- 'Property' - quantity name that should be added to IfcCostItem from 'Query' elements.  
+If 'Query' is provided, 'Property' can be left empty or set to "COUNT" to count queried elements.
+- 'Query' - selector query for elements to assign to IfcCostItem.  
+If query is provided, it takes priority over 'Quantity' (using both is not supported).
+
+**Exported informational columns (not used for import)**
 - 'Hierarchy' is just an informational column that doesn't affect the import.  
 E.g. '1', '1.1', '1.1.1', etc.
-- 'Index' column is a hierarchy depth that's used for building hierarchy during csv import, starting from 1.  
+- 'Id' - IfcCostItem.id
-E.g. root items of the same level have index '1', their children have '2', etc.
+- 'RateSubtotal' - all IfcCostItem specific costs, not including subitem costs.
-- 'Index' was preferred for import hierarchy source over 'Hierarchy' as it's easier to edit from the table view.
+- 'TotalPrice' - IfcCostItem total cost, including subitem sum calculations.
 
 
-Simple example:
-```python
-import ifc5d.csv2ifc
-
-csv2ifc = ifc5d.csv2ifc.Csv2Ifc(csv_filepath, ifc_file)
-csv2ifc.execute()
-```
 
 
 ## Usage IFC to CSV, ODS, XSLS

src/ifc5d/ifc5d/csv2ifc.py

@@ -32,27 +32,30 @@ from typing import Any, Union, Optional, TypedDict, NotRequired
 class CsvHeader(TypedDict):
     Index: int
     Name: int
-    Quantity: int
     Unit: int
     Identification: NotRequired[int]
     Value: NotRequired[int]
 
     # Not schedule of rates:
-    Property: int
+    Quantity: NotRequired[int]
-    Query: int
+    Property: NotRequired[int]
+    Query: NotRequired[int]
 
 
 class CostItem(TypedDict):
     children: list[CostItem]
-    assignments: dict[str, Any]
     ifc: NotRequired[ifcopenshell.entity_instance]
 
     Identification: Union[str, None]
     Name: Union[str, None]
     Unit: Union[str, None]
-    Quantity: Union[float, None]
     CostValues: Union[dict[str, float], float, None]
 
+    # Only might be available in non-SOR.
+    Quantity: Union[float, None]
+    Property: Union[str, None]
+    Query: Union[str, None]
+
 
 class Csv2Ifc:
     # Inputs.
@@ -118,12 +121,9 @@ class Csv2Ifc:
                 # parse header
                 if not self.headers:
                     self.has_categories = True
-                    for i, col in enumerate(row):
+                    self.headers = {col: i for i, col in enumerate(row) if col}
-                        if not col:
+                    if "Value" in self.headers:
-                            continue
+                        self.has_categories = False
-                        if col == "Value":
-                            self.has_categories = False
-                        self.headers[col] = i
 
                     # validate header
                     mandatory_fields = {"Name", "Quantity", "Unit"}
@@ -169,18 +169,14 @@ class Csv2Ifc:
     def get_row_cost_data(self, row: list[str]) -> CostItem:
         name = row[self.headers["Name"]]
         identification = row[self.headers["Identification"]] if "Identification" in self.headers else None
-        quantity = row[self.headers["Quantity"]]
+        quantity = row[(self.headers["Quantity"])] if "Quantity" in self.headers else None
         unit = row[self.headers["Unit"]]
-        if not self.is_schedule_of_rates:
+        if self.is_schedule_of_rates:
-            assignments = {
+            property_name, query = None, None
-                "PropertyName": row[self.headers["Property"]],
-                "Query": row[self.headers["Query"]],
-            }
         else:
-            assignments = {
+            property_name = row[(self.headers["Property"])] if "Property" in self.headers else None
-                "PropertyName": None,
+            query = row[(self.headers["Query"])] if "Query" in self.headers else None
-                "Query": None,
+
-            }
         if self.has_categories:
             cost_values = {
                 k: locale.atof(row[v])
@@ -195,10 +191,11 @@ class Csv2Ifc:
         return {
             "Identification": str(identification) if identification else None,
             "Name": str(name) if name else None,
-            "Quantity": float(quantity) if quantity else None,
             "Unit": str(unit) if unit else None,
             "CostValues": cost_values,
-            "assignments": assignments,
+            "Quantity": float(quantity) if quantity else None,
+            "Property": property_name,
+            "Query": query,
             "children": [],
         }
 
@@ -229,13 +226,16 @@ class Csv2Ifc:
         cost_item["ifc"].Name = cost_item["Name"]
         cost_item["ifc"].Identification = cost_item["Identification"]
 
-        if not cost_item["CostValues"] and cost_item["children"]:
+        cost_values = cost_item["CostValues"]
+        if ((isinstance(cost_values, dict) and len(cost_values) == 0) or (cost_values is None)) and cost_item[
+            "children"
+        ]:
             if not self.is_schedule_of_rates:
                 cost_value = ifcopenshell.api.cost.add_cost_value(self.file, parent=cost_item["ifc"])
                 cost_value.Category = "*"
         elif self.has_categories:
-            assert isinstance(cost_item["CostValues"], dict)
+            assert isinstance(cost_values, dict)
-            for category, value in cost_item["CostValues"].items():
+            for category, value in cost_values.items():
                 cost_value = ifcopenshell.api.cost.add_cost_value(self.file, parent=cost_item["ifc"])
                 cost_value.AppliedValue = self.file.createIfcMonetaryMeasure(value)
                 if category != "Rate" or category != "Price":
@@ -243,7 +243,7 @@ class Csv2Ifc:
                         category = category.replace("Rate", "")
                         category = category.strip()
                     cost_value.Category = category
-        elif cost_item["CostValues"]:
+        elif cost_values:
             cost_value = ifcopenshell.api.cost.add_cost_value(self.file, parent=cost_item["ifc"])
             cost_value.AppliedValue = self.file.createIfcMonetaryMeasure(cost_item["CostValues"])
             if self.is_schedule_of_rates:
@@ -267,10 +267,9 @@ class Csv2Ifc:
 
         quantity = None
         quantity_class = ifcopenshell.util.unit.get_symbol_quantity_class(cost_item["Unit"])
-        if not cost_item["assignments"]["PropertyName"] or cost_item["assignments"]["PropertyName"].upper() == "COUNT":
+        prop_name = cost_item["Property"]
+        if not prop_name or prop_name.upper() == "COUNT":
             prop_name = ""
-        else:
-            prop_name = cost_item["assignments"]["PropertyName"]
 
         if not self.is_schedule_of_rates and cost_item["Quantity"] is not None:
             quantity = ifcopenshell.api.cost.add_cost_item_quantity(
@@ -281,8 +280,8 @@ class Csv2Ifc:
             if prop_name:
                 quantity.Name = prop_name
 
-        if cost_item["assignments"]["Query"]:
+        if cost_item["Query"]:
-            results = ifcopenshell.util.selector.filter_elements(self.file, cost_item["assignments"]["Query"])
+            results = ifcopenshell.util.selector.filter_elements(self.file, cost_item["Query"])
             results = [r for r in results if has_property(self.file, r, prop_name)]
             # NOTE: currently we do not support count quantities that have
             # both defined quantity in .csv "Quantity" column

src/ifc5d/ifc5d/ifc5Dspreadsheet.py

@@ -17,6 +17,7 @@
 # along with Ifc5D.  If not, see <http://www.gnu.org/licenses/>.
 
 
+from __future__ import annotations
 import os
 import time
 import argparse
@@ -31,17 +32,32 @@ from typing import Union, Optional, Any, TypedDict, NotRequired
 
 
 class CostItem(TypedDict):
+    # Exported columns.
     Index: int
     Hierarchy: str
     Id: int
     Identification: Union[str, None]
     Name: Union[str, None]
     Unit: str
-    Quantity: int
+    Quantity: Union[float, None]
-    ChildrenData: list["CostItem"]
+    RateSubtotal: float
-    # Total Price: float
+    TotalPrice: float
-    # * Cost: str
+
-    # Rate Subtotal: float
+    # Internal.
+    cost_categories: dict[str, float]
+
+
+class CostItemQuantity(TypedDict):
+    quantity: Union[float, None]
+
+
+class CostValue(TypedDict):
+    id: int
+    """Cost Value id."""
+    name: Union[str, None]
+    applied_value: float
+    unit: str
+    category: str
 
 
 class IfcDataGetter:
@@ -64,10 +80,10 @@ class IfcDataGetter:
         return ifcopenshell.util.cost.get_root_cost_items(cost_schedule)
 
     @staticmethod
-    def get_cost_item_values(cost_item: Union[ifcopenshell.entity_instance, None]) -> Union[list[dict[str, Any]], None]:
+    def get_cost_item_values(cost_item: Union[ifcopenshell.entity_instance, None]) -> Union[list[CostValue], None]:
         if not cost_item:
             return None
-        values = []
+        values: list[CostValue] = []
         for cost_value in cost_item.CostValues or []:
             name = cost_value.Name
             applied_value = ifcopenshell.util.cost.calculate_applied_value(cost_item, cost_value)
@@ -84,48 +100,33 @@ class IfcDataGetter:
         return values
 
     @staticmethod
-    def process_categories(cost_item: ifcopenshell.entity_instance, categories: set[str]) -> set[str]:
+    def get_cost_items_data(
-        """
-        :param categories: A set to fill with categories.
-        """
-        for cost_value in cost_item.CostValues or []:
-            if cost_value.Category:
-                categories.add("{}{}".format(cost_value.Category, " Cost"))
-        return categories
-
-    @staticmethod
-    def process_cost_item_categories(cost_item: ifcopenshell.entity_instance, categories: set[str]) -> set[str]:
-        IfcDataGetter.process_categories(cost_item, categories)
-        for child in ifcopenshell.util.cost.get_nested_cost_items(cost_item):
-            IfcDataGetter.process_cost_item_categories(child, categories)
-        return categories
-
-    @staticmethod
-    def get_cost_rates_categories(schedule: ifcopenshell.entity_instance) -> set[str]:
-        """
-        :param categories: A set to fill with categories.
-        """
-        categories: set[str] = set()
-        for cost_item in IfcDataGetter.get_root_costs(schedule):
-            IfcDataGetter.process_cost_item_categories(cost_item, categories)
-        return categories
-
-    @staticmethod
-    def process_cost_data(
         file: ifcopenshell.file,
         cost_item: ifcopenshell.entity_instance,
-        cost_items_data: list[CostItem],
         index: int = 1,
         hierarchy: str = "1",
-    ) -> None:
+    ) -> list[CostItem]:
         """
         :param cost_items_data: A list to fill with cost items.
         :param index: Current hierarchy depth.
         """
+        cost_items_data: list[CostItem] = []
 
         quantity_data = IfcDataGetter.get_cost_item_quantity(file, cost_item)
         cost_values_data = IfcDataGetter.get_cost_item_values(cost_item)
 
+        rate_subtotal = 0.0
+        total_price = 0.0
+        cost_categories: dict[str, float] = {}
+        for cost_value in cost_values_data or []:
+            category = cost_value["category"]
+            if cost_value["category"] == "*":  # A sum.
+                total_price = cost_value["applied_value"]
+            else:
+                cost_category = "{}{}".format(category, " Cost")
+                cost_categories[cost_category] = cost_value["applied_value"]
+                rate_subtotal += cost_value["applied_value"]
+
         data: CostItem = {
             "Index": index,
             "Hierarchy": hierarchy,
@@ -133,35 +134,25 @@ class IfcDataGetter:
             "Identification": cost_item.Identification,
             "Name": cost_item.Name,
             "Unit": cost_values_data[0]["unit"] if cost_values_data else "",
-            "Quantity": quantity_data["quantity"]["total_quantity"],
+            "Quantity": quantity_data["quantity"],
-            "ChildrenData": [],
+            "RateSubtotal": rate_subtotal,
+            "TotalPrice": total_price,
+            "cost_categories": cost_categories,
         }
-        for cost_value in cost_values_data or []:
-            cost_category = "{}{}".format(cost_value["category"], " Cost")
-            data[cost_category] = cost_value["applied_value"]
-        if data.get("* Cost", None):
-            data["Total Price"] = data["* Cost"]
-            data["* Cost"] = ""
-        rate_subtotal = 0
-        for key, value in data.items():
-            if "Cost" in key and not "*" in key:
-                rate_subtotal += value
-
-        data["Rate Subtotal"] = rate_subtotal
-
         cost_items_data.append(data)
 
         index += 1
         child_hierarchy = hierarchy + ".1"
         for i, nested_cost in enumerate(ifcopenshell.util.cost.get_nested_cost_items(cost_item), 1):
             child_hierarchy = f"{hierarchy}.{i}"
-            IfcDataGetter.process_cost_data(file, nested_cost, cost_items_data, index, child_hierarchy)
+            cost_items_data.extend(IfcDataGetter.get_cost_items_data(file, nested_cost, index, child_hierarchy))
+        return cost_items_data
 
     @staticmethod
-    def get_cost_items_data(file: ifcopenshell.file, schedule: ifcopenshell.entity_instance) -> list[CostItem]:
+    def get_schedule_cost_items_data(file: ifcopenshell.file, schedule: ifcopenshell.entity_instance) -> list[CostItem]:
         cost_items_data: list[CostItem] = []
         for cost_item in IfcDataGetter.get_root_costs(schedule):
-            IfcDataGetter.process_cost_data(file, cost_item, cost_items_data)
+            cost_items_data.extend(IfcDataGetter.get_cost_items_data(file, cost_item))
         return cost_items_data
 
     @staticmethod
@@ -184,40 +175,45 @@ class IfcDataGetter:
         return IfcDataGetter.format_unit(unit.UnitComponent)
 
     @staticmethod
-    def get_cost_item_quantity(file: ifcopenshell.file, cost_item: ifcopenshell.entity_instance) -> dict[str, Any]:
+    def get_cost_item_quantity(file: ifcopenshell.file, cost_item: ifcopenshell.entity_instance) -> CostItemQuantity:
+        accounted_for: set[ifcopenshell.entity_instance] = set()
+
+        # NOTE: take_off_name is not used anywhere.
+        take_off_name: str = ""
+
         # TODO: handle multiple quantities, THOSE WHHICH ARE JUYST ASSIGNED TO THE COST ITEM DIRECTLY, NOT THROUGH OBJECTS.
         def add_quantity(quantity: ifcopenshell.entity_instance, take_off_name: str) -> float:
-            accounted_for.append(quantity)
+            accounted_for.add(quantity)
             if take_off_name == "":
+                # 0 IfcPhysicalSimpleQuantity.Name
                 take_off_name = quantity[0]
-                if quantity[0] != take_off_name:
+            if quantity[0] != take_off_name:
-                    take_off_name = "mixed-takeoff-quantities"
+                take_off_name = "mixed-takeoff-quantities"
+            # 3 IfcPhysicalSimpleQuantity.Value
             return quantity[3]
 
-        take_off_name = ""
+        cost_item_quantities: list[ifcopenshell.entity_instance] = cost_item.CostQuantities
-        total_cost_quantity = 0
-        accounted_for = []
-        cost_item_quantities = cost_item.CostQuantities
         if cost_item_quantities:
-            for rel in cost_item.Controls or []:
+            total_cost_quantity = 0.0
-                for related_object in rel.RelatedObjects:
+            # Add quantities from cost assignments.
-                    qtos = ifcopenshell.util.element.get_psets(related_object, qtos_only=True)
+            for related_object in ifcopenshell.util.cost.get_cost_assignments_by_type(cost_item):
-                    for quantities in qtos.values() or []:
+                qtos = ifcopenshell.util.element.get_psets(related_object, qtos_only=True)
-                        qto = file.by_id(quantities["id"])
+                for quantities in qtos.values() or []:
-                        for quantity in qto.Quantities:
+                    qto = file.by_id(quantities["id"])
-                            if quantity not in cost_item_quantities:
+                    for quantity in qto.Quantities:
-                                continue
+                        if quantity not in cost_item_quantities:
-                            total_cost_quantity += add_quantity(quantity, take_off_name)
+                            continue
-                            accounted_for.append(quantity)
+                        total_cost_quantity += add_quantity(quantity, take_off_name)
 
+            # Add cost item quantities assigned to the cost item directly.
             for quantity in cost_item_quantities:
-                if not quantity in accounted_for:
+                if quantity not in accounted_for:
                     total_cost_quantity += add_quantity(quantity, take_off_name)
+        else:
+            total_cost_quantity = None
 
         return {
-            "id": cost_item.id(),
+            "quantity": total_cost_quantity,
-            "name": cost_item.Name,
-            "quantity": {"take_off_name": take_off_name, "total_quantity": total_cost_quantity},
         }
 
 
@@ -282,31 +278,42 @@ class Ifc5Dwriter:
         counter: Counter[str] = Counter()
         for cost_schedule in self.cost_schedules:
             sheet_id = cost_schedule.id()
-            self.sheet_data[sheet_id] = sheet_data = SheetData()
+            cost_items = IfcDataGetter.get_schedule_cost_items_data(self.file, cost_schedule)
-            sheet_data["headers"] = [
+            headers: list[str] = [
                 "Id",
                 "Hierarchy",
                 "Index",
                 "Identification",
                 "Name",
-                "Quantity",
                 "Unit",
             ]
-            cost_rate_categories = IfcDataGetter.get_cost_rates_categories(cost_schedule)
+            if cost_schedule.PredefinedType != "SCHEDULEOFRATES":
-            sheet_data["headers"].extend(list(cost_rate_categories))
+                headers.insert(-1, "Quantity")
-            sheet_data["headers"].extend(["Rate Subtotal", "Total Price", "Children"])
+            headers.extend(["RateSubtotal", "TotalPrice"])
-            sheet_data["cost_items"] = IfcDataGetter.get_cost_items_data(self.file, cost_schedule)
-            sheet_data["UpdateDate"] = IfcDataGetter.canonicalise_time(
-                ifcopenshell.util.date.ifc2datetime(cost_schedule.UpdateDate)
-            )
-            sheet_data["PredefinedType"] = cost_schedule.PredefinedType
-            schedule_name = cost_schedule.Name or "Unnamed"
 
+            # Handle cost categories.
+            categories: set[str] = set()
+            for cost_item in cost_items:
+                for category, value in cost_item["cost_categories"].items():
+                    categories.add(category)
+                    cost_item[category] = value
+            assert not (intersection := categories.intersection(headers)), intersection
+            headers.extend(categories)
+
+            schedule_name = cost_schedule.Name or "Unnamed"
             counter[schedule_name] += 1
             if (count := counter[schedule_name]) > 1:
                 schedule_name = f"{schedule_name}_{count - 1}"
 
-            sheet_data["Name"] = schedule_name
+            self.sheet_data[sheet_id] = {
+                "Name": schedule_name,
+                "headers": headers,
+                "cost_items": cost_items,
+                "UpdateDate": IfcDataGetter.canonicalise_time(
+                    ifcopenshell.util.date.ifc2datetime(cost_schedule.UpdateDate)
+                ),
+                "PredefinedType": cost_schedule.PredefinedType,
+            }
 
     def multiply_cells(self, cell1, cell2):
         return "={}*{}".format(cell1, cell2)