[TASK] plotting scripts drafted

Author: Sulimankhail

src/dt4acc/custom_epics/scripts/__init__.py

@@ -0,0 +1,80 @@
+from p4p.client.thread import Context
+import matplotlib.pyplot as plt
+import numpy as np
+import threading
+
+# --- Configuration ---
+pv_base = "waheed:beam:twiss"  # base name prefix
+update_interval = 0.2  # seconds
+
+# PVs to monitor (relative to base)
+pv_names = [
+     "x:alpha", "x:beta", "x:nu",
+     "y:alpha", "y:beta", "y:nu",
+]
+
+# --- Data structures ---
+ctx = Context("pva")
+latest = {pv: None for pv in pv_names}
+subs = {}
+lock = threading.Lock()
+
+# --- Callback factory ---
+def make_callback(pv):
+    def _cb(value):
+        with lock:
+            try:
+                arr = np.array(value["value"], dtype=float)
+            except Exception:
+                arr = np.array(value, dtype=float)
+            latest[pv] = arr
+    return _cb
+
+# --- Subscribe to all PVs ---
+for pv in pv_names:
+    full = f"{pv_base}:{pv}"
+    subs[pv] = ctx.monitor(full, make_callback(pv))
+    print(f"Subscribed to {full}")
+
+# --- Plot setup ---
+plt.ion()
+fig, axes = plt.subplots(len(pv_names), 1, figsize=(8, 12), sharex=False)
+lines = {}
+
+for ax, pv in zip(axes, pv_names):
+    (line,) = ax.plot([], [], lw=1.3)
+    ax.set_title(pv)
+    ax.set_ylabel("Value")
+    ax.grid(True)
+    lines[pv] = (ax, line)
+
+axes[-1].set_xlabel("Index")
+fig.suptitle(f"{pv_base}:* (live monitor)", fontsize=14)
+fig.tight_layout(rect=[0, 0, 1, 0.97])
+
+# --- Live update loop ---
+print("Monitoring all PVs... close the figure to stop.")
+try:
+    while plt.fignum_exists(fig.number):
+        updated = False
+        with lock:
+            for pv, arr in latest.items():
+                if arr is not None and len(arr) > 0:
+                    ax, line = lines[pv]
+                    x = np.arange(len(arr))
+                    line.set_data(x, arr)
+                    ax.relim()
+                    ax.autoscale_view()
+                    updated = True
+        if updated:
+            plt.pause(update_interval)
+        else:
+            plt.pause(0.1)
+
+except KeyboardInterrupt:
+    pass
+finally:
+    for s in subs.values():
+        s.close()
+    ctx.close()
+    print("Stopped monitoring.")

src/dt4acc/custom_epics/scripts/play_better.py

@@ -0,0 +1,77 @@
+from p4p.client.thread import Context
+import matplotlib.pyplot as plt
+import numpy as np
+import threading
+
+# --- Configuration ---
+pv_x = "waheed:beam:twiss:x:beta"
+pv_y = "waheed:beam:twiss:y:beta"
+update_interval = 0.1  # seconds
+
+# --- Shared data ---
+latest_x = None
+latest_y = None
+lock = threading.Lock()
+
+# --- P4P context ---
+ctx = Context("pva")
+
+def on_update_x(value):
+    global latest_x
+    with lock:
+        try:
+            latest_x = np.array(value["value"], dtype=float)
+        except Exception:
+            latest_x = np.array(value, dtype=float)
+
+def on_update_y(value):
+    global latest_y
+    with lock:
+        try:
+            latest_y = np.array(value["value"], dtype=float)
+        except Exception:
+            latest_y = np.array(value, dtype=float)
+
+# Subscribe to both PVs
+sub_x = ctx.monitor(pv_x, on_update_x)
+sub_y = ctx.monitor(pv_y, on_update_y)
+
+# --- Matplotlib setup ---
+plt.ion()
+fig, ax = plt.subplots()
+(line_x,) = ax.plot([], [], label="x plane", lw=1.5)
+(line_y,) = ax.plot([], [], label="y plane", lw=1.5)
+ax.set_title("waheed:beam:twiss:beta (x & y live)")
+ax.set_xlabel("Index")
+ax.set_ylabel(r"$\beta_{x, y}$")
+ax.legend()
+ax.grid(True)
+
+print(f"Monitoring:\n  {pv_x}\n  {pv_y}\n(close the window to stop)")
+
+try:
+    while plt.fignum_exists(fig.number):
+        with lock:
+            updated = False
+            if latest_x is not None:
+                x_axis = np.arange(len(latest_x))
+                line_x.set_data(x_axis, latest_x)
+                updated = True
+            if latest_y is not None:
+                y_axis = np.arange(len(latest_y))
+                line_y.set_data(y_axis, latest_y)
+                updated = True
+
+            if updated:
+                ax.relim()
+                ax.autoscale_view()
+
+        plt.pause(update_interval)
+
+except KeyboardInterrupt:
+    pass
+finally:
+    sub_x.close()
+    sub_y.close()
+    ctx.close()
+    print("Stopped monitoring.")