import os.path import pydoc from pathlib import Path from typing import List, Optional from ee import EeException, StopToolException 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 __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: List[Hit] = [] 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(fact_keys.value) if value_fact: value_fact = value_fact.valueProp for supplier in suppliers: if value_fact is not None: pass # 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: if len(set(hit.part.uri for hit in 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 StopToolException("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 bp: ( bp.selected_part.supplier, bp.selected_part.uri) if bp.selected_part else None) out_file = project.report_dir / (os.path.splitext(out_path.name)[0] + ".rst") make_report(out_file, unresolved_parts, bom_parts, supplier_parts) if len(unresolved_parts) and fail_on_missing_parts: raise StopToolException("The bom has parts that can't be found from any supplier") 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)