Files
dev/script/Inventory.py
2018-01-19 10:56:53 +01:00

153 lines
4.2 KiB
Python
Executable File

import requests
def inventory_request(url, attributes):
"""
Thin wrapper for PSI Inventory system (I4). This wrapper does nothing
more than sanitises output (decompresses dict) and takes care of dict
to json conversion for attributes and json to dict conversion for
response.
Use: https://inventory.psi.ch/#action=Introspection for details
Example usage:
Query:
query = {"query": [
{"Field": "Holy List Name", "Operator": "Is", "Value": holy_name},
{"Field": 'Type', "Operator": "Is", "Value": type}
], "columns": ["Label"]}
attr = {"search": query}
r = inventory_request('https://inventory.psi.ch/DataAccess.asmx/FindObjects', attr)
Fetch atributes for part:
attributes = inventory_request('https://inventory.psi.ch/DataAccess.asmx/GetPartAttributes', {'psiLabel': part_id})
:param url: Inventory URL endpoint
:param attributes: Python dict of attributes
:return:
"""
r = requests.post(url, json=attributes)
if not r.ok:
raise ConnectionError("Inventory returned error: ", r.json())
return r.json()['d']
def find_partid_by_holy(holy_name, type="DSCR"):
"""
:param holy_name:
:param type:
:return: Returns first partId that matches holy_name and type
"""
query = {"query": [
{"Field": "Holy List Name", "Operator": "Is", "Value": holy_name},
{"Field": 'Type', "Operator": "Is", "Value": type}
], "columns": ["Label"]}
attr = {"search": query}
r = inventory_request('https://inventory.psi.ch/DataAccess.asmx/FindObjects', attr)
try:
id = r['Rows'][0][0]
except:
return None
return id
def find_all_by_type(type="DSCR", column = "Holy List Name"):
"""
:param type: Type to match
:param column: Label, Holy List Name, ...
:return: List
"""
query = {"query": [
{"Field": 'Type', "Operator": "Is", "Value": type}
], "columns": [column]}
attr = {"search": query}
r = inventory_request('https://inventory.psi.ch/DataAccess.asmx/FindObjects', attr)
rows = []
for i in r['Rows']:
rows.append(i[0])
return rows
def get_part_attributes_from_inventory(part_label = None, holy_name = None):
if holy_name:
part_label = find_partid_by_holy(holy_name, "DSCR")
if not part_label: raise RuntimeError("Could not find inventory part for >>{}<<".format(holy_name))
return inventory_request('https://inventory.psi.ch/DataAccess.asmx/GetPartAttributes', {'psiLabel': part_label})
def get_logbook_from_inventory(part_label = None, holy_name = None):
import re
attributes = get_part_attributes_from_inventory(part_label, holy_name)
elog_link = None
for a in attributes:
if a['Name'] == 'Elog':
try:
elog_link = a['Value']
elog_link = re.findall('http(?:s|)://[^\s]*')
if elog_link:
elog_link = elog_link[0]
else:
elog_link = None
except TypeError:
elog_link = None
return elog_link
def get_calib_from_invetory(part_label = None, holy_name = None):
horizontal_dist = 0.0
vertical_dist = 0.0
horizontal_tilt = 0.0
vertical_tilt = 0.0
attributes = get_part_attributes_from_inventory(part_label, holy_name)
for a in attributes:
if a['Name'] == 'Crystal angle in x (e-beam system) [deg]':
try:
horizontal_tilt = float(a['Value'])
except TypeError:
pass
if a['Name'] == 'Crystal angle in y (e-beam system) [deg]':
try:
vertical_tilt = float(a['Value'])
except TypeError:
pass
if a['Name'] == 'Mark distance in x (e-beam system) [mm]':
try:
horizontal_dist = float(a['Value']) * 1e3
except TypeError:
pass
if a['Name'] == 'Mark distance in y (e-beam system) [mm]':
try:
vertical_dist = float(a['Value']) * 1e3
except TypeError:
pass
return horizontal_dist, vertical_dist, horizontal_tilt, vertical_tilt
print get_calib_from_invetory(None, "SINEG01-DSCR190")