From 97c8bb9db96e27051f8746865f657408263db0b8 Mon Sep 17 00:00:00 2001 From: Trygve Laugstøl Date: Tue, 26 Feb 2019 23:08:19 +0100 Subject: o Creating a PartDb that manages a file system directory with one xml file per part. o Switching xml-based code to use PartDb. --- src/ee/digikey/import_parts.py | 100 ---------------------------------------- src/ee/digikey/refresh_parts.py | 97 -------------------------------------- src/ee/digikey/search_parts.py | 94 +++++++++++++++++++++++++++++++++++++ 3 files changed, 94 insertions(+), 197 deletions(-) delete mode 100644 src/ee/digikey/import_parts.py delete mode 100644 src/ee/digikey/refresh_parts.py create mode 100644 src/ee/digikey/search_parts.py (limited to 'src/ee/digikey') diff --git a/src/ee/digikey/import_parts.py b/src/ee/digikey/import_parts.py deleted file mode 100644 index 748bbef..0000000 --- a/src/ee/digikey/import_parts.py +++ /dev/null @@ -1,100 +0,0 @@ -import os -from typing import List, MutableMapping - -from ee.xml import bomFile -from ee.xml.bom_file_utils import * -from ee.xml.uris import DIGIKEY_URI - -__all__ = ["import_parts"] - - -class Entry(object): - def __init__(self, new: bool, part: bomFile.Part): - self.new = new - self.part = part - - self.pn = find_pn(part) - self.dpn = find_dpn(part, DIGIKEY_URI) - - -def import_parts(in_path, out_path): - print("in: {}, out: {}".format(in_path, out_path)) - - in_file = bomFile.parse(in_path, True) - if in_file.partsProp is None: - in_file.partsProp = bomFile.PartList() - in_part_list = in_file.partsProp.partProp # type: List[bomFile.Part] - - print("in file: {} parts".format(len(in_part_list))) - - if os.path.isfile(out_path): - out_file = bomFile.parse(out_path, True) - else: - out_file = bomFile.BomFile() - - if out_file.partsProp is None: - out_file.partsProp = bomFile.PartList() - out_part_list = out_file.partsProp.partProp # type: List[bomFile.Part] - print("out file: {} parts".format(len(out_part_list))) - - existing_parts = [] # type: List[Entry] - pn_index = {} # type: MutableMapping[str, Entry] - dpn_index = {} # type: MutableMapping[str, Entry] - new_entry_added = 0 - - def add_entry(e: Entry): - existing_parts.append(e) - pn_index[e.pn] = e - dpn_index[e.dpn] = e - - if e.new: - out_part_list.append(e.part) - nonlocal new_entry_added - new_entry_added = new_entry_added + 1 - - print("len(out_part_list)={}".format(len(out_part_list))) - for part in out_part_list: # type: bomFile.Part - entry = Entry(False, part) - add_entry(entry) - - print("loaded {} existing parts".format(len(existing_parts))) - - for part in in_part_list: - pn_value = find_pn(part) - - if pn_value is None: - print("Skipping part with no part number: id={}".format(part.idProp)) - continue - - entry = pn_index.get(pn_value) - - if entry is not None: - print("Already imported pn_value={}".format(pn_value)) - continue - - print("Importing {}".format(pn_value)) - - pns = bomFile.PartNumberList() - - if pn_value is not None: - pns.add_part_number(bomFile.PartNumber(value=pn_value)) - - dpn_value = find_dpn(part, DIGIKEY_URI) - if dpn_value is not None: - pns.add_part_number(bomFile.PartNumber(value=dpn_value, distributor=DIGIKEY_URI)) - - if len(pns.part_numberProp) == 0: - continue - - new_part = bomFile.Part(part_numbers=pns) - entry = Entry(True, new_part) - add_entry(entry) - - if new_entry_added: - print("Imported {} entries".format(new_entry_added)) - tmp_path = out_path + ".tmp" - with open(tmp_path, "w") as f: - out_file.export(f, 0, name_="bom-file") - os.rename(tmp_path, out_path) - else: - print("no new entries") diff --git a/src/ee/digikey/refresh_parts.py b/src/ee/digikey/refresh_parts.py deleted file mode 100644 index 87edf2f..0000000 --- a/src/ee/digikey/refresh_parts.py +++ /dev/null @@ -1,97 +0,0 @@ -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)) diff --git a/src/ee/digikey/search_parts.py b/src/ee/digikey/search_parts.py new file mode 100644 index 0000000..62bb4ee --- /dev/null +++ b/src/ee/digikey/search_parts.py @@ -0,0 +1,94 @@ +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 +from ee.xml import bomFile, bom_file_utils +from ee.xml.bomFile import DigikeyDistributorInfo +from ee.xml.uris import DIGIKEY_URI + +__all__ = ["search_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 search_parts(in_dir: Path, out_dir: Path, cache_dir: Path, force_refresh: bool): + print("in: {}, out: {}".format(in_dir, out_dir)) + + in_db = load_db(in_dir) + out_parts = PartDb() + + parser = DigikeyParser(Digikey()) + client = DigikeyClient(cache_dir) + + for part in in_db.iterparts(): + 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_parts.add_entry(part, True) + + print("Saving {} work parts".format(out_parts.size())) + save_db(out_dir, out_parts) -- cgit v1.2.3