121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
import requests
|
|
import json
|
|
|
|
|
|
class PShellClient:
|
|
def __init__(self, url):
|
|
self.url = url
|
|
|
|
def get_response(self, response, is_json=True):
|
|
if response.status_code != 200:
|
|
raise Exception(response.text)
|
|
return json.loads(response.text) if is_json else response.text
|
|
|
|
def get_binary_response(self, response):
|
|
if response.status_code != 200:
|
|
raise Exception(response.text)
|
|
return response.raw.read()
|
|
|
|
def get_version(self):
|
|
return self.get_response(requests.get(url=self.url+"/version"), False)
|
|
|
|
def get_config(self):
|
|
return self.get_response(requests.get(url=self.url+"/config"))
|
|
|
|
def get_state(self):
|
|
return self.get_response(requests.get(url=self.url+"/state"))
|
|
|
|
def get_logs(self):
|
|
return self.get_response(requests.get(url=self.url+"/logs"))
|
|
|
|
def get_history(self, index):
|
|
return self.get_response(requests.get(url=self.url+"/history/"+str(index)), False)
|
|
|
|
def get_script(self, path):
|
|
return self.get_response(requests.get(url=self.url+"/script/"+str(path)), False)
|
|
|
|
def get_devices(self):
|
|
return self.get_response(requests.get(url=self.url+"/devices"))
|
|
|
|
def abort(self, command=None):
|
|
if command is None:
|
|
requests.get(url=self.url+"/abort")
|
|
else:
|
|
return requests.get(url=self.url+"/abort/"+str(command))
|
|
|
|
def reinit(self):
|
|
requests.get(url=self.url+"/reinit")
|
|
|
|
def stop(self):
|
|
requests.get(url=self.url+"/stop")
|
|
|
|
def update(self):
|
|
requests.get(url=self.url+"/update")
|
|
|
|
def eval(self,statement):
|
|
return self.get_response(requests.get(url=self.url+"/eval/"+statement), False)
|
|
|
|
def run(self,script, pars=None, background=False, async=False):
|
|
return self.get_response(requests.put(url=self.url+"/run", json={"script":script, "pars":pars, "background":background, "async":async }))
|
|
|
|
def help(self, input = "<builtins>"):
|
|
return self.get_response(requests.get(url=self.url+"/autocompletion/" + input))
|
|
|
|
def get_contents(self, path=None):
|
|
return self.get_response(requests.get(url=self.url+ "/contents" + ("" if path is None else ( "/"+path))), False)
|
|
|
|
def get_data(self, path=None, type="txt"):
|
|
if type == "json":
|
|
return self.get_response(requests.get(url=self.url+ "/data-json/"+path), True)
|
|
elif type == "bin":
|
|
return self.get_binary_response(requests.get(url=self.url+"/data-bin/"+path, stream=True))
|
|
elif type == "bs":
|
|
from collections import OrderedDict
|
|
bs = self.get_binary_response(requests.get(url=self.url+"/data-bs/"+path, stream=True))
|
|
index=0
|
|
msg = []
|
|
for i in range(4):
|
|
size =int.from_bytes(bs[index:index+4], byteorder='big', signed=False)
|
|
index=index+4
|
|
msg.append(bs[index:index+size])
|
|
index=index+size
|
|
[main_header, data_header, data, timestamp] = msg
|
|
main_header = json.loads(main_header, object_pairs_hook=OrderedDict)
|
|
data_header = json.loads(data_header, object_pairs_hook=OrderedDict)
|
|
channel = data_header["channels"][0]
|
|
channel["encoding"] = "<" if channel.get("encoding", "little") else ">"
|
|
from bsread.data.helpers import get_channel_reader
|
|
channel_value_reader = get_channel_reader(channel)
|
|
return channel_value_reader(data)
|
|
|
|
return self.get_response(requests.get(url=self.url+ "/data" + ("" if path is None else ( "/"+path))), False)
|
|
|
|
def print_logs(self):
|
|
for l in self.get_logs():
|
|
print ("%s %s %-20s %-8s %s" % tuple(l))
|
|
|
|
def print_devices(self):
|
|
for l in self.get_devices():
|
|
print ("%-16s %-32s %-10s %-32s %s" % tuple(l))
|
|
|
|
def print_help(self, input = "<builtins>"):
|
|
for l in self.help(input):
|
|
print (l)
|
|
|
|
pshell = PShellClient('http://Alexandres-MBP.psi.ch:8080')
|
|
print (pshell.get_state())
|
|
print (pshell.get_version())
|
|
pshell.print_devices()
|
|
print (pshell.get_config())
|
|
pshell.print_logs()
|
|
print (pshell.get_history(1))
|
|
print pshell.eval("1+1&")
|
|
print (pshell.get_script("test/DataLink.py"))
|
|
pars = ["S30CB09-DWSC440","X1", [-200, 200, -200, 200],3,200,["BPM1", "BPM2"],["BLM1", "BLM2"],10,None,False]
|
|
print pshell.run('test/WireScanMock.py', pars, True)
|
|
pshell.print_help()
|
|
print (pshell.get_data("2018_01/20180112/20180112_134943_Tutorial_12_AreaScan.h5 /scan 1/ai1", "json")["data"]["sliceData"])
|
|
|
|
|
|
|