import os from pathlib import Path from typing import List from ee.digikey import Digikey, DigikeyParser, DigikeyClient, SearchResponseTypes, DigikeyProduct from ee.xml import bomFile, bom_file_utils from ee.xml.bomFile import DigikeyDistributorInfo from ee.xml.uris import DIGIKEY_URI __all__ = ["refresh_parts"] def resolved(di: DigikeyDistributorInfo, part: bomFile.Part, p: DigikeyProduct): di.stateProp = "resolved" fact_set = bom_file_utils.find_fact_set(part, DIGIKEY_URI, create=True) # Remove the old list fact_set.factsProp = bomFile.FactList() facts: List[bomFile.Fact] = fact_set.factsProp.factProp for a in p.attributes: facts.append(bomFile.Fact(key=a.attribute_type.id, label=a.attribute_type.label, value=a.value)) def refresh_parts(in_path: Path, out_path: Path, cache_dir: Path, force_refresh: bool): print("in: {}, out: {}".format(in_path, out_path)) in_file = bomFile.parse(str(in_path), True) if in_file.partsProp is None: in_file.partsProp = bomFile.PartList() parser = DigikeyParser(Digikey()) client = DigikeyClient(cache_dir) for part in in_file.partsProp.partProp: # type: bomFile.Part dpn = bom_file_utils.find_dpn(part, DIGIKEY_URI) mpn = bom_file_utils.find_pn(part) is_mpn = query = None if dpn is not None: query = dpn is_mpn = False elif mpn is not None: query = mpn is_mpn = True if query is None: print("could not find pn or dpn: part.id={}".format(part.idProp)) continue di = part.distributor_infoProp # type: DigikeyDistributorInfo if di is None: di = bomFile.DigikeyDistributorInfo() di.extensiontype_ = "DigikeyDistributorInfo" di.original_tagname_ = "distributor-info" part.distributor_infoProp = di if force_refresh or di.stateProp != "resolved": text = client.search(query) response = parser.parse_string(text) if response.response_type == SearchResponseTypes.SINGLE: resolved(di, part, 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: resolved(di, part, 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_path = in_path out_file = in_file tmp_path = str(out_path) + ".tmp" with open(tmp_path, "w") as f: out_file.export(f, 0, name_="bom-file") os.rename(tmp_path, str(out_path))