Task that creates input for Flint plot
Use case in Bliss+Flint (see execute_workflow
for what the workflow needs to return):
from time import sleep
import numpy
from bliss.common.plot import get_flint
def execute_workflow():
return {
"plot1": {
"x": numpy.arange(100),
"y": numpy.random.random(100),
"xlabel": "Energy (eV)",
"ylabel": "mu",
},
"plot2": {
"x": numpy.arange(100),
"y": numpy.random.random(100) + 10,
"xlabel": "Wavenumber (Å)",
"ylabel": "k^2 χ(k) (Å^2)",
},
"plot3": {
"x": numpy.arange(100),
"y": numpy.random.random(100) + 20,
"xlabel": "Radial Distance (Å)",
"ylabel": "|χ(R)| (Å^3)",
},
"plot4": {
"x": numpy.arange(100),
"y": numpy.random.random(100) + 30,
"xlabel": "Energy (eV)",
"ylabel": "Noise",
},
}
def prepare_scan_plot(plots, scan, data):
for name, p in plots.items():
with p.transaction():
p.add_curve_item("x", "y", legend=scan)
p.add_curve_item("x", "y", legend=scan)
p.xlabel = data[name]["xlabel"]
p.ylabel = data[name]["ylabel"]
def refresh_scan_plot(plots, data):
for name, p in plots.items():
with p.transaction():
d = data[name]
p.set_data(x=d["x"], y=d["y"])
def plot_exafs(scan="1.1"):
f = get_flint()
plots = dict()
plots["plot1"] = f.get_plot("curve", unique_name="plot1")
plots["plot2"] = f.get_plot("curve", unique_name="plot2")
plots["plot3"] = f.get_plot("curve", unique_name="plot3")
plots["plot4"] = f.get_plot("curve", unique_name="plot4")
# loop until the end of the scan (interval of n seconds)
for i in range(10):
data = execute_workflow()
if i == 0:
prepare_scan_plot(plots, scan, data)
refresh_scan_plot(plots, data)
sleep(1)
return f, plots
Or with a grid bliss/bliss!4744 (merged)