From 327debee6f120ad7a3036a07ddb7da36c757bd17 Mon Sep 17 00:00:00 2001 From: spike Date: Sun, 2 Jul 2023 17:44:07 +0200 Subject: [PATCH] Add support for part query from TME --- kintree/config/inventree/suppliers.yaml | 5 +- kintree/config/settings.py | 4 + kintree/config/tme/tme_api.yaml | 4 + kintree/config/tme/tme_config.yaml | 10 ++ kintree/database/inventree_interface.py | 8 +- kintree/gui/views/settings.py | 37 +++++ kintree/search/tme_api.py | 174 ++++++++++++++++++++++++ run_tests.py | 11 +- 8 files changed, 250 insertions(+), 3 deletions(-) create mode 100644 kintree/config/tme/tme_api.yaml create mode 100644 kintree/config/tme/tme_config.yaml create mode 100644 kintree/search/tme_api.py diff --git a/kintree/config/inventree/suppliers.yaml b/kintree/config/inventree/suppliers.yaml index f4e6c33d..231c7b5d 100644 --- a/kintree/config/inventree/suppliers.yaml +++ b/kintree/config/inventree/suppliers.yaml @@ -15,4 +15,7 @@ Newark: name: Newark LCSC: enable: true - name: LCSC \ No newline at end of file + name: LCSC +TME: + enable: true + name: TME \ No newline at end of file diff --git a/kintree/config/settings.py b/kintree/config/settings.py index 0679a828..95f87368 100644 --- a/kintree/config/settings.py +++ b/kintree/config/settings.py @@ -175,6 +175,10 @@ def load_suppliers(): CONFIG_LCSC = config_interface.load_file(os.path.join(CONFIG_USER_FILES, 'lcsc_config.yaml')) CONFIG_LCSC_API = os.path.join(CONFIG_USER_FILES, 'lcsc_api.yaml') +# TME user configuration +CONFIG_TME = config_interface.load_file(os.path.join(CONFIG_USER_FILES, 'tme_config.yaml')) +CONFIG_TME_API = os.path.join(CONFIG_USER_FILES, 'tme_api.yaml') + # Automatic category match confidence level (from 0 to 100) CATEGORY_MATCH_RATIO_LIMIT = CONFIG_SEARCH_API.get('CATEGORY_MATCH_RATIO_LIMIT', 100) # Search results caching (stored in files) diff --git a/kintree/config/tme/tme_api.yaml b/kintree/config/tme/tme_api.yaml new file mode 100644 index 00000000..76645ddb --- /dev/null +++ b/kintree/config/tme/tme_api.yaml @@ -0,0 +1,4 @@ +TME_API_TOKEN: NULL +TME_API_SECRET: NULL +TME_API_COUNTRY: NULL +TME_API_LANGUAGE: NULL \ No newline at end of file diff --git a/kintree/config/tme/tme_config.yaml b/kintree/config/tme/tme_config.yaml new file mode 100644 index 00000000..b0ef4a9e --- /dev/null +++ b/kintree/config/tme/tme_config.yaml @@ -0,0 +1,10 @@ +SUPPLIER_INVENTREE_NAME: TME +SEARCH_NAME: null +SEARCH_DESCRIPTION: null +SEARCH_REVISION: null +SEARCH_KEYWORDS: null +SEARCH_SKU: null +SEARCH_MANUFACTURER: null +SEARCH_MPN: null +SEARCH_SUPPLIER_URL: null +SEARCH_DATASHEET: null \ No newline at end of file diff --git a/kintree/database/inventree_interface.py b/kintree/database/inventree_interface.py index 5df6965a..ce230171 100644 --- a/kintree/database/inventree_interface.py +++ b/kintree/database/inventree_interface.py @@ -5,7 +5,7 @@ from ..common.tools import cprint from ..config import config_interface from ..database import inventree_api -from ..search import search_api, digikey_api, mouser_api, element14_api, lcsc_api +from ..search import search_api, digikey_api, mouser_api, element14_api, lcsc_api, tme_api category_separator = '/' @@ -351,6 +351,8 @@ def get_value_from_user_key(user_key: str, default_key: str, default_value=None) user_search_key = settings.CONFIG_ELEMENT14.get(user_key, None) elif supplier == 'LCSC': user_search_key = settings.CONFIG_LCSC.get(user_key, None) + elif supplier == 'TME': + user_search_key = settings.CONFIG_TME.get(user_key, None) else: return default_value @@ -373,6 +375,8 @@ def get_value_from_user_key(user_key: str, default_key: str, default_value=None) default_search_keys = element14_api.get_default_search_keys() elif supplier == 'LCSC': default_search_keys = lcsc_api.get_default_search_keys() + elif supplier == 'TME': + default_search_keys = tme_api.get_default_search_keys() else: # Empty array of default search keys default_search_keys = [''] * len(digikey_api.get_default_search_keys()) @@ -424,6 +428,8 @@ def supplier_search(supplier: str, part_number: str, test_mode=False) -> dict: part_info = element14_api.fetch_part_info(part_number, supplier) elif supplier == 'LCSC': part_info = lcsc_api.fetch_part_info(part_number) + elif supplier == 'TME': + part_info = tme_api.fetch_part_info(part_number) # Check supplier data exist if not part_info: diff --git a/kintree/gui/views/settings.py b/kintree/gui/views/settings.py index a5007b1f..f4a7c233 100644 --- a/kintree/gui/views/settings.py +++ b/kintree/gui/views/settings.py @@ -81,6 +81,28 @@ ft.TextField(), None, ] + elif supplier == 'TME': + tme_api_settings = config_interface.load_file(global_settings.CONFIG_TME_API) + supplier_settings[supplier]['API Token'] = [ + tme_api_settings['TME_API_TOKEN'], + ft.TextField(), + None, + ] + supplier_settings[supplier]['API Secret'] = [ + tme_api_settings['TME_API_SECRET'], + ft.TextField(), + None, + ] + supplier_settings[supplier]['API Country'] = [ + tme_api_settings['TME_API_COUNTRY'], + ft.TextField(), + None, + ] + supplier_settings[supplier]['API Language'] = [ + tme_api_settings['TME_API_LANGUAGE'], + ft.TextField(), + None, + ] SETTINGS = { 'User Settings': { @@ -641,6 +663,18 @@ def save_s(self, e: ft.ControlEvent, supplier: str, show_dialog=True): } lcsc_settings = {**settings_from_file, **updated_settings} config_interface.dump_file(lcsc_settings, global_settings.CONFIG_LCSC_API) + elif supplier == 'TME': + # Load settings from file + settings_from_file = config_interface.load_file(global_settings.CONFIG_TME_API) + # Update settings values + updated_settings = { + 'TME_API_TOKEN': SETTINGS[self.title][supplier]['API Token'][1].value, + 'TME_API_SECRET': SETTINGS[self.title][supplier]['API Secret'][1].value, + 'TME_API_COUNTRY': SETTINGS[self.title][supplier]['API Country'][1].value, + 'TME_API_LANGUAGE': SETTINGS[self.title][supplier]['API Language'][1].value, + } + tme_settings = {**settings_from_file, **updated_settings} + config_interface.dump_file(tme_settings, global_settings.CONFIG_TME_API) if show_dialog: self.show_dialog( @@ -665,6 +699,9 @@ def test_s(self, e: ft.ControlEvent, supplier: str): elif supplier == 'LCSC': from ...search import lcsc_api result = lcsc_api.test_api() + elif supplier == 'TME': + from ...search import tme_api + result = tme_api.test_api() if result: self.show_dialog( diff --git a/kintree/search/tme_api.py b/kintree/search/tme_api.py new file mode 100644 index 00000000..d1eddd3b --- /dev/null +++ b/kintree/search/tme_api.py @@ -0,0 +1,174 @@ +import base64 +import collections +import hashlib +import hmac +import os +import urllib.parse +import urllib.request + +from ..common.tools import download +from ..config import config_interface, settings + + +def get_default_search_keys(): + return [ + 'Symbol', + 'Description', + '', # Revision + 'Category', + 'Symbol', + 'Producer', + 'OriginalSymbol', + 'ProductInformationPage', + 'Datasheet', + 'Photo', + ] + + +def check_environment() -> bool: + TME_API_TOKEN = os.environ.get('TME_API_TOKEN', None) + TME_API_SECRET = os.environ.get('TME_API_SECRET', None) + + if not TME_API_TOKEN or not TME_API_SECRET: + return False + + return True + + +def setup_environment(force=False) -> bool: + if not check_environment() or force: + tme_api_settings = config_interface.load_file(settings.CONFIG_TME_API) + os.environ['TME_API_TOKEN'] = tme_api_settings['TME_API_TOKEN'] + os.environ['TME_API_SECRET'] = tme_api_settings['TME_API_SECRET'] + + return check_environment() + + +# Based on TME API snippets mentioned in API documentation: https://developers.tme.eu/documentation/download +# https://github.com/tme-dev/TME-API/blob/master/Python/call.py +def tme_api_request(endpoint, tme_api_settings, part_number, api_host='https://api.tme.eu', format='json'): + params = collections.OrderedDict() + params['Country'] = tme_api_settings['TME_API_COUNTRY'] + params['Language'] = tme_api_settings['TME_API_LANGUAGE'] + params['SymbolList[0]'] = part_number + params['Token'] = tme_api_settings['TME_API_TOKEN'] + + url = api_host + endpoint + '.' + format + encoded_params = urllib.parse.urlencode(params, quote_via=urllib.parse.quote) + signature_base = 'POST' + '&' + urllib.parse.quote(url, '') + '&' + urllib.parse.quote(encoded_params, '') + hmac_value = hmac.new( + tme_api_settings['TME_API_SECRET'].encode(), + signature_base.encode(), + hashlib.sha1 + ).digest() + api_signature = base64.encodebytes(hmac_value).rstrip() + params['ApiSignature'] = api_signature + + data = urllib.parse.urlencode(params).encode() + headers = { + "Content-type": "application/x-www-form-urlencoded", + } + return urllib.request.Request(url, data, headers) + + +def fetch_part_info(part_number: str) -> dict: + tme_api_settings = config_interface.load_file(settings.CONFIG_TME_API) + response = download(tme_api_request('/Products/GetProducts', tme_api_settings, part_number)) + if response is None or response['Status'] != 'OK': + return {} + # in the case if multiple parts returned + # (for e.g. if we looking for NE555A we could have NE555A and NE555AB in the results) + found = False + index = 0 + for product in response['Data']['ProductList']: + if product['Symbol'] == part_number: + found = True + break + index = index + 1 + + if not found: + return {} + part_info = response['Data']['ProductList'][index] + part_info['Photo'] = "http:" + part_info['Photo'] + part_info['ProductInformationPage'] = "http:" + part_info['ProductInformationPage'] + part_info['category'] = part_info['Category'] + part_info['subcategory'] = None + + # query the parameters + response = download(tme_api_request('/Products/GetParameters', tme_api_settings, part_number)) + # check if accidentally no data returned + if response is None or response['Status'] != 'OK': + return part_info + + found = False + index = 0 + for product in response['Data']['ProductList']: + if product['Symbol'] == part_number: + found = True + break + index = index + 1 + + if not found: + return part_info + + part_info['parameters'] = {} + for param in response['Data']['ProductList'][index]["ParameterList"]: + part_info['parameters'][param['ParameterName']] = param['ParameterValue'] + + # Query the files associated to the product + response = download(tme_api_request('/Products/GetProductsFiles', tme_api_settings, part_number)) + # check if accidentally no products returned + if response is None or response['Status'] != 'OK': + return part_info + + found = False + index = 0 + for product in response['Data']['ProductList']: + if product['Symbol'] == part_number: + found = True + break + index = index + 1 + + if not found: + return part_info + + for doc in response['Data']['ProductList'][index]['Files']['DocumentList']: + if doc['DocumentType'] == 'DTE': + part_info['Datasheet'] = 'http:' + doc['DocumentUrl'] + break + return part_info + + +def test_api(check_content=False) -> bool: + ''' Test method for API ''' + setup_environment() + + test_success = True + expected = { + 'Description': 'Capacitor: ceramic; MLCC; 33pF; 50V; C0G; ±5%; SMD; 0402', + 'Symbol': 'CL05C330JB5NNNC', + 'Producer': 'SAMSUNG', + 'OriginalSymbol': 'CL05C330JB5NNNC', + 'ProductInformationPage': 'http://www.tme.eu/en/details/cl05c330jb5nnnc/mlcc-smd-capacitors/samsung/', + 'Datasheet': 'http://www.tme.eu/Document/7da762c1dbaf553c64ad9c40d3603826/mlcc_samsung.pdf', + 'Photo': 'http://ce8dc832c.cloudimg.io/v7/_cdn_/8D/4E/00/00/0/58584_1.jpg?width=640&height=480&wat=1&wat_url=_tme-wrk_%2Ftme_new.png&wat_scale=100p&ci_sign=be42abccf5ef8119c2a0d945a27afde3acbeb699', + } + + test_part = fetch_part_info('CL05C330JB5NNNC') + + # Check for response + if not test_part: + test_success = False + + if not check_content: + return test_success + + # Check content of response + if test_success: + for key, value in expected.items(): + if test_part[key] != value: + print(f'{test_part[key]} != {value}') + test_success = False + break + + return test_success diff --git a/run_tests.py b/run_tests.py index 06eeabf4..f639f8d6 100644 --- a/run_tests.py +++ b/run_tests.py @@ -6,7 +6,7 @@ from kintree.config import config_interface from kintree.database import inventree_api, inventree_interface from kintree.kicad import kicad_interface -from kintree.search import digikey_api, mouser_api, element14_api, lcsc_api +from kintree.search import digikey_api, mouser_api, element14_api, lcsc_api, tme_api from kintree.search.snapeda_api import test_snapeda_api from kintree.setup_inventree import setup_inventree @@ -120,6 +120,15 @@ def check_result(status: str, new_part: bool) -> bool: else: cprint('[ PASS ]') + # Test TME API + if 'TME' in settings.SUPPORTED_SUPPLIERS_API: + pretty_test_print('[MAIN]\tTME API Test') + if not tme_api.test_api(): + cprint('[ FAIL ]') + sys.exit(-1) + else: + cprint('[ PASS ]') + # Test SnapEDA API methods pretty_test_print('[MAIN]\tSnapEDA API Test') if not test_snapeda_api():