import os.path import pydoc from pathlib import Path from typing import List, Optional from ee import EeException from ee.db import ObjDb from ee.logging import log from ee.part import PartDb, load_db, save_db, Part, fact_keys from ee.project import Project, report, SupplierDescriptor from ee.xml import types from ee.xml.uris import make_fact_key __all__ = ["create_bom"] class Hit(object): def __init__(self, part: Part, method: str): self.part = part self.method = method class BomPart(object): def __init__(self, part: Part): self.part = part ref = self.part.get_only_schematic_reference() self.ref = ref.referenceProp if ref else None self.hits = [] self.selected_part: Optional[Part] = None def add_hit(self, part, method): self.hits.append(Hit(part, method)) def make_report(out_file, unresolved_parts, bom_parts: ObjDb[BomPart], supplier_parts: ObjDb[Path]): kwargs = { "bom_parts": bom_parts, "supplier_parts": supplier_parts, "unresolved_parts": unresolved_parts, } report.save_report("ee", "bom.rst.j2", out_file, **kwargs) def default_strategy(x): return x def create_bom(project: Project, schematic_path: Path, out_path: Path, part_dbs: List[Path], fail_on_missing_parts: bool, strategy_name: Optional[str]): strategy = default_strategy if strategy_name: strategy = pydoc.locate(strategy_name) if not callable(strategy): raise EeException("Not a callable: {}, is a {}".format(strategy_name, type(strategy))) log.info("Using strategy '{}'".format(strategy_name)) supplier_parts = ObjDb[Part]() supplier_parts.add_unique_index("uri", lambda p: p.uri) supplier_parts.add_index("spn", lambda p: [ref.valueProp for ref in p.get_spns()], multiple=True) supplier_pn_idx = supplier_parts.add_multi_index("supplier,pn", lambda p: [ (p.supplier, ref.valueProp) for ref in p.get_mpns()], multiple=True) supplier_spn_idx = supplier_parts.add_multi_index("supplier,spn", lambda p: [ (p.supplier, ref.valueProp) for ref in p.get_spns()], multiple=True) suppliers: List[SupplierDescriptor] = [project.get_supplier_by_key(path.parent.name) for path in part_dbs] for path in part_dbs: for xml in load_db(path).iterparts(): if not xml.supplierProp: continue supplier_parts.add(Part(xml)) sch_db = load_db(schematic_path) bom_parts: ObjDb[BomPart] = ObjDb[BomPart]() bom_parts.add_multi_index("supplier,pn", lambda op: [ (op.part.supplierProp, ref.valueProp) for ref in op.part.get_mpns()] if op.part.supplier else None, multiple=True) for sch_part in sch_db.iterparts(): part = Part(sch_part) part = strategy(part) if part is None: continue include_in_bom = part.facts.get_value(fact_keys.include_in_bom) or "yes" if include_in_bom == "no": log.debug("Skipping {}, marked as schematic only".format(part.printable_reference)) continue bom_parts.add(BomPart(part)) for bom_part in bom_parts: sch_part_numbers = [pn.valueProp for pn in bom_part.part.get_mpns()] sch_supplier_part_numbers = [spn.valueProp for spn in bom_part.part.get_spns()] value_fact = bom_part.part.find_fact(make_fact_key("value")) if value_fact: value_fact = value_fact.valueProp for supplier in suppliers: # Part number search pns = supplier_pn_idx.get(supplier.uri) for sch_pn in sch_part_numbers: for supplier_part in pns.get(sch_pn, []): bom_part.add_hit(supplier_part, "pn") if value_fact: for supplier_part in pns.get(value_fact, []): bom_part.add_hit(supplier_part, "pn_value_fact") # Supplier number search spns = supplier_spn_idx.get(supplier.uri) for sch_spn in sch_supplier_part_numbers: for supplier_part in spns.get(sch_spn, []): bom_part.add_hit(supplier_part, "spn") if value_fact: for supplier_part in spns.get(value_fact, []): bom_part.add_hit(supplier_part, "spn_value_fact") unresolved_parts = [] for bom_part in bom_parts: if len(bom_part.hits) == 0: unresolved_parts.append(bom_part) elif len(bom_part.hits) == 1: bom_part.selected_part = bom_part.hits[0].part else: references = [hit.part.printable_reference for hit in bom_part.hits] raise EeException("Multiple hits when looking for part: {}". format(bom_part.ref, ",".join(references))) bom_parts.add_index("uri", lambda bp: bp.selected_part.uri if bp.selected_part else None) bom_parts.add_multi_index("supplier,part", lambda op: ( op.selected_part.supplier, op.selected_part.uri) if op.selected_part else None) if len(unresolved_parts) and fail_on_missing_parts: raise EeException("The bom has parts that can't be found from any supplier") out_file = project.report_dir / (os.path.splitext(out_path.name)[0] + ".rst") make_report(out_file, unresolved_parts, bom_parts, supplier_parts) out_parts = PartDb() found_parts = 0 for bom_part in bom_parts: if not bom_part.selected_part: log.info("No part selected for {}".format(bom_part.part.printable_reference)) supplier_part = None else: supplier_part = bom_part.selected_part uri = None # TODO: generate part = Part(types.Part(uri=uri)) # TODO: this should use the part's uri instead of schematic reference. However, right now there is no way to # differentiate between two part-reference objects. part.add_schematic_reference(bom_part.part.get_exactly_one_schematic_reference().referenceProp) if supplier_part: part.add_part_reference(supplier_part.uri) found_parts += 1 out_parts.add_entry(part, True) log.info("Found {} of {} parts (skipped {} that where not to be included in BOM), missing {}". format(found_parts, len(bom_parts), len(sch_db) - len(bom_parts), len(unresolved_parts))) save_db(out_path, out_parts)