Add support for part query from TME

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

@@ -173,6 +173,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 = '/'
 
@@ -350,6 +350,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
 
@@ -372,6 +374,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())
@@ -423,6 +427,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': {
@@ -635,6 +657,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(
@@ -659,6 +693,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,189 @@
+import os
+
+from ..database import inventree_api
+from ..database import inventree_interface
+
+from .tme.tmeapi import Client
+from ..config import settings, config_interface
+from ..common.tools import download
+
+
+def create_category_recursive(category, parent_pk, category_counter):
+    pk = inventree_api.create_category_force(parent_pk, category['Name'])
+    category_counter = category_counter + 1
+    if not category['SubTree'] is None:
+        for sub_tree in category['SubTree']:
+            create_category_recursive(category, pk, category_counter)
+
+
+def createAllCategories() -> int:
+    inventree_connect = inventree_interface.connect_to_server()
+
+    if not inventree_connect:
+        return 0
+
+    category_counter = 0
+    tme_api_settings = config_interface.load_file(settings.CONFIG_TME_API)
+    tme_client = Client(tme_api_settings['TME_API_TOKEN'], tme_api_settings['TME_API_SECRET'])
+    params = {
+        'Country': tme_api_settings['TME_API_COUNTRY'],
+        'Language': tme_api_settings['TME_API_LANGUAGE'],
+        'Tree': True
+    }
+
+    response = download(tme_client.request('/Products/GetCategories', params))
+    if response is None or response['Status'] != 'OK':
+        return 0
+
+    for category in response['Data']['CategoryTree']['SubTree']:
+        create_category_recursive(category, None, category_counter)
+    return category_counter
+
+
+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()
+
+
+def fetch_part_info(part_number: str) -> dict:
+    tme_api_settings = config_interface.load_file(settings.CONFIG_TME_API)
+    tme_client = Client(tme_api_settings['TME_API_TOKEN'], tme_api_settings['TME_API_SECRET'])
+    get_params_params = get_prod_params = get_files_params = {
+        'Country': tme_api_settings['TME_API_COUNTRY'],
+        'Language': tme_api_settings['TME_API_LANGUAGE'],
+        'SymbolList[0]': part_number,
+    }
+
+    response = download(tme_client.request('/Products/GetProducts', get_prod_params))
+    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_client.request('/Products/GetParameters', get_params_params))
+    # 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_client.request('/Products/GetProductsFiles', get_files_params))
+    # 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
+    '''
+    # Parameters
+    tmePart['parameters'] = {}
+    [parameter_key, name_key, value_key] = PARAMETERS_MAP
+    try:
+        for parameter in range(len(part[parameter_key])):
+            parameter_name = part[parameter_key][parameter][name_key]
+            parameter_value = part[parameter_key][parameter][value_key]
+            # Append to parameters dictionary
+            part_info['parameters'][parameter_name] = parameter_value
+    except TypeError:
+        # Parameter list is empty
+        pass'''
+    return part_info
+
+
+def get_lang_list(token, secret) -> list:
+    tme_client = Client(token, secret)
+    response = download(tme_client.request('/Utils/GetLanguages', {}))
+    if response['Status'] == "OK":
+        return response['Data']['LanguageList']
+    return []
+
+
+def get_country_list(token, secret) -> []:
+    tme_client = Client(token, secret)
+    response = download(tme_client.request('/Utils/GetCountries', {'Language': 'EN'}))
+    if response['Status'] == "OK":
+        ret = []
+        for country in response['Data']['CountryList']:
+            ret.append(country['CountryId'])
+        return ret
+    return []
+
+
+def test_api() -> bool:
+    ''' Test method for API '''
+    setup_environment()
+    tme_client = Client(os.environ['TME_API_TOKEN'], os.environ['TME_API_SECRET'])
+    jsonResponse = download(tme_client.request('/Utils/Ping', {}))
+    return jsonResponse['Status'] == "OK"

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
 
@@ -118,6 +118,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():