from pathlib import Path from typing import List from ee.digikey import Digikey, DigikeyParser, DigikeyClient, SearchResponseTypes, DigikeyProduct from ee.part import PartDb, load_db, save_db, Part from ee.project import Project from ee.xml import types from ee.xml.uris import DIGIKEY_URI, make_digikey_fact_key __all__ = ["search_parts"] def resolved(p: DigikeyProduct) -> Part: # TODO: fix uri xml = types.Part(uri="https://digikey.com/pn#{}".format(p.part_number), supplier=DIGIKEY_URI, description=p.description, distributor_info=types.DistributorInfo(state="resolved"), links=types.LinkList(), facts=types.FactList(), references=types.ReferencesList()) part = Part(xml) if p.url: part.get_links().append(types.Link(url=p.url, relation="canonical", media_type="text/html")) for d in p.documents: title = "{}: {}".format(d.kind, d.title) part.get_links().append(types.Link(url=d.url, relation="http://purl.org/ee/link-relation#documentation", media_type="text/html", title=title)) part.add_spn(p.part_number) if p.mpn: part.add_mpn(p.mpn) facts: List[types.Fact] = xml.factsProp.factProp for a in p.attributes: key = make_digikey_fact_key(a.attribute_type.id) facts.append(types.Fact(key=key, label=a.attribute_type.label, value=a.value)) if len(p.price_breaks): xml.price_breaksProp = types.PriceBreakList() price_breaks: List[types.PriceBreak] = xml.price_breaksProp.price_break for pb in p.price_breaks: amount = types.Amount(value=str(pb.per_piece_price.amount), currency=pb.per_piece_price.currency) price_breaks.append(types.PriceBreak(pb.quantity, amount=amount)) return part def search_parts(in_path: Path, out_path: Path, cache_dir: Path): in_db = load_db(in_path) out_parts = PartDb() parser = DigikeyParser(Digikey()) client = DigikeyClient(cache_dir) for xml in in_db.iterparts(): if xml.supplierProp is not None and xml.supplierProp != DIGIKEY_URI: continue part = Part(xml) dpn = part.get_only_spn() mpn = part.get_only_mpn() is_mpn = query = None if dpn is not None: query = dpn.valueProp is_mpn = False elif mpn is not None: query = mpn.valueProp is_mpn = True if query is None: # TODO: use schematic reference print("could not find pn or dpn: part.uri={}".format(xml.uriProp)) continue out_id = query out_part = types.Part(uri=out_id, distributor_info=types.DistributorInfo(), supplier=DIGIKEY_URI, references=xml.referencesProp) di = out_part.distributor_infoProp text = client.search(query) response = parser.parse_string(text) if response.response_type == SearchResponseTypes.SINGLE: out_part = resolved(response.products[0]) elif response.response_type == SearchResponseTypes.MANY: # find those with an exact match. Digikey uses a prefix search so a query for "FOO" will return "FOO" # and "FOOT". def get_field(p): return p.mpn if is_mpn else p.part_number filtered_products = [p for p in response.products if get_field(p) == query] if len(filtered_products) == 0: di.stateProp = "not-found" else: dpn = sorted(filtered_products, key=lambda p: p.part_number)[0].part_number response = parser.parse_string(client.search(dpn)) if response.response_type == SearchResponseTypes.SINGLE: out_part = resolved(response.products[0]) else: di.stateProp = "many" elif response.response_type == SearchResponseTypes.TOO_MANY: di.stateProp = "too-many" elif response.response_type == SearchResponseTypes.NO_MATCHES: di.stateProp = "not-found" out_parts.add_entry(out_part, True) print("Saving {} work parts".format(out_parts.size())) save_db(out_path, out_parts, sort=True)