Bump mypy from 1.13.0 to 1.20.1

Bumps [mypy](https://github.com/python/mypy) from 1.13.0 to 1.20.1.
- [Changelog](https://github.com/python/mypy/blob/master/CHANGELOG.md)
- [Commits](https://github.com/python/mypy/compare/v1.13.0...v1.20.1)

---
updated-dependencies:
- dependency-name: mypy
  dependency-version: 1.20.1
  dependency-type: direct:development
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

README.md

@@ -1,6 +1,14 @@
 # Speckle → IFC 4.3 Exporter (Rhino)
 
-A [Speckle Automate](https://automate.speckle.dev/) function that converts Speckle models (from Rhino and other authoring tools) into IFC 4.3 files.
+## 🚧 Project Status: WIP
+Hey there! This project is still under active development, so expect changes, bugs, and incomplete features.
+If you have any questions or suggestions, don’t hesitate to reach out at: **nikos@speckle.systems**
+
+A [Speckle Automate](https://docs.speckle.systems/developers/automate/introduction) function that converts Speckle models from Rhino into IFC 4X3 files.
+
+> ⚠️ **Note on Model Uploads**
+>
+> Large models (greater than 200MB) may fail to upload due to current file size limitations. The team is actively working on resolving this issue.
 
 ## How It Works
 
@@ -9,7 +17,19 @@ A [Speckle Automate](https://automate.speckle.dev/) function that converts Speck
 3. **Classifies** each element into its IFC type (e.g. `IfcColumn`, `IfcWall`) using the `Attributes.type` property.
 4. **Exports geometry** — meshes, instances (block definitions), and curves — into IFC representations.
 5. **Clones properties** — attributes, property sets, and quantities — from the Speckle object onto the IFC entity.
-6. **Writes** the resulting `.ifc` file.
+6. **Writes** the resulting `.ifc` file, compressed as a `.zip`.
+
+## Performance Optimizations
+
+The exporter is optimized for file size and speed:
+
+- **Geometry deduplication** — identical meshes are hashed (MD5 of vertex + face data) and shared via `IfcRepresentationMap` + `IfcMappedItem`, so instances reuse a single geometry copy.
+- **Shared property sets** — identical `IfcPropertySet` / `IfcElementQuantity` entities are created once and linked to all elements via batched `IfcRelDefinesByProperties`.
+- **Batched spatial containment** — `IfcRelContainedInSpatialStructure` and type assignments are written in bulk at the end, not per-element.
+- **Vertex deduplication & rounding** — near-coincident vertices are merged (0.01mm tolerance) and coordinates rounded to 3 decimal places.
+- **Direction & value caching** — `IfcDirection`, `IfcCartesianPoint`, and `IfcNominalValue` entities are reused across the file.
+- **Lazy material creation** — `IfcSurfaceStyle` entities are only created when actually assigned to geometry.
+- **ZIP compression** — output is compressed before upload.
 
 ## Supported Property Formats
 
@@ -43,6 +63,16 @@ properties:
 
 The exporter automatically detects the format and unflattens dot-notation keys into the nested structure before processing.
 
+## Instance Handling
+
+Speckle InstanceProxy objects (block instances) are exported using the IFC mapped representation pattern:
+
+- Each unique block definition becomes an `IfcRepresentationMap` (geometry stored once).
+- Each instance becomes an `IfcMappedItem` with an `IfcCartesianTransformationOperator3DnonUniform` encoding the full 4x4 transform (explicit Axis3 for correct orientation with mirrors and non-orthogonal transforms).
+- Content-based geometry hashing ensures that different definition IDs with identical geometry share the same `IfcRepresentationMap`.
+
+Supports both Revit-format instances (hex hash definitionId, mm units) and speckleifc-format instances (`DEFINITION:` prefix, metre units).
+
 ## Function Inputs
 
 | Input | Description |
@@ -60,19 +90,18 @@ utils/
   traversal.py          # Walks the Speckle Collection tree
   mapper.py             # Maps Speckle objects → IFC entity classes
   properties.py         # Extracts & writes attributes, property sets, quantities
-  geometry.py           # Mesh → IFC geometry conversion
-  instances.py          # Block instance / definition handling
-  curves.py             # Curve geometry (Polycurve, Line, Arc)
+  geometry.py           # Mesh → IFC geometry conversion (IfcPolygonalFaceSet)
+  instances.py          # Block instance / definition handling (RepMap + MappedItem)
+  curves.py             # Curve geometry (Polycurve, Line, Arc → IfcIndexedPolyCurve)
   writer.py             # IFC scaffold creation, storey management
   type_manager.py       # IfcTypeObject creation & assignment
-  materials.py          # Material mapping
+  materials.py          # Material colour mapping (IfcSurfaceStyle)
   helpers.py            # Shared utilities (_get, unit scales)
-  receiver.py           # Speckle server connection & data retrieval
 ```
 
 ## Getting Started
 
-## Using with Speckle Automate
+### Using with Speckle Automate
 
 1. Go to the Automations tab in your project
 2. Click New Automation

main.py

@@ -1,3 +1,4 @@
+import zipfile
 from datetime import datetime
 
 import ifcopenshell.api
@@ -11,6 +12,7 @@ from utils.instances import (
 )
 from utils.properties import (
     get_building_storey, get_element_name, write_all_properties,
+    PropertySetManager,
 )
 from utils.curves import curve_to_ifc
 from utils.writer import create_ifc_scaffold, StoreyManager
@@ -89,6 +91,7 @@ def automate_function(
     material_manager = MaterialManager(ifc, base)
     material_manager.build_definition_material_map(definition_map)
     type_manager = TypeManager(ifc)
+    property_manager = PropertySetManager(ifc)
 
     # ------------------------------------------------------------------ #
     # 4. Traverse & export
@@ -138,7 +141,7 @@ def automate_function(
                 continue
             element = _create_element(ifc, ifc_class, name, rep, placement,
                                       storey, storey_manager=storey_manager)
-            write_all_properties(ifc, element, obj)
+            write_all_properties(ifc, element, obj, property_manager)
             type_manager.assign(element, obj, ifc_class)
             instance_count += 1
             total += 1
@@ -159,7 +162,7 @@ def automate_function(
             if rep:
                 element = _create_element(ifc, ifc_class, name, rep, placement,
                                           storey, storey_manager=storey_manager)
-                write_all_properties(ifc, element, obj)
+                write_all_properties(ifc, element, obj, property_manager)
                 type_manager.assign(element, obj, ifc_class)
                 total += 1
 
@@ -177,7 +180,7 @@ def automate_function(
                     ifc, ifc_class, name, inst_rep, inst_placement,
                     storey, storey_manager=storey_manager,
                 )
-                write_all_properties(ifc, inst_element, obj)
+                write_all_properties(ifc, inst_element, obj, property_manager)
                 type_manager.assign(inst_element, obj, ifc_class)
                 instance_count += 1
                 total += 1
@@ -191,7 +194,7 @@ def automate_function(
                 if rep:
                     element = _create_element(ifc, ifc_class, name, rep, placement,
                                               storey, storey_manager=storey_manager)
-                    write_all_properties(ifc, element, obj)
+                    write_all_properties(ifc, element, obj, property_manager)
                     type_manager.assign(element, obj, ifc_class)
                     total += 1
 
@@ -209,6 +212,8 @@ def automate_function(
     storey_manager.flush()
     print("Flushing type relationships...")
     type_manager.flush()
+    print("Flushing shared property sets...")
+    property_manager.flush()
 
     file_name = function_inputs.file_name
     timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -217,11 +222,17 @@ def automate_function(
 
     ifc.write(ifc_filename)
     print(f"\nIFC file written: {ifc_filename}")
+
+    zip_filename = f"{file_name}_{timestamp}.zip"
+    with zipfile.ZipFile(zip_filename, "w", zipfile.ZIP_DEFLATED) as zf:
+        zf.write(ifc_filename)
+    print(f"Zipped: {zip_filename}")
+
     try:
-        automate_context.mark_run_success("Success! You can download the IF file below.")
-        automate_context.store_file_result(f"./{ifc_filename}")
+        automate_context.mark_run_success("Success! You can download the IFC file below.")
+        automate_context.store_file_result(f"./{zip_filename}")
     except Exception as e:
-        print(f"  ⚠️  Could not upload file result (network issue?): {e}")
+        print(f"  Could not upload file result (network issue?): {e}")
         automate_context.mark_run_failed(f"Something went wrong when storing file result. Exception detail: {e}") 
 
     print(f"\n{'=' * 60}")
@@ -234,6 +245,7 @@ def automate_function(
     print(f"  Levels             : {', '.join(storey_manager.names)}")
     print_instance_stats()
     material_manager.print_stats()
+    property_manager.print_stats()
     print(f"{'=' * 60}\n")
 
 def _create_element(ifc, ifc_class, name, rep, placement, storey,

pyproject.toml

@@ -15,7 +15,7 @@ dependencies = ["specklepy==3.1.0",
 
 [project.optional-dependencies]
 dev = [
-    "mypy==1.13.0",
+    "mypy==1.20.1",
     "pytest==7.4.4",
     "ruff==0.11.12",
 ]

utils/curves.py

@@ -38,10 +38,10 @@ def _resolve_scale(obj, fallback: float) -> float:
 
 
 def _point_coords(pt, scale: float) -> tuple:
-    """Extract (x, y, z) from a Speckle Point, scaled to mm."""
-    x = float(_get(pt, "x") or 0) * scale
-    y = float(_get(pt, "y") or 0) * scale
-    z = float(_get(pt, "z") or 0) * scale
+    """Extract (x, y, z) from a Speckle Point, scaled to mm and rounded."""
+    x = round(float(_get(pt, "x") or 0) * scale, 3)
+    y = round(float(_get(pt, "y") or 0) * scale, 3)
+    z = round(float(_get(pt, "z") or 0) * scale, 3)
     return x, y, z
 
 
@@ -112,9 +112,9 @@ def _extract_polycurve(obj, scale: float) -> tuple:
             values = list(raw_value) if not isinstance(raw_value, list) else raw_value
             indices = []
             for vi in range(0, len(values) - 2, 3):
-                x = float(values[vi]) * seg_scale
-                y = float(values[vi + 1]) * seg_scale
-                z = float(values[vi + 2]) * seg_scale
+                x = round(float(values[vi]) * seg_scale, 3)
+                y = round(float(values[vi + 1]) * seg_scale, 3)
+                z = round(float(values[vi + 2]) * seg_scale, 3)
                 key = (round(x * 100), round(y * 100), round(z * 100))
                 if key in point_map:
                     idx = point_map[key]

utils/geometry.py

@@ -76,6 +76,13 @@ def build_ifc_facesets(ifc, verts_scaled: list, face_groups: list) -> list:
     if not valid_faces or not deduped_verts:
         return []
 
+    # Round vertex coordinates to reduce IFC text file size
+    # 3 decimal places = 0.001mm precision (more than sufficient)
+    for v in deduped_verts:
+        v[0] = round(v[0], 3)
+        v[1] = round(v[1], 3)
+        v[2] = round(v[2], 3)
+
     # Build IFC entities
     try:
         point_list = ifc.createIfcCartesianPointList3D(deduped_verts)
@@ -311,9 +318,9 @@ def _get_shared(ifc):
 
 
 def _make_placement(ifc, x: float, y: float, z: float):
-    """Create an IfcLocalPlacement at absolute world coordinates (metres)."""
+    """Create an IfcLocalPlacement at absolute world coordinates (mm)."""
     shared = _get_shared(ifc)
-    origin = ifc.createIfcCartesianPoint([x, y, z])
+    origin = ifc.createIfcCartesianPoint([round(x, 3), round(y, 3), round(z, 3)])
     a2p    = ifc.createIfcAxis2Placement3D(origin, shared["z_axis"], shared["x_axis"])
     return ifc.createIfcLocalPlacement(PlacementRelTo=None, RelativePlacement=a2p)
 

utils/instances.py

@@ -20,7 +20,9 @@
 # sharing the same definition reference a single copy of the geometry.
 # =============================================================================
 
+import hashlib
 import math
+import struct
 import ifcopenshell.api
 from specklepy.objects.base import Base
 from utils.helpers import _get, MM_SCALES
@@ -249,24 +251,54 @@ _mesh_data_cache: dict = {}
 # All instances sharing the same definition reuse one geometry copy.
 _rep_map_cache: dict = {}
 
+# Cache: geometry content hash → IfcRepresentationMap
+# Enables sharing across different definitionIds that have identical geometry.
+_geometry_hash_cache: dict = {}
+
 # Shared identity placement for all instances (keyed by ifc file id)
 _identity_placement_cache: dict[int, object] = {}
 
 
+# --------------------------------------------------------------------------- #
+# Geometry content hashing
+# --------------------------------------------------------------------------- #
+
+def _hash_mesh_data(mesh_data_list: list, material_key: str = "") -> str:
+    """Compute a content hash from mesh geometry data for deduplication.
+
+    mesh_data_list: list of (verts_local, face_groups) tuples
+    material_key:   string identifying the material (included in hash)
+    Returns: hex digest string
+    """
+    h = hashlib.md5(usedforsecurity=False)
+    for verts_local, face_groups in mesh_data_list:
+        # Hash rounded vertices as packed floats (faster than str conversion)
+        for i in range(0, len(verts_local), 3):
+            h.update(struct.pack("3f",
+                round(verts_local[i], 3),
+                round(verts_local[i+1], 3),
+                round(verts_local[i+2], 3),
+            ))
+        # Hash face indices
+        for face in face_groups:
+            h.update(struct.pack(f"{len(face)}i", *face))
+        # Separator between meshes
+        h.update(b"|")
+    if material_key:
+        h.update(material_key.encode())
+    return h.hexdigest()
+
+
 # --------------------------------------------------------------------------- #
 # IfcRepresentationMap builder — geometry created once per definition
 # --------------------------------------------------------------------------- #
 
-def _build_rep_map(ifc, body_context, meshes: list, ifc_format: bool,
-                   material_manager=None, fallback_app_ids: list = None,
-                   definition_id: str = None):
-    """
-    Build an IfcRepresentationMap from definition meshes.
-    Geometry is in local coordinates (mm, no instance transform applied).
-    Returns IfcRepresentationMap or None if no valid geometry.
-    """
-    geom_items = []
+def _collect_mesh_data(meshes: list, ifc_format: bool) -> list:
+    """Unpack, scale, and cache mesh vertex/face data.
 
+    Returns list of (mesh_obj, verts_local, face_groups) tuples.
+    """
+    result = []
     for mesh in meshes:
         mesh_id = _get(mesh, "id") or _get(mesh, "applicationId")
         if mesh_id and mesh_id in _mesh_data_cache:
@@ -288,19 +320,62 @@ def _build_rep_map(ifc, body_context, meshes: list, ifc_format: bool,
                 print(f"  Warning: Instance face decode: {e}")
                 continue
 
-            # Scale vertices once and cache the result
             verts_local = [float(v) * ms for v in verts]
 
             if mesh_id:
                 _mesh_data_cache[mesh_id] = (verts_local, face_groups)
 
-        mesh_facesets = build_ifc_facesets(ifc, verts_local, face_groups)
+        result.append((mesh, verts_local, face_groups))
+    return result
 
+
+def _resolve_material_key(meshes_data: list, material_manager, fallback_app_ids, definition_id) -> str:
+    """Build a material cache key string for geometry hashing."""
+    if not material_manager:
+        return ""
+    parts = []
+    for mesh, _, _ in meshes_data:
+        mesh_app_id = _get(mesh, "applicationId")
+        style = material_manager.get_style_with_fallbacks(
+            primary_app_id=str(mesh_app_id) if mesh_app_id else None,
+            fallback_app_ids=fallback_app_ids,
+            definition_id=definition_id,
+        )
+        parts.append(str(id(style)) if style else "")
+    return "|".join(parts)
+
+
+def _build_rep_map(ifc, body_context, meshes: list, ifc_format: bool,
+                   material_manager=None, fallback_app_ids: list = None,
+                   definition_id: str = None):
+    """
+    Build an IfcRepresentationMap from definition meshes.
+    Uses content-based hashing to reuse identical geometry across different
+    definitionIds. Returns IfcRepresentationMap or None if no valid geometry.
+    """
+    # Step 1: Collect and cache raw mesh data (no IFC entities created yet)
+    meshes_data = _collect_mesh_data(meshes, ifc_format)
+    if not meshes_data:
+        return None
+
+    # Step 2: Compute content hash to check for identical geometry
+    mat_key = _resolve_material_key(meshes_data, material_manager, fallback_app_ids, definition_id)
+    geom_hash = _hash_mesh_data(
+        [(verts, faces) for _, verts, faces in meshes_data],
+        material_key=mat_key,
+    )
+
+    if geom_hash in _geometry_hash_cache:
+        return _geometry_hash_cache[geom_hash]
+
+    # Step 3: No match — build IFC geometry entities
+    geom_items = []
+
+    for mesh, verts_local, face_groups in meshes_data:
+        mesh_facesets = build_ifc_facesets(ifc, verts_local, face_groups)
         if not mesh_facesets:
             continue
 
-        # Apply material style to each faceset
-        # Try: mesh applicationId → fallback IDs → definitionId mapping
         if material_manager:
             mesh_app_id = _get(mesh, "applicationId")
             style = material_manager.get_style_with_fallbacks(
@@ -322,13 +397,12 @@ def _build_rep_map(ifc, body_context, meshes: list, ifc_format: bool,
         geom_items.extend(mesh_facesets)
 
     if not geom_items:
+        _geometry_hash_cache[geom_hash] = None
         return None
 
-    # Mapping origin = identity (local coords origin) — reuse shared origin
     shared = _get_shared(ifc)
     a2p    = ifc.createIfcAxis2Placement3D(shared["origin_0"], None, None)
 
-    # The mapped representation holds the actual geometry
     mapped_rep = ifc.createIfcShapeRepresentation(
         ContextOfItems=body_context,
         RepresentationIdentifier="Body",
@@ -336,7 +410,9 @@ def _build_rep_map(ifc, body_context, meshes: list, ifc_format: bool,
         Items=geom_items,
     )
 
-    return ifc.createIfcRepresentationMap(a2p, mapped_rep)
+    rep_map = ifc.createIfcRepresentationMap(a2p, mapped_rep)
+    _geometry_hash_cache[geom_hash] = rep_map
+    return rep_map
 
 
 # --------------------------------------------------------------------------- #
@@ -347,6 +423,22 @@ def _vec_magnitude(x, y, z):
     return math.sqrt(x*x + y*y + z*z)
 
 
+# Cache: rounded direction tuple → IfcDirection entity (keyed by ifc file id)
+_direction_cache: dict[int, dict] = {}
+
+def _get_or_create_direction(ifc, dx, dy, dz):
+    """Return a cached IfcDirection or create and cache a new one."""
+    fid = id(ifc)
+    if fid not in _direction_cache:
+        _direction_cache[fid] = {}
+    cache = _direction_cache[fid]
+    # Round to 6 decimals — sufficient for unit vectors
+    key = (round(dx, 6), round(dy, 6), round(dz, 6))
+    if key not in cache:
+        cache[key] = ifc.createIfcDirection([key[0], key[1], key[2]])
+    return cache[key]
+
+
 def _make_transform_operator(ifc, t: list, ts: float):
     """
     Convert a row-major 4x4 matrix + translation scale into an
@@ -355,22 +447,20 @@ def _make_transform_operator(ifc, t: list, ts: float):
     t:  16 floats, row-major [r00,r01,r02,tx, r10,r11,r12,ty, r20,r21,r22,tz, 0,0,0,1]
     ts: scale factor for translation components (e.g. 1000.0 for m→mm)
 
-    The matrix acts as: p' = M * p + translation, where M rows are:
-      row0 = (t[0], t[1], t[2])
-      row1 = (t[4], t[5], t[6])
-      row2 = (t[8], t[9], t[10])
-
     IfcCartesianTransformationOperator axes represent the COLUMNS of M:
       Axis1 = column 0 = where local X maps → (t[0], t[4], t[8])
       Axis2 = column 1 = where local Y maps → (t[1], t[5], t[9])
       Axis3 = column 2 = where local Z maps → (t[2], t[6], t[10])
 
+    Always uses the non-uniform variant with explicit Axis3 to ensure
+    correct orientation for all transform types (mirrors, non-orthogonal, etc.).
+
     Returns the IFC entity, or None if the transform is degenerate.
     """
     # Extract COLUMNS of the 3x3 rotation/scale sub-matrix
-    ax1 = (float(t[0]), float(t[4]), float(t[8]))    # column 0: X-axis direction
-    ax2 = (float(t[1]), float(t[5]), float(t[9]))    # column 1: Y-axis direction
-    ax3 = (float(t[2]), float(t[6]), float(t[10]))   # column 2: Z-axis direction
+    ax1 = (float(t[0]), float(t[4]), float(t[8]))
+    ax2 = (float(t[1]), float(t[5]), float(t[9]))
+    ax3 = (float(t[2]), float(t[6]), float(t[10]))
 
     s1 = _vec_magnitude(*ax1)
     s2 = _vec_magnitude(*ax2)
@@ -379,24 +469,28 @@ def _make_transform_operator(ifc, t: list, ts: float):
     if s1 < 1e-10 or s2 < 1e-10 or s3 < 1e-10:
         return None  # degenerate transform
 
-    # Normalized direction vectors
-    d1 = ifc.createIfcDirection([ax1[0]/s1, ax1[1]/s1, ax1[2]/s1])
-    d2 = ifc.createIfcDirection([ax2[0]/s2, ax2[1]/s2, ax2[2]/s2])
-    d3 = ifc.createIfcDirection([ax3[0]/s3, ax3[1]/s3, ax3[2]/s3])
+    # Normalized direction vectors — reuse cached IfcDirection entities
+    d1 = _get_or_create_direction(ifc, ax1[0]/s1, ax1[1]/s1, ax1[2]/s1)
+    d2 = _get_or_create_direction(ifc, ax2[0]/s2, ax2[1]/s2, ax2[2]/s2)
+    d3 = _get_or_create_direction(ifc, ax3[0]/s3, ax3[1]/s3, ax3[2]/s3)
 
-    # Translation, scaled to mm
-    tx = float(t[3])  * ts
-    ty = float(t[7])  * ts
-    tz = float(t[11]) * ts
+    # Translation, scaled and rounded to mm
+    tx = round(float(t[3])  * ts, 3)
+    ty = round(float(t[7])  * ts, 3)
+    tz = round(float(t[11]) * ts, 3)
     origin = ifc.createIfcCartesianPoint([tx, ty, tz])
 
-    # Use non-uniform variant to handle mirrors and non-uniform scale
+    # Round scales for cleaner output
+    s1 = round(s1, 6)
+    s2 = round(s2, 6)
+    s3 = round(s3, 6)
+
     return ifc.createIfcCartesianTransformationOperator3DnonUniform(
         d1,      # Axis1
         d2,      # Axis2
         origin,  # LocalOrigin
         s1,      # Scale
-        d3,      # Axis3
+        d3,      # Axis3 (explicit — never derived)
         s2,      # Scale2
         s3,      # Scale3
     )
@@ -529,12 +623,18 @@ def print_instance_stats():
     print(f"  Instance resolution: {_stats['found']}/{total} definitions found")
     if _stats["not_found"] > 0:
         print(f"  Warning: {_stats['not_found']} instances had no definition geometry")
+    unique_defs = len(_rep_map_cache)
+    unique_geom = len([v for v in _geometry_hash_cache.values() if v is not None])
+    if unique_defs > unique_geom:
+        print(f"  Geometry dedup: {unique_defs} definitions -> {unique_geom} unique geometries")
 
 
 def reset_caches():
     """Reset module-level caches (call at start of each export run)."""
     _mesh_data_cache.clear()
     _rep_map_cache.clear()
+    _geometry_hash_cache.clear()
     _identity_placement_cache.clear()
+    _direction_cache.clear()
     _stats["found"] = 0
     _stats["not_found"] = 0

utils/properties.py

@@ -13,6 +13,7 @@
 
 import ifcopenshell
 import ifcopenshell.api
+import ifcopenshell.guid
 from specklepy.objects.base import Base
 from utils.helpers import _get
 
@@ -121,17 +122,34 @@ def get_element_name(obj) -> str:
 # IFC value creation
 # ---------------------------------------------------------------------------
 
+_ifc_value_cache: dict[int, dict] = {}
+
+
 def _make_ifc_value(ifc, value):
-    """Create an IFC nominal value entity from a Python value, detecting type."""
+    """Create a cached IFC nominal value entity from a Python value, detecting type.
+
+    Identical values reuse the same IFC entity to reduce file size.
+    """
+    fid = id(ifc)
+    if fid not in _ifc_value_cache:
+        _ifc_value_cache[fid] = {}
+    cache = _ifc_value_cache[fid]
+
     if isinstance(value, bool):
-        return ifc.create_entity("IfcBoolean", wrappedValue=value)
-    if isinstance(value, int):
-        return ifc.create_entity("IfcInteger", wrappedValue=value)
-    if isinstance(value, float):
-        return ifc.create_entity("IfcReal", wrappedValue=value)
-    if isinstance(value, list):
-        return ifc.create_entity("IfcLabel", wrappedValue=", ".join(str(v) for v in value))
-    return ifc.create_entity("IfcLabel", wrappedValue=str(value))
+        cache_key = ("IfcBoolean", value)
+    elif isinstance(value, int):
+        cache_key = ("IfcInteger", value)
+    elif isinstance(value, float):
+        cache_key = ("IfcReal", value)
+    elif isinstance(value, list):
+        cache_key = ("IfcLabel", ", ".join(str(v) for v in value))
+    else:
+        cache_key = ("IfcLabel", str(value))
+
+    if cache_key not in cache:
+        entity_type, wrapped = cache_key
+        cache[cache_key] = ifc.create_entity(entity_type, wrappedValue=wrapped)
+    return cache[cache_key]
 
 
 # ---------------------------------------------------------------------------
@@ -202,50 +220,7 @@ def set_element_attributes(ifc, element, obj):
 
 
 # ---------------------------------------------------------------------------
-# Write property sets from _properties.Property Sets
-# ---------------------------------------------------------------------------
-
-def write_property_sets(ifc, element, obj):
-    """Write all property sets from _properties.Property Sets."""
-    props = get_properties(obj)
-    properties_section = _to_dict(props.get("Property Sets"))
-    if not properties_section:
-        return
-
-    for pset_name, pset_data in properties_section.items():
-        pset_dict = _to_dict(pset_data) if not isinstance(pset_data, dict) else pset_data
-        if not pset_dict:
-            continue
-
-        ifc_props = []
-        for prop_name, prop_value in pset_dict.items():
-            if prop_name == "id":
-                continue
-            if prop_value is None:
-                continue
-            try:
-                nominal = _make_ifc_value(ifc, prop_value)
-                p = ifc.create_entity(
-                    "IfcPropertySingleValue",
-                    Name=str(prop_name),
-                    NominalValue=nominal,
-                )
-                ifc_props.append(p)
-            except Exception:
-                continue
-
-        if ifc_props:
-            try:
-                pset = ifcopenshell.api.run(
-                    "pset.add_pset", ifc, product=element, name=pset_name
-                )
-                pset.HasProperties = ifc_props
-            except Exception as e:
-                print(f"  Warning: {pset_name}: {e}")
-
-
-# ---------------------------------------------------------------------------
-# Write quantity sets from _properties.Quantities
+# PropertySetManager — shared property sets across elements
 # ---------------------------------------------------------------------------
 
 def _try_float(value):
@@ -260,8 +235,261 @@ def _try_float(value):
     return None
 
 
-def write_quantity_sets(ifc, element, obj):
-    """Write all quantity sets from _properties.Quantities."""
+def _pset_content_key(pset_name: str, items: list) -> str:
+    """Build a hashable key from pset name + sorted property name-value pairs."""
+    return repr((pset_name, sorted(items)))
+
+
+def _qto_content_key(qto_name: str, items: list) -> str:
+    """Build a hashable key from qto name + sorted quantity tuples."""
+    return repr((qto_name, sorted(items)))
+
+
+class PropertySetManager:
+    """Creates shared IfcPropertySet / IfcElementQuantity entities.
+
+    Instead of creating one pset per element, identical psets are created
+    once and linked to all elements that share them via a single
+    IfcRelDefinesByProperties (written at flush time).
+    """
+
+    def __init__(self, ifc: ifcopenshell.file):
+        self._ifc = ifc
+        # content_key → IfcPropertySet / IfcElementQuantity entity
+        self._pset_cache: dict[str, object] = {}
+        # pset entity id → [element, ...]
+        self._pending: dict[int, list] = {}
+        self._pset_count = 0
+        self._shared_count = 0
+
+    def queue_property_sets(self, element, obj):
+        """Extract Property Sets from obj and queue shared assignment to element."""
+        props = get_properties(obj)
+        properties_section = _to_dict(props.get("Property Sets"))
+        if not properties_section:
+            return
+
+        ifc = self._ifc
+        for pset_name, pset_data in properties_section.items():
+            pset_dict = _to_dict(pset_data) if not isinstance(pset_data, dict) else pset_data
+            if not pset_dict:
+                continue
+
+            # Build content items for hashing (skip id and None values)
+            content_items = []
+            for prop_name, prop_value in pset_dict.items():
+                if prop_name == "id" or prop_value is None:
+                    continue
+                content_items.append((str(prop_name), repr(prop_value)))
+
+            if not content_items:
+                continue
+
+            key = _pset_content_key(pset_name, content_items)
+
+            if key not in self._pset_cache:
+                # Create the shared IfcPropertySet entity
+                ifc_props = []
+                for prop_name, prop_value in pset_dict.items():
+                    if prop_name == "id" or prop_value is None:
+                        continue
+                    try:
+                        nominal = _make_ifc_value(ifc, prop_value)
+                        p = ifc.create_entity(
+                            "IfcPropertySingleValue",
+                            Name=str(prop_name),
+                            NominalValue=nominal,
+                        )
+                        ifc_props.append(p)
+                    except Exception:
+                        continue
+
+                if not ifc_props:
+                    continue
+
+                pset = ifc.create_entity(
+                    "IfcPropertySet",
+                    GlobalId=ifcopenshell.guid.new(),
+                    Name=pset_name,
+                    HasProperties=ifc_props,
+                )
+                self._pset_cache[key] = pset
+                self._pset_count += 1
+            else:
+                self._shared_count += 1
+
+            pset = self._pset_cache[key]
+            pid = pset.id()
+            if pid not in self._pending:
+                self._pending[pid] = []
+            self._pending[pid].append(element)
+
+    def queue_quantity_sets(self, element, obj):
+        """Extract Quantities from obj and queue shared assignment to element."""
+        props = get_properties(obj)
+        quantities_raw = props.get("Quantities")
+        if quantities_raw is None:
+            return
+        quantities_section = _to_dict(quantities_raw)
+        if not quantities_section:
+            return
+
+        ifc = self._ifc
+        for qto_name, qto_data in quantities_section.items():
+            qto_dict = _to_dict(qto_data) if not isinstance(qto_data, dict) else qto_data
+            if not qto_dict:
+                continue
+
+            # Build content items for hashing
+            content_items = []
+            for qty_key, qty_entry in qto_dict.items():
+                if qty_key == "id":
+                    continue
+                if isinstance(qty_entry, dict):
+                    name = qty_entry.get("name", qty_key)
+                    units = (qty_entry.get("units") or "").strip().lower()
+                    value = _try_float(qty_entry.get("value"))
+                else:
+                    value = _try_float(qty_entry)
+                    if value is None:
+                        continue
+                    name = qty_key
+                    units = ""
+                if value is not None:
+                    content_items.append((name, units, value))
+
+            if not content_items:
+                continue
+
+            key = _qto_content_key(qto_name, content_items)
+
+            if key not in self._pset_cache:
+                quantities = []
+                for qty_key, qty_entry in qto_dict.items():
+                    if qty_key == "id":
+                        continue
+                    if isinstance(qty_entry, dict):
+                        name = qty_entry.get("name", qty_key)
+                        units = (qty_entry.get("units") or "").strip().lower()
+                        value = _try_float(qty_entry.get("value"))
+                    else:
+                        value = _try_float(qty_entry)
+                        if value is None:
+                            continue
+                        name = qty_key
+                        units = ""
+                    if value is None:
+                        continue
+
+                    try:
+                        mapping = _UNIT_QTY_MAP.get(units)
+                        if not mapping:
+                            name_lower = name.lower()
+                            for keyword, m in _NAME_QTY_MAP.items():
+                                if keyword in name_lower:
+                                    mapping = m
+                                    break
+                        if mapping:
+                            qty_type, value_attr = mapping
+                            qty = ifc.create_entity(
+                                qty_type, Name=name, **{value_attr: value}
+                            )
+                        else:
+                            qty = ifc.create_entity(
+                                "IfcQuantityCount", Name=name, CountValue=int(value)
+                            )
+                        quantities.append(qty)
+                    except Exception as e:
+                        print(f"  Warning: quantity {name}: {e}")
+                        continue
+
+                if not quantities:
+                    continue
+
+                qto = ifc.create_entity(
+                    "IfcElementQuantity",
+                    GlobalId=ifcopenshell.guid.new(),
+                    Name=qto_name,
+                    Quantities=quantities,
+                )
+                self._pset_cache[key] = qto
+                self._pset_count += 1
+            else:
+                self._shared_count += 1
+
+            qto = self._pset_cache[key]
+            qid = qto.id()
+            if qid not in self._pending:
+                self._pending[qid] = []
+            self._pending[qid].append(element)
+
+    def flush(self):
+        """Write all batched IfcRelDefinesByProperties relationships."""
+        ifc = self._ifc
+        for pset_id, elements in self._pending.items():
+            pset = ifc.by_id(pset_id)
+            try:
+                ifc.create_entity(
+                    "IfcRelDefinesByProperties",
+                    GlobalId=ifcopenshell.guid.new(),
+                    RelatedObjects=elements,
+                    RelatingPropertyDefinition=pset,
+                )
+            except Exception as e:
+                print(f"  Warning: pset rel: {e}")
+        self._pending.clear()
+
+    def print_stats(self):
+        total = self._pset_count + self._shared_count
+        print(f"  Property sets: {self._pset_count} unique / {total} total ({self._shared_count} shared)")
+
+
+# ---------------------------------------------------------------------------
+# Public API — called from main.py
+# ---------------------------------------------------------------------------
+
+def write_all_properties(ifc, element, obj, property_manager=None):
+    """Write all properties, quantities, and attributes from _properties."""
+    set_element_attributes(ifc, element, obj)
+    if property_manager:
+        property_manager.queue_property_sets(element, obj)
+        property_manager.queue_quantity_sets(element, obj)
+    else:
+        # Fallback: direct per-element creation (no sharing)
+        _write_property_sets_direct(ifc, element, obj)
+        _write_quantity_sets_direct(ifc, element, obj)
+
+
+def _write_property_sets_direct(ifc, element, obj):
+    """Legacy per-element property set writing (fallback)."""
+    props = get_properties(obj)
+    properties_section = _to_dict(props.get("Property Sets"))
+    if not properties_section:
+        return
+    for pset_name, pset_data in properties_section.items():
+        pset_dict = _to_dict(pset_data) if not isinstance(pset_data, dict) else pset_data
+        if not pset_dict:
+            continue
+        ifc_props = []
+        for prop_name, prop_value in pset_dict.items():
+            if prop_name == "id" or prop_value is None:
+                continue
+            try:
+                nominal = _make_ifc_value(ifc, prop_value)
+                p = ifc.create_entity("IfcPropertySingleValue", Name=str(prop_name), NominalValue=nominal)
+                ifc_props.append(p)
+            except Exception:
+                continue
+        if ifc_props:
+            try:
+                pset = ifcopenshell.api.run("pset.add_pset", ifc, product=element, name=pset_name)
+                pset.HasProperties = ifc_props
+            except Exception:
+                pass
+
+
+def _write_quantity_sets_direct(ifc, element, obj):
+    """Legacy per-element quantity set writing (fallback)."""
     props = get_properties(obj)
     quantities_raw = props.get("Quantities")
     if quantities_raw is None:
@@ -269,20 +497,14 @@ def write_quantity_sets(ifc, element, obj):
     quantities_section = _to_dict(quantities_raw)
     if not quantities_section:
         return
-
     for qto_name, qto_data in quantities_section.items():
         qto_dict = _to_dict(qto_data) if not isinstance(qto_data, dict) else qto_data
         if not qto_dict:
             continue
-
         quantities = []
         for qty_key, qty_entry in qto_dict.items():
             if qty_key == "id":
                 continue
-
-            # Quantity entries can be:
-            #   - {name, units, value} dicts (ArchiCAD / IFC-native)
-            #   - plain numbers or numeric strings (Grasshopper)
             if isinstance(qty_entry, dict):
                 name = qty_entry.get("name", qty_key)
                 units = (qty_entry.get("units") or "").strip().lower()
@@ -293,50 +515,26 @@ def write_quantity_sets(ifc, element, obj):
                     continue
                 name = qty_key
                 units = ""
-
             if value is None:
                 continue
-
             try:
                 mapping = _UNIT_QTY_MAP.get(units)
                 if not mapping:
-                    # Infer quantity type from name keywords
-                    name_lower = name.lower()
                     for keyword, m in _NAME_QTY_MAP.items():
-                        if keyword in name_lower:
+                        if keyword in name.lower():
                             mapping = m
                             break
                 if mapping:
                     qty_type, value_attr = mapping
-                    qty = ifc.create_entity(
-                        qty_type, Name=name, **{value_attr: value}
-                    )
+                    qty = ifc.create_entity(qty_type, Name=name, **{value_attr: value})
                 else:
-                    # CountValue requires int
-                    qty = ifc.create_entity(
-                        "IfcQuantityCount", Name=name, CountValue=int(value)
-                    )
+                    qty = ifc.create_entity("IfcQuantityCount", Name=name, CountValue=int(value))
                 quantities.append(qty)
-            except Exception as e:
-                print(f"  Warning: quantity {name}: {e}")
+            except Exception:
                 continue
-
         if quantities:
             try:
-                qto = ifcopenshell.api.run(
-                    "pset.add_qto", ifc, product=element, name=qto_name
-                )
+                qto = ifcopenshell.api.run("pset.add_qto", ifc, product=element, name=qto_name)
                 qto.Quantities = quantities
-            except Exception as e:
-                print(f"  Warning: {qto_name}: {e}")
-
-
-# ---------------------------------------------------------------------------
-# Public API — called from main.py
-# ---------------------------------------------------------------------------
-
-def write_all_properties(ifc, element, obj):
-    """Write all properties, quantities, and attributes from _properties."""
-    set_element_attributes(ifc, element, obj)
-    write_property_sets(ifc, element, obj)
-    write_quantity_sets(ifc, element, obj)
+            except Exception:
+                pass