mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 03:31:50 +02:00
feat(plots/image): basic image visualisation, getting data are based on stream_connector (deprecated)
This commit is contained in:
wyzula-jan
@ -271,6 +271,78 @@ class BECFigure(RPCBase, BECFigureClientMixin):
|
||||
**axis_kwargs(dict): Additional axis properties to set on the widget after creation.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def add_image(
|
||||
self,
|
||||
widget_id: "str" = None,
|
||||
row: "int" = None,
|
||||
col: "int" = None,
|
||||
config=None,
|
||||
color_map: "str" = "magma",
|
||||
color_bar: "Literal['simple', 'full']" = "full",
|
||||
vrange: "tuple[float, float]" = None,
|
||||
**axis_kwargs
|
||||
) -> "BECImageShow":
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def plot(
|
||||
self,
|
||||
x_name: "str" = None,
|
||||
y_name: "str" = None,
|
||||
x_entry: "str" = None,
|
||||
y_entry: "str" = None,
|
||||
x: "list | np.ndarray" = None,
|
||||
y: "list | np.ndarray" = None,
|
||||
color: "Optional[str]" = None,
|
||||
label: "Optional[str]" = None,
|
||||
validate: "bool" = True,
|
||||
**axis_kwargs
|
||||
) -> "BECWaveform1D":
|
||||
"""
|
||||
Add a 1D waveform plot to the figure.
|
||||
Args:
|
||||
x_name(str): The name of the device for the x-axis.
|
||||
y_name(str): The name of the device for the y-axis.
|
||||
x_entry(str): The name of the entry for the x-axis.
|
||||
y_entry(str): The name of the entry for the y-axis.
|
||||
x(list | np.ndarray): Custom x data to plot.
|
||||
y(list | np.ndarray): Custom y data to plot.
|
||||
color(str): The color of the curve.
|
||||
label(str): The label of the curve.
|
||||
validate(bool): If True, validate the device names and entries.
|
||||
**axis_kwargs: Additional axis properties to set on the widget after creation.
|
||||
|
||||
Returns:
|
||||
BECWaveform1D: The waveform plot widget.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def image(
|
||||
self,
|
||||
monitor: "str" = None,
|
||||
color_bar: "Literal['simple', 'full']" = "full",
|
||||
color_map: "str" = "magma",
|
||||
data: "np.ndarray" = None,
|
||||
vrange: "tuple[float, float]" = None,
|
||||
**axis_kwargs
|
||||
) -> "BECImageShow":
|
||||
"""
|
||||
Add an image to the figure.
|
||||
Args:
|
||||
monitor(str): The name of the monitor to display.
|
||||
color_bar(Literal["simple","full"]): The type of color bar to display.
|
||||
color_map(str): The color map to use for the image.
|
||||
data(np.ndarray): Custom data to display.
|
||||
vrange(tuple[float, float]): The range of values to display.
|
||||
**axis_kwargs: Additional axis properties to set on the widget after creation.
|
||||
|
||||
Returns:
|
||||
BECImageShow: The image widget.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def remove(
|
||||
self,
|
||||
@ -392,3 +464,74 @@ class BECCurve(RPCBase):
|
||||
Returns:
|
||||
tuple[np.ndarray,np.ndarray]: X and Y data of the curve.
|
||||
"""
|
||||
|
||||
|
||||
class BECImageShow(RPCBase):
|
||||
@rpc_call
|
||||
def set_vrange(self, vmin: "float", vmax: "float"):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_monitor(self, monitor: "str" = None) -> "None":
|
||||
"""
|
||||
Set/update monitor device.
|
||||
Args:
|
||||
monitor(str): Name of the monitor.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_color_map(self, cmap: "str" = "magma"):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_image(self, data: "np.ndarray"):
|
||||
"""
|
||||
Set the image to be displayed.
|
||||
Args:
|
||||
data(np.ndarray): The image to be displayed.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def set_processing(
|
||||
self,
|
||||
fft: "bool" = False,
|
||||
log: "bool" = False,
|
||||
rotation: "int" = None,
|
||||
transpose: "bool" = False,
|
||||
):
|
||||
"""
|
||||
Set the processing of the monitor data.
|
||||
Args:
|
||||
fft(bool): Whether to perform FFT on the monitor data.
|
||||
log(bool): Whether to perform log on the monitor data.
|
||||
rotation(int): The rotation angle of the monitor data before displaying.
|
||||
transpose(bool): Whether to transpose the monitor data before displaying.
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def enable_fft(self, enable: "bool" = True):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def enable_log(self, enable: "bool" = True):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def rotate(self, angle: "int"):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@rpc_call
|
||||
def transpose(self):
|
||||
"""
|
||||
None
|
||||
"""
|
||||
|
||||
@ -106,13 +106,13 @@ class {class_name}(RPCBase):"""
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import os
|
||||
|
||||
# Assuming ClientGenerator is defined in this script or imported correctly
|
||||
from bec_widgets.widgets.figure import BECFigure
|
||||
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D, BECCurve
|
||||
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D, BECImageShow # ,BECCurve
|
||||
from bec_widgets.widgets.plots.waveform1d import BECCurve
|
||||
|
||||
current_path = os.path.dirname(__file__)
|
||||
client_path = os.path.join(current_path, "client.py")
|
||||
clss = [BECPlotBase, BECWaveform1D, BECFigure, BECCurve]
|
||||
clss = [BECPlotBase, BECWaveform1D, BECFigure, BECCurve, BECImageShow]
|
||||
generator = ClientGenerator()
|
||||
generator.generate_client(clss)
|
||||
generator.write(client_path)
|
||||
|
||||
@ -5,11 +5,11 @@ from bec_lib import MessageEndpoints, messages
|
||||
from bec_widgets.utils import BECDispatcher
|
||||
from bec_widgets.utils.bec_connector import BECConnector
|
||||
from bec_widgets.widgets.figure import BECFigure
|
||||
from bec_widgets.widgets.plots import BECCurve, BECWaveform1D
|
||||
from bec_widgets.widgets.plots import BECCurve, BECWaveform1D, BECImageShow
|
||||
|
||||
|
||||
class BECWidgetsCLIServer:
|
||||
WIDGETS = [BECWaveform1D, BECFigure, BECCurve]
|
||||
WIDGETS = [BECWaveform1D, BECFigure, BECCurve, BECImageShow]
|
||||
|
||||
def __init__(self, gui_id: str = None, dispatcher: BECDispatcher = None) -> None:
|
||||
self.dispatcher = BECDispatcher() if dispatcher is None else dispatcher
|
||||
|
||||
@ -38,6 +38,7 @@ class BECConnector:
|
||||
|
||||
if config:
|
||||
self.config = config
|
||||
self.config.widget_class = self.__class__.__name__
|
||||
else:
|
||||
print(
|
||||
f"No initial config found for {self.__class__.__name__}.\n"
|
||||
|
||||
@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
import itertools
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from typing import Literal, Optional
|
||||
from typing import Literal, Optional, Type
|
||||
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
@ -22,7 +22,7 @@ from bec_widgets.widgets.plots import (
|
||||
WidgetConfig,
|
||||
BECImageShow,
|
||||
)
|
||||
from bec_widgets.widgets.plots.image import BECImageShowWithHistogram
|
||||
from bec_widgets.widgets.plots.image import ImageConfig
|
||||
|
||||
|
||||
class FigureConfig(ConnectionConfig):
|
||||
@ -43,7 +43,7 @@ class WidgetHandler:
|
||||
self.widget_factory = {
|
||||
"PlotBase": (BECPlotBase, WidgetConfig),
|
||||
"Waveform1D": (BECWaveform1D, Waveform1DConfig),
|
||||
"ImShow": (BECImageShow, WidgetConfig),
|
||||
"ImShow": (BECImageShow, ImageConfig),
|
||||
}
|
||||
|
||||
def create_widget(
|
||||
@ -73,6 +73,8 @@ class WidgetHandler:
|
||||
raise ValueError(f"Unsupported widget type: {widget_type}")
|
||||
|
||||
widget_class, config_class = entry
|
||||
if config is not None and isinstance(config, config_class):
|
||||
config = config.model_dump()
|
||||
widget_config_dict = {
|
||||
"widget_class": widget_class.__name__,
|
||||
"parent_id": parent_id,
|
||||
@ -91,7 +93,16 @@ class WidgetHandler:
|
||||
|
||||
|
||||
class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
|
||||
USER_ACCESS = ["add_plot", "remove", "change_layout", "change_theme", "clear_all"]
|
||||
USER_ACCESS = [
|
||||
"add_plot",
|
||||
"add_image",
|
||||
"plot",
|
||||
"image",
|
||||
"remove",
|
||||
"change_layout",
|
||||
"change_theme",
|
||||
"clear_all",
|
||||
]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@ -148,9 +159,127 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
|
||||
**axis_kwargs,
|
||||
)
|
||||
|
||||
def add_image(
|
||||
self, widget_id: str = None, row: int = None, col: int = None, config=None, **axis_kwargs
|
||||
def plot(
|
||||
self,
|
||||
x_name: str = None,
|
||||
y_name: str = None,
|
||||
x_entry: str = None,
|
||||
y_entry: str = None,
|
||||
x: list | np.ndarray = None,
|
||||
y: list | np.ndarray = None,
|
||||
color: Optional[str] = None,
|
||||
label: Optional[str] = None,
|
||||
validate: bool = True,
|
||||
**axis_kwargs,
|
||||
) -> BECWaveform1D:
|
||||
"""
|
||||
Add a 1D waveform plot to the figure.
|
||||
Args:
|
||||
x_name(str): The name of the device for the x-axis.
|
||||
y_name(str): The name of the device for the y-axis.
|
||||
x_entry(str): The name of the entry for the x-axis.
|
||||
y_entry(str): The name of the entry for the y-axis.
|
||||
x(list | np.ndarray): Custom x data to plot.
|
||||
y(list | np.ndarray): Custom y data to plot.
|
||||
color(str): The color of the curve.
|
||||
label(str): The label of the curve.
|
||||
validate(bool): If True, validate the device names and entries.
|
||||
**axis_kwargs: Additional axis properties to set on the widget after creation.
|
||||
|
||||
Returns:
|
||||
BECWaveform1D: The waveform plot widget.
|
||||
"""
|
||||
waveform = self._find_first_widget_by_class(BECWaveform1D, can_fail=True)
|
||||
if waveform is not None:
|
||||
if axis_kwargs:
|
||||
waveform.set(**axis_kwargs)
|
||||
else:
|
||||
waveform = self.add_plot(**axis_kwargs)
|
||||
|
||||
# User wants to add scan curve
|
||||
if x_name is not None and y_name is not None and x is None and y is None:
|
||||
waveform.add_curve_scan(
|
||||
x_name=x_name,
|
||||
y_name=y_name,
|
||||
x_entry=x_entry,
|
||||
y_entry=y_entry,
|
||||
validate=validate,
|
||||
color=color,
|
||||
label=label,
|
||||
)
|
||||
# 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,
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
"Invalid input. Provide either device names (x_name, y_name) or custom data."
|
||||
)
|
||||
return waveform
|
||||
|
||||
def image(
|
||||
self,
|
||||
monitor: str = None,
|
||||
color_bar: Literal["simple", "full"] = "full",
|
||||
color_map: str = "magma",
|
||||
data: np.ndarray = None,
|
||||
vrange: tuple[float, float] = None,
|
||||
**axis_kwargs,
|
||||
) -> BECImageShow:
|
||||
"""
|
||||
Add an image to the figure.
|
||||
Args:
|
||||
monitor(str): The name of the monitor to display.
|
||||
color_bar(Literal["simple","full"]): The type of color bar to display.
|
||||
color_map(str): The color map to use for the image.
|
||||
data(np.ndarray): Custom data to display.
|
||||
vrange(tuple[float, float]): The range of values to display.
|
||||
**axis_kwargs: Additional axis properties to set on the widget after creation.
|
||||
|
||||
Returns:
|
||||
BECImageShow: The image widget.
|
||||
"""
|
||||
image = self._find_first_widget_by_class(BECImageShow, can_fail=True)
|
||||
if image is not None:
|
||||
if axis_kwargs:
|
||||
image.set(**axis_kwargs)
|
||||
else:
|
||||
image = self.add_image(color_bar=color_bar, **axis_kwargs)
|
||||
|
||||
# Setting appearance
|
||||
if vrange is not None:
|
||||
image.set_vrange(vmin=vrange[0], vmax=vrange[1])
|
||||
if color_map is not None:
|
||||
image.set_color_map(color_map)
|
||||
|
||||
# Setting data
|
||||
if monitor is not None and data is None:
|
||||
image.set_monitor(monitor)
|
||||
elif data is not None and monitor is None:
|
||||
image.set_image(data)
|
||||
else:
|
||||
raise ValueError("Invalid input. Provide either monitor name or custom data.")
|
||||
|
||||
return image
|
||||
|
||||
def add_image(
|
||||
self,
|
||||
widget_id: str = None,
|
||||
row: int = None,
|
||||
col: int = None,
|
||||
config=None,
|
||||
color_map: str = "magma",
|
||||
color_bar: Literal["simple", "full"] = "full",
|
||||
vrange: tuple[float, float] = None,
|
||||
**axis_kwargs,
|
||||
) -> BECImageShow:
|
||||
if config is None:
|
||||
# config = ImageConfig(color_map=color_map, color_bar=color_bar, vrange=vrange)
|
||||
config = ImageConfig(color_map=color_map, color_bar=color_bar, vrange=vrange)
|
||||
return self.add_widget(
|
||||
widget_type="ImShow",
|
||||
widget_id=widget_id,
|
||||
@ -248,6 +377,25 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
|
||||
else:
|
||||
raise ValueError("Must provide either widget_id or coordinates for removal.")
|
||||
|
||||
def _find_first_widget_by_class(
|
||||
self, widget_class: Type[BECPlotBase], can_fail: bool = True
|
||||
) -> BECPlotBase | None:
|
||||
"""
|
||||
Find the first widget of a given class in the figure.
|
||||
Args:
|
||||
widget_class(Type[BECPlotBase]): The class of the widget to find.
|
||||
can_fail(bool): If True, the method will return None if no widget is found. If False, it will raise an error.
|
||||
Returns:
|
||||
BECPlotBase: The widget of the given class.
|
||||
"""
|
||||
for widget_id, widget in self.widgets.items():
|
||||
if isinstance(widget, widget_class):
|
||||
return widget
|
||||
if can_fail:
|
||||
return None
|
||||
else:
|
||||
raise ValueError(f"No widget of class {widget_class} found.")
|
||||
|
||||
def _remove_by_coordinates(self, row: int, col: int) -> None:
|
||||
"""
|
||||
Remove a widget from the figure by its coordinates.
|
||||
@ -421,16 +569,6 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
|
||||
|
||||
sys.exit(app.exec_())
|
||||
|
||||
def add_image_with_histogram(self, image_data, widget_id=None, row=None, col=None):
|
||||
# Create the custom image show widget
|
||||
image_widget = BECImageShowWithHistogram()
|
||||
|
||||
# Set the image data
|
||||
image_widget.setImage(image_data)
|
||||
|
||||
# Add the widget to BECFigure
|
||||
self.addItem(image_widget, row=row, col=col)
|
||||
|
||||
|
||||
##################################################
|
||||
##################################################
|
||||
@ -493,14 +631,15 @@ class DebugWindow(QWidget): # pragma: no cover:
|
||||
|
||||
def _init_figure(self):
|
||||
self.figure.add_widget(widget_type="Waveform1D", row=0, col=0, title="Widget 1")
|
||||
self.figure.add_widget(widget_type="Waveform1D", row=1, col=0, title="Widget 2")
|
||||
self.figure.add_widget(widget_type="Waveform1D", row=0, col=1, title="Widget 3")
|
||||
self.figure.add_widget(widget_type="Waveform1D", row=1, col=1, title="Widget 4")
|
||||
# self.figure.add_image(title="Image", row=1, col=1)
|
||||
self.figure.add_widget(widget_type="Waveform1D", row=0, col=1, title="Widget 2")
|
||||
self.figure.add_image(
|
||||
title="Image", row=1, col=0, color_map="viridis", color_bar="simple", vrange=(0, 100)
|
||||
)
|
||||
self.figure.add_image(title="Image", row=1, col=1, vrange=(0, 100))
|
||||
|
||||
self.w1 = self.figure[0, 0]
|
||||
self.w2 = self.figure[1, 0]
|
||||
self.w3 = self.figure[0, 1]
|
||||
self.w2 = self.figure[0, 1]
|
||||
self.w3 = self.figure[1, 0]
|
||||
self.w4 = self.figure[1, 1]
|
||||
|
||||
# curves for w1
|
||||
@ -522,14 +661,14 @@ class DebugWindow(QWidget): # pragma: no cover:
|
||||
)
|
||||
|
||||
# curves for w3
|
||||
self.w3.add_curve_scan("samx", "bpm4i", pen_style="dash")
|
||||
self.w3.add_curve_custom(
|
||||
x=[1, 2, 3, 4, 5],
|
||||
y=[1, 2, 3, 4, 5],
|
||||
label="curve-custom",
|
||||
color="blue",
|
||||
pen_style="dashdot",
|
||||
)
|
||||
# self.w3.add_curve_scan("samx", "bpm4i", pen_style="dash")
|
||||
# self.w3.add_curve_custom(
|
||||
# x=[1, 2, 3, 4, 5],
|
||||
# y=[1, 2, 3, 4, 5],
|
||||
# label="curve-custom",
|
||||
# color="blue",
|
||||
# pen_style="dashdot",
|
||||
# )
|
||||
|
||||
# curves for w4
|
||||
# self.w4.add_curve_scan("samx", "bpm4i", pen_style="dash")
|
||||
@ -541,6 +680,14 @@ class DebugWindow(QWidget): # pragma: no cover:
|
||||
# pen_style="dashdot",
|
||||
# )
|
||||
|
||||
# Image setting for w3
|
||||
self.w3.set_monitor("eiger")
|
||||
# self.w3.add_color_bar("simple")
|
||||
|
||||
# Image setting for w4
|
||||
self.w4.set_monitor("eiger")
|
||||
# self.w4.add_color_bar("full")
|
||||
|
||||
def add_debug_histo(self):
|
||||
image_data = np.random.normal(loc=100, scale=50, size=(100, 100)) # Example image data
|
||||
self.figure.add_image_with_histogram(image_data, row=2, col=0)
|
||||
|
||||
@ -1,3 +1,3 @@
|
||||
from .plot_base import AxisConfig, WidgetConfig, BECPlotBase
|
||||
from .waveform1d import Waveform1DConfig, BECWaveform1D, BECCurve
|
||||
from .image import BECImageShow, BECImageShowConfig, BECImageItem
|
||||
from .image import BECImageShow, ImageItemConfig, BECImageItem
|
||||
|
||||
@ -1,45 +1,75 @@
|
||||
from __future__ import annotations
|
||||
import scipy as sp
|
||||
|
||||
from collections import defaultdict
|
||||
from typing import Literal, Optional, Any
|
||||
from typing import Literal, Optional
|
||||
|
||||
import numpy as np
|
||||
import pyqtgraph as pg
|
||||
from PyQt6.QtWidgets import QMainWindow
|
||||
from pydantic import Field, BaseModel
|
||||
from qtpy.QtCore import QThread
|
||||
from pydantic import Field, BaseModel, ValidationError
|
||||
from pyqtgraph import mkBrush
|
||||
from qtpy import QtCore
|
||||
from qtpy.QtCore import Signal as pyqtSignal
|
||||
from qtpy.QtCore import Slot as pyqtSlot
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_lib import MessageEndpoints, RedisConnector
|
||||
from bec_lib.scan_data import ScanData
|
||||
from bec_widgets.utils import Colors, ConnectionConfig, BECConnector, EntryValidator, BECDispatcher
|
||||
from bec_widgets.utils import ConnectionConfig, BECConnector, BECDispatcher
|
||||
from bec_widgets.widgets.plots import BECPlotBase, WidgetConfig
|
||||
|
||||
|
||||
class ImageConfig(ConnectionConfig):
|
||||
pass
|
||||
class MonitorConfig(BaseModel):
|
||||
monitor: Optional[str] = Field(None, description="The name of the monitor.")
|
||||
fft: Optional[bool] = Field(False, description="Whether to perform FFT on the monitor data.")
|
||||
log: Optional[bool] = Field(False, description="Whether to perform log on the monitor data.")
|
||||
center_of_mass: Optional[bool] = Field(
|
||||
False, description="Whether to calculate the center of mass of the monitor data."
|
||||
)
|
||||
transpose: Optional[bool] = Field(
|
||||
False, description="Whether to transpose the monitor data before displaying."
|
||||
)
|
||||
rotation: Optional[int] = Field(
|
||||
None, description="The rotation angle of the monitor data before displaying."
|
||||
)
|
||||
|
||||
# TODO Decide if usefully to include port and host
|
||||
host: Optional[str] = Field(None, description="The host of the monitor.")
|
||||
port: Optional[int] = Field(None, description="The port of the monitor.")
|
||||
|
||||
|
||||
class BECImageShowConfig(WidgetConfig):
|
||||
pass
|
||||
class ImageItemConfig(ConnectionConfig):
|
||||
# color_map: Optional[str] = Field("magma", description="The color map of the image.")
|
||||
source: Optional[str] = Field(None, description="The source of the curve.")
|
||||
signals: MonitorConfig = Field(
|
||||
default_factory=MonitorConfig, description="The configuration of the monitor."
|
||||
)
|
||||
|
||||
|
||||
class BECImageItem(BECConnector, pg.ImageItem):
|
||||
class ImageConfig(WidgetConfig):
|
||||
color_map: Optional[str] = Field("magma", description="The color map of the image.")
|
||||
color_bar: Optional[Literal["simple", "full"]] = Field(
|
||||
"simple", description="The type of the color bar."
|
||||
)
|
||||
vrange: Optional[tuple[int, int]] = Field(
|
||||
None, description="The range of the color bar. If None, the range is automatically set."
|
||||
)
|
||||
images: dict[str, ImageItemConfig] = Field(
|
||||
{},
|
||||
description="The configuration of the images. The key is the name of the image.",
|
||||
)
|
||||
# TODO Decide if implement or not
|
||||
# x_transpose_axis:
|
||||
# y_transpose_axis:
|
||||
|
||||
|
||||
class BECImageItem(BECConnector, pg.ImageItem): # TODO decide how complex it should be
|
||||
USER_ACCESS = []
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: Optional[ImageConfig] = None,
|
||||
config: Optional[ImageItemConfig] = None,
|
||||
gui_id: Optional[str] = None,
|
||||
**kwargs,
|
||||
):
|
||||
if config is None:
|
||||
config = ImageConfig(widget_class=self.__class__.__name__)
|
||||
config = ImageItemConfig(widget_class=self.__class__.__name__)
|
||||
self.config = config
|
||||
else:
|
||||
self.config = config
|
||||
@ -52,14 +82,29 @@ class BECImageItem(BECConnector, pg.ImageItem):
|
||||
self.set(**kwargs)
|
||||
|
||||
def apply_config(self):
|
||||
pass
|
||||
...
|
||||
# self.set_color_map(self.config.color_map)
|
||||
|
||||
def set(self, **kwargs):
|
||||
pass
|
||||
|
||||
def set_color_map(self, cmap: str = "magma"):
|
||||
self.setColorMap(cmap)
|
||||
# self.config.color_map = cmap
|
||||
|
||||
|
||||
class BECImageShow(BECPlotBase):
|
||||
USER_ACCESS = ["show_image"]
|
||||
USER_ACCESS = [
|
||||
"set_vrange",
|
||||
"set_monitor",
|
||||
"set_color_map",
|
||||
"set_image",
|
||||
"set_processing",
|
||||
"enable_fft",
|
||||
"enable_log",
|
||||
"rotate",
|
||||
"transpose",
|
||||
]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@ -68,82 +113,185 @@ class BECImageShow(BECPlotBase):
|
||||
config: Optional[WidgetConfig] = None,
|
||||
client=None,
|
||||
gui_id: Optional[str] = None,
|
||||
monitor: Optional[str] = None,
|
||||
):
|
||||
if config is None:
|
||||
config = BECImageShowConfig(widget_class=self.__class__.__name__)
|
||||
config = ImageConfig(widget_class=self.__class__.__name__)
|
||||
super().__init__(
|
||||
parent=parent, parent_figure=parent_figure, config=config, client=client, gui_id=gui_id
|
||||
)
|
||||
|
||||
# Items to be added to the plot
|
||||
self.image = None
|
||||
self.color_bar = None
|
||||
|
||||
# Args to pass
|
||||
self.monitor = monitor
|
||||
|
||||
# init image and image thread
|
||||
self._init_image()
|
||||
self._init_image_thread(monitor=self.monitor)
|
||||
|
||||
def find_widget_by_id(self, item_id: str):
|
||||
if self.image.gui_id == item_id:
|
||||
return self.image
|
||||
|
||||
def apply_config(self): # TODO implement
|
||||
...
|
||||
|
||||
def _init_image(self):
|
||||
self.image = BECImageItem()
|
||||
self.addItem(self.image)
|
||||
self.addColorBar(self.image, values=(0, 100))
|
||||
# self.add_histogram()
|
||||
self.plot_item.addItem(self.image)
|
||||
self.config.images["device_monitor"] = self.image.config
|
||||
|
||||
# set mock data
|
||||
# self.image.setImage(np.random.rand(100, 100))
|
||||
# self.image.setOpts(axisOrder="row-major")
|
||||
# customising ImageItem
|
||||
self._add_color_bar(style=self.config.color_bar, vrange=self.config.vrange)
|
||||
self.set_color_map(cmap=self.config.color_map)
|
||||
|
||||
self.debug_stream()
|
||||
def _init_image_thread(self, monitor: str = None):
|
||||
self.monitor = monitor
|
||||
self.image.config.signals.monitor = monitor
|
||||
self.image_thread = ImageThread(client=self.client, monitor=monitor)
|
||||
self.image_thread.config = self.image.config.signals
|
||||
self.proxy_update_plot = pg.SignalProxy(
|
||||
self.image_thread.image_updated, rateLimit=25, slot=self.on_image_update
|
||||
)
|
||||
|
||||
def debug_stream(self):
|
||||
device = "eiger"
|
||||
self.image_thread = ImageThread(client=self.client, monitor=device)
|
||||
# self.image_thread.start()
|
||||
self.image_thread.image_updated.connect(self.on_image_update)
|
||||
def _add_color_bar(
|
||||
self, style: Literal["simple,full"] = "simple", vrange: tuple[int, int] = (0, 100)
|
||||
):
|
||||
if style == "simple":
|
||||
self.color_bar = pg.ColorBarItem(colorMap=self.config.color_map)
|
||||
if vrange is not None:
|
||||
self.color_bar.setLevels(low=vrange[0], high=vrange[1])
|
||||
self.color_bar.setImageItem(self.image)
|
||||
self.addItem(self.color_bar, row=0, col=1)
|
||||
self.config.color_bar = "simple"
|
||||
elif style == "full":
|
||||
# Setting histogram
|
||||
self.color_bar = pg.HistogramLUTItem()
|
||||
self.color_bar.setImageItem(self.image)
|
||||
self.color_bar.gradient.loadPreset(self.config.color_map)
|
||||
if vrange is not None:
|
||||
self.color_bar.setLevels(min=vrange[0], max=vrange[1])
|
||||
self.color_bar.setHistogramRange(
|
||||
vrange[0] - 0.1 * vrange[0], vrange[1] + 0.1 * vrange[1]
|
||||
)
|
||||
|
||||
def add_color_bar(self, vmap: tuple[int, int] = (0, 100)):
|
||||
self.addColorBar(self.image, values=vmap)
|
||||
# Adding histogram to the layout
|
||||
self.addItem(self.color_bar, row=0, col=1)
|
||||
|
||||
def add_histogram(self):
|
||||
# Create HistogramLUTWidget
|
||||
self.histogram = pg.HistogramLUTWidget()
|
||||
# save settings
|
||||
self.config.color_bar = "full"
|
||||
else:
|
||||
raise ValueError("style should be 'simple' or 'full'")
|
||||
|
||||
# Link HistogramLUTWidget to ImageItem
|
||||
self.histogram.setImageItem(self.image)
|
||||
# def color_bar_switch(self, style: Literal["simple,full"] = "simple"): #TODO check if possible
|
||||
# if style == "simple" and self.config.color_bar == "full":
|
||||
# self.color_bar.remove()
|
||||
|
||||
# def show_image(
|
||||
# self,
|
||||
# image: np.ndarray,
|
||||
# scale: Optional[tuple] = None,
|
||||
# pos: Optional[tuple] = None,
|
||||
# auto_levels: Optional[bool] = True,
|
||||
# auto_range: Optional[bool] = True,
|
||||
# lut: Optional[list] = None,
|
||||
# opacity: Optional[float] = 1.0,
|
||||
# auto_downsample: Optional[bool] = True,
|
||||
# ):
|
||||
# self.image.setImage(
|
||||
# image,
|
||||
# scale=scale,
|
||||
# pos=pos,
|
||||
# autoLevels=auto_levels,
|
||||
# autoRange=auto_range,
|
||||
# lut=lut,
|
||||
# opacity=opacity,
|
||||
# autoDownsample=auto_downsample,
|
||||
# )
|
||||
#
|
||||
# def remove(self):
|
||||
# self.image.clear()
|
||||
# self.removeItem(self.image)
|
||||
# self.image = None
|
||||
# super().remove()
|
||||
def set_vrange(self, vmin: float, vmax: float):
|
||||
self.image.setLevels([vmin, vmax])
|
||||
if self.color_bar is not None:
|
||||
if self.config.color_bar == "simple":
|
||||
self.color_bar.setLevels(low=vmin, high=vmax)
|
||||
elif self.config.color_bar == "full":
|
||||
self.color_bar.setLevels(min=vmin, max=vmax)
|
||||
self.color_bar.setHistogramRange(vmin - 0.1 * vmin, vmax + 0.1 * vmax)
|
||||
|
||||
def set_monitor(self, monitor: str = None): ...
|
||||
def set_monitor(self, monitor: str = None) -> None:
|
||||
"""
|
||||
Set/update monitor device.
|
||||
Args:
|
||||
monitor(str): Name of the monitor.
|
||||
"""
|
||||
self.image_thread.set_monitor(monitor)
|
||||
self.image.config.signals.monitor = monitor
|
||||
|
||||
def set_zmq(self, address: str = None): ...
|
||||
def set_color_map(self, cmap: str = "magma"):
|
||||
self.image.set_color_map(cmap)
|
||||
if self.color_bar is not None:
|
||||
if self.config.color_bar == "simple":
|
||||
self.color_bar.setColorMap(cmap)
|
||||
elif self.config.color_bar == "full":
|
||||
self.color_bar.gradient.loadPreset(cmap)
|
||||
|
||||
@pyqtSlot(np.ndarray) # TODO specify format
|
||||
# def set_zmq(self, address: str = None): # TODO to be implemented
|
||||
# ...
|
||||
|
||||
def set_processing(
|
||||
self, fft: bool = False, log: bool = False, rotation: int = None, transpose: bool = False
|
||||
):
|
||||
"""
|
||||
Set the processing of the monitor data.
|
||||
Args:
|
||||
fft(bool): Whether to perform FFT on the monitor data.
|
||||
log(bool): Whether to perform log on the monitor data.
|
||||
rotation(int): The rotation angle of the monitor data before displaying.
|
||||
transpose(bool): Whether to transpose the monitor data before displaying.
|
||||
"""
|
||||
self.image.config.signals.fft = fft
|
||||
self.image.config.signals.log = log
|
||||
self.image.config.signals.rotation = rotation
|
||||
self.image.config.signals.transpose = transpose
|
||||
self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
def enable_fft(self, enable: bool = True): # TODO enable processing of already taken images
|
||||
self.image.config.signals.fft = enable
|
||||
self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
def enable_log(self, enable: bool = True):
|
||||
self.image.config.signals.log = enable
|
||||
self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
def rotate(self, angle: int): # TODO fine tune, can be byt any angle not just by 90deg?
|
||||
self.image.config.signals.rotation = angle
|
||||
self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
def transpose(self): # TODO do enable or not?
|
||||
self.image.config.signals.transpose = not self.image.config.signals.transpose
|
||||
self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
# def enable_center_of_mass(self, enable: bool = True): #TODO check and enable
|
||||
# self.image.config.signals.center_of_mass = enable
|
||||
# self.image_thread.update_config(self.image.config.signals)
|
||||
|
||||
@pyqtSlot(np.ndarray)
|
||||
def on_image_update(self, image):
|
||||
self.image.updateImage(image)
|
||||
self.image.updateImage(image[0])
|
||||
|
||||
def set_image(self, data: np.ndarray):
|
||||
"""
|
||||
Set the image to be displayed.
|
||||
Args:
|
||||
data(np.ndarray): The image to be displayed.
|
||||
"""
|
||||
self.imageItem.setImage(data)
|
||||
self.image_thread.set_monitor(None)
|
||||
|
||||
def cleanup(self): # TODO test
|
||||
self.image_thread.quit()
|
||||
self.image_thread.wait()
|
||||
self.image_thread.deleteLater()
|
||||
self.image_thread = None
|
||||
self.image.remove()
|
||||
self.color_bar.remove()
|
||||
super().cleanup()
|
||||
|
||||
|
||||
class ImageThread(QThread):
|
||||
image_updated = pyqtSignal(np.ndarray) # TODO add type
|
||||
image_updated = pyqtSignal(np.ndarray)
|
||||
|
||||
def __init__(self, parent=None, client=None, monitor: str = None, port: int = None):
|
||||
super().__init__()
|
||||
def __init__(
|
||||
self,
|
||||
parent=None,
|
||||
client=None,
|
||||
monitor: str = None,
|
||||
monitor_config: MonitorConfig = None,
|
||||
host: str = None,
|
||||
port: int | str = None,
|
||||
):
|
||||
super().__init__(parent=parent)
|
||||
|
||||
bec_dispatcher = BECDispatcher()
|
||||
self.client = bec_dispatcher.client if client is None else client
|
||||
@ -153,27 +301,48 @@ class ImageThread(QThread):
|
||||
|
||||
# Monitor Device
|
||||
self.monitor = monitor
|
||||
self.config = monitor_config
|
||||
|
||||
# Connection
|
||||
self.port = port
|
||||
if self.port is None:
|
||||
self.port = self.client.connector.host
|
||||
# self.connector = RedisConnector(self.port)
|
||||
self.connector = RedisConnector("localhost:6379")
|
||||
self.stream_consumer = None
|
||||
self.host = host
|
||||
self.port = str(port)
|
||||
if self.host is None:
|
||||
self.host = self.client.connector.host
|
||||
self.port = self.client.connector.port
|
||||
self.connector = RedisConnector(f"{self.host}:{self.port}")
|
||||
|
||||
self.stream_consumer = None
|
||||
if self.monitor is not None:
|
||||
self.connect_stream_consumer(self.monitor)
|
||||
|
||||
def set_monitor(self, monitor: str = None) -> None:
|
||||
def update_config(self, config: MonitorConfig | dict): # TODO include monitor update in config?
|
||||
"""
|
||||
Update the monitor configuration.
|
||||
Args:
|
||||
config(MonitorConfig|dict): The new monitor configuration.
|
||||
"""
|
||||
if isinstance(config, dict):
|
||||
config = MonitorConfig(**config)
|
||||
self.config = config
|
||||
|
||||
def set_monitor(self, monitor: str = None) -> None: # TODO check monitor update with the config
|
||||
"""
|
||||
Set/update monitor device.
|
||||
Args:
|
||||
monitor(str): Name of the monitor.
|
||||
"""
|
||||
self.monitor = monitor
|
||||
if self.monitor is not None:
|
||||
self.connect_stream_consumer(self.monitor)
|
||||
elif monitor is None:
|
||||
self.stream_consumer.shutdown()
|
||||
|
||||
def connect_stream_consumer(self, device):
|
||||
"""
|
||||
Connect to the stream consumer for the device.
|
||||
Args:
|
||||
device(str): Name of the device.
|
||||
"""
|
||||
if self.stream_consumer is not None:
|
||||
self.stream_consumer.shutdown()
|
||||
|
||||
@ -187,11 +356,31 @@ class ImageThread(QThread):
|
||||
|
||||
print(f"Stream consumer started for device: {device}")
|
||||
|
||||
def process_FFT(self, data: np.ndarray) -> np.ndarray:
|
||||
return np.fft.fft2(data)
|
||||
def process_FFT(self, data: np.ndarray) -> np.ndarray: # TODO check functionality
|
||||
return np.abs(np.fft.fftshift(np.fft.fft2(data)))
|
||||
|
||||
def center_of_mass(self, data: np.ndarray) -> tuple:
|
||||
return np.unravel_index(np.argmax(data), data.shape)
|
||||
def rotation(self, data: np.ndarray, angle: int) -> np.ndarray:
|
||||
return np.rot90(data, k=angle, axes=(0, 1))
|
||||
|
||||
def transpose(self, data: np.ndarray) -> np.ndarray:
|
||||
return np.transpose(data)
|
||||
|
||||
def log(self, data: np.ndarray) -> np.ndarray:
|
||||
return np.log10(np.abs(data))
|
||||
|
||||
# def center_of_mass(self, data: np.ndarray) -> tuple: # TODO check functionality
|
||||
# return np.unravel_index(np.argmax(data), data.shape)
|
||||
|
||||
def post_processing(self, data: np.ndarray) -> np.ndarray:
|
||||
if self.config.fft:
|
||||
data = self.process_FFT(data)
|
||||
if self.config.rotation is not None:
|
||||
data = self.rotation(data, self.config.rotation)
|
||||
if self.config.transpose:
|
||||
data = self.transpose(data)
|
||||
if self.config.log:
|
||||
data = self.log(data)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _streamer_cb(msg, *, parent, **_kwargs) -> None:
|
||||
@ -199,48 +388,7 @@ class ImageThread(QThread):
|
||||
metadata = msg_device.metadata
|
||||
|
||||
data = msg_device.content["data"]
|
||||
|
||||
data = parent.post_processing(data)
|
||||
|
||||
parent.image_updated.emit(data)
|
||||
|
||||
|
||||
class BECImageShowWithHistogram(pg.GraphicsLayoutWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent=parent)
|
||||
|
||||
# Create ImageItem and HistogramLUTItem
|
||||
self.imageItem = pg.ImageItem()
|
||||
self.histogram = pg.HistogramLUTItem()
|
||||
|
||||
# Link Histogram to ImageItem
|
||||
self.histogram.setImageItem(self.imageItem)
|
||||
|
||||
# Create a layout within the GraphicsLayoutWidget
|
||||
self.layout = self
|
||||
|
||||
# Add ViewBox and Histogram to the layout
|
||||
self.viewBox = self.addViewBox(row=0, col=0)
|
||||
self.viewBox.addItem(self.imageItem)
|
||||
self.viewBox.setAspectLocked(True) # Lock the aspect ratio
|
||||
|
||||
# Add Histogram to the layout in the same cell
|
||||
self.addItem(self.histogram, row=0, col=1)
|
||||
self.histogram.setMaximumWidth(200) # Adjust the width of the histogram to fit
|
||||
|
||||
def setImage(self, image):
|
||||
"""Set the image to be displayed."""
|
||||
self.imageItem.setImage(image)
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# import sys
|
||||
# from qtpy.QtWidgets import QApplication
|
||||
#
|
||||
# bec_dispatcher = BECDispatcher()
|
||||
# client = bec_dispatcher.client
|
||||
# client.start()
|
||||
#
|
||||
# app = QApplication(sys.argv)
|
||||
# win = QMainWindow()
|
||||
# img = BECImageShow(client=client)
|
||||
# win.setCentralWidget(img)
|
||||
# win.show()
|
||||
# sys.exit(app.exec_())
|
||||
|
||||
Reference in New Issue
Block a user