Add initial implementation of /parts endpoint

Author: ubruhin

.gitignore

@@ -4,4 +4,5 @@
 *.autosave
 *.swp
 .DS_Store
+config/
 Thumbs.db

Dockerfile

@@ -0,0 +1,27 @@
+ARG ALPINE_TAG
+FROM alpine:$ALPINE_TAG
+
+# Install packages.
+RUN apk add --no-cache \
+  python3 \
+  py3-flask \
+  py3-flask-pyc \
+  py3-gunicorn \
+  py3-gunicorn-pyc \
+  py3-requests \
+  py3-requests-pyc
+
+# Copy files.
+COPY app.py app/
+COPY static/ app/static/
+WORKDIR app
+
+# Set entrypoint.
+ENTRYPOINT [ \
+    "gunicorn", \
+    "--access-logfile=-", \
+    "--bind=0.0.0.0:8000", \
+    "--forwarded-allow-ips=*", \
+    "--workers=4", \
+    "app:app" \
+]

README.md

@@ -1,5 +1,44 @@
 # librepcb-api-server
 
+Official server-side implementation of the
+[LibrePCB API](https://developers.librepcb.org/d1/dcb/doc_server_api.html)
+as accessed by the LibrePCB application. Note that some older API paths are
+implemented in a different way and might be migrated to this repository
+later.
+
+## Requirements
+
+Only Docker Compose is needed to run this server on a Linux machine.
+
+## Configuration
+
+To make all features working, a configuration file `config/api.json` is
+required with the following content:
+
+```json
+{
+     /* Config for endpoint '/parts' */
+     "parts_operational": false,
+     "parts_query_url": "",
+     "parts_query_token": ""
+}
+```
+
+## Usage
+
+For local development, the server can be run with this command:
+
+```bash
+docker-compose up --build
+```
+
+Afterwards, the API runs on http://localhost:8000/:
+
+```bash
+curl -X POST -H "Content-Type: application/json" -d @demo-request.json \
+     'http://localhost:8000/api/v1/parts/query' | jq '.'
+```
+
 ## License
 
 The content in this repository is published under the

app.py

@@ -0,0 +1,307 @@
+# -*- coding: utf-8 -*-
+
+import json
+
+import requests
+from flask import (Flask, g, make_response, request, send_from_directory,
+                   url_for)
+from werkzeug.middleware.proxy_fix import ProxyFix
+
+app = Flask(__name__)
+app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
+
+PARTS_MAX_COUNT = 10
+PARTS_QUERY_TIMEOUT = 8.0
+PARTS_QUERY_FRAGMENT = """
+ fragment f on Stock {
+  products {
+   basic {
+    manufacturer
+    mfgpartno
+    status
+   }
+   url
+   imageUrl
+   datasheetUrl
+  }
+  summary {
+   inStockInventory
+   medianPrice
+   suppliersInStock
+  }
+ }
+"""
+PARTS_QUERY_STATUS_MAP = {
+    'active': 'Active',
+    'active-unconfirmed': 'Active',
+    'nrfnd': 'NRND',
+    'eol': 'Obsolete',
+    'obsolete': 'Obsolete',
+    'discontinued': 'Obsolete',
+    'transferred': 'Obsolete',
+    'contact mfr': None,  # Not supported, but here to avoid warning.
+}
+MANUFACTURER_REPLACEMENTS = {
+    'ä': 'ae',
+    'ö': 'oe',
+    'ü': 'ue',
+    'texas instruments': 'ti',
+    'stmicroelectronics': 'st',
+}
+MANUFACTURER_REMOVALS = set([
+    'contact',
+    'devices',
+    'electronics',
+    'inc.',
+    'inc',
+    'incorporated',
+    'integrated',
+    'international',
+    'limited',
+    'ltd.',
+    'ltd',
+    'microelectronics',
+    'semiconductor',
+    'semiconductors',
+    'solutions',
+    'systems',
+    'technology',
+    'usa',
+])
+
+
+def _get_config(key, fallback=None):
+    if 'config' not in g:
+        try:
+            with open('/config/api.json', 'rb') as f:
+                g.config = json.load(f)
+        except Exception as e:
+            app.logger.critical(str(e))
+            g.config = dict()
+    return g.config.get(key, fallback)
+
+
+def _build_headers():
+    return {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json, multipart/mixed',
+        'Authorization': 'Bearer {}'.format(_get_config('parts_query_token')),
+    }
+
+
+def _build_request(parts):
+    args = []
+    queries = []
+    variables = {}
+    for i in range(len(parts)):
+        args.append('$mpn{}:String!'.format(i))
+        queries.append('q{}:findStocks(mfgpartno:$mpn{}){{...f}}'.format(i, i))
+        variables['mpn{}'.format(i)] = parts[i]['mpn']
+    query = 'query Stocks({}) {{\n{}\n}}'.format(
+        ','.join(args),
+        '\n'.join(queries)
+    ) + PARTS_QUERY_FRAGMENT
+    return dict(query=query, variables=variables)
+
+
+def _get_basic_value(product, key):
+    if type(product) is dict:
+        basic = product.get('basic')
+        if type(basic) is dict:
+            value = basic.get(key)
+            if type(value) is str:
+                return value
+    return ''
+
+
+def _normalize_manufacturer(mfr):
+    mfr = mfr.lower()
+    for old, new in MANUFACTURER_REPLACEMENTS.items():
+        mfr = mfr.replace(old, new)
+    terms = [s for s in mfr.split(' ') if s not in MANUFACTURER_REMOVALS]
+    return ' '.join(terms)
+
+
+def _calc_product_match_score(p, mpn_n, mfr_n):
+    score = 0
+
+    status_p = PARTS_QUERY_STATUS_MAP.get(_get_basic_value(p, 'status'))
+    if status_p == 'Active':
+        score += 200
+    elif status_p == 'NRND':
+        score += 100
+
+    mpn_p = _get_basic_value(p, 'mfgpartno').lower()
+    if mpn_p == mpn_n:
+        score += 20  # MPN matches exactly.
+    elif mpn_p.replace(' ', '') == mpn_n.replace(' ', ''):
+        score += 10  # MPN matches when ignoring whitespaces.
+    else:
+        return 0  # MPN does not match!
+
+    mfr_p = _normalize_manufacturer(_get_basic_value(p, 'manufacturer'))
+    if mfr_p == mfr_n:
+        score += 4  # Manufacturer matches exactly.
+    elif mfr_n in mfr_p:
+        score += 3  # Manufacturer matches partially.
+    elif mfr_n.replace(' ', '') in mfr_p.replace(' ', ''):
+        score += 2  # Manufacturer matches partially when ignoring whitespaces.
+    elif mfr_n.split(' ')[0] in mfr_p:
+        score += 1  # The first term of the manufacturer matches.
+    else:
+        return 0  # Manufacturer does not match!
+
+    return score
+
+
+def _get_product(data, mpn, manufacturer):
+    products = (data.get('products') or [])
+    for p in products:
+        p['_score'] = _calc_product_match_score(
+            p, mpn.lower(), _normalize_manufacturer(manufacturer))
+    products = sorted([p for p in products if p['_score'] > 0],
+                      key=lambda p: p['_score'], reverse=True)
+    return products[0] if len(products) else None
+
+
+def _add_pricing_url(out, data):
+    value = data.get('url')
+    if value is not None:
+        out['pricing_url'] = value
+
+
+def _add_image_url(out, data):
+    value = data.get('imageUrl')
+    if value is not None:
+        out['picture_url'] = value
+
+
+def _add_status(out, data):
+    status = data.get('status') or ''
+    status_n = status.lower()
+    value = PARTS_QUERY_STATUS_MAP.get(status_n.lower())
+    if value is not None:
+        out['status'] = value
+    elif len(status_n) and (status_n not in PARTS_QUERY_STATUS_MAP):
+        app.logger.warning('Unknown part lifecycle status: {}'.format(status))
+
+
+def _stock_to_availability(stock):
+    if stock > 100000:
+        return 10  # Very Good
+    elif stock > 5000:
+        return 5  # Good
+    elif stock > 200:
+        return 0  # Normal
+    elif stock > 0:
+        return -5  # Bad
+    else:
+        return -10  # Very Bad
+
+
+def _suppliers_to_availability(suppliers):
+    if suppliers > 30:
+        return 10  # Very Good
+    elif suppliers > 9:
+        return 5  # Good
+    elif suppliers > 1:
+        return 0  # Normal
+    elif suppliers > 0:
+        return -5  # Bad
+    else:
+        return -10  # Very Bad
+
+
+def _add_availability(out, data):
+    stock = data.get('inStockInventory')
+    suppliers = data.get('suppliersInStock')
+    values = []
+    if type(stock) is int:
+        values.append(_stock_to_availability(stock))
+    if type(suppliers) is int:
+        values.append(_suppliers_to_availability(suppliers))
+    if len(values):
+        out['availability'] = min(values)
+
+
+def _add_prices(out, summary):
+    value = summary.get('medianPrice')
+    if type(value) in [float, int]:
+        out['prices'] = [dict(quantity=1, price=float(value))]
+
+
+def _add_resources(out, data):
+    value = data.get('datasheetUrl')
+    if value is not None:
+        out['resources'] = [
+            dict(name="Datasheet", mediatype="application/pdf", url=value),
+        ]
+
+
+@app.route('/api/v1/parts', methods=['GET'])
+def parts():
+    enabled = _get_config('parts_operational', False)
+    response = make_response(dict(
+        provider_name='Partstack',
+        provider_url='https://partstack.com',
+        provider_logo_url=url_for('parts_static',
+                                  filename='parts-provider-partstack.png',
+                                  _external=True),
+        info_url='https://api.librepcb.org/api',
+        query_url=url_for('parts_query', _external=True) if enabled else None,
+        max_parts=PARTS_MAX_COUNT,
+    ))
+    response.headers['Cache-Control'] = 'max-age=300'
+    return response
+
+
+@app.route('/api/v1/parts/query', methods=['POST'])
+def parts_query():
+    # Get requested parts.
+    payload = request.get_json()
+    parts = payload['parts'][:PARTS_MAX_COUNT]
+
+    # Query parts from information provider.
+    query_response = requests.post(
+        _get_config('parts_query_url'),
+        headers=_build_headers(),
+        json=_build_request(parts),
+        timeout=PARTS_QUERY_TIMEOUT,
+    )
+    query_json = query_response.json()
+    data = query_json.get('data') or {}
+    errors = query_json.get('errors') or []
+    if (len(data) == 0) and (type(query_json.get('message')) is str):
+        errors.append(query_json['message'])
+    for error in errors:
+        app.logger.warning("GraphQL Error: " + str(error))
+
+    # Convert query response data and return it to the client.
+    tx = dict(parts=[])
+    for i in range(len(parts)):
+        mpn = parts[i]['mpn']
+        manufacturer = parts[i]['manufacturer']
+        part_data = data.get('q' + str(i)) or {}
+        product = _get_product(part_data, mpn, manufacturer)
+        part = dict(
+            mpn=mpn,
+            manufacturer=manufacturer,
+            results=0 if product is None else 1,
+        )
+        if product is not None:
+            basic = product.get('basic') or {}
+            summary = part_data.get('summary') or {}
+            _add_pricing_url(part, product)
+            _add_image_url(part, product)
+            _add_status(part, basic)
+            _add_availability(part, summary)
+            _add_prices(part, summary)
+            _add_resources(part, product)
+        tx['parts'].append(part)
+    return tx
+
+
+@app.route('/api/v1/parts/static/<filename>', methods=['GET'])
+def parts_static(filename):
+    return send_from_directory(
+        'static', filename, mimetype='image/png', max_age=24*3600)

demo-request.json

@@ -0,0 +1,8 @@
+{
+    "parts": [
+      {
+        "mpn": "1N4148",
+        "manufacturer": "Vishay"
+      }
+    ]
+}