Add support for part query from TME

kintree/config/config_interface.py

@@ -71,7 +71,7 @@ def load_config(path):
 
             dump_file(user_settings, os.path.join(path_to_user_files, filename))
 
-    for dir in ['user', 'inventree', 'kicad', 'digikey', 'mouser', 'element14', 'lcsc']:
+    for dir in ['user', 'inventree', 'kicad', 'digikey', 'mouser', 'element14', 'lcsc', 'tme']:
         try:
             # Load configuration
             config_files = os.path.join(path_to_root, dir, '')

kintree/config/inventree/suppliers.yaml

@@ -15,4 +15,7 @@ Newark:
   name: Newark
 LCSC:
   enable: true
-  name: LCSC
+  name: LCSC
+TME:
+  enable: true
+  name: TME

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)

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

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

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:

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(

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

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():