bec/pxiii_bec
0
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Compare commits

base: bec:main
Branches Tags
bec:main bec:fix/samcam bec:development bec:feature/fittebale-autoplotting bec:feature/bec-demo-pxiii bec:feature/simple-grid-scan-demo bec:pxi_liveplotting
..
compare: bec:feature/fittebale-autoplotting
Branches Tags
bec:fix/samcam bec:main bec:development bec:feature/fittebale-autoplotting bec:feature/bec-demo-pxiii bec:feature/simple-grid-scan-demo bec:pxi_liveplotting

7 Commits
main ... feature/fi

Author SHA1 Message Date
Unknown MX Person
d1844eaeba WIP 2025-03-25 17:42:55 +01:00
gac-x06da
d53ec4c14c WIP 2025-03-21 09:57:25 +01:00
gac-x06da
4091f11b0a WIP 2025-03-21 09:27:29 +01:00
gac-x06da
29ae5c196b WIP 2025-03-14 16:10:16 +01:00
gac-x06da
74521da7b3 Blacking 2025-03-14 12:40:23 +01:00
gac-x06da
a5f844b816 Keyword scans for 2D 2025-03-14 12:32:57 +01:00
gac-x06da
34d4d6ef8c Added AttributeError to updates 2025-03-14 12:32:57 +01:00
48 changed files with 436 additions and 1217 deletions
Expand all files Collapse all files

9
.copier-answers.yml
View File

@@ -1,9 +0,0 @@
# Do not edit this file!
# It is needed to track the repo template version, and editing may break things.
# This file will be overwritten by copier on template updates.
_commit: v1.2.2
_src_path: https://github.com/bec-project/plugin_copier_template.git
make_commit: false
project_name: pxiii_bec
widget_plugins_input: null

3
.git_hooks/post-commit
View File

@@ -1,3 +0,0 @@
SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
semantic-release changelog -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__
semantic-release version -D version_variable=$SCRIPT_DIR/../../semantic_release/__init__.py:__version__

3
.git_hooks/pre-commit
View File

@@ -1,3 +0,0 @@
black --line-length=100 $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
isort --line-length=100 --profile=black --multi-line=3 --trailing-comma $(git diff --cached --name-only --diff-filter=ACM -- '*.py')
git add $(git diff --cached --name-only --diff-filter=ACM -- '*.py')

97
.gitea/workflows/ci.yml
View File

@@ -1,97 +0,0 @@
name: CI for pxiii_bec
on:
push:
pull_request:
workflow_dispatch:
inputs:
BEC_WIDGETS_BRANCH:
description: "Branch of BEC Widgets to install"
required: false
type: string
default: "main"
BEC_CORE_BRANCH:
description: "Branch of BEC Core to install"
required: false
type: string
default: "main"
OPHYD_DEVICES_BRANCH:
description: "Branch of Ophyd Devices to install"
required: false
type: string
default: "main"
BEC_PLUGIN_REPO_BRANCH:
description: "Branch of the BEC Plugin Repository to install"
required: false
type: string
default: "main"
PYTHON_VERSION:
description: "Python version to use"
required: false
type: string
default: "3.11"
permissions:
pull-requests: write
jobs:
test:
runs-on: ubuntu-latest
env:
QTWEBENGINE_DISABLE_SANDBOX: 1
QT_QPA_PLATFORM: "offscreen"
steps:
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "${{ inputs.PYTHON_VERSION || '3.11' }}"
- name: Checkout BEC Core
uses: actions/checkout@v4
with:
repository: bec/bec
ref: "${{ inputs.BEC_CORE_BRANCH || 'main' }}"
path: ./bec
- name: Checkout Ophyd Devices
uses: actions/checkout@v4
with:
repository: bec/ophyd_devices
ref: "${{ inputs.OPHYD_DEVICES_BRANCH || 'main' }}"
path: ./ophyd_devices
- name: Checkout BEC Widgets
uses: actions/checkout@v4
with:
repository: bec/bec_widgets
ref: "${{ inputs.BEC_WIDGETS_BRANCH || 'main' }}"
path: ./bec_widgets
- name: Checkout BEC Plugin Repository
uses: actions/checkout@v4
with:
repository: bec/pxiii_bec
ref: "${{ inputs.BEC_PLUGIN_REPO_BRANCH || github.head_ref || github.sha }}"
path: ./pxiii_bec
- name: Install dependencies
shell: bash
run: |
sudo apt-get update
sudo apt-get install -y libgl1 libegl1 x11-utils libxkbcommon-x11-0 libdbus-1-3 xvfb
sudo apt-get -y install libnss3 libxdamage1 libasound2t64 libatomic1 libxcursor1
- name: Install Python dependencies
shell: bash
run: |
pip install uv
uv pip install --system -e ./ophyd_devices
uv pip install --system -e ./bec/bec_lib[dev]
uv pip install --system -e ./bec/bec_ipython_client
uv pip install --system -e ./bec/bec_server[dev]
uv pip install --system -e ./bec_widgets[dev,pyside6]
uv pip install --system -e ./pxiii_bec
- name: Run Pytest with Coverage
id: coverage
run: pytest --random-order --cov=./pxiii_bec --cov-config=./pxiii_bec/pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail ./pxiii_bec/tests/ || test $? -eq 5

3
.gitignore vendored
View File

@@ -8,9 +8,6 @@
**/.pytest_cache
**/*.egg*
# recovery_config files
recovery_config_*
# file writer data
**.h5

12
.gitlab-ci.yml
View File

@@ -1,7 +1,7 @@
include:
- file: /templates/plugin-repo-template.yml
inputs:
name: pxiii_bec
target: pxiii_bec
branch: $CHILD_PIPELINE_BRANCH
project: bec/awi_utils
- project: bec/awi_utils
file: /templates/plugin-repo-template.yml
inputs:
name: "pxiii"
target: "pxiii_bec"
branch: $CHILD_PIPELINE_BRANCH

5
LICENSE
View File

@@ -1,7 +1,6 @@
BSD 3-Clause License
Copyright (c) 2025, Paul Scherrer Institute
Copyright (c) 2024, Paul Scherrer Institute
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
@@ -26,4 +25,4 @@ DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

1
bin/.gitignore vendored
View File

@@ -1 +0,0 @@
# Add anything you don't want to check in to git, e.g. very large files

11
deployment/autodeploy_versions Normal file
View File

@@ -0,0 +1,11 @@
# This file is used to select the BEC and Ophyd Devices version for the auto deployment process.
# Do not edit this file unless you know what you are doing!
# The version can be a git tag, branch or commit hash.
# BEC version to use
BEC_AUTODEPLOY_VERSION="master"
# ophyd_devices version to use
OPHYD_DEVICES_AUTODEPLOY_VERSION="master"

18
deployment/bec-server-config.yaml Normal file
View File

@@ -0,0 +1,18 @@
redis:
host: localhost
port: 6379
mongodb:
host: localhost
port: 27017
scibec:
host: http://[::1]
port: 3030
beamline: "PXIII"
service_config:
general:
reset_queue_on_cancel: True
enforce_ACLs: False
file_writer:
plugin: default_NeXus_format
base_path: ./

29
deployment/deploy.sh Executable file
View File

@@ -0,0 +1,29 @@
# deployment script to be translated to Ansible
# can be removed once we have the autodeployment in place
BEAMLINE_REPO=gitlab.psi.ch:bec/pxiii-bec.git
git clone git@$BEAMLINE_REPO
module add psi-python311/2024.02
# start redis
docker run --network=host --name redis-bec -d redis
# alternative:
# conda install -y redis; redis-server &
# get the target versions for ophyd_devices and BEC
source ./pxiii-bec/deployment/autodeploy_versions
git clone -b $OPHYD_DEVICES_AUTODEPLOY_VERSION https://gitlab.psi.ch/bec/ophyd_devices.git
git clone -b $BEC_AUTODEPLOY_VERSION https://gitlab.psi.ch/bec/bec.git
# install BEC
cd bec
source ./bin/install_bec_dev.sh
cd ../
pip install -e ./pxiii-bec
# start the BEC server
bec-server start --config ./pxiii-bec/deployment/bec-server-config.yaml

15
pxiii_bec/bec_ipython_client/startup/pre_startup.py
View File

@@ -1,14 +1,10 @@
"""
Pre-startup script for BEC client. This script is executed before the BEC client
is started. It can be used to add additional command line arguments.
is started. It can be used to add additional command line arguments.
"""
import os
from bec_lib.service_config import ServiceConfig
import pxiii_bec
def extend_command_line_args(parser):
"""
@@ -22,11 +18,6 @@ def extend_command_line_args(parser):
def get_config() -> ServiceConfig:
"""
Create and return the ServiceConfig for the plugin repository
Create and return the service configuration.
"""
deployment_path = os.path.dirname(os.path.dirname(os.path.dirname(pxiii_bec.__file__)))
files = os.listdir(deployment_path)
if "bec_config.yaml" in files:
return ServiceConfig(config_path=os.path.join(deployment_path, "bec_config.yaml"))
else:
return ServiceConfig(redis={"host": "localhost", "port": 6379})
return ServiceConfig(redis={"host": "x06da-bec-001", "port": 6379})

135
pxiii_bec/bec_widgets/ComissioningGUIDraft.ui
View File

@@ -1,135 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>1801</width>
<height>1459</height>
</rect>
</property>
<property name="minimumSize">
<size>
<width>1801</width>
<height>1459</height>
</size>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QTabWidget" name="tabWidget">
<property name="currentIndex">
<number>0</number>
</property>
<widget class="QWidget" name="tab">
<attribute name="title">
<string>Control Panel</string>
</attribute>
<layout class="QGridLayout" name="gridLayout" rowstretch="3,4" columnstretch="2,5">
<item row="0" column="0">
<widget class="Waveform" name="waveform"/>
</item>
<item row="0" column="1">
<widget class="ScanControl" name="scan_control"/>
</item>
<item row="1" column="0">
<widget class="ScanHistory" name="scan_history"/>
</item>
<item row="1" column="1">
<widget class="BECQueue" name="bec_queue"/>
</item>
</layout>
</widget>
<widget class="QWidget" name="tab_2">
<attribute name="title">
<string>Logbook</string>
</attribute>
<layout class="QGridLayout" name="gridLayout_3">
<item row="0" column="0">
<widget class="QLabel" name="label">
<property name="font">
<font>
<pointsize>24</pointsize>
</font>
</property>
<property name="text">
<string>Coming soon...</string>
</property>
</widget>
</item>
</layout>
</widget>
<widget class="QWidget" name="tab_3">
<attribute name="title">
<string>Take a break</string>
</attribute>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="Minesweeper" name="minesweeper"/>
</item>
<item row="0" column="1">
<spacer name="horizontalSpacer">
<property name="orientation">
<enum>Qt::Orientation::Horizontal</enum>
</property>
<property name="sizeHint" stdset="0">
<size>
<width>1073</width>
<height>20</height>
</size>
</property>
</spacer>
</item>
<item row="1" column="0">
<spacer name="verticalSpacer">
<property name="orientation">
<enum>Qt::Orientation::Vertical</enum>
</property>
<property name="sizeHint" stdset="0">
<size>
<width>20</width>
<height>40</height>
</size>
</property>
</spacer>
</item>
</layout>
</widget>
</widget>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>ScanControl</class>
<extends>QWidget</extends>
<header>scan_control</header>
</customwidget>
<customwidget>
<class>Waveform</class>
<extends>QWidget</extends>
<header>waveform</header>
</customwidget>
<customwidget>
<class>BECQueue</class>
<extends>QWidget</extends>
<header>bec_queue</header>
</customwidget>
<customwidget>
<class>Minesweeper</class>
<extends>QWidget</extends>
<header>minesweeper</header>
</customwidget>
<customwidget>
<class>ScanHistory</class>
<extends>QWidget</extends>
<header>scan_history</header>
</customwidget>
</customwidgets>
<resources/>
<connections/>
</ui>

2
pxiii_bec/bec_widgets/__init__.py
View File

@@ -1 +1 @@
from .auto_updates import PlotUpdate

173
pxiii_bec/bec_widgets/auto_updates.py Normal file
View File

@@ -0,0 +1,173 @@
from bec_widgets.cli.auto_updates import AutoUpdates, ScanInfo
from bec_widgets.cli.rpc.rpc_base import RPCResponseTimeoutError
class PlotUpdate(AutoUpdates):
create_default_dock = True
dock_name = "default_dock"
enabled = True
_scan_msg = None
# def __init__(self, gui: BECDockArea):
# super().__init__(gui)
def start_default_dock(self):
"""
Create a default dock for the auto updates.
"""
self._default_dock = self.gui.add_dock(self.dock_name)
self._default_dock.add_widget("BECWaveformWidget")
self._default_fig = self._default_dock.widget_list[0]
def do_update(self, msg):
"""Save the original scan message for future use"""
self._scan_msg = msg
return super().do_update(msg)
# def simple_line_scan(self, info: ScanInfo) -> None:
# """
# Simple line scan.
# """
# dev_x = info.scan_report_devices[0]
# dev_y = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
# if not dev_y:
# return
# self.figure.clear_all()
# plt = self.figure.plot(dev_x, dev_y)
# plt.set(title=f"PXIII: Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
def plot_handler(self, info: ScanInfo) -> None:
"""Simple keyword handler for 'plot'
This simple keyword handler looks for the keyword 'plot' in the scan arguments.
This allows the user to explictly specify the desired data source. Useful for alignment
scans.
"""
print(info.scan_report_devices)
dev_x = info.scan_report_devices[0]
# Keyword lookup for 'plot' and 'fit'
if "kwargs" in self._scan_msg.info:
signals = self._scan_msg.info["kwargs"].get("plot", None)
fit = self._scan_msg.info["kwargs"].get("fit", None)
else:
signals = None
fit = None
if not signals:
return
if isinstance(signals, str):
signals = [signals]
if isinstance(fit, str):
fit = [fit]
# try:
# self.gui.clear_all()
# except RPCResponseTimeoutError:
# pass
# try:
# self._default_dock = self.gui.add_dock(self.dock_name)
# except RPCResponseTimeoutError:
# pass
# try:
# self._default_fig = self._default_dock.add_widget("BECWaveformWidget")
# except RPCResponseTimeoutError:
# pass
# try:
# self._default_fig = self._default_dock.widget_list[0]
# except RPCResponseTimeoutError:
# pass
# # dck1 = self._default_dock
# # dck1.clear_all()
plt1 = self.get_default_figure()
# # Clear figure
# try:
# old_data = yield plt1.get_all_data()
# except RPCResponseTimeoutError:
# old_data = [1,2,3,4,5,6,7,8,9]
# print(old_data)
# for _ in range(len(old_data)):
# try:
# plt1.remove_curve(-1)
# except RPCResponseTimeoutError:
# pass
# except Exception as ex:
# print(f"{ex}\t{type(ex)}")
# clear_all() will throw RPCResponseTimeoutError
# try:
# plt1.clear_all()
# except RPCResponseTimeoutError:
# pass
print(type(plt1), plt1)
print(f"Plotted signals: {signals}")
# plot() will throw RPCResponseTimeoutError
try:
if len(info.scan_report_devices) == 2:
# 2D plot
dev_x = info.scan_report_devices[0]
dev_y = info.scan_report_devices[1]
plt1.plot(
x_name=dev_x,
y_name=dev_y,
z_name=signals[0],
title=f"Scan {info.scan_number}",
x_label=dev_x,
y_label=dev_y,
z_label=signals[0],
)
elif len(info.scan_report_devices) == 1:
# 1D plot
dev_x = info.scan_report_devices[0]
for sig in signals:
try:
plt1.plot(
x_name=dev_x,
y_name=sig,
title=f"Scan {info.scan_number}",
x_label=dev_x,
y_label=sig,
)
except RPCResponseTimeoutError:
pass
else:
# Default is 1D
dev_x = info.scan_report_devices[0]
for sig in signals:
try:
plt1.plot(
x_name=dev_x,
y_name=sig,
title=f"Scan {info.scan_number}",
x_label=dev_x,
y_label=sig,
)
except RPCResponseTimeoutError:
pass
except RPCResponseTimeoutError:
pass
if fit is not None and len(fit):
dev_x = info.scan_report_devices[0]
sig = signals[0]
plt1.add_dap(dev_x, sig, dap=fit[0])
def handler(self, info: ScanInfo) -> None:
"""Dock configuration handler"""
# EXAMPLES:
# if info.scan_name == "line_scan" and info.scan_report_devices:
# self.simple_line_scan(info)
# return
# if info.scan_name == "grid_scan" and info.scan_report_devices:
# self.run_grid_scan_update(info)
# return
super().handler(info)
self.plot_handler(info)

1
pxiii_bec/bec_widgets/auto_updates/__init__.py
View File

@@ -1 +0,0 @@
from .auto_updates import AutoUpdates

71
pxiii_bec/bec_widgets/auto_updates/auto_updates.py
View File

@@ -1,71 +0,0 @@
from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates
from bec_lib.messages import ScanStatusMessage
from bec_widgets.cli.rpc.rpc_base import RPCResponseTimeoutError
class PlotUpdate(AutoUpdates):
#######################################################################
################# GUI Callbacks #######################################
#######################################################################
def on_start(self) -> None:
"""
Procedure to run when the auto updates are enabled.
"""
self.start_default_dock()
def on_stop(self) -> None:
"""
Procedure to run when the auto updates are disabled.
"""
def on_scan_open(self, msg: ScanStatusMessage) -> None:
"""
Procedure to run when a scan starts.
Args:
msg (ScanStatusMessage): The scan status message.
"""
if msg.scan_name == "line_scan" and msg.scan_report_devices:
return self.simple_line_scan(msg)
if msg.scan_name == "grid_scan" and msg.scan_report_devices:
return self.simple_grid_scan(msg)
dev_x = msg.scan_report_devices[0]
if "kwargs" in msg.request_inputs:
dev_y = msg.request_inputs["kwargs"].get("plot", None)
if dev_y is not None:
# Set the dock to the waveform widget
wf = self.set_dock_to_widget("Waveform")
# Clear the waveform widget and plot the data
wf.clear_all()
wf.plot(
x_name=dev_x,
y_name=dev_y,
label=f"Scan {msg.info.scan_number} - {dev_y}",
title=f"Scan {msg.info.scan_number}",
x_label=dev_x,
y_label=dev_y,
)
elif msg.scan_report_devices:
return self.best_effort(msg)
return None
def on_scan_closed(self, msg: ScanStatusMessage) -> None:
"""
Procedure to run when a scan ends.
Args:
msg (ScanStatusMessage): The scan status message.
"""
def on_scan_abort(self, msg: ScanStatusMessage) -> None:
"""
Procedure to run when a scan is aborted.
Args:
msg (ScanStatusMessage): The scan status message.
"""

0
pxiii_bec/bec_widgets/widgets/__init__.py
View File

40
pxiii_bec/bec_widgets/widgets/client.py
View File

@@ -1,40 +0,0 @@
# This file was automatically generated by generate_cli.py
# type: ignore
from __future__ import annotations
from bec_lib.logger import bec_logger
from bec_widgets.cli.rpc.rpc_base import RPCBase, rpc_call
logger = bec_logger.logger
# pylint: skip-file
_Widgets = {
"ScanHistory": "ScanHistory",
}
class ScanHistory(RPCBase):
@rpc_call
def select_scan_from_history(self, value: "int") -> "None":
"""
Set scan from CLI.
Args:
value (int) : value from history -1 ...-10000
"""
@rpc_call
def add_scan_from_history(self) -> "None":
"""
Load selected scan from history.
"""
@rpc_call
def clear_plot(self) -> "None":
"""
Delete all curves on the plot.
"""

0
pxiii_bec/bec_widgets/widgets/scan_history/__init__.py
View File

15
pxiii_bec/bec_widgets/widgets/scan_history/register_scan_history.py
View File

@@ -1,15 +0,0 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from pxiii_bec.bec_widgets.widgets.scan_history.scan_history_plugin import ScanHistoryPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(ScanHistoryPlugin())
if __name__ == "__main__": # pragma: no cover
main()

191
pxiii_bec/bec_widgets/widgets/scan_history/scan_history.py
View File

@@ -1,191 +0,0 @@
from __future__ import annotations
import os
from typing import TYPE_CHECKING, TypedDict
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.ui_loader import UILoader
from qtpy.QtWidgets import QVBoxLayout, QWidget
logger = bec_logger.logger
if TYPE_CHECKING:
from qtpy.QtWidgets import QPushButton, QLabel, QSpinBox
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.editors.text_box.text_box import TextBox
class ScanHistoryUIComponents(TypedDict):
waveform: Waveform
metadata_text_box: TextBox
monitor_label: QLabel
monitor_combobox: DeviceComboBox
history_label: QLabel
history_spin_box: QSpinBox
history_add: QPushButton
history_clear: QPushButton
class ScanHistory(BECWidget, QWidget):
USER_ACCESS = ["select_scan_from_history", "add_scan_from_history", "clear_plot"]
PLUGIN = True
ui_file = "./scan_history.ui"
components: ScanHistoryUIComponents
def __init__(self, parent=None, **kwargs):
super().__init__(parent=parent, **kwargs)
self._load_ui()
def _load_ui(self):
current_path = os.path.dirname(__file__)
self.ui = UILoader(self).loader(os.path.join(current_path, self.ui_file))
layout = QVBoxLayout()
layout.addWidget(self.ui)
self.setLayout(layout)
self.components: ScanHistoryUIComponents = {
"waveform" : self.ui.waveform,
"metadata_text_box" : self.ui.metadata_text_box,
"monitor_label" : self.ui.monitor_label,
"monitor_combobox" : self.ui.monitor_combobox,
"history_label" : self.ui.history_label,
"history_spin_box" : self.ui.history_spin_box,
"history_add" : self.ui.history_add,
"history_clear" : self.ui.history_clear,
}
icon_options = {"size": (16, 16), "convert_to_pixmap": False}
self.components['monitor_combobox'].apply_filter = False
self.components['monitor_combobox'].devices = ['dccm_diode_bottom', 'dccm_diode_top', 'dccm_xbpm', 'ssxbpm', 'xbox_diode']
self.components['history_spin_box'].setMinimum(-10000)
self.components['history_spin_box'].setMaximum(-1)
self.components['history_spin_box'].valueChanged.connect(self._scan_history_selected)
self._scan_history_selected(-1)
self.components['history_spin_box'].setValue(-1)
self.components['history_add'].setText("Load")
self.components['history_add'].setStyleSheet(
"background-color: #129490; color: white; font-weight: bold; font-size: 12px;"
)
self.components['history_clear'].setText("Clear")
self.components['history_clear'].setStyleSheet(
"background-color: #065143; color: white; font-weight: bold; font-size: 12px;"
)
self.components['history_add'].clicked.connect(self._refresh_plot)
self.components['history_clear'].clicked.connect(self.clear_plot)
self.setWindowTitle("Scan History")
self._scan_history_selected(-1)
@SafeSlot()
def add_scan_from_history(self) -> None:
"""Load selected scan from history."""
self.components['history_add'].click()
@SafeSlot()
def clear_plot(self) -> None:
"""Delete all curves on the plot."""
self.components['waveform'].clear_all()
@SafeSlot()
def _refresh_plot(self) -> None:
"""Refresh plot."""
spin_box_value = self.components['history_spin_box'].value()
self._check_scan_in_history(spin_box_value)
# Get the data from the client
data = self.client.history[spin_box_value]
# Check that the plot does not already have a curve with the same data
scan_number = int(data.metadata.bec['scan_number'])
monitor_name = self.components['monitor_combobox'].currentText()
# Get signal hints
signal_name = getattr(self.client.device_manager.devices, monitor_name)._hints
signal_name = signal_name[0] if len(signal_name)>0 else signal_name
curve_label = f"Scan-{scan_number}-{monitor_name}-{signal_name}"
if len([curve for curve in self.components['waveform'].curves if curve.config.label == curve_label]):
return
if not hasattr(data.devices, monitor_name):
raise ValueError(f"Device {monitor_name} not found in data.")
# Get scan motors and check that the plot x_axis motor is the same as the scan motor, if not, clear the plot
scan_motors = [motor.decode() for motor in data.metadata.bec['scan_motors']]
x_motor_name = self.components['waveform'].x_mode
if x_motor_name not in scan_motors:
self.clear_plot()
self.components['waveform'].x_mode = x_motor_name = scan_motors[0]
# fetching the data
monitor_data = getattr(data.devices, monitor_name).read()[signal_name]['value']
motor_data = getattr(data.devices, x_motor_name).read()[x_motor_name]['value']
# Plot custom curve, with custom label
self.components['waveform'].plot(x=motor_data, y=monitor_data, label=curve_label)
x_label = f"{x_motor_name} / [{getattr(self.client.device_manager.devices, x_motor_name).egu()}]"
self.components['waveform'].x_label = x_label
def _check_scan_in_history(self, history_value:int) -> None:
"""
Check if scan is in history.
Args:
history_value (int): Value from history -1...-10000
"""
if len(self.client.history) < abs(history_value):
self.components['metadata_text_box'].set_plain_text(f"Scan history does not have the request scan {history_value} of history with length: {len(self.client.history)}")
return
def select_scan_from_history(self, value:int) -> None:
"""
Set scan from CLI.
Args:
value (int) : value from history -1 ...-10000
"""
if value >=0:
raise ValueError(f"Value must be smaller or equal -1, provided {value}")
self.components['history_spin_box'].setValue(value)
@SafeSlot(int)
def _scan_history_selected(self, spin_box_value:int) -> None:
self._check_scan_in_history(spin_box_value)
data = self.client.history[spin_box_value]
data.metadata.bec['scan_motors'][0].decode()
text = str(data)
scan_motor_text = "\n" + "Scan Motors: "
for motor in data.metadata.bec['scan_motors']:
scan_motor_text += f" {motor.decode()}"
self.components['metadata_text_box'].set_plain_text(text + scan_motor_text)
@SafeSlot(str)
def _set_x_axis(self, device_x:str) -> None:
self.components['waveform'].x_mode = device_x
@SafeSlot(str)
def _plot_new_device(self, device:str) -> None:
# if len(curve for curve in self.components["waveform"].curves if curve.config.label == f"{device}-{device}":
self.components["waveform"].plot(device)
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = ScanHistory()
widget.show()
sys.exit(app.exec_())

1
pxiii_bec/bec_widgets/widgets/scan_history/scan_history.pyproject
View File

@@ -1 +0,0 @@
{'files': ['scan_history.py']}

115
pxiii_bec/bec_widgets/widgets/scan_history/scan_history.ui
View File

@@ -1,115 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>955</width>
<height>796</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2" stretch="9,3">
<item>
<widget class="Waveform" name="waveform">
<property name="sizePolicy">
<sizepolicy hsizetype="Preferred" vsizetype="Preferred">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="minimumSize">
<size>
<width>0</width>
<height>0</height>
</size>
</property>
</widget>
</item>
<item>
<layout class="QHBoxLayout" name="horizontalLayout_2">
<item>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QLabel" name="monitor_label">
<property name="font">
<font/>
</property>
<property name="text">
<string>BPM Monitor</string>
</property>
</widget>
</item>
<item>
<widget class="DeviceComboBox" name="monitor_combobox"/>
</item>
<item>
<widget class="QLabel" name="history_label">
<property name="text">
<string>Scan History</string>
</property>
</widget>
</item>
<item>
<widget class="QSpinBox" name="history_spin_box"/>
</item>
<item>
<widget class="QPushButton" name="history_add">
<property name="text">
<string>Add scan</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="history_clear">
<property name="text">
<string>clear all</string>
</property>
</widget>
</item>
</layout>
</item>
<item>
<widget class="TextBox" name="metadata_text_box">
<property name="sizePolicy">
<sizepolicy hsizetype="Preferred" vsizetype="MinimumExpanding">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="minimumSize">
<size>
<width>795</width>
<height>191</height>
</size>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</widget>
<customwidgets>
<customwidget>
<class>TextBox</class>
<extends>QWidget</extends>
<header>text_box</header>
</customwidget>
<customwidget>
<class>DeviceComboBox</class>
<extends>QComboBox</extends>
<header>device_combobox</header>
</customwidget>
<customwidget>
<class>Waveform</class>
<extends>QWidget</extends>
<header>waveform</header>
</customwidget>
</customwidgets>
<resources/>
<connections/>
</ui>

54
pxiii_bec/bec_widgets/widgets/scan_history/scan_history_plugin.py
View File

@@ -1,54 +0,0 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from bec_widgets.utils.bec_designer import designer_material_icon
from pxiii_bec.bec_widgets.widgets.scan_history.scan_history import ScanHistory
DOM_XML = """
<ui language='c++'>
<widget class='ScanHistory' name='scan_history'>
</widget>
</ui>
"""
class ScanHistoryPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = ScanHistory(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(ScanHistory.ICON_NAME)
def includeFile(self):
return "scan_history"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "ScanHistory"
def toolTip(self):
return "ScanHistory"
def whatsThis(self):
return self.toolTip()

0
pxiii_bec/deployments/__init__.py
View File

0
pxiii_bec/deployments/device_server/__init__.py
View File

11
pxiii_bec/deployments/device_server/startup.py
View File

@@ -1,11 +0,0 @@
import os
def setup_epics_ca():
# os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "NO"
# os.environ["EPICS_CA_ADDR_LIST"] = "129.129.122.255 sls-x12sa-cagw.psi.ch:5836"
os.environ["PYTHONIOENCODING"] = "latin1"
def run():
setup_epics_ca()

126
pxiii_bec/device_configs/x06da_device_config.yaml
View File

@@ -1,11 +1,12 @@
sls_current:
description: sls current
description: SLS current
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'ARS07-DPCT-0100:CURR', auto_monitor: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
deviceTags:
- ring
- fe
readOnly: true
softwareTrigger: false
@@ -36,7 +37,7 @@ abs_press:
sldi_cenx:
description: FE slit-diaphragm horizontal center
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENTERX'}
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENX'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -57,8 +58,8 @@ sldi_sizex:
softwareTrigger: false
sldi_ceny:
description: FE slit-diaphragm vertical center
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENTERY'}
deviceClass: ophyd_devices.EpicsMotorEC
deviceConfig: {prefix: 'X06DA-FE-SLDI:CENY'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -78,50 +79,6 @@ sldi_sizey:
readOnly: false
softwareTrigger: false
fecmi_try:
description: FE collimating mirror try
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-MI1:TRY'}
onFailure: buffer
enabled: true
readoutPriority: monitored
deviceTags:
- fe
readOnly: false
softwareTrigger: false
fecmi_pitch:
description: FE collimating mirror pitch
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-MI1:PITCH'}
onFailure: buffer
enabled: true
readoutPriority: monitored
deviceTags:
- fe
readOnly: false
softwareTrigger: false
fecmi_bend:
description: FE collimating mirror bend
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-FE-MI1:BEND1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
deviceTags:
- fe
readOnly: false
softwareTrigger: false
slh_press:
description: OP slit pressure
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-OP-SLH-VMFR-1010:PRESSURE', auto_monitor: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: true
softwareTrigger: false
slh_trxr:
description: OP slit inner blade motion
deviceClass: ophyd.EpicsMotor
@@ -152,23 +109,14 @@ fi1_try:
dccm_theta1:
description: Monochromator pitch 1
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:THETA1'}
deviceConfig: {prefix: 'X06DA-OP-DCCM:PITCH1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
dccm_diode_top:
description: Top diode between mono crystals
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-OP-XPM1:TOP:READOUT', auto_monitor: true}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: true
softwareTrigger: false
dccm_diode_bottom:
description: Bottom diode between mono crystals
dccm_diode:
description: Diode between mono crystals
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-OP-XPM1:BOT:READOUT', auto_monitor: true}
onFailure: buffer
@@ -179,7 +127,7 @@ dccm_diode_bottom:
dccm_theta2:
description: Monochromator pitch 2
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-OP-DCCM:THETA2'}
deviceConfig: {prefix: 'X06DA-OP-DCCM:PITCH2'}
onFailure: buffer
enabled: true
readoutPriority: monitored
@@ -230,15 +178,15 @@ ssxbpm_try:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
ssxbpm:
description: XBPM before secondary source
deviceClass: ophyd.EpicsSignalRO
deviceConfig: {read_pv: 'X06DA-ES-SSBPM1:SumAll:MeanValue_RBV'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: true
softwareTrigger: false
# ssxbpm:
# description: XBPM before secondary source
# deviceClass: ophyd.EpicsSignalRO
# deviceConfig: {read_pv: 'X06DA-ES-SSBPM1:SumAll:MeanValue_RBV'}
# onFailure: buffer
# enabled: true
# readoutPriority: monitored
# readOnly: true
# softwareTrigger: false
ssslit_trxr:
description: Secondary source blade motion
deviceClass: ophyd.EpicsMotor
@@ -641,7 +589,24 @@ backlight:
readoutPriority: baseline
readOnly: false
softwareTrigger: false
det_y:
description: Pilatus height
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-ES-DET:TRY1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
det_z:
description: Pilatus translation
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-ES-DET:TRZ1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
@@ -736,22 +701,3 @@ phi:
readoutPriority: monitored
readOnly: false
softwareTrigger: false
det_y:
description: Pilatus height
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-ES-DET:TRY1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false
det_z:
description: Pilatus translation
deviceClass: ophyd.EpicsMotor
deviceConfig: {prefix: 'X06DA-ES-DET:TRZ1'}
onFailure: buffer
enabled: true
readoutPriority: monitored
readOnly: false
softwareTrigger: false

198
pxiii_bec/devices/A3200.py
View File

@@ -51,9 +51,9 @@ Examples
"""
import time
from ophyd import Device, Component, EpicsSignal, EpicsSignalRO, Kind
from ophyd import Component, EpicsSignal, EpicsSignalRO, Kind
from ophyd.status import SubscriptionStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from ophyd_devices.interfaces.base_classes.bec_device_base import BECDeviceBase, CustomPrepare
try:
from .A3200enums import AbrCmd, AbrMode
@@ -67,7 +67,90 @@ logger = bec_logger.logger
# pylint: disable=logging-fstring-interpolation
class AerotechAbrStage(PSIDeviceBase, Device):
class AerotechAbrMixin(CustomPrepare):
"""Configuration class for the Aerotech A3200 controller for the ABR stage"""
def on_stage(self):
"""
NOTE: Zac's request is that stage is essentially ARM, i.e. get ready and don't do anything.
"""
logger.warning(f"Configuring {self.parent.scaninfo.scan_msg.info['scan_name']} on ABR")
d = {}
if self.parent.scaninfo.scan_type in ("measure", "measurement", "fly"):
scanargs = self.parent.scaninfo.scan_msg.info["kwargs"]
scanname = self.parent.scaninfo.scan_msg.info["scan_name"]
if scanname in (
"standardscan",
"helicalscan",
"helicalscan1",
"helicalscan2",
"helicalscan3",
):
d["scan_command"] = AbrCmd.MEASURE_STANDARD
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["range"]
d["var_3"] = scanargs["move_time"]
d["var_4"] = scanargs.get("ready_rate", 500)
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("verticallinescan", "vlinescan"):
d["scan_command"] = AbrCmd.VERTICAL_LINE_SCAN
d["var_1"] = scanargs["range"] / scanargs["steps"]
d["var_2"] = scanargs["steps"]
d["var_3"] = scanargs["exp_time"]
d["var_4"] = 0
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("screeningscan"):
d["scan_command"] = AbrCmd.SCREENING
d["var_1"] = scanargs["start"]
d["var_2"] = scanargs["oscrange"]
d["var_3"] = scanargs["exp_time"]
d["var_4"] = scanargs["range"] / scanargs["steps"]
d["var_5"] = scanargs["steps"]
d["var_6"] = scanargs.get("delta", 0.5)
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("rasterscan", "rastersimplescan"):
d["scan_command"] = AbrCmd.RASTER_SCAN_SIMPLE
d["var_1"] = scanargs["exp_time"]
d["var_2"] = scanargs["range_x"] / scanargs["steps_x"]
d["var_3"] = scanargs["range_y"] / scanargs["steps_y"]
d["var_4"] = scanargs["steps_x"]
d["var_5"] = scanargs["steps_y"]
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
# Reconfigure if got a valid scan config
if len(d) > 0:
self.parent.configure(d)
# Stage the parent
self.parent.bluestage()
def on_kickoff(self):
"""Kick off parent"""
self.parent.bluekickoff()
def on_unstage(self):
"""Unstage the ABR controller"""
self.parent.blueunstage()
class AerotechAbrStage(BECDeviceBase):
"""Standard PX stage on A3200 controller
This is the wrapper class for the standard rotation stage layout for the PX
@@ -78,7 +161,8 @@ class AerotechAbrStage(PSIDeviceBase, Device):
it via 10+1 global variables.
"""
USER_ACCESS = ["reset", "kickoff", "complete", "set_axis_mode", "arm", "disarm"]
custom_prepare_cls = AerotechAbrMixin
USER_ACCESS = ["reset", "kickoff", "bluekickoff", "complete", "set_axis_mode", "arm", "disarm"]
taskStop = Component(EpicsSignal, "-AERO:TSK-STOP", put_complete=True, kind=Kind.omitted)
status = Component(EpicsSignal, "-AERO:STAT", put_complete=True, kind=Kind.omitted)
@@ -130,30 +214,6 @@ class AerotechAbrStage(PSIDeviceBase, Device):
task4 = Component(EpicsSignalRO, "-AERO:TSK4-DONE", auto_monitor=True)
scan_done = Component(EpicsSignal, "-GRD:SCAN-DONE", kind=Kind.config)
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
scan_info=None,
**kwargs,
):
# super() will call the mixin class
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
scan_info=scan_info,
**kwargs,
)
def set_axis_mode(self, mode: str, settle_time=0.1) -> None:
"""Set axis mode to direct/measurement mode.
@@ -170,82 +230,6 @@ class AerotechAbrStage(PSIDeviceBase, Device):
if mode == "measuring":
self.axisAxesMode.set(AbrMode.MEASURING, settle_time=settle_time).wait()
def on_stage(self):
"""
NOTE: Zac's request is that stage is essentially ARM, i.e. get ready and don't do anything.
"""
d = {}
# FIXME: I don't care about how we fish out config parameters from scan info
scan_args = {
**self.scan_info.msg.request_inputs["inputs"],
**self.scan_info.msg.request_inputs["kwargs"],
**self.scan_info.msg.scan_parameters,
}
scanname = self.scan_info.msg.scan_name
if scanname in (
"standardscan",
"helicalscan",
"helicalscan1",
"helicalscan2",
"helicalscan3",
):
d["scan_command"] = AbrCmd.MEASURE_STANDARD
d["var_1"] = scan_args["start"]
d["var_2"] = scan_args["range"]
d["var_3"] = scan_args["move_time"]
d["var_4"] = scan_args.get("ready_rate", 500)
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("verticallinescan", "vlinescan"):
d["scan_command"] = AbrCmd.VERTICAL_LINE_SCAN
d["var_1"] = scan_args["range"] / scan_args["steps"]
d["var_2"] = scan_args["steps"]
d["var_3"] = scan_args["exp_time"]
d["var_4"] = 0
d["var_5"] = 0
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("screeningscan"):
d["scan_command"] = AbrCmd.SCREENING
d["var_1"] = scan_args["start"]
d["var_2"] = scan_args["oscrange"]
d["var_3"] = scan_args["exp_time"]
d["var_4"] = scan_args["range"] / scan_args["steps"]
d["var_5"] = scan_args["steps"]
d["var_6"] = scan_args.get("delta", 0.5)
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
if scanname in ("rasterscan", "rastersimplescan"):
d["scan_command"] = AbrCmd.RASTER_SCAN_SIMPLE
d["var_1"] = scan_args["exp_time"]
d["var_2"] = scan_args["range_x"] / scan_args["steps_x"]
d["var_3"] = scan_args["range_y"] / scan_args["steps_y"]
d["var_4"] = scan_args["steps_x"]
d["var_5"] = scan_args["steps_y"]
d["var_6"] = 0
d["var_7"] = 0
# d["var_8"] = 0
# d["var_9"] = 0
# Reconfigure if got a valid scan config
if len(d) > 0:
self.configure(d)
# Stage the ABR stage
self.arm()
def on_unstage(self):
"""Unstage the ABR controller"""
self.disarm()
def configure(self, d: dict) -> tuple:
""" " Configure the exposure scripts
@@ -300,14 +284,14 @@ class AerotechAbrStage(PSIDeviceBase, Device):
new = self.read_configuration()
return old, new
def arm(self):
def bluestage(self):
"""Bluesky-style stage
Since configuration synchronization is not guaranteed, this does
nothing. The script launched by kickoff().
"""
def on_kickoff(self, timeout=1) -> SubscriptionStatus:
def bluekickoff(self, timeout=1) -> SubscriptionStatus:
"""Kick off the set program"""
self.start_command.set(1).wait()
@@ -320,7 +304,7 @@ class AerotechAbrStage(PSIDeviceBase, Device):
status.wait()
# return status
def disarm(self, settle_time=0.1):
def blueunstage(self, settle_time=0.1):
"""Stops current script and releases the axes"""
# Disarm commands
self.scan_command.set(AbrCmd.NONE, settle_time=settle_time).wait()

3
pxiii_bec/devices/StdDaqPreview.py
View File

@@ -102,6 +102,8 @@ class StdDaqPreviewMixin(CustomDetectorMixin):
self.parent.array_counter.put(header["frame"], force=True)
self.parent.ndimensions.put(len(header["shape"]), force=True)
self.parent.array_size.put(header["shape"], force=True)
self.parent.array_average.put(np.mean(image), force=True)
# self.parent.array_data.put(data, force=True)
self.parent.shaped_image.put(image, force=True)
@@ -152,6 +154,7 @@ class StdDaqPreviewDetector(PSIDetectorBase):
url = Component(Signal, kind=Kind.config, metadata={"write_access": False})
throttle = Component(Signal, value=0.25, kind=Kind.config)
# Streamed data status
array_average = Component(Signal, kind=Kind.hinted, metadata={"write_access": False})
array_counter = Component(Signal, kind=Kind.hinted, metadata={"write_access": False})
ndimensions = Component(Signal, kind=Kind.normal, metadata={"write_access": False})
array_size = Component(Signal, kind=Kind.normal, metadata={"write_access": False})

6
pxiii_bec/macros/README.md
View File

@@ -1,6 +0,0 @@
# Macros
This directory is intended to store macros which will be loaded automatically when starting BEC.
Macros are small functions to make repetitive tasks easier. Functions defined in python files in this directory will be accessible from the BEC console.
Please do not put any code outside of function definitions here. If you wish for code to be automatically run when starting BEC, see the startup script at pxiii_bec/bec_ipython_client/startup/post_startup.py
For a guide on writing macros, please see: https://bec.readthedocs.io/en/latest/user/command_line_interface.html#how-to-write-a-macro

0
pxiii_bec/macros/__init__.py
View File

0
pxiii_bec/scans/metadata_schema/__init__.py
View File

12
pxiii_bec/scans/metadata_schema/metadata_schema_registry.py
View File

@@ -1,12 +0,0 @@
# from .metadata_schema_template import ExampleSchema
METADATA_SCHEMA_REGISTRY = {
# Add models which should be used to validate scan metadata here.
# Make a model according to the template, and import it as above
# Then associate it with a scan like so:
# "example_scan": ExampleSchema
}
# Define a default schema type which should be used as the fallback for everything:
DEFAULT_SCHEMA = None

34
pxiii_bec/scans/metadata_schema/metadata_schema_template.py
View File

@@ -1,34 +0,0 @@
# # By inheriting from BasicScanMetadata you can define a schema by which metadata
# # supplied to a scan must be validated.
# # This schema is a Pydantic model: https://docs.pydantic.dev/latest/concepts/models/
# # but by default it will still allow you to add any arbitrary information to it.
# # That is to say, when you run a scan with which such a model has been associated in the
# # metadata_schema_registry, you can supply any python dictionary with strings as keys
# # and built-in python types (strings, integers, floats) as values, and these will be
# # added to the experiment metadata, but it *must* contain the keys and values of the
# # types defined in the schema class.
# #
# #
# # For example, say that you would like to enforce recording information about sample
# # pretreatment, you could define the following:
# #
#
# from bec_lib.metadata_schema import BasicScanMetadata
#
#
# class ExampleSchema(BasicScanMetadata):
# treatment_description: str
# treatment_temperature_k: int
#
#
# # If this was used according to the example in metadata_schema_registry.py,
# # then when calling the scan, the user would need to write something like:
# >>> scans.example_scan(
# >>> motor,
# >>> 1,
# >>> 2,
# >>> 3,
# >>> metadata={"treatment_description": "oven overnight", "treatment_temperature_k": 575},
# >>> )
#
# # And the additional metadata would be saved in the HDF5 file created for the scan.

49
pxiii_bec/scripts/alignment_fit.py
View File

@@ -1,49 +0,0 @@
import numpy as np
from scipy.ndimage import gaussian_filter1d
from lmfit.models import GaussianModel
def alignment_fit_and_plot(
history_index: int,
device_name: str,
signal_name: str | None = None,
smoothing_sigma: float = 2.0,
):
"""
Get data for a completed scan from the BEC history, apply smoothing, gaussian fit,
gradient, and plot all the results.
Args:
history_index (int): scan to fetch, e.g. -1 for the most recent scan
device_name (str): the device for which to get the monitoring data
"""
# Fetch scan data from the history
# by default, signal = device name, unless otherwise specified
signal = signal_name or device_name
scan = bec.history[history_index]
md = scan.metadata["bec"]
data = scan.devices[device_name][signal].read()["value"]
# motor name is a bytes object in the metadata, so make a string
motor_name = md["scan_motors"][0].decode()
# Create a plot and a text box to display results
dock_area = bec.gui.new()
wf = dock_area.new().new(bec.gui.available_widgets.Waveform)
wf.title = f"Scan {md['scan_number']}: {md['scan_name']} of {motor_name}"
text = dock_area.new(position="right").new(widget=bec.gui.available_widgets.TextBox)
# Calculate some processed data and add everything to the plot
wf.plot(data, label="Raw data")
smoothed_data = gaussian_filter1d(data, smoothing_sigma)
wf.plot(smoothed_data, label="Smoothed")
gradient = np.gradient(smoothed_data)
wf.plot(gradient, label="gradient")
# Fit a Gaussian model to the smoothed data and show the fitting parameters in the textbox
x_data = scan.devices[motor_name][motor_name].read()["value"]
model = GaussianModel()
result = model.fit(smoothed_data, x=x_data)
text.set_plain_text(f"Fit parameters: \n{result.params.pretty_repr()}")
return result

8
pxiii_bec/scripts/beamlinescripts.py
View File

@@ -2,8 +2,6 @@
# import bec
# import bec_lib.devicemanager.DeviceContainer as dev
import time
def rock(steps, exp_time, scan_start=None, scan_end=None, datasource=None, visual=True, **kwargs):
"""Demo step scan with plotting
@@ -69,9 +67,3 @@ def rock(steps, exp_time, scan_start=None, scan_end=None, datasource=None, visua
# TODO: Move to fitted maximum
return s, firt_par
def monitor(device, steps, t=1):
for _ in range(steps):
print(device.read())
time.sleep(t)

30
pxiii_bec/scripts/kat.py
View File

@@ -1,30 +0,0 @@
def scan_theta2(scan_start, scan_end, stepno, exp):
# Save the motor starting position
start_value = dev.dccm_theta2.read()['dccm_theta2']['value']
print(f"Motor position is {start_value}")
# Run the scan
s = scans.line_scan(dev.dccm_theta2, scan_start, scan_end, steps=stepno, exp_time=exp, relative=True)
# data = s.devices[dccm_xbpm][dccm_xbpm].read()["value"]
# Move motor back to starting position and print XBPM reading
umv(dev.dccm_theta2, start_value)
xbpm_reading = dev.dccm_xbpm.read()['dccm_xbpm']['value']
print(f"Moving dccm_theta2 back to start position of where XBPM Reading is {xbpm_reading}")
end_value = dev.dccm_theta2.read()['dccm_theta2']['value']
print(f"Motor was at {start_value} before the scan, now at {end_value}")
# # Create a plot to display the results
dock_area = bec.gui.new()
wf = dock_area.new().new(bec.gui.available_widgets.Waveform)
wf.title = f"Scan of DCCM_theta2"
wf.plot(x_name='dccm_theta2', y_name='dccm_xbpm')
wf.add_dap_curve(device_label='dccm_xbpm-dccm_xbpm', dap_name='GaussianModel')
print(dap_xbpm.dap_params)

17
pxiii_bec/scripts/scanwrappers.py
View File

@@ -1,4 +1,5 @@
# pylint: disable=undefined-variable
# import bec
# import bec_lib.devicemanager.DeviceContainer as dev
def bl_check_beam():
@@ -7,7 +8,15 @@ def bl_check_beam():
def ascan(
motor, scan_start, scan_end, steps, exp_time, plot=None, visual=True, relative=False, **kwargs
motor,
scan_start,
scan_end,
steps,
exp_time,
plot=None,
visual=True,
relative=False,
**kwargs,
):
"""Demo step scan with plotting
@@ -58,7 +67,9 @@ def ascan(
firt_par = plt1.get_dap_params()
else:
# Fitting without GUI
firt_par = bec.dap.LinearModel.fit(s, motor.name, motor.name, plot.name, plot.name)
firt_par = bec.dap.LinearModel.fit(
s, motor.name, motor.name, plot.name, plot.name
)
# # Some basic fit
# dkey = datasource.full_name

0
pxiii_bec/services/__init__.py
View File

16
pyproject.toml
View File

@@ -5,7 +5,7 @@ build-backend = "hatchling.build"
[project]
name = "pxiii_bec"
version = "0.0.0"
description = "A plugin repository for BEC"
description = "Custom device implementations based on the ophyd hardware abstraction layer"
requires-python = ">=3.10"
classifiers = [
"Development Status :: 3 - Alpha",
@@ -29,21 +29,19 @@ dependencies = [
[project.optional-dependencies]
dev = [
"black",
"copier",
"isort",
"coverage",
"pylint",
"pytest",
"pytest-random-order",
"ophyd_devices",
"bec_server",
"pytest-redis",
]
[project.entry-points."bec"]
plugin_bec = "pxiii_bec"
[project.entry-points."bec.deployment.device_server"]
plugin_ds_startup = "pxiii_bec.deployments.device_server.startup:run"
plugin_ds_startup = "pxiii_bec.deployment.device_server.startup:run"
[project.entry-points."bec.file_writer"]
plugin_file_writer = "pxiii_bec.file_writer"
@@ -51,18 +49,12 @@ plugin_file_writer = "pxiii_bec.file_writer"
[project.entry-points."bec.scans"]
plugin_scans = "pxiii_bec.scans"
[project.entry-points."bec.scans.metadata_schema"]
plugin_metadata_schema = "pxiii_bec.scans.metadata_schema"
[project.entry-points."bec.ipython_client_startup"]
plugin_ipython_client_pre = "pxiii_bec.bec_ipython_client.startup.pre_startup"
plugin_ipython_client_post = "pxiii_bec.bec_ipython_client.startup"
[project.entry-points."bec.widgets.auto_updates"]
plugin_widgets_update = "pxiii_bec.bec_widgets.auto_updates"
[project.entry-points."bec.widgets.user_widgets"]
plugin_widgets = "pxiii_bec.bec_widgets.widgets"
plugin_widgets_update = "pxiii_bec.bec_widgets.auto_updates:PlotUpdate"
[tool.hatch.build.targets.wheel]
include = ["*"]

21
tests/tests_bec_ipython_client/README.md
View File

@@ -1,34 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your _python environment_.
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

21
tests/tests_bec_widgets/README.md
View File

@@ -1,34 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your _python environment_.
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

21
tests/tests_dap_services/README.md
View File

@@ -1,34 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your _python environment_.
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

21
tests/tests_devices/README.md
View File

@@ -1,34 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your _python environment_.
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

34
tests/tests_file_writer/README.md
View File

@@ -1,34 +0,0 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
pip install pytest
```
in your _python environment_.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).

21
tests/tests_scans/README.md
View File

@@ -1,34 +1,31 @@
# Getting Started with Testing using pytest
BEC is using the [pytest](https://docs.pytest.org/en/latest/) framework.
It can be installed via
```bash
BEC is using the [pytest](https://docs.pytest.org/en/8.0.x/) framework.
It can be install via
``` bash
pip install pytest
```
in your _python environment_.
in your *python environment*.
We note that pytest is part of the optional-dependencies `[dev]` of the plugin package.
## Introduction
Tests in this package should be stored in the `tests` directory.
We suggest to sort tests of different submodules, i.e. `scans` or `devices` in the respective folder structure, and to folow a naming convention of `<test_module_name.py>`.
It is mandatory for test files to begin with `test_` for pytest to discover them.
To run all tests, navigate to the directory of the plugin from the command line, and run the command
To run all tests, navigate to the directory of the plugin from the command line, and run the command
```bash
``` bash
pytest -v --random-order ./tests
```
Note, the python environment needs to be active.
The additional arg `-v` allows pytest to run in verbose mode which provides more detailed information about the tests being run.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
The argument `--random-order` instructs pytest to run the tests in random order, which is the default in the CI pipelines.
## Test examples
Writing tests can be quite specific for the given function.
Writing tests can be quite specific for the given function.
We recommend writing tests as isolated as possible, i.e. try to test single functions instead of full classes.
A very useful class to enable isolated testing is [MagicMock](https://docs.python.org/3/library/unittest.mock.html).
In addition, we also recommend to take a look at the [How-to guides from pytest](https://docs.pytest.org/en/8.0.x/how-to/index.html).