parteye

script to parse data from QR and Bar codes and post it to Partkeepr.
git clone https://git.e1e0.net/parteye.git
Log | Files | Refs | README

commit b33d064e907ab7f10c260ef45446a8414483f074
Author: Paco Esteban <paco@onna.be>
Date:   Thu,  8 Mar 2018 19:55:31 +0100

initial commit

Diffstat:
A.gitignore | 1+
AREADME.md | 64++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Abeep.wav | 0
Aparteye.py | 248+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Arequest.json | 125+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
5 files changed, 438 insertions(+), 0 deletions(-)

diff --git a/.gitignore b/.gitignore @@ -0,0 +1 @@ +config.ini diff --git a/README.md b/README.md @@ -0,0 +1,64 @@ +# parteye.py # + +Acts as a glue between [zbar](http://zbar.sourceforge.net/) and +[Partkeepr](https://www.partkeepr.org/). +A good way of repurpose an old webcam ! + +For now it only works with [TME](https://www.tme.eu/), the polish electronics +parts distributor. On their orders, they place a little QR code on each +package with some info. With that and the distributor's API ... no need to type +anymore ! + +I'll probably extend it to work with [Farnell/Element14](http://farnell.com) if +I place an order soon, as they too have a reasonably good API and maybe they do +something similar with their packages + +# requirements # + +As far as I know, python3 and [requests +](http://docs.python-requests.org/en/master/) library. + +# usage # + +First configure the thing. Create a file named `config.ini` in the same folder +as the script. It should look like this: + +``` +[tme] +token = your_tme_api_token +secret = your_tme_api_secret + +[partkeepr] +user = your_pk_username +pwd = your_pk_password +url = http://your_pk_url + +``` + +And then basically, pass the output of zbar to the script via pipe: + +``` +zbarcam --raw /dev/video0 | ./parteye.py +``` + +On [Arch Linux](https://www.archlinux.org/) you may need a little bit of +tweaking because of zbar ...: + +``` +LD_PRELOAD=/usr/lib/libv4l/v4l1compat.so zbarcam --raw /dev/video0 | ./parteye.py +``` + +# notes # + +The script places the new part under a category at root level named `00 - QR-reader`. +For the location uses one named _"Sense-Ordenar"_. +You can change that on request.json. + +Partkeepr API doc is literally non-existent ... So if you want to tweak this, +you'll have to make good use of _Developer tools_ on your browser. + +# license # + +For what it's worth .. this goes under MIT license. + +Beep sound by _kalisemorrison_. Find more [here](https://freesound.org/people/kalisemorrison/) diff --git a/beep.wav b/beep.wav Binary files differ. diff --git a/parteye.py b/parteye.py @@ -0,0 +1,248 @@ +#! /usr/bin/env python +# -*- coding: utf-8 -*- +# vim:fenc=utf-8 +# +# Copyright © 2018 Paco Esteban <paco@onna.be> +# +# Distributed under terms of the MIT license. + +import base64 +import collections +import configparser +import hashlib +import hmac +import json +import re +import sys +import urllib.parse +import urllib.request +from datetime import datetime +from subprocess import run + +import requests +from requests.auth import HTTPBasicAuth + +config = configparser.ConfigParser() +config.read('config.ini') + + +def tme_api_call(action, params): + """calls TME API and returns the json response + + it handles all the hmac signature and all that ... + :action: string action url after the main TME api domain + :params: dict with params for the request + """ + + api_url = 'https://api.tme.eu/' + action + '.json' + params['Token'] = config["tme"]["token"] + + # params need to be ordered + params = collections.OrderedDict(sorted(params.items())) + encoded_params = urllib.parse.urlencode(params, '') + signature_base = ('POST' + '&' + urllib.parse.quote(api_url, '') + '&' + + urllib.parse.quote(encoded_params, '')) + + api_signature = base64.encodestring( + hmac.new( + bytes(config["tme"]["secret"], 'UTF-8'), + bytes(signature_base, 'UTF-8'), hashlib.sha1).digest()).rstrip() + params['ApiSignature'] = api_signature + + try: + r = requests.post(api_url, params) + r.raise_for_status() + except requests.exceptions.HTTPError as err: + print(err) + sys.exit(1) + + return r.json() + + +def pk_api_call(method, url, **kwargs): + """calls Partkeepr API + + :method: requst method + :url: part of the url to call (without base) + :data: tata to pass to the request if any + :returns: requests object + + """ + pk_user = config["partkeepr"]["user"] + pk_pwd = config["partkeepr"]["pwd"] + pk_url = config["partkeepr"]["url"] + try: + r = requests.request( + method, + pk_url + url, + **kwargs, + auth=HTTPBasicAuth(pk_user, pk_pwd), + ) + r.raise_for_status() + except requests.exceptions.HTTPError as err: + print(err) + sys.exit(1) + + return r + + +def read_in(): + """ + Reads a line from stdin and returns it + """ + return sys.stdin.readline().strip() + + +def parse_tme(raw_in): + """ + Parses the raw data comming from stdin (barcode reader) + :raw_in: string in the form: + QTY:1 PN:HA50151V4 MFR:SUNON MPN:HA50151V4-000U-999 + PO:5094268/9 https://www.tme.eu/details/HA50151V4 + Where : + FIELD NAME Desc + 0 QTY Quantity + 1 PN Part Number + 2 MFR Manufacturer + 3 MPN Manufacturer part number + 4 PO Order Number (at TME) + 5 URL Url of the product at vendor(TME) + """ + part = { + 'PN': raw_in[1].split(":")[1], + 'Quantity': raw_in[0].split(":")[1], + 'Files': [], + 'Case': '', + 'PO': raw_in[4].split(":")[1] + } + + params = {'SymbolList[0]': part["PN"], 'Country': 'ES', 'Language': 'EN'} + + run(["/usr/bin/play", "-q", "./beep.wav"]) + print("Looking for part: {}".format(part["PN"])) + + # first we get the description of the part + product = tme_api_call('Products/GetProducts', params) + part["Desc"] = product["Data"]["ProductList"][0]["Description"] + + # then we get the footprint name if any + parameters = tme_api_call('Products/GetParameters', params) + symbols = parameters["Data"]["ProductList"][0]["ParameterList"] + for param in symbols: + if param["ParameterId"] == "35" or param["ParameterId"] == "2932": + part["Case"] = param["ParameterValue"] + + # finally we get all pdfs related to this part + files = tme_api_call('Products/GetProductsFiles', params) + docs = files["Data"]["ProductList"][0]["Files"]["DocumentList"] + for d in docs: + if d["DocumentUrl"][-3:] == "pdf": + part["Files"].append("https:" + d["DocumentUrl"]) + + part["Files"] = list(set(part["Files"])) + + return part + + +def generate_footprint(fp): + """Checks for footprint if it exists + + :fp: string footprint name + :returns: json structure to attach to part creation or None + """ + if len(fp) == 0: + return None + + params = { + 'filter': + '{{"property":"name","operator":"=","value":"{}"}}'.format(fp) + } + r = pk_api_call('get', '/api/footprints', params=params) + + rj = r.json() + if len(rj["hydra:member"]) > 0: + return rj["hydra:member"][0] + + return None + + +def upload_attachments(files): + """Check if there's any file for the part and uploads it to partkeepr + + :files: files arry + :returns: json structure to attach to part creation or None + """ + uploads = [] + if len(files) > 0: + for f in files: + r = pk_api_call( + 'post', '/api/temp_uploaded_files/upload', data={'url': f}) + uploads.append(r.json()["response"]) + + return uploads + + return None + + +def insert_part(part): + """ Inserts the part in partkeepr. + If found, it just adds stock + + :part: dict representing the part + """ + # we first look for the part name + params = { + 'filter': + '{{"property":"name","operator":"=","value":"{}"}}'.format(part["PN"]) + } + r = pk_api_call('get', '/api/parts', params=params) + + rj = r.json() + # if found, just add stock and return + if len(rj["hydra:member"]) > 0: + r = pk_api_call( + 'put', + '{}/addStock'.format(rj["hydra:member"][0]["@id"]), + data={ + 'quantity': part["Quantity"], + 'comment': part["PO"] + }) + print("{} - Increased stock in {} units".format( + part["PN"], part["Quantity"])) + return + + # if not, we prepare the json payload for insert + with open('request.json') as json_data: + d = json.load(json_data) + + date = datetime.now() + + d["name"] = part["PN"] + d["description"] = part["Desc"] + d["stockLevels"][0]["stockLevel"] = part["Quantity"] + d["createDate"] = date.strftime("%Y-%m-%dT%H:%M:%S.000Z") + d["footprint"] = generate_footprint(part["Case"]) + # here files are uploaded to tmp + d["attachments"] = upload_attachments(part["Files"]) + + r = pk_api_call('post', '/api/parts', json=d) + r.raise_for_status() + + print("Part {} ({} new units) loaded to Partkeepr".format( + part["PN"], part["Quantity"])) + + +while True: + line = read_in() # read from stdin + + r = re.compile(r'^QTY:\d+ PN:.*tme\.eu.*') + if r.match(line) is not None: # is this from TME ? + my_part = parse_tme(line.split(" ")) + else: + if not line: + print("bye !") + sys.exit(0) + print("Unrecognized raw data format") + sys.exit(1) + + insert_part(my_part) diff --git a/request.json b/request.json @@ -0,0 +1,125 @@ +{ + "name": "foo", + "description": "bar", + "comment": "", + "stockLevel": 0, + "minStockLevel": 0, + "averagePrice": 0, + "status": "", + "needsReview": false, + "partCondition": "", + "createDate": "2018-03-04T15:51:57.778Z", + "internalPartNumber": "", + "removals": false, + "lowStock": false, + "category": { + "@id": "/api/part_categories/56", + "@type": "PartCategory", + "parent": "/api/part_categories/1", + "children": [], + "categoryPath": "Root Category ➤ 00 - QR-reader", + "expanded": true, + "name": "00 - QR-reader", + "description": "", + "parentId": "/api/part_categories/1", + "index": 0, + "depth": 2, + "expandable": true, + "checked": null, + "leaf": false, + "cls": "", + "iconCls": "", + "icon": "", + "root": 0, + "isLast": false, + "isFirst": true, + "allowDrop": true, + "allowDrag": true, + "loaded": true, + "loading": false, + "href": "", + "hrefTarget": "", + "qtip": "", + "qtitle": "", + "qshowDelay": 0, + "visible": true, + "text": "", + "lft": 0, + "rgt": 0, + "lvl": 0 + }, + "partUnit": { + "@id": "/api/part_measurement_units/1", + "@type": "PartMeasurementUnit", + "name": "Pieces", + "shortName": "pcs", + "default": true + }, + "footprint": null, + "storageLocation": { + "@id": "/api/storage_locations/30", + "@type": "StorageLocation", + "name": "Sense-Ordenar", + "image": null, + "categoryPath": "Root Category", + "category": { + "@id": "/api/storage_location_categories/1", + "@type": "StorageLocationCategory", + "parent": null, + "categoryPath": "Root Category", + "expanded": true, + "name": "Root Category", + "description": "", + "parentId": null, + "index": -1, + "depth": 0, + "expandable": true, + "checked": null, + "leaf": false, + "cls": "", + "iconCls": "", + "icon": "", + "root": 0, + "isLast": false, + "isFirst": false, + "allowDrop": true, + "allowDrag": true, + "loaded": false, + "loading": false, + "href": "", + "hrefTarget": "", + "qtip": "", + "qtitle": "", + "qshowDelay": 0, + "children": null, + "visible": true, + "text": "", + "lft": 0, + "rgt": 0, + "lvl": 0 + } + }, + "partPartKeeprPartBundleEntityPartAttachments": [], + "attachments": [], + "partPartKeeprPartBundleEntityPartDistributors": [], + "distributors": [], + "partPartKeeprPartBundleEntityPartImages": [], + "partPartKeeprPartBundleEntityPartManufacturers": [], + "manufacturers": [], + "partPartKeeprPartBundleEntityPartParameters": [], + "parameters": [], + "partPartKeeprProjectBundleEntityProjectParts": [], + "projectParts": [], + "partPartKeeprStockBundleEntityStockEntries": [], + "stockLevels": [ + { + "stockLevel": 100, + "price": 0, + "dateTime": null, + "correction": false, + "comment": "", + "user": null + } + ], + "partPartKeeprProjectBundleEntityProjectReports": [] +}