import logging from itertools import groupby from operator import itemgetter from typing import List import ee.digikey as dk from ee.doit import DoItConfig from ee.ds import DataSet logger = logging.getLogger(__name__) doit_config = DoItConfig() def resolve_schematic_components(output: DataSet, in_ds: DataSet): def find(field, value): return [o for o in output.items() if o.object_type.name == "component-to-part-mapping" and o.get(field) == value] def save(refs: List[str], p: dk.DigikeyProduct): logger.info("Found part, dpn={}, mpn={}, refs={}".format(p.part_number, p.mpn, ", ".join(refs))) for ref in refs: # Make the key a bit long so we can have several parts that matches the MPN from a single seller key = "digikey-{}-{}".format(ref, p.part_number) output.create_object("component-to-part-mapping", key, replace=True). \ set("seller", "digikey"). \ set("ref", ref). \ set("part-number", p.part_number). \ set("mpn", p.mpn). \ set("description", p.description). \ set("quantity-available", p.quantity_available). \ set("url", p.url) digikey = dk.Digikey() client = dk.DigikeyClient(digikey, on_download=logger.info) components = [] # TODO: support searching by value and digikey part number directly. Priority should be "digikey", "mpn" and # "value", first field present should be used. for o in in_ds.items(): if o.object_type.name != "component": continue ref, = o.values("ref", required=True) if not ref: raise Exception("Bad component: object key={}, missing required field 'ref'".format(o.key)) mpn, = o.values("mpn", strip=True) # We ignore components without mpn. if not mpn: logger.debug("Skipping component without") continue components.append([mpn, ref]) components = sorted(components, key=itemgetter(0)) for mpn, components in groupby(components, key=itemgetter(0)): references = [c[1] for c in components] dk_components = find("mpn", mpn) if len(dk_components): logger.info("Already resolved {} to {}".format(mpn, ", ".join( sorted(set([c.get("part-number") for c in dk_components]))))) continue logger.info("Looking up MPN: {}, used by {}".format(mpn, ", ".join(sorted(references)))) response = client.search(mpn) if response.response_type == dk.SearchResponseTypes.SINGLE: save(references, response.products[0]) elif response.response_type == dk.SearchResponseTypes.MANY: # A search for "FOO" might return products "FOO" and "FOOZ" so we pick out the ones with the matching mpn # This will often be more than one product, but digikey shows the same part in different packagings. viable_products = [p for p in response.products if p.mpn == mpn] if len(viable_products) == 0: logger.warning("BUG: Got multiple hits ({}) but didn't find anyone that matched the MPN. Strange!". format(len(response.products))) else: if len(viable_products) == 1: part = viable_products[0] else: # Pick the first one, should be as good as any part = sorted(viable_products, key=lambda x: x.part_number)[0] logger.info("Got multiple hits ({})".format(len(viable_products))) save(references, part) elif response.response_type == dk.SearchResponseTypes.TOO_MANY: logger.warning("to many matches") elif response.response_type == dk.SearchResponseTypes.NO_MATCHES: logger.warning("no matches") def task_digikey_resolve_schematic_components(): out_data_set, in_data_sets = doit_config.data_sets_for(task_digikey_resolve_schematic_components) def action(): in_ds = doit_config.dsm.load_data_sets(in_data_sets) with doit_config.dsm.create_rw(out_data_set, clean=False) as output: resolve_schematic_components(output, in_ds) return dict( file_dep=[doit_config.dsm.cookie_for_ds(ds) for ds in in_data_sets], actions=[action], targets=[doit_config.dsm.cookie_for_ds(out_data_set)], ) doit_config.set_data_sets_for(task_digikey_resolve_schematic_components, "digikey-resolved-parts", "components") def download_part_facts(output: DataSet, in_ds: DataSet): digikey = dk.Digikey() client = dk.DigikeyClient(digikey, on_download=logger.debug) parts = [o for o in in_ds.items() if o.object_type.name == "component-to-part-mapping" and o.get("seller") == "digikey"] for pn in sorted({part.get("part-number") for part in parts}): logger.info("Downloading facts for {}".format(pn)) response = client.search(pn) if response.response_type == dk.SearchResponseTypes.SINGLE: product = response.products[0] o = output.create_object("digikey-part", pn, replace=True). \ set("part-number", product.part_number). \ set("url", product.url). \ set("mpn", product.mpn) for a in product.attributes: key = "{}/{}".format(a.attribute_type.id, a.attribute_type.label) key = key.replace("%", "_") value = a.value.replace("%", "%%") o.set(key, value) def task_digikey_fetch_full_part_facts(): out_data_set, in_data_sets = doit_config.data_sets_for(task_digikey_fetch_full_part_facts) def action(): in_ds = doit_config.dsm.load_data_sets(in_data_sets) with doit_config.dsm.create_rw(out_data_set, clean=False) as output: download_part_facts(output, in_ds) return dict( file_dep=[doit_config.dsm.cookie_for_ds(ds) for ds in in_data_sets], actions=[action], targets=[doit_config.dsm.cookie_for_ds(out_data_set)], ) doit_config.set_data_sets_for(task_digikey_fetch_full_part_facts, "digikey-parts", "digikey-resolved-parts")