diff options
Diffstat (limited to 'src/ee/digikey/refresh_parts.py')
-rw-r--r-- | src/ee/digikey/refresh_parts.py | 97 |
1 files changed, 97 insertions, 0 deletions
diff --git a/src/ee/digikey/refresh_parts.py b/src/ee/digikey/refresh_parts.py new file mode 100644 index 0000000..87edf2f --- /dev/null +++ b/src/ee/digikey/refresh_parts.py @@ -0,0 +1,97 @@ +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)) |