f67d9a0422
Phase 1 implementation: - New bom_sync.py: extract cross-document App::Link components from Assembly, resolve SiloItemId UUIDs to part numbers, diff against server BOM, apply adds/qty updates via individual CRUD calls. - Hook _push_bom_after_upload into Silo_Save and Silo_Commit (same non-blocking pattern as DAG sync). - Hook _update_manifest_revision to write revision_hash into .kc manifest after successful upload (#277). - Add bom_merged SSE signal + dispatch + Activity pane handler. - Add merge_bom_json to SiloClient (forward-looking for Phase 2). Merge rules: auto-add, auto-update qty, NEVER auto-delete removed entries (warn only). Refs: #276, #277
318 lines
10 KiB
Python
318 lines
10 KiB
Python
"""BOM extraction engine for FreeCAD Assembly documents.
|
|
|
|
Extracts cross-document ``App::Link`` components from an Assembly,
|
|
resolves Silo UUIDs to part numbers, diffs against the server BOM,
|
|
and applies adds/quantity updates via individual API calls.
|
|
|
|
No GUI dependencies -- usable in both desktop and headless mode.
|
|
|
|
Public API
|
|
----------
|
|
extract_bom_entries(doc) -> List[BomEntry]
|
|
resolve_entries(entries, client) -> (resolved, unresolved)
|
|
diff_bom(local_entries, remote_entries) -> BomDiff
|
|
apply_bom_diff(diff, parent_pn, client) -> BomResult
|
|
sync_bom_after_upload(doc, part_number, client) -> Optional[BomResult]
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import FreeCAD
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data structures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class BomEntry:
|
|
"""A single BOM line extracted from an Assembly."""
|
|
|
|
silo_item_id: Optional[str] # UUID from SiloItemId property
|
|
part_number: Optional[str] # resolved via get_item_by_uuid()
|
|
label: str # FreeCAD Label of the linked object
|
|
doc_path: str # FileName of the linked document
|
|
quantity: int # summed from ElementCount + individual links
|
|
consolidation_warning: bool = False # multiple individual links to same source
|
|
|
|
|
|
@dataclass
|
|
class BomDiff:
|
|
"""Result of diffing local assembly BOM against server BOM."""
|
|
|
|
added: List[Dict[str, Any]] = field(default_factory=list)
|
|
removed: List[Dict[str, Any]] = field(default_factory=list)
|
|
quantity_changed: List[Dict[str, Any]] = field(default_factory=list)
|
|
unchanged: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class BomResult:
|
|
"""Summary of a BOM sync operation."""
|
|
|
|
added_count: int = 0
|
|
updated_count: int = 0
|
|
unreferenced_count: int = 0
|
|
unresolved_count: int = 0
|
|
errors: List[str] = field(default_factory=list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Assembly detection helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _find_assembly(doc):
|
|
"""Return the first ``Assembly::AssemblyObject`` in *doc*, or ``None``."""
|
|
for obj in doc.Objects:
|
|
if obj.isDerivedFrom("Assembly::AssemblyObject"):
|
|
return obj
|
|
return None
|
|
|
|
|
|
def _get_silo_item_id(doc):
|
|
"""Read ``SiloItemId`` from the tracked object in *doc*.
|
|
|
|
Returns ``None`` if the document has no tracked object or no UUID.
|
|
"""
|
|
for obj in doc.Objects:
|
|
if hasattr(obj, "SiloItemId") and obj.SiloItemId:
|
|
return obj.SiloItemId
|
|
return None
|
|
|
|
|
|
def _link_count(link_obj) -> int:
|
|
"""Return the instance count for an ``App::Link``.
|
|
|
|
``ElementCount > 0`` means link array; otherwise single link (qty 1).
|
|
"""
|
|
element_count = getattr(link_obj, "ElementCount", 0)
|
|
return element_count if element_count > 0 else 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def extract_bom_entries(doc) -> List[BomEntry]:
|
|
"""Walk the Assembly in *doc* and collect cross-document link entries.
|
|
|
|
Returns an empty list if the document has no Assembly or no
|
|
cross-document links. Only first-level children are extracted;
|
|
sub-assemblies commit their own BOMs separately.
|
|
"""
|
|
assembly = _find_assembly(doc)
|
|
if assembly is None:
|
|
return []
|
|
|
|
# Key by linked document path to merge duplicates.
|
|
entries: Dict[str, BomEntry] = {}
|
|
|
|
for obj in assembly.Group:
|
|
# Accept App::Link (single or array) and App::LinkElement.
|
|
if not (
|
|
obj.isDerivedFrom("App::Link") or obj.isDerivedFrom("App::LinkElement")
|
|
):
|
|
continue
|
|
|
|
linked = obj.getLinkedObject()
|
|
if linked is None:
|
|
continue
|
|
|
|
# Skip in-document links (construction / layout geometry).
|
|
if linked.Document == doc:
|
|
continue
|
|
|
|
linked_doc = linked.Document
|
|
doc_path = linked_doc.FileName or linked_doc.Name
|
|
|
|
qty = _link_count(obj)
|
|
|
|
if doc_path in entries:
|
|
entries[doc_path].quantity += qty
|
|
entries[doc_path].consolidation_warning = True
|
|
else:
|
|
entries[doc_path] = BomEntry(
|
|
silo_item_id=_get_silo_item_id(linked_doc),
|
|
part_number=None,
|
|
label=linked.Label,
|
|
doc_path=doc_path,
|
|
quantity=qty,
|
|
)
|
|
|
|
return list(entries.values())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def resolve_entries(
|
|
entries: List[BomEntry], client
|
|
) -> Tuple[List[BomEntry], List[BomEntry]]:
|
|
"""Resolve ``SiloItemId`` UUIDs to part numbers via the API.
|
|
|
|
Returns ``(resolved, unresolved)``. Unresolved entries have no
|
|
``SiloItemId`` or the UUID lookup failed.
|
|
"""
|
|
resolved: List[BomEntry] = []
|
|
unresolved: List[BomEntry] = []
|
|
|
|
for entry in entries:
|
|
if not entry.silo_item_id:
|
|
unresolved.append(entry)
|
|
continue
|
|
|
|
try:
|
|
item = client.get_item_by_uuid(entry.silo_item_id)
|
|
entry.part_number = item["part_number"]
|
|
resolved.append(entry)
|
|
except Exception:
|
|
unresolved.append(entry)
|
|
|
|
return resolved, unresolved
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diff
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def diff_bom(
|
|
local_entries: List[BomEntry],
|
|
remote_entries: List[Dict[str, Any]],
|
|
) -> BomDiff:
|
|
"""Diff local assembly BOM against server BOM.
|
|
|
|
*local_entries*: resolved ``BomEntry`` list.
|
|
*remote_entries*: raw dicts from ``client.get_bom()`` with keys
|
|
``child_part_number`` and ``quantity``.
|
|
"""
|
|
local_map = {e.part_number: e.quantity for e in local_entries}
|
|
remote_map = {e["child_part_number"]: e.get("quantity", 1) for e in remote_entries}
|
|
|
|
diff = BomDiff()
|
|
|
|
for pn, qty in local_map.items():
|
|
if pn not in remote_map:
|
|
diff.added.append({"part_number": pn, "quantity": qty})
|
|
elif remote_map[pn] != qty:
|
|
diff.quantity_changed.append(
|
|
{
|
|
"part_number": pn,
|
|
"local_quantity": qty,
|
|
"remote_quantity": remote_map[pn],
|
|
}
|
|
)
|
|
else:
|
|
diff.unchanged.append({"part_number": pn, "quantity": qty})
|
|
|
|
for pn, qty in remote_map.items():
|
|
if pn not in local_map:
|
|
diff.removed.append({"part_number": pn, "quantity": qty})
|
|
|
|
return diff
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Apply
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def apply_bom_diff(diff: BomDiff, parent_pn: str, client) -> BomResult:
|
|
"""Apply adds and quantity updates to the server BOM.
|
|
|
|
Uses individual CRUD calls (Phase 1 fallback). Phase 2 replaces
|
|
this with a single ``POST /items/{pn}/bom/merge`` call.
|
|
|
|
Removed entries are NEVER deleted -- only warned about.
|
|
Each call is individually wrapped so one failure does not block others.
|
|
"""
|
|
result = BomResult()
|
|
|
|
for entry in diff.added:
|
|
try:
|
|
client.add_bom_entry(
|
|
parent_pn,
|
|
entry["part_number"],
|
|
quantity=entry["quantity"],
|
|
rel_type="component",
|
|
)
|
|
result.added_count += 1
|
|
except Exception as e:
|
|
result.errors.append(f"add {entry['part_number']}: {e}")
|
|
|
|
for entry in diff.quantity_changed:
|
|
try:
|
|
client.update_bom_entry(
|
|
parent_pn,
|
|
entry["part_number"],
|
|
quantity=entry["local_quantity"],
|
|
)
|
|
result.updated_count += 1
|
|
except Exception as e:
|
|
result.errors.append(f"update {entry['part_number']}: {e}")
|
|
|
|
result.unreferenced_count = len(diff.removed)
|
|
if diff.removed:
|
|
pns = ", ".join(e["part_number"] for e in diff.removed)
|
|
FreeCAD.Console.PrintWarning(
|
|
f"BOM sync: {result.unreferenced_count} server entries "
|
|
f"not in assembly (not deleted): {pns}\n"
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Top-level orchestrator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def sync_bom_after_upload(doc, part_number: str, client) -> Optional[BomResult]:
|
|
"""Full BOM sync pipeline: extract, resolve, diff, apply.
|
|
|
|
Returns ``None`` if *doc* is not an assembly or has no cross-document
|
|
links. Returns a ``BomResult`` with summary counts otherwise.
|
|
"""
|
|
entries = extract_bom_entries(doc)
|
|
if not entries:
|
|
return None
|
|
|
|
resolved, unresolved = resolve_entries(entries, client)
|
|
|
|
# Log consolidation warnings.
|
|
for entry in entries:
|
|
if entry.consolidation_warning:
|
|
FreeCAD.Console.PrintWarning(
|
|
f"BOM sync: {entry.label} ({entry.doc_path}) has multiple "
|
|
f"individual links. Consider using a link array "
|
|
f"(ElementCount) for cleaner assembly management.\n"
|
|
)
|
|
|
|
# Log unresolved components.
|
|
for entry in unresolved:
|
|
FreeCAD.Console.PrintWarning(
|
|
f"BOM sync: {entry.label} ({entry.doc_path}) has no Silo "
|
|
f"part number -- excluded from BOM.\n"
|
|
)
|
|
|
|
if not resolved:
|
|
result = BomResult()
|
|
result.unresolved_count = len(unresolved)
|
|
return result
|
|
|
|
# Fetch current server BOM.
|
|
try:
|
|
remote = client.get_bom(part_number)
|
|
except Exception:
|
|
remote = []
|
|
|
|
diff = diff_bom(resolved, remote)
|
|
result = apply_bom_diff(diff, part_number, client)
|
|
result.unresolved_count = len(unresolved)
|
|
return result
|