bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

feat(waveform): data are taken directly from ScanItem which is defined from scan_status endpoint; scan update is triggered from scan_segment; plots can be added just specifying y_name -> best effort for setting x reported device

Browse Source
This commit is contained in:
wyzula_j
2024-07-09 11:50:51 +02:00
parent 0aa317aae5
commit b8717f1327
5 changed files with 133 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
bec_widgets/cli/client.py
View File

@ -19,6 +19,9 @@ class Widgets(str, enum.Enum):
BECMotorMapWidget = "BECMotorMapWidget"
BECQueue = "BECQueue"
BECStatusBox = "BECStatusBox"
DeviceBox = "DeviceBox"
DeviceComboBox = "DeviceComboBox"
DeviceLineEdit = "DeviceLineEdit"
RingProgressBar = "RingProgressBar"
ScanControl = "ScanControl"
StopButton = "StopButton"
@ -1507,6 +1510,7 @@ class BECWaveform(RPCBase):
label: "str | None" = None,
validate: "bool" = True,
dap: "str | None" = None,
**kwargs,
) -> "BECCurve":
"""
Plot a curve to the plot widget.
@ -1752,6 +1756,24 @@ class BECWaveform(RPCBase):
"""
class DeviceBox(RPCBase):
@property
@rpc_call
def _config_dict(self) -> "dict":
"""
Get the configuration of the widget.
Returns:
dict: The configuration of the widget.
"""
@rpc_call
def _get_all_rpc(self) -> "dict":
"""
Get all registered RPC objects.
"""
class DeviceComboBox(RPCBase):
@property
@rpc_call

20
bec_widgets/examples/jupyter_console/jupyter_console_window.py
View File

@ -40,10 +40,8 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
"w1": self.w1,
"w2": self.w2,
"w3": self.w3,
"w1_c": self.w1_c,
"w2_c": self.w2_c,
"w3_c": self.w3_c,
"w4": self.w4,
"w5": self.w5,
"d0": self.d0,
"d1": self.d1,
"d2": self.d2,
@ -73,20 +71,18 @@ class JupyterConsoleWindow(QWidget): # pragma: no cover:
self.console_layout.addWidget(self.console)
def _init_figure(self):
self.figure.plot(x_name="samx", y_name="samy", z_name="bpm4i", color_map_z="cividis")
self.figure.motor_map("samx", "samy")
self.figure.image("eiger", color_map="viridis", vrange=(0, 100))
self.figure.plot(
self.w1 = self.figure.plot(
x_name="samx", y_name="samy", z_name="bpm4i", color_map_z="cividis"
)
self.w2 = self.figure.motor_map("samx", "samy")
self.w3 = self.figure.image("eiger", color_map="viridis", vrange=(0, 100))
self.w4 = self.figure.plot(
x_name="samx", y_name="samy", z_name="bpm4i", color_map_z="magma", new=True
)
self.w5 = self.figure.plot(y_name="bpm4i", new=True)
self.figure.change_layout(2, 2)
self.w1 = self.figure[0, 0]
self.w2 = self.figure[0, 1]
self.w3 = self.figure[1, 0]
self.w4 = self.figure[1, 1]
# Plot Customisation
self.w1.set_title("Waveform 1")
self.w1.set_x_label("Motor Position (samx)")

54
bec_widgets/widgets/figure/figure.py
View File

@ -283,26 +283,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
if x is not None and y is not None:
waveform.add_curve_custom(x=x, y=y, color=color, label=label)
return waveform
# User wants to add scan curve -> 1D Waveform
if x_name is not None and y_name is not None and z_name is None and x is None and y is None:
waveform.plot(
x_name=x_name,
y_name=y_name,
x_entry=x_entry,
y_entry=y_entry,
validate=validate,
color=color,
label=label,
dap=dap,
)
# User wants to add scan curve -> 2D Waveform Scatter
if (
x_name is not None
and y_name is not None
and z_name is not None
and x is None
and y is None
):
if y_name is not None:
waveform.plot(
x_name=x_name,
y_name=y_name,
@ -316,6 +297,39 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
validate=validate,
dap=dap,
)
# # User wants to add scan curve -> 1D Waveform
# if x_name is not None and y_name is not None and z_name is None and x is None and y is None:
# waveform.plot(
# x_name=x_name,
# y_name=y_name,
# x_entry=x_entry,
# y_entry=y_entry,
# validate=validate,
# color=color,
# label=label,
# dap=dap,
# )
# # User wants to add scan curve -> 2D Waveform Scatter
# if (
# x_name is not None
# and y_name is not None
# and z_name is not None
# and x is None
# and y is None
# ):
# waveform.plot(
# x_name=x_name,
# y_name=y_name,
# z_name=z_name,
# x_entry=x_entry,
# y_entry=y_entry,
# z_entry=z_entry,
# color=color,
# color_map_z=color_map_z,
# label=label,
# validate=validate,
# dap=dap,
# )
# User wants to add custom curve
elif x is not None and y is not None and x_name is None and y_name is None:
waveform.add_curve_custom(x=x, y=y, color=color, label=label)

98
bec_widgets/widgets/figure/plots/waveform/waveform.py
View File

@ -78,10 +78,13 @@ class BECWaveform(BECPlotBase):
self._curves_data = defaultdict(dict)
self.old_scan_id = None
self.scan_id = None
self.scan_item = None
self.dap = None
self.scan_motors = None # TODO maybe not needed actually, can be fetched direclty from scan_item -> msg['info']['scan_report_devices']
# Scan segment update proxy
self.proxy_update_plot = pg.SignalProxy(
self.scan_signal_update, rateLimit=25, slot=self._update_scan_segment_plot
self.scan_signal_update, rateLimit=25, slot=self._update_scan_curves
)
self.proxy_update_dap = pg.SignalProxy(
@ -92,6 +95,7 @@ class BECWaveform(BECPlotBase):
# Connect dispatcher signals
self.bec_dispatcher.connect_slot(self.on_scan_segment, MessageEndpoints.scan_segment())
self.bec_dispatcher.connect_slot(self.on_scan_status, MessageEndpoints.scan_status())
self.entry_validator = EntryValidator(self.dev)
@ -319,8 +323,8 @@ class BECWaveform(BECPlotBase):
def add_curve_scan(
self,
x_name: str,
y_name: str,
x_name: Optional[str] = None,
y_name: Optional[str] = None,
z_name: Optional[str] = None,
x_entry: Optional[str] = None,
y_entry: Optional[str] = None,
@ -351,6 +355,8 @@ class BECWaveform(BECPlotBase):
Returns:
BECCurve: The curve object.
"""
if y_name is None:
raise ValueError("y_name must be provided.")
# Check if curve already exists
curve_source = source
@ -385,7 +391,7 @@ class BECWaveform(BECPlotBase):
source=curve_source,
signals=Signal(
source=curve_source,
x=SignalData(name=x_name, entry=x_entry),
x=SignalData(name=x_name, entry=x_entry) if x_name else None,
y=SignalData(name=y_name, entry=y_entry),
z=SignalData(name=z_name, entry=z_entry) if z_name else None,
dap=dap,
@ -486,8 +492,8 @@ class BECWaveform(BECPlotBase):
def _validate_signal_entries(
self,
x_name: str,
y_name: str,
x_name: str | None,
y_name: str | None,
z_name: str | None,
x_entry: str | None,
y_entry: str | None,
@ -510,8 +516,10 @@ class BECWaveform(BECPlotBase):
tuple[str,str,str|None]: Validated x, y, z entries.
"""
if validate_bec:
x_entry = self.entry_validator.validate_signal(x_name, x_entry)
y_entry = self.entry_validator.validate_signal(y_name, y_entry)
if x_name:
x_entry = self.entry_validator.validate_signal(x_name, x_entry)
if y_name:
y_entry = self.entry_validator.validate_signal(y_name, y_entry)
if z_name:
z_entry = self.entry_validator.validate_signal(z_name, z_entry)
else:
@ -593,15 +601,15 @@ class BECWaveform(BECPlotBase):
else:
raise IndexError(f"Curve order {N} out of range.")
@pyqtSlot(dict, dict)
def on_scan_segment(self, msg: dict, metadata: dict):
@pyqtSlot(dict)
def on_scan_status(self, msg):
"""
Handle new scan segments and saves data to a dictionary. Linked through bec_dispatcher.
Handle the scan status message.
Args:
msg (dict): Message received with scan data.
metadata (dict): Metadata of the scan.
msg(dict): Message received with scan status.
"""
current_scan_id = msg.get("scan_id", None)
if current_scan_id is None:
return
@ -609,14 +617,28 @@ class BECWaveform(BECPlotBase):
if current_scan_id != self.scan_id:
self.old_scan_id = self.scan_id
self.scan_id = current_scan_id
self.scan_segment_data = self.queue.scan_storage.find_scan_by_ID(
self.scan_id
) # TODO do scan access through BECFigure
self.setup_dap(self.old_scan_id, self.scan_id)
self.scan_item = self.queue.scan_storage.find_scan_by_ID(self.scan_id)
if self._curves_data["DAP"]:
self.setup_dap(self.old_scan_id, self.scan_id)
if self._curves_data["async"]:
print("setting async")
# for curve in self._curves_data["async"]:
# self.setup_async(curve.config.signals.y.name)
@pyqtSlot(dict, dict)
def on_scan_segment(self, msg: dict, metadata: dict):
"""
Handle new scan segments and saves data to a dictionary. Linked through bec_dispatcher.
Used only for triggering scan segment update from the BECClient scan storage.
Args:
msg (dict): Message received with scan data.
metadata (dict): Metadata of the scan.
"""
self.scan_signal_update.emit()
def setup_dap(self, old_scan_id, new_scan_id):
def setup_dap(self, old_scan_id: str | None, new_scan_id: str | None):
"""
Setup DAP for the new scan.
@ -633,6 +655,15 @@ class BECWaveform(BECPlotBase):
self.update_dap, MessageEndpoints.dap_response(new_scan_id)
)
def setup_async(self, device: str):
self.bec_dispatcher.disconnect_slot(
self.update_dap, MessageEndpoints.device_async_readback(self.old_scan_id, device)
)
if len(self._curves_data["async"]) > 0:
self.bec_dispatcher.connect_slot(
self.update_dap, MessageEndpoints.device_async_readback(self.scan_id, device)
)
def refresh_dap(self):
"""
Refresh the DAP curves with the latest data from the DAP model MessageEndpoints.dap_response().
@ -676,24 +707,31 @@ class BECWaveform(BECPlotBase):
self.dap_params_update.emit(curve.dap_params)
break
def _update_scan_segment_plot(self):
"""Update the plot with the data from the scan segment."""
data = self.scan_segment_data.data
self._update_scan_curves(data)
@pyqtSlot(dict, dict)
def update_async(self, msg, metadata):
print("async")
print(f"msg: {msg}")
print(f"metadata: {metadata}")
def _update_scan_curves(self, data: ScanData):
@pyqtSlot()
def _update_scan_curves(self):
"""
Update the scan curves with the data from the scan segment.
Args:
data(ScanData): Data from the scan segment.
"""
data = self.scan_item.data
data_x = None
data_y = None
data_z = None
for curve_id, curve in self._curves_data["scan_segment"].items():
x_name = curve.config.signals.x.name
x_entry = curve.config.signals.x.entry
if curve.config.signals.x:
x_name = curve.config.signals.x.name
x_entry = curve.config.signals.x.entry
else:
x_name = self.scan_item.status_message.info["scan_report_devices"][0]
x_entry = self.entry_validator.validate_signal(x_name, None)
y_name = curve.config.signals.y.name
y_entry = curve.config.signals.y.entry
if curve.config.signals.z:
@ -765,8 +803,8 @@ class BECWaveform(BECPlotBase):
self.scan_id = scan_id
self.setup_dap(self.old_scan_id, self.scan_id)
data = self.queue.scan_storage.find_scan_by_ID(self.scan_id).data
self._update_scan_curves(data)
self.scan_item = self.queue.scan_storage.find_scan_by_ID(self.scan_id)
self.scan_signal_update.emit()
def get_all_data(self, output: Literal["dict", "pandas"] = "dict") -> dict | pd.DataFrame:
"""

2
bec_widgets/widgets/figure/plots/waveform/waveform_curve.py
View File

@ -28,7 +28,7 @@ class Signal(BaseModel):
"""The configuration of a signal in the 1D waveform widget."""
source: str
x: SignalData # TODO maybe add metadata for config gui later
x: Optional[SignalData] = None
y: SignalData
z: Optional[SignalData] = None
dap: Optional[str] = None