core_tools/GUI/script_runner/commands.py

@@ -2,14 +2,27 @@ import inspect
 import logging
 import os
 from enum import Enum
+from typing import Any
 
 from abc import ABC, abstractmethod
 
 try:
     from spyder_kernels.customize.spydercustomize import runcell
+    runcell_version = 5
 except Exception:
     runcell = None
 
+try:
+    from IPython import get_ipython  # pyright: ignore
+
+    ipython = get_ipython()
+    if ipython is not None:
+        runcell = ipython.magics_manager.magics['line']['runcell']
+        runcell_version = 6
+except KeyError:
+    # no runcell magic
+    pass
+
 
 logger = logging.getLogger(__name__)
 
@@ -48,10 +61,20 @@ class Cell(Command):
             raise Exception('runcell not available. Upgrade to Spyder 4+ to use Cell()')
 
     def __call__(self):
-        command = f"runcell({self.cell}, '{self.python_file}')"
-        # print(command)
-        logger.info(command)
-        runcell(self.cell, self.python_file)
+        if runcell_version == 5:
+            command = f"runcell({self.cell}, '{self.python_file}')"
+            print(command)
+            logger.info(command)
+            runcell(self.cell, self.python_file)
+        elif runcell_version == 6:
+            if isinstance(self.cell, int):
+                command = f"-i {self.cell} {self.python_file}"
+            else:
+                command = f"-n '{self.cell}' {self.python_file}"
+            print("runcell", command)
+            logger.info(f"runcell {command}")
+            result = runcell(command)
+            print(result)
 
 
 class Function(Command):
@@ -74,7 +97,7 @@ class Function(Command):
         kwargs: default arguments to pass with function.
     '''
 
-    def __init__(self, func: any, command_name: str | None = None, **kwargs):
+    def __init__(self, func: Any, command_name: str | None = None, **kwargs):
         if command_name is None:
             command_name = func.__name__
         signature = inspect.signature(func)

core_tools/GUI/script_runner/script_runner_main.py

@@ -2,6 +2,7 @@ import logging
 import inspect
 import os
 from enum import Enum
+from typing import Any
 
 from PyQt5 import QtCore, QtWidgets
 
@@ -63,7 +64,7 @@ class ScriptRunner(QtWidgets.QMainWindow, Ui_MainWindow):
         if not instance_ready:
             self.app.exec()
 
-    def add_function(self, func: any, command_name: str | None = None, **kwargs):
+    def add_function(self, func: Any, command_name: str | None = None, **kwargs):
         '''
         Adds a function to be run as command in ScriptRunner.
 
@@ -228,7 +229,7 @@ if __name__ == "__main__":
     def fit(x: float, mode: Mode):
         print(f'fit {x}, {mode}')
 
-    path = os.path.dirname(__file__)
+    path = os.path.dirname(__file__).replace('\\', '/')
 
     ui = ScriptRunner()
     ui.add_function(sayHi)
@@ -237,11 +238,11 @@ if __name__ == "__main__":
     ui.add_function(fit, 'Fit it', mode=Mode.CENTER, x=1.0)
     ui.add_function(fit, 'Fit it', mode='center', x=1.0)
     ui.add_function(fit, 'Fit it', mode='CENTER', x=1.0)
-    # ui.add_cell('Say Hi', path+'/test_script.py')
-    # ui.add_cell(2, path+'/test_script.py', 'Magic Button')
-    # ui.add_cell('Oops', path+'/test_script.py')
-    # ui.add_cell('Syntax Error', path+'/test_script.py')
+    ui.add_cell('Say Hi', path+'/test_script.py')
+    ui.add_cell(2, path+'/test_script.py', 'Magic Button')
+    ui.add_cell('Oops', path+'/test_script.py')
+    ui.add_cell('Syntax Error', path+'/test_script.py')
 
     # NOTE:
     # To start servicing http requests run:
-    ui.run_server()
+    # ui.run_server()

core_tools/GUI/script_runner/web_server.py

@@ -108,8 +108,9 @@ class WebRequestHandler(BaseHTTPRequestHandler):
         )
 
 
-def run_web_server(command_list: list[Command]):
-    server_address = ('127.0.0.1', 8001)
+def run_web_server(command_list: list[Command], server_address: tuple[str, int] | None = None):
+    if server_address is None:
+        server_address = ('0.0.0.0', 8001)
     httpd = HTTPServer(server_address, WebRequestHandler)
     httpd.commands = command_list
     try:

examples/data/data_writer.py

-import numpy as np
-import pulse_lib.segments.utility.looping as lp
-
-from core_tools.data.ds.data_writer import DataWriter
-
-frequency = lp.linspace(100e6, 200e6, 11, 'frequency', unit='Hz')
-amplitude = lp.array([1, 3, 5, 12], 'amplitude', unit='mV')
-
-
-with DataWriter('WriterTest') as dw:
-    # add references and metadata....
-    # dw.add_ds_reference(ds_uuid, 'raw data')
-    # dw.add_metadata('abc', dict_x)
-
-    dw.add_gridded('x', unit='a.u.', label='x data',
-                   axes=[frequency, amplitude],
-                   data=np.arange(44).reshape(11, 4))
-
-    y = dw.grid_stream('y', unit='a.u.', label='x data', axes=[amplitude])
-
-    for i in range(4):
-        y.append(i + 5)
-
-    # --- scattered ---
-
-    dw.add_scattered('x', unit='a.u.', label='x data', axes=[frequency, amplitude],
-                     data=dict(x=np.arange(44).reshape(11, 4),
-                               amplitude=[1] * 44,
-                               frequency=[2] * 44))
-
-    y = dw.scattered_stream('y', unit='a.u.', label='x data', axes=[amplitude])
-    for i in range(4):
-        y.append(i + 5, amplitude=amplitude[4 - i])
-
-
-dw.write_data(
-    Var('x', unit='a.u.', label='x data', data=np.arange(44).reshape(11, 4)),
-    Var('y', unit='a.u.', label='y data', data=np.arange(44).reshape(11, 4)),
-    axes=(frequency, amplitude))
-
-
-'''
-wrap/extend Dataset.
-ds = xr.Dataset()
-
-ds.add(nane, label=, unit=, axes=[], data=)
-
-x = ds.add_stream(nane, label=, unit=, axes=[])
-
-ds.close()
-
-Also save to file? database?
-
-'''

examples/data/databrowser.py

-from core_tools.data.SQL.connector import set_up_local_storage, set_up_remote_storage, set_up_local_and_remote_storage
-from core_tools.data.gui.data_browser import data_browser
-
-##################################################
-# uncomment the option that is applicable to you #
-##################################################
-
-# in case you are only using a local server.
-# set_up_local_storage('local_usernam', 'local_passwd', 
-# 	'local_dbname', 'project_name', 'set_up_name', 'sample_name')
-
-# in case you are only the remote server.
-# set_up_remote_storage('ipaddr_rem_server', 5432, 
-# 	'remote_usernam', 'remote_passwd', 'remote_dbname', 
-# 	'project_name', 'set_up_name', 'sample_name')
-
-# in case you are using both a local and remote server.
-# set_up_local_and_remote_storage('ipaddr_rem_server', 5432,
-# 	'local_usernam', 'local_passwd', 'local_dbname',
-# 	'remote_usernam', 'remote_passwd', 'remote_dbname',
-# 	'project_name', 'set_up_name', 'sample_name')
-
-db = data_browser()
\ No newline at end of file

examples/data/measurement_script.py

-from core_tools.data.SQL.connector import set_up_local_storage,\
-	set_up_remote_storage, set_up_local_and_remote_storage
-
-##################################################
-# uncomment the option that is applicable to you #
-##################################################
-
-# in case you are only using a local server.
-# set_up_local_storage('local_usernam', 'local_passwd', 
-# 	'local_dbname', 'project_name', 'set_up_name', 'sample_name')
-
-# in case you are only the remote server.
-# set_up_remote_storage('ipaddr_rem_server', 5432, 
-# 	'remote_usernam', 'remote_passwd', 'remote_dbname', 
-# 	'project_name', 'set_up_name', 'sample_name')
-
-# in case you are using both a local and remote server.
-# set_up_local_and_remote_storage('ipaddr_rem_server', 5432,
-# 	'local_usernam', 'local_passwd', 'local_dbname',
-# 	'remote_usernam', 'remote_passwd', 'remote_dbname',
-# 	'project_name', 'set_up_name', 'sample_name')
-
-
-# when you want to do sweeps
-from core_tools.sweeps.sweeps import do0D, do1D, do2D
-
-ds = do2D(param, start, stop, n_points, delay).run()
-
-# see what is in the dataset
-print(ds)
-
-# inspecting data (extract arrays, labels, units):
-x_data, y_data, z_data = ds.m1.x(), ds.m1.y(), ds.m1()
-x_label, y_label, z_label = ds.m1.x.label, ds.m1.y.label, ds.m1.label
-x_unit, y_unit, z_unit = ds.m1.x.unit, ds.m1.y.unit, ds.m1.unit
-
-# when you want to plot a dataset
-from core_tools.data.gui.plot_mgr import data_plotter
-plot = data_plotter(ds)
-
-# load a dataset by id or uuid
-from core_tools.data.ds.data_set import load_by_id, load_by_uuid
-
-ds = load_by_id(101)

examples/demo_station/dataset_webclient.py

+import json
+import requests
+from pprint import pprint
+
+
+base = "http://127.0.0.1:8002"
+
+
+# %%
+
+r = requests.get(base + "/latest",
+                 params={
+                     "start_time": "2025-03-06 15:30",
+                     })
+print(r.status_code)
+pprint(json.loads(r.content))

examples/demo_station/dataset_webserver.py

+import json
+from functools import cached_property
+from http.cookies import SimpleCookie
+from http.server import HTTPServer, BaseHTTPRequestHandler, HTTPStatus
+from urllib.parse import parse_qsl, urlparse
+
+import core_tools as ct
+from core_tools.data.ds.reader import load_by_uuid
+from core_tools.data.SQL.queries.dataset_gui_queries import query_for_measurement_results
+
+from core_tools.data.ds.ds2xarray import ds2xarray
+
+
+"""
+TIP:
+    Switch project name to 'Game', so all data is for Game
+"""
+
+
+class DatasetRequestHandler(BaseHTTPRequestHandler):
+
+    @cached_property
+    def url(self):
+        return urlparse(self.path)
+
+    @cached_property
+    def query_data(self):
+        return dict(parse_qsl(self.url.query))
+
+    @cached_property
+    def post_data(self):
+        content_length = int(self.headers.get("Content-Length", 0))
+        return self.rfile.read(content_length)
+
+    @cached_property
+    def form_data(self):
+        return dict(parse_qsl(self.post_data.decode("utf-8")))
+
+    @cached_property
+    def cookies(self):
+        return SimpleCookie(self.headers.get("Cookie"))
+
+    def log_request(self, code='-', size='-'):
+        pass
+
+    def do_GET(self):
+        if self.url.path != "/latest":
+            self.send_error(HTTPStatus.NOT_FOUND)
+        else:
+            start_time = self.query_data.get("start_time")
+            ds = self.get_last_ds(start_time)
+            if ds is None:
+                self.send_error(HTTPStatus.NO_CONTENT)
+            else:
+                response = self.get_ds_json(ds).encode("utf-8")
+                self.send_response(HTTPStatus.OK)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(response)
+
+    def get_last_ds(self, start_time: str | None):
+        res = query_for_measurement_results.search_query(
+                start_time=start_time,  # optional
+                name=None,  # part of name optional
+                project=self.server.project_name,  # optional
+                keywords=None,  # optional
+                )
+        if len(res) == 0:
+            return None
+        uuid = res[-1].uuid
+        ds = load_by_uuid(uuid)
+        return ds
+
+    def get_ds_json(self, ds, variables: list[str] | None = None):
+        xds = ds2xarray(ds, snapshot=False)
+        for m_param in ds:
+            for name, descr in m_param:
+                var_name = descr.param_name
+                xds[var_name].attrs["written"] = descr.written()
+
+        if variables is not None:
+            xds = xds[variables]
+
+        d = xds.to_dict()
+        return json.dumps(d, indent=1)
+
+
+def run_web_server(project_name: str | None, server_address: tuple[str, int] | None = None):
+    if server_address is None:
+        server_address = ('0.0.0.0', 8002)
+    httpd = HTTPServer(server_address, DatasetRequestHandler)
+    httpd.project_name = project_name
+    try:
+        print(f"Server running at http://{server_address[0]}:{server_address[1]}")
+        print("Interrupt kernel to stop server")
+        httpd.serve_forever()
+    except KeyboardInterrupt:
+        httpd.shutdown()
+
+
+ # %%
+
+ct.configure('./setup_config/ct_config_laptop.yaml')
+
+run_web_server(None)

examples/demo_station/remote_script.py

+import json
+import requests
+from pprint import pprint
+
+
+base = "http://127.0.0.1:8001"
+
+# %%
+r = requests.get(base + "/functions")
+print(r.status_code)
+pprint(json.loads(r.content))
+
+
+# %%
+
+r = requests.post(base + "/run",
+                  data={
+                      "__name__": "sayHi",
+                      "name": "Oriol",
+                      "times": 2,
+                      })
+print(r.status_code)
+pprint(json.loads(r.content))
+
+r = requests.post(base + "/run",
+                  data={
+                      "__name__": "Fit it",
+                      "x": 13.0,
+                      "mode": "RIGHT",
+                      })
+print(r.status_code)
+pprint(json.loads(r.content))