diff --git a/daq/src/mxlibs3/pshell_client.py b/daq/src/mxlibs3/pshell_client.py index aaa223bd..e81a62c3 100644 --- a/daq/src/mxlibs3/pshell_client.py +++ b/daq/src/mxlibs3/pshell_client.py @@ -1,26 +1,247 @@ -import json -from urllib.parse import quote - +import threading +import time +import sys import requests +import json +try: + from urllib import quote # Python 2 +except ImportError: + from urllib.parse import quote # Python 3 + + +class TimeoutException(Exception): + pass + + +try: + from sseclient import SSEClient +except: + SSEClient = None + +class SSEReceiver: + def __init__(self, url, subscribed_events): + if SSEClient is None: + raise Exception("sseclient library is not installed: server events are not available") + self.url = url + self.events = subscribed_events + self._lock = threading.Lock() + self._stop = threading.Event() + self.session = None + self.client = None + self.debug = False + self._subscribers = {} + self.thread = threading.Thread(target=self.task, kwargs={}) + self.thread.daemon = True + self.thread.start() + + def task(self): + try: + while not self._stop.is_set(): + try: + self.session = requests.Session() + self.client = SSEClient(self.url, session=self.session) + for msg in self.client: + if self.is_closed(): + break + event_name = msg.event or "message" + + if (self.events is None) or (event_name in self.events): + try: + data = json.loads(msg.data) + except: + data = str(msg.data) + #if self.debug: + # print (event_name, data) + with self._lock: + subs = list(self._subscribers.values()) + + for events, callback in subs: + if events is None or event_name in events: + try: + callback(event_name, data) + except Exception as e: + if self.debug: + print(f"[SSEManager] Error in callback {callback}: {e}") + + except IOError as e: + # print(e) + pass + except: + if self.debug: + print("Error:", sys.exc_info()[1]) + finally: + self._close_client() + if self.is_closed(): + break + else: + time.sleep(1.0) + finally: + if self.debug: + print("Exit SSE loop task") + + def subscribe(self, callback, events=None): + """ + Subscribe to SSE events. + + Args: + callback: function(event_name, data) + events: None (all events), str (one event), or list[str] (multiple events) + """ + if isinstance(events, str): + events = [events] + if events is not None: + events = set(events) + + with self._lock: + self._subscribers[id(callback)] = (events, callback) + + def unsubscribe(self, callback): + """Unsubscribe a previously subscribed callback.""" + with self._lock: + self._subscribers.pop(id(callback), None) + + def wait_events(self, events={}, timeout=-1): + """Wait any of the events matching the value (value None for any). + + Args: + events (dict event name->value) + timeout: + Returns: + (event, value) or None if timeout + """ + rx = {} + condition = threading.Condition() + def callback(name, value): + with condition: + rx[name] = value + condition.notify_all() + + self.subscribe(callback, events.keys()) + try: + start = time.time() + with condition: + while True: + for name in events.keys(): + if name in rx.keys(): + values, rx_value = events[name], rx[name] + if values is not None and type(values) is not list: + values = [values] + if values is None or rx_value in values: + return name,rx_value + + remaining = None + if timeout >= 0: + remaining = max(0, timeout - (time.time() - start)) + if remaining <= 0: + return None + condition.wait(timeout=remaining) + finally: + self.unsubscribe(callback) + + + def _close_client(self): + self.client = None + """ + if self.client is not None: + try: + if hasattr(self.client.resp, "raw"): + conn = getattr(self.client.resp.raw, "_connection", None) + if conn and hasattr(conn, "sock") and conn.sock: + conn.sock.shutdown(2) + conn.sock.close() + self.client.resp.close() + self.client = None + except: + pass + """ + if self.session is not None: + try: + self.session.close() + self.session = None + except: + pass + + def close(self): + self._stop.set() + self._close_client() + if self.debug: + print("closed") + + def is_closed(self): + return self._stop.is_set() + class PShellClient: def __init__(self, url): + if not url.endswith('/'): + url = url + "/" self.url = url self.sse_event_loop_thread = None - self.subscribed_events = None - self.event_callback = None + self.sse_client = None + self.plot_defaults = {"format": "png", "width": 600, "height": 400} + self.debug = False + self.polling_interval = 0.1 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def _get(self, url, stream=False): + url = self.url + url + if self.debug: + print("GET " + url) + return requests.get(url=url, stream=stream) + + def _put(self, url, json_data=None): + url = self.url + url + if self.debug: + print("PUT " + url + " -> " + json.dumps(json_data)) + return requests.put(url=url, json=json_data) + + def _del(self, url): + url = self.url + url + if self.debug: + print("DEL " + url) + return requests.delete(url=url) def _get_response(self, response, is_json=True): - if response.status_code != 200: - raise Exception(response.text) + if self.debug == True or self.debug == "rx": + print(" -> " + str(response.status_code) + ((" - " + response.text) if self.debug == "rx" else "")) + try: + response.raise_for_status() + except: + print(response.text) + raise 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) + response.raise_for_status() return response.raw.read() + def get_plot_defaults(self): + """Return plot default properties. + + Args: + + Returns: + Dictionary + + """ + return self.plot_defaults.copy() + + def set_plot_defaults(self, defaults): + """Update plot default properties. + + Args: + Dictionary that will updated into the plot default properties. + Returns: + + """ + self.plot_defaults.update(defaults) + def get_version(self): """Return application version. @@ -30,7 +251,7 @@ class PShellClient: String with application version. """ - return self._get_response(requests.get(url=self.url + "/version"), False) + return self._get_response(self._get("version"), False) def get_config(self): """Return application configuration. @@ -40,7 +261,7 @@ class PShellClient: Returns: Dictionary. """ - return self._get_response(requests.get(url=self.url + "/config")) + return self._get_response(self._get("config")) def get_state(self): """Return application state. @@ -50,7 +271,39 @@ class PShellClient: Returns: String: Invalid, Initializing,Ready, Paused, Busy, Disabled, Closing, Fault, Offline """ - return self._get_response(requests.get(url=self.url + "/state")) + return self._get_response(self._get("state")) + + def wait_state(self, state, timeout=-1): + """Wait application state equals. + + Args: + state (string or list of strings) + timeout(number) wait timeout in seconds. If less or equal 0 then wait forever. + Returns: + """ + if type(state) == str: + state = [state] + start = time.time() + while self.get_state() not in state: + if (timeout >= 0) and ((time.time() - start) > timeout): + raise TimeoutException(f"Timeout waiting state {state}") + time.sleep(self.polling_interval) + + def wait_state_not(self, state, timeout=-1): + """Wait application state different than. + + Args: + state (string or list of strings) + timeout(number) wait timeout in seconds. If less or equal 0 then wait forever. + Returns: + """ + if type(state) == str: + state = [state] + start = time.time() + while self.get_state() in state: + if (timeout >= 0) and ((time.time() - start) > timeout): + raise TimeoutException(f"Timeout waiting state not {state}") + time.sleep(self.polling_interval) def get_logs(self): """Return application logs. @@ -62,7 +315,7 @@ class PShellClient: Format of each log: [date, time, origin, level, description] """ - return self._get_response(requests.get(url=self.url + "/logs")) + return self._get_response(self._get("logs")) def get_history(self, index): """Access console command history. @@ -74,7 +327,7 @@ class PShellClient: History entry """ - return self._get_response(requests.get(url=self.url + "/history/" + str(index)), False) + return self._get_response(self._get("history/" + str(index)), False) def get_script(self, path): """Return script. @@ -86,7 +339,7 @@ class PShellClient: String with file contents. """ - return self._get_response(requests.get(url=self.url + "/script/" + str(path)), False) + return self._get_response(self._get("script/" + str(path)), False) def get_devices(self): """Return global devices. @@ -98,7 +351,7 @@ class PShellClient: Format of each device record: [name, type, state, value, age] """ - return self._get_response(requests.get(url=self.url + "/devices")) + return self._get_response(self._get("devices")) def abort(self, command_id=None): """Abort execution of command @@ -111,9 +364,29 @@ class PShellClient: """ if command_id is None: - requests.get(url=self.url + "/abort") + self._get("abort") else: - return requests.get(url=self.url + "/abort/" + str(command_id)) + return self._get("abort/" + str(command_id)) + + def pause(self): + """Pause execution of command + + Args: + + Returns: + + """ + self._get("pause") + + def resume(self): + """Resume execution of command + + Args: + + Returns: + + """ + self._get("resume") def reinit(self): """Reinitialize the software. @@ -123,7 +396,7 @@ class PShellClient: Returns: """ - requests.get(url=self.url + "/reinit") + self._get("reinit") def stop(self): """Stop all devices implementing the 'Stoppable' interface. @@ -133,7 +406,7 @@ class PShellClient: Returns: """ - requests.get(url=self.url + "/stop") + self._get("stop") def update(self): """Update all global devices. @@ -143,7 +416,7 @@ class PShellClient: Returns: """ - requests.get(url=self.url + "/update") + self._get("update") def eval(self, statement): """Evaluates a statement in the interpreter. @@ -158,7 +431,7 @@ class PShellClient: If an exception is produces in the interpretor, it is re-thrown here. """ statement = quote(statement) - return self._get_response(requests.get(url=self.url + "/eval/" + statement), False) + return self._get_response(self._get("eval/" + statement), False) def run(self, script, pars=None, background=False): """Executes script in the interpreter. @@ -174,10 +447,7 @@ class PShellClient: If an exception is produces in the interpretor, it is re-thrown here. """ return self._get_response( - requests.put( - url=self.url + "/run", json={"script": script, "pars": pars, "background": background, "async": False} - ) - ) + self._put("run", {"script": script, "pars": pars, "background": background, "async": False})) def start_eval(self, statement): """Starts evaluation of a statement in the interpreter. @@ -191,7 +461,58 @@ class PShellClient: Command id (int), which is used to retrieve command execution status/result (get_result). """ statement = quote(statement) - return int(self._get_response(requests.get(url=self.url + "/evalAsync/" + statement), False)) + return int(self._get_response(self._get("evalAsync/" + statement), False)) + + def eval_json(self, statement): + """Evaluates a statement in the interpreter. + Args: + statement(str): input statement + + Returns: + Return object decoded from JSON string + """ + statement = quote(statement) + return self._get_response(self._get("eval-json/" + statement), True) + + def eval_then(self, statement, on_success=True, on_exception=True): + """Set a next execution stage for the interpreter - the statement is executed + after the foreground task concludes, keeping application state busy. + + Args: + statement(str): statement for next execution stage + on_successs(bool): statement is executed if foreground task completes successfully. + on_exception(bool): statement is executed if foreground task throws exception. + """ + return self._get_response( + self._put("then", {"statement": statement, "onSuccess": on_success, "onException": on_exception})) + + def run_then(self, script, pars=None, on_success=True, on_exception=True): + """Set a next execution stage for the interpreter - the script is executed + after the foreground task concludes, keeping application state busy. + + Args: + script(str): name of the script (absolute or relative to the script base folder). Extension may be omitted. + pars(optional, list or dict): if a list is given, it sets sys.argv for the script. + If a dict is given, it sets global variable for the script. + on_successs(bool): statement is executed if foreground task completes successfully. + on_exception(bool): statement is executed if foreground task throws exception. + """ + cmd = f"run('{script}', {pars})" + return self.eval_then(cmd, on_success, on_exception) + + def set_var(self, name, value): + """Sets interpreter variable. + Args: + name(str): variable name + value(obj): value - must be JSON compatible + + Returns: + + """ + data = {} + data["name"] = name + data["value"] = value + return self._get_response(self._put("set-var", data), False) def start_run(self, script, pars=None, background=False): """Starts execution of a script in the interpreter. @@ -205,14 +526,8 @@ class PShellClient: Returns: Command id (int), which is used to retrieve command execution status/result (get_result). """ - return int( - self._get_response( - requests.put( - url=self.url + "/run", - json={"script": script, "pars": pars, "background": background, "async": True}, - ) - ) - ) + return int(self._get_response( + self._put("run", {"script": script, "pars": pars, "background": background, "async": True}))) def get_result(self, command_id=-1): """Gets status/result of a command executed asynchronously (start_eval and start_run). @@ -227,7 +542,7 @@ class PShellClient: 'return' (obj): if status equals 'completed', holds return value of script (start_run) or console return (start_eval) """ - return self._get_response(requests.get(url=self.url + "/result/" + str(command_id))) + return self._get_response(self._get("result/" + str(command_id))) def help(self, input=""): """Returns help or auto-completion strings. @@ -243,7 +558,7 @@ class PShellClient: List """ - return self._get_response(requests.get(url=self.url + "/autocompletion/" + input)) + return self._get_response(self._get("autocompletion/" + input)) def get_contents(self, path=None): """Returns contents of data path. @@ -259,9 +574,7 @@ class PShellClient: List of contents """ - return self._get_response( - requests.get(url=self.url + "/contents" + ("" if path is None else ("/" + path))), False - ) + return self._get_response(self._get("contents" + ("" if path is None else ("/" + path))), False) def get_data(self, path, type="txt"): """Returns data on a given path. @@ -277,16 +590,205 @@ class PShellClient: """ if type == "json": - return self._get_response(requests.get(url=self.url + "/data-json/" + path), True) + return self._get_response(self._get("data-json/" + path), True) elif type == "bin": - return self._get_binary_response(requests.get(url=self.url + "/data-bin/" + path, stream=True)) + return self._get_binary_response(self._get("data-bin/" + path, stream=True)) + elif type == "bs": + from collections import OrderedDict + bs = self._get_binary_response(self._get("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) + return self._get_response(self._get("data" + ("" if path is None else ("/" + path))), False) + + def get_data_attrs(self, path): + return self._get_response(self._get("data-attr/" + path), True) + + def get_data_info(self, path): + return self._get_response(self._get("data-info/" + path), True) + + def get_scan_data(self, layout, path, group, device, type="txt"): + """Returns scan data of a device. + + Args: + layout(str): data layout + path(str): scan path + group(str): scan group + device(str): device name + type(optional, str): txt, "json", "bin" + + Returns: + Data accordind to selected format. + + """ + if layout is None or layout.strip() == "" or path is None: + raise Exception("Invalid scan persistence path or layout") + path = path.replace("/", "
") + path = path.replace("|", "

") + group = group.replace("/", "
") + layout = layout.replace(".", "
") + url = layout + "/" + path + "/" + group + "/" + device + if type == "json": + url = "scandata-json/" + url + return self._get_response(self._get(url), True) + elif type == "bin": + url = "scandata-bin/" + url + return self._get_binary_response(self._get(url, stream=True)) + url = "scandata/" + url + return self._get_response(self._get(url), False) + + def get_plot_contexts(self): + """Return list of plot contexts + + Args: + + Returns: + List of names + + """ + return self._get_response(self._get("plots")) + + def delete_plot_context(self, title): + """ + Delete a plotting context. + + Args: + title(str): name of the plotting context + + Returns: + + """ + return self._get_response(self._del("plots/" + title), False) + + def get_num_plots(self, title=None): + """Return number of plots in a given plotting context. + + Args: + title(str): name of the plotting context + + Returns: + Number of plots + + """ + if title is None: + title = "null" + return int(self._get_response(self._get("plots/" + title))) + + def get_plot(self, title=None, index=0, format="png", width=None, height=None): + """Return a plot as a given image type. + + Args: + title(str): name of the plotting context + index(int): plot index (0-based) + format(str): plot format ("jpg", "png", "gif", "tif") + width(int): plot width (if 0 gets plot staddard size) + height(int): plot height (if 0 gets plot staddard size) + Returns: + Image file byte array + + """ + if title is None: + title = "null" + if format is None: + format = self.plot_defaults["format"] + if width is None: + width = self.plot_defaults["width"] + if height is None: + height = self.plot_defaults["height"] + + url = "plot/" + title + "/" + str(index) + "/" + format + "/" + str(width) + "/" + str(height) + return self._get_binary_response(self._get(url, stream=True)) def print_logs(self): - for log_line in self.get_logs(): - print("%s %s %-20s %-8s %s" % tuple(log_line)) + for l in self.get_logs(): + print("%s %s %-20s %-8s %s" % tuple(l)) def print_devices(self): - for log_line in self.get_devices(): - print("%-16s %-32s %-10s %-32s %s" % tuple(log_line)) + for l in self.get_devices(): + print("%-16s %-32s %-10s %-32s %s" % tuple(l)) + + def print_help(self, input=""): + for l in self.help(input): + print(l) + + def _get_sse(self): + if self.sse_client is None: + self.sse_client = SSEReceiver(self.url + "events", None) + self.sse_client.debug = self.debug + return self.sse_client + + def subscribe(self, callback=None, events=None): + """ + Subscribe to SSE events. + + Args: + callback: function(event_name, data), If None, calls self.on_event + events: None (all events), str (one event), or list[str] (multiple events) + + Usage example: + def on_event(name, value): + if name == "state": + print ("State changed: ", value) + elif name == "record": + print ("Received scan record: ", value) + + pc.subscribe(["state", "record"], on_event) + """ + if callback == None: + callback = self.on_event + self._get_sse().subscribe(callback, events) + + def unsubscribe(self, callback): + """Unsubscribe a previously subscribed callback.""" + self._get_sse().unsubscribe(callback) + + def wait_events(self, events={}, timeout=-1): + """Wait any of the events matching the value (value None for any). + + Args: + events (dict event name->value) + timeout(number) wait timeout in seconds. If less or equal 0 then wait forever. + Returns: + (event, value) or raises TimeoutException + + Usage example: + def on_event(name, value): + if name == "state": + print ("State changed: ", value) + elif name == "record": + print ("Received scan record: ", value) + + pc.subscribe(["state", "record"], on_event) + + """ + ret = self._get_sse().wait_events(events, timeout) + if ret is None: + raise TimeoutException(f"Timeout waiting for events {events}") + return ret + + def on_event(self, name, value): + """ + Default event callback + Args: + name: event name. + value: event value. + """ + pass + + def close(self): + if self.sse_client is not None: + self.sse_client.close() +