tests: add test for smargon

Reviewed-on: #8

.copier-answers.yml

@@ -2,8 +2,9 @@
 # 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: v0.3.3
-_src_path: https://gitea.psi.ch/bec/bec_plugin_copier_template.git
+_commit: v1.2.8
+_src_path: https://github.com/bec-project/plugin_copier_template.git
+make_commit: false
 project_name: pxii_bec
 widget_plugins_input:
 -   name: scan_history

.gitea/workflows/ci.yml

@@ -0,0 +1,102 @@
+name: CI for pxii_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.12"
+
+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.12' }}"
+
+      - name: Checkout BEC Plugin Repository
+        uses: actions/checkout@v4
+        with:
+          repository: bec/pxii_bec
+          ref: "${{ inputs.BEC_PLUGIN_REPO_BRANCH || github.head_ref || github.sha }}"
+          path: ./pxii_bec
+
+      - name: Lint for merge conflicts from template updates
+        shell: bash
+        # Find all Copier conflicts except this line
+        run: '! grep -r "<<<<<<< before updating" | grep -v "grep -r \"<<<<<<< before updating"'
+
+      - 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: 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 ./pxii_bec
+
+      - name: Run Pytest with Coverage
+        id: coverage
+        run: pytest --random-order --cov=./pxii_bec --cov-config=./pxii_bec/pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail ./pxii_bec/tests/ || test $? -eq 5

.gitea/workflows/create_update_pr.yml

@@ -0,0 +1,62 @@
+name: Create template upgrade PR for pxii_bec
+on:
+  workflow_dispatch:
+
+permissions:
+  pull-requests: write
+
+jobs:
+  create_update_branch_and_pr:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+
+    steps:
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+    
+      - name: Install tools
+        run: |
+          pip install copier PySide6
+
+      - name: Checkout
+        uses: actions/checkout@v4
+    
+      - name: Perform update
+        run: |
+          git config --global user.email "bec_ci_staging@psi.ch"
+          git config --global user.name "BEC automated CI"
+
+          branch="chore/update-template-$(python -m uuid)"
+          echo "switching to branch $branch"
+          git checkout -b $branch
+
+          echo "Running copier update..."
+          output="$(copier update --trust --defaults --conflict inline 2>&1)"
+          echo "$output"
+          msg="$(printf '%s\n' "$output" | head -n 1)"
+
+          if ! grep -q "make_commit: true" .copier-answers.yml ; then
+            echo "Autocommit not made, committing..."
+            git add -A
+            git commit -a -m "$msg"
+          fi
+
+          if diff-index --quiet HEAD ; then
+            echo "No changes detected"
+            exit 0
+          fi
+
+          git push -u origin $branch
+          curl -X POST "https://gitea.psi.ch/api/v1/repos/${{ gitea.repository }}/pulls" \
+            -H "Authorization: token ${{ secrets.CI_REPO_WRITE }}" \
+            -H "Content-Type: application/json" \
+            -d "{
+                  \"title\": \"Template: $(echo $msg)\",
+                  \"body\": \"This PR was created by Gitea Actions\",
+                  \"head\": \"$(echo $branch)\",
+                  \"base\": \"main\"
+                }"

.gitlab-ci.yml

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

pxii_bec/bec_ipython_client/startup/post_startup.py

@@ -34,3 +34,4 @@ to setup the prompts.
 """
 
 # pylint: disable=invalid-name, unused-import, import-error, undefined-variable, unused-variable, unused-argument, no-name-in-module
+init_positioned_devices()

pxii_bec/bec_ipython_client/startup/pre_startup.py

@@ -3,8 +3,12 @@ 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.
 """
 
+import os
+
 from bec_lib.service_config import ServiceConfig
 
+import pxii_bec
+
 
 def extend_command_line_args(parser):
     """
@@ -14,3 +18,14 @@ def extend_command_line_args(parser):
     # parser.add_argument("--session", help="Session name", type=str, default="cSAXS")
 
     return parser
+
+def get_config() -> ServiceConfig:
+    """
+    Create and return the ServiceConfig for the plugin repository
+    """
+    deployment_path = os.path.dirname(os.path.dirname(os.path.dirname(pxii_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})

pxii_bec/device_configs/device_config.yaml-tst

@@ -1,543 +0,0 @@
-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:
-    - SLS
-  readOnly: true
-  softwareTrigger: false
-
-gap:
-  description: 'U19 gap'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-UIND:GAP'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - SLS
-  readOnly: false 
-  softwareTrigger: false
-
-ps1_press:
-  description: Pumpstand1 pressure
-  deviceClass: ophyd.EpicsSignalRO
-  deviceConfig: {read_pv: 'X10SA-FE-PUM1-VPIG-1020:PRESSURE', auto_monitor: true}
-  onFailure: buffer
-  enabled: true
-  readoutPriority: monitored
-  deviceTags:
-    - fe
-  readOnly: true
-  softwareTrigger: false
-
-#xbpm1_temp:
-#  description: XBPM1 temp
-#  deviceClass: ophyd.EpicsSignalRO
-#  deviceConfig: {read_pv: 'X10SA-FE-XBPM1-ETTC-0010:TEMP', auto_monitor: true}
-#  onFailure: buffer
-#  enabled: true
-#  readoutPriority: monitored
-#  deviceTags:
-#    - fe
-#  readOnly: true
-#  softwareTrigger: false
-
-s1_xw:
-  description: 'BSF slit outboard'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLH:TRXW'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_xr:
-  description: 'BSF slit inboard'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLH:TRXR'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_yt:
-  description: 'BSF slit top'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLV:TRYT'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_yb:
-  description: 'BSF slit bottom'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLV:TRYB'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_xcen:
-  description: 'BSF X centre'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLH:CENTER'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_xsize:
-  description: 'BSF X size'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLH:SIZE'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-s1_ycen:
-  description: 'BSF Y centre'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-SLV:CENTER'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-opf1_y:
-  description: 'BSF Filter 1 Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-FI1:TRY'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-opf2_y:
-  description: 'BSF Filter 2 Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-OP-FI2:TRY'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - bsf
-  readOnly: false 
-  softwareTrigger: false
-ssbpm_x:
-  description: 'SS BPM X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSBPM1:TRX1'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-ssbpm_y:
-  description: 'SS BPM Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSBPM1:TRY1'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_xw:
-  description: 'SS slit wall'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSH1:TRXW'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_xr:
-  description: 'SS slit ring'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSH1:TRXR'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_xcen:
-  description: 'SS slit X centre'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSH1:CENTER'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_xsize:
-  description: 'SS slit X size'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSH1:CENTER'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_yt:
-  description: 'SS slit top'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSV1:TRYT'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_yb:
-  description: 'SS slit bottom'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSV1:TRYB'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_ycen:
-  description: 'SS slit Y centre'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSV1:CENTER'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-s2_ysize:
-  description: 'SS slit Y size'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSSV1:SIZE'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-xeye_x:
-  description: 'SS X-ray eye X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSXI1:TRX1'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-xeye_y:
-  description: 'SS X-ray eye Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-SSXI1:TRY1'}
-  onFailure: buffer
-  enabled: True
-  readoutPriority: monitored
-  deviceTags:
-    - ss
-  readOnly: false 
-  softwareTrigger: false
-vfm_xu:
-  description: 'VFM Upstream X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRXU'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_xd:
-  description: 'VFM Downstream X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRXD'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_yur:
-  description: 'VFM Upstream Ring Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRYUR'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_yw:
-  description: 'VFM Wall Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRYW'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_ydr:
-  description: 'VFM Downstream Ring Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRYDR'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_bu:
-  description: 'VFM Upstream Bender'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:BNDU'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_bd:
-  description: 'VFM Downstream Bender'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:BNDD'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_yaw:
-  description: 'VFM Virtual Yaw'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:YAW'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_roll:
-  description: 'VFM Virtual Roll'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:ROLL'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_pitch:
-  description: 'VFM Virtual Pitch'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:PITCH'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_lat:
-  description: 'VFM Virtual X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRX'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-vfm_vert:
-  description: 'VFM Virtual Y '
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-VFM:TRY'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - vfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_xu:
-  description: 'HFM Upstream X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRXU'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_xd:
-  description: 'HFM Downstream X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRXD'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_yuw:
-  description: 'HFM Upstream Wall Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRYUW'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_yr:
-  description: 'HFM Ring Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRYR'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_ydw:
-  description: 'HFM Downstream Wall Y'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRYDW'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_bu:
-  description: 'HFM Upstream Bender'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:BNDU'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_bd:
-  description: 'HFM Downstream Bender'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:BNDD'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_yaw:
-  description: 'HFM Virtual Yaw'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:YAW'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_roll:
-  description: 'HFM Virtual Roll'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:ROLL'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_pitch:
-  description: 'HFM Virtual Pitch'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:PITCH'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_lat:
-  description: 'HFM Virtual X'
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRX'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false
-hfm_vert:
-  description: 'HFM Virtual Y '
-  deviceClass: ophyd.EpicsMotor
-  deviceConfig: {prefix: 'X10SA-ES-HFM:TRY'}
-  onFailure: buffer
-  enabled: False
-  readoutPriority: monitored
-  deviceTags:
-    - hfm
-  readOnly: false 
-  softwareTrigger: false

pxii_bec/device_configs/x10sa_device_config.yaml

@@ -1,13 +1,58 @@
 base_config:
-  - !include ./device_config.yaml
-
+  - !include ./pxii-autogenerated.yaml
 id_gap:
   readoutPriority: baseline
   description: undulator gap
   deviceClass: pxii_bec.devices.undulator.UndulatorGap
   deviceConfig:
-    prefix: 'X10SA-UIND:'
+     prefix: 'X10SA-UIND:'
   onFailure: buffer
   enabled: true
   readOnly: false
   softwareTrigger: false
+coll_x:
+  description: Collimator X
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-ES-COL:TRX'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+    - se
+  readOnly: false 
+  softwareTrigger: false
+dcm_fpitch:
+  description: DCM 2nd crystal fine pitch
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-OP-DCM:PITCH-C2'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+   - dcm
+  readOnly: false 
+  softwareTrigger: false
+dcm_froll:
+  description: DCM 2nd crystal fine roll
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-OP-DCM:ROLL-C2'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+    - dcm
+  readOnly: false 
+  softwareTrigger: false
+
+smargon:
+  description: REST-based device which connects to Smargopolo
+  deviceClass: pxii_bec.devices.smargopolo_smargon.Smargon
+  deviceConfig: {prefix: 'http://x10sa-smargopolo.psi.ch:3000'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+    - smargon
+    - motors
+  readOnly: false 
+  softwareTrigger: false

pxii_bec/device_configs/x10sa_device_config_testing.yaml

@@ -0,0 +1,99 @@
+base_config:
+  - !include ./pxii-autogenerated.yaml
+id_gap:
+  readoutPriority: baseline
+  description: undulator gap
+  deviceClass: pxii_bec.devices.undulator.UndulatorGap
+  deviceConfig:
+     prefix: 'X10SA-UIND:'
+  onFailure: buffer
+  enabled: true
+  readOnly: false
+  softwareTrigger: false
+coll_x:
+  description: Collimator X
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-ES-COL:TRX'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+    - se
+  readOnly: false 
+  softwareTrigger: false
+dcm_fpitch:
+  description: DCM 2nd crystal fine pitch
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-OP-DCM:PITCH-C2'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+   - dcm
+  readOnly: false 
+  softwareTrigger: false
+dcm_froll:
+  description: DCM 2nd crystal fine roll
+  deviceClass: ophyd.EpicsMotor
+  deviceConfig: {prefix: 'X10SA-OP-DCM:ROLL-C2'}
+  onFailure: buffer
+  enabled: True
+  readoutPriority: baseline
+  deviceTags:
+    - dcm
+  readOnly: false 
+  softwareTrigger: false
+
+bcu_xasym:
+  description: horizontal asymmetry 
+  deviceClass: ophyd_devices.ComputedSignal
+  deviceConfig:
+    compute_method: "def compute_xasym(signal1, signal2, signal3, signal4):\n    return (signal3.get()+signal4.get() - signal1.get()- signal2.get())/(sum((signal1.get(), signal2.get(), signal3.get(), signal4.get())))"
+    input_signals: 
+      - "bcu_bpm1"
+      - "bcu_bpm2"
+      - "bcu_bpm3"
+      - "bcu_bpm4"
+  enabled: true
+  readOnly: false
+  readoutPriority: baseline
+
+bcu_yasym:
+  description: vertical asymmetry 
+  deviceClass: ophyd_devices.ComputedSignal
+  deviceConfig:
+    compute_method: "def compute_xasym(signal1, signal2, signal3, signal4):\n    return (signal1.get()+signal2.get() - signal3.get()- signal4.get())/(sum((signal1.get(), signal2.get(), signal3.get(), signal4.get())))"
+    input_signals: 
+      - "bcu_bpm1"
+      - "bcu_bpm2"
+      - "bcu_bpm3"
+      - "bcu_bpm4"
+  enabled: true
+  readOnly: false
+  readoutPriority: baseline
+
+bcu_xpos:
+  description: horizontal position
+  deviceClass: ophyd_devices.ComputedSignal
+  deviceConfig:
+    compute_method: "def compute_xpos(signal1):\n    import numpy as np\n    return  0.131786+  np.arctanh((signal1.get()-0.007105) /0.99342) / 9.5597 "
+    input_signals: 
+      - "bcu_xasym"
+  enabled: true
+  readOnly: false
+  readoutPriority: baseline
+
+bcu_ypos:
+  description: vertical position
+  deviceClass: ophyd_devices.ComputedSignal
+  deviceConfig:
+    compute_method: "def compute_ypos(signal1):\n    import numpy as np\n    return -0.20283 +  np.arctanh((signal1.get()- (-0.19936)) /0.80653) / (-13.18539)"
+    input_signals: 
+      - "bcu_yasym"
+  enabled: true
+  readOnly: false
+  readoutPriority: baseline
+
+  
+
+

pxii_bec/devices/smargopolo_smargon.py

@@ -0,0 +1,165 @@
+import time
+from threading import Event, RLock, Thread
+from typing import Any
+
+from ophyd import Component as Cpt
+from ophyd import OphydObject
+from ophyd_devices import PSIDeviceBase
+from ophyd_devices.utils.socket import SocketSignal
+from requests import Response, get, put
+
+_TIMESTAMP_ID = "__timestamp"
+_POLL_INTERVAL_SLOW = 0.1
+
+
+class HttpRestError(Exception):
+    """Error for rest calls from a HttpRestSignal."""
+
+    def __init__(self, resp: Response, *args: object, value: Any | None = None) -> None:
+        method, url = resp.request.method, resp.request.url
+        data = f"{str(value)} to " if value is not None else ""
+        super().__init__(
+            f"Could not {method} {data}{url}. Code: {resp.status_code}. Reason: {resp.reason}.",
+            *args,
+        )
+
+
+class SmargonSignal(SocketSignal):
+    """Ophyd signal which gets and puts to a REST API rather than EPICS PVs, mediated through the SmargonController"""
+
+    def __init__(self, *args, axis_identifier: str, **kwargs):
+        super().__init__(*args, **kwargs)
+        controller: SmargonController | None = getattr(self.root, "controller", None)
+        if controller is None:
+            raise TypeError("SmargonSignal must be used in a device with a SmargonController")
+        self._controller = controller
+        self._axis_id = axis_identifier
+        self._controller.register(self._axis_id)
+
+    def _socket_get(self):  # type: ignore
+        self._readback, self.metadata["timestamp"] = self._controller.get_readback(
+            self._axis_id
+        ) or (0.0, 0.0)
+        return self._readback
+
+    def _socket_set(self, val: float):
+        self._controller.put(self._axis_id, val)
+
+    def get(self, **kwargs):
+        if self._controller.monitor_stopped():
+            self._controller.start_monitor()
+        return super().get(**kwargs)
+
+
+class SmargonController(OphydObject):
+    """Controller to consolidate polling loops and other REST calls for the smargon"""
+
+    def __init__(self, *, prefix, **kwargs):
+        self._prefix = prefix
+        self._readback_endpoint = "/readbackSCS"
+        self._target_endpoint = "/targetSCS"
+        self._targets = {}
+        self._signal_registry: set[str] = set()
+        self._readback_poll_interval: float = _POLL_INTERVAL_SLOW
+
+        super().__init__(**kwargs)
+        self._setup_readback()
+
+    def _setup_readback(self):
+        self._readbacks: dict[str, float] = {}
+        self._stop_monitor_readback_event = Event()
+        self._readback_lock = RLock()
+        self._monitor_readback_thread = Thread(
+            target=self._monitor,
+            args=[
+                self._readback_endpoint,
+                self._stop_monitor_readback_event,
+                self._readback_lock,
+                self._readbacks,
+            ],
+        )
+
+    def manual_update(self):
+        self._update_reading(self._readback_endpoint, self._readback_lock, self._readbacks)
+
+    def _update_reading(self, endpoint: str, lock: RLock, buffer: dict):
+        data = self._rest_get(endpoint)
+        timestamp = time.monotonic()
+        with lock:
+            buffer.update(data)
+            buffer["__timestamp"] = timestamp
+
+    def _monitor(self, endpoint: str, event: Event, lock: RLock, buffer: dict):
+        while not event.is_set():
+            self._update_reading(endpoint, lock, buffer)
+            time.sleep(self._readback_poll_interval)
+
+    def _clean_monitor(self):
+        if self._monitor_readback_thread.is_alive():
+            self._stop_monitor_readback_event.set()
+            self._monitor_readback_thread.join(timeout=2)
+            if self._monitor_readback_thread.is_alive():
+                raise RuntimeError("Failed to clean up Smargon monitor thread.")
+
+    def register(self, axis_id: str):
+        self._signal_registry.add(axis_id)
+
+    def _rest_get(self, endpoint):
+        resp = get(self._prefix + endpoint)
+        if not resp.ok:
+            raise HttpRestError(resp)
+        return resp.json()
+
+    def _rest_put(self, val: dict[str, float]):
+        resp = put(self._prefix + self._target_endpoint, params=val)
+        if not resp.ok:
+            raise HttpRestError(resp, value=val)
+
+    def start_monitor(self):
+        """Start or restart the automonitor thread."""
+        self._clean_monitor()
+        self._setup_readback()
+        self._monitor_readback_thread.start()
+
+    def monitor_stopped(self):
+        return not self._monitor_readback_thread.is_alive()
+
+    def get_readback(self, axis_id: str) -> tuple[float, float] | None:
+        with self._readback_lock:
+            if axis_id not in self._readbacks or _TIMESTAMP_ID not in self._readbacks:
+                return None
+            return self._readbacks.get(axis_id), self._readbacks.get(_TIMESTAMP_ID)  # type: ignore
+
+    def put(self, axis: str, val: float):
+        self._rest_put({axis: val})
+
+    def stop(self):
+        # There doesn't appear to be a stop endpoint on the server
+        # Best effort: set the target to the current position
+        self._rest_put(self._readbacks)
+
+
+class Smargon(PSIDeviceBase):
+
+    x = Cpt(SmargonSignal, axis_identifier="SHX", tolerance=0.01)
+    y = Cpt(SmargonSignal, axis_identifier="SHY", tolerance=0.01)
+    z = Cpt(SmargonSignal, axis_identifier="SHZ", tolerance=0.01)
+    phi = Cpt(SmargonSignal, axis_identifier="PHI", tolerance=0.01)
+    chi = Cpt(SmargonSignal, axis_identifier="CHI", tolerance=0.01)
+
+    def __init__(
+        self, *, name: str, prefix: str = "", scan_info=None, device_manager=None, **kwargs
+    ):
+        self.controller = SmargonController(prefix=prefix)
+        super().__init__(
+            name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
+        )
+
+    def wait_for_connection(self, **kwargs):  # type: ignore
+        self.controller.start_monitor()
+        self.controller.manual_update()
+        return super().wait_for_connection(**kwargs)
+
+    def stop(self, *, success: bool = False) -> None:
+        self.controller.stop()
+        return super().stop(success=success)

pxii_bec/macros/README.md

@@ -0,0 +1,6 @@
+# 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 pxii_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

pxii_bec/macros/Undu_Helpers.py

@@ -0,0 +1,319 @@
+#!/usr/bin/env python
+from pylab import *
+import sys, os
+
+# import pandas as pd
+import matplotlib.pyplot as plt
+import numpy as np
+import scipy.interpolate
+from scipy.optimize import curve_fit
+from scipy.ndimage import gaussian_filter1d
+
+
+
+def fit_harm(harm, n, order):
+    x = harm[0, :].astype(float)
+    y = harm[1, :].astype(
+        float
+    )  ## else funny object that might contain funny strings ...
+    coeff = np.polyfit(x, y, order)  # 3 in general i.e., 4 params
+    polynomial = np.poly1d(coeff)
+    x_fit = np.linspace(min(x), max(x), 100)
+    y_fit = polynomial(x_fit)
+    print("Polynomial coefficients of the harmonic: ", n, coeff)
+    plot(x_fit, y_fit, color="blue")
+
+    return (
+        x / n,
+        y,
+    )  # get the normalized energy gap relation for fitting the Halbach coeff
+
+
+######### simple exp fit ############
+def exponential_func0(x, a, b, c):
+    # fit 3 params a,b,c
+
+    return a * np.exp(b * x + c * x ** 2)
+
+
+######### inverse exp fit ############
+def exponential_func1(x, a, b, c):
+    # fit 3 params a,b,c
+
+    return 1 / (1 + (a * np.exp(b * x + c * x ** 2)))
+
+
+######### inverse exp fit plus E_max ############
+def exponential_func2(x, a, b, c, d):
+    # fit 4 params a,b,c, e.g.,  fit energy of storage ring as well
+    return d / (1 + (a * np.exp(b * x + c * x ** 2)))
+
+
+##################################
+
+
+def return_harmon():
+    # from 3rd to 21th Energy range from 6 keV to 30 keV
+    # harmonics, July 2025
+    import numpy as np
+
+    nharm = 13
+    h = [np.array([]) for _ in range(nharm + 1)]
+    h[0] = np.array([0])
+    h[1] = np.array([0])
+    h[2] = np.array([0])
+    h[3] = np.array([1.07141725, -0.52834258])  # 3rd harmon
+    h[4] = np.array([0.007111, -0.13678409, 1.52295567, -1.06882785])
+    h[5] = np.array([0.00315051, -0.0766359, 1.13107469, -0.86442062])
+
+    h[6] = np.array([0.00162862, -0.04452336, 0.82948259, -0.37329674])
+    h[7] = np.array([0.00080764, -0.02418882, 0.59212947, 0.15702886])
+    h[8] = np.array([1.16699242e-03, -4.68207543e-02, 9.43972396e-01, -1.94201019e00])
+    h[9] = np.array([4.84194444e-04, -2.01064762e-02, 5.55141139e-01, -3.81800667e-01])
+    h[10] = np.array([2.11583333e-04, -8.03503571e-03, 3.43141595e-01, 6.02318000e-01])
+    h[11] = np.array([3.34391667e-03, -1.84053357e-01, 3.58338994e00, -1.93640241e01])
+    h[12] = np.array([3.20520798e-05, 2.39253145e-03, 8.09198503e-02, 2.22897377e00])
+    h[13] = np.array([0.00278744, 0.07979874, 2.05143916])
+
+    # fitpar_u19 = np.array([ 2.17078531,  0.519452,   -0.00720255]) # measured from U19
+
+    return h
+
+def plot_harmon(e_start, e_end, h_no, pr_out = False):
+
+    enarr=  np.arange(e_start, e_end+1, 0.5)
+
+    h_all = return_harmon()
+    h=h_all[h_no] 
+    polynomial = np.poly1d(h)
+    gaps = polynomial(enarr)
+
+    if (pr_out):
+        print("en =", enarr )
+        print("gaps =", gaps ) 
+    
+    plt.ion()
+    plt.figure()
+    
+    plt.plot(enarr,gaps,'*')
+    plt.title(f"harmonic no {h_no}")
+    plt.xlabel("E / keV")
+    plt.ylabel("Gap / mm")
+    plt.show()
+
+def setu19(en, *harm_no, detune=0):
+    """
+    set the U19 to the gaps defined in Jul2025, or the "theoretical" ones for higher
+    harmonics
+    USAGE:
+    setu19(en, *harm_no, detune=0)
+    en in keV, possibly select a special harmonics, or detune [0/1] to a value 
+    with a nicer beam shape but less flux 
+    """
+    g0 = dev.id_gap.readback.get()
+
+    fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255])  # measured from U19
+
+    h_all = return_harmon()
+    # dep on energy -> select harmonics
+    if harm_no:
+        harm = int(harm_no[0])
+        print("Selected harmonics is:", harm)
+
+    else:
+        print("Using the default optimised harmonics")
+        if en <= 7:
+            harm = 3  #   h = h[3]
+        elif 7 < en <= 10:
+            harm = 5
+        elif 10 < en <= 13:
+            harm = 7
+        elif 13 < en <= 16:
+            harm = 9
+        elif 16 < en <= 19:
+            harm = 11
+        elif 19 < en <= 22:
+            harm = 13
+
+    if en <= 22:
+        h = h_all[harm]
+        print(f"harmonics is {harm}.")
+
+        polynomial = np.poly1d(h)
+        g = polynomial(en)
+    #   print(f"For energy {en} the used harmonic is {polynomial} for gap {g}")
+    ## for time being: if E > 20 : use the "theoretical", old measured and interpolated values up to E = 30 keV
+    # 21, 22  : 13
+    # 23, 24, 25: 15;
+    # 26: 15  or 17
+    # 27, 28: 17
+    # 29: 15, 17, 19
+    # 30: 19
+    ##
+    if 23 < en <= 25:
+        n = 15
+        enorm = en / n
+        g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
+    elif 25 < en <= 29:
+        n = 17
+        enorm = en / n
+        g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
+    elif 29 < en:
+        n = 19
+        enorm = en / n
+        g = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
+
+    print("Undulator has been ", g0, " mm")
+    if 4.5 <= g <= 20:
+        print("Moving Undulator gap to ", g, " mm")
+    else:
+        print("not a valid gap, do nothing")
+         
+    if detune:
+        g =g *0.996
+        print("moving to detuned gap value, slightly below max, about 0.15 % ")
+    #print("move disabled!!")
+    res =  scans.umv(dev.id_gap, g, relative=False)
+
+    return
+
+##################################
+def harmon_walk(estart=7.5, end_en=13):
+    import time
+
+    en = estart
+    ans ='y'
+    while en < end_en+0.5 and ans == 'y':
+        print(en) 
+        setu19(en, 5)
+        time.sleep(2)
+        sete(en)  
+        en = en+0.5
+        ans = input("Next energy? y/n: ")
+        
+    
+##################################
+def gap_harm(e=12.4):
+    fitpar_u19 = np.array([2.17078531, 0.519452, -0.00720255])
+    fitpar_u17 = np.array([2.17078572, 0.46477267, -0.005766])
+    e = float(input("Please enter an energy between 6 and 30: "))
+
+    for n in range(3, 23):
+
+        # n = int(input('Please enter a harmonic n, n between 3 and 15 '))
+        enorm = e / n
+        g_19 = exponential_func0(enorm, fitpar_u19[0], fitpar_u19[1], fitpar_u19[2])
+        g_17 = exponential_func0(enorm, fitpar_u17[0], fitpar_u17[1], fitpar_u17[2])
+        print(f"harmonic {n}, gap for U19 = {g_19:.5f}, for U17 = {g_17:.5f}")
+        # print(f"harmonic {n}, gap for U17 = {g_17}")
+
+    return
+
+
+##################################
+
+
+def long_gscan(estart=7, end_en=20.5, g_low=4.5, g_high=9.0, nsteps=1500):
+    import time
+    import numpy as np
+
+    dirname = "/home/gac-x10sa/Data/"
+
+    print(
+        f"scanning the U19 gap from {estart} keV to {end_en} keV, for a gapsize from {g_low} to {g_high}"
+    )
+    resol = (g_high - g_low) / nsteps
+    print(f"nsteps = {nsteps}; resolution is {resol} mm")
+    dock_area = bec.gui.new("LongGapScan")
+    wr = dock_area.new(bec.gui.available_widgets.Waveform)
+    mot = dev.id_gap
+    det = dev.lu_bpmsum
+    wr.plot(device_x=mot.name, device_y=det.name)  ## names first !
+    wr.x_label = mot.name
+    wr.y_label = det.name
+    g0 = dev.id_gap.readback.get()
+
+    # g_low = 4.5  # 4.5
+    # g_high = 9.0  # 9.0
+    # nsteps = 1500 # res = 3 um
+
+    en = estart
+
+    while en < end_en:
+        sete(en)
+        time.sleep(1)
+        rock()
+        print(f"setting energy to {en}")
+        time.sleep(2)
+        ds = scans.line_scan(
+            dev.id_gap, g_low, g_high, steps=nsteps, exp_time=0.8, relative=False
+        )
+        gap_data = ds.scan.live_data.id_gap.id_gap.val
+        bpm_data = ds.scan.live_data.lu_bpmsum.lu_bpmsum.val
+        wr.plot(x=gap_data, y=bpm_data)
+        # writing output to simple data file for later analysis:
+        combined = np.column_stack((gap_data, bpm_data))
+        filename = dirname + "gaps" + str(en) + ".txt"
+        with open(filename, "w") as f:
+            np.savetxt(f, combined, delimiter=",", fmt="%5f")
+
+        en += 1
+    return
+
+
+##################################
+def gscan(centre=0, gomax=0, detune=0):
+    """
+    Scan the ID GAP and go
+    to the max of lu_bpm intensity
+    gscan(centre=0): just scan
+    gscan(centre=1): go to centre of fit max
+    gscan(centre=1, gomax=1): go to max of intensity
+    gscan(centre=1,detune=1): position of slightly less flux with nicer beam shape
+         
+    """ 
+    import time
+
+    dock_area = bec.gui.new()
+    wr = dock_area.new(bec.gui.available_widgets.Waveform)
+    mot = dev.id_gap
+    det = dev.lu_bpmsum
+    wr.plot(device_x=mot.name, device_y=det.name)  ## names first !
+    # wr.plot(x=mot.name,y=det.name)           ### this comes later
+    wr.x_label = mot.name
+    wr.y_label = det.name
+
+    g0 = dev.id_gap.readback.get()
+    deltag = 0.05
+    ds = scans.line_scan(
+        dev.id_gap, -deltag, deltag, steps=30, exp_time=0.5, relative=True
+    )
+    gap_data = ds.scan.live_data.id_gap.id_gap.val
+    bpm_data = ds.scan.live_data.lu_bpmsum.lu_bpmsum.val
+
+    #maxy = max(bpm_data)
+    #indmax = np.argmax(bpm_data)
+    #gm = gap_data[indmax]
+
+    gcen, xm = fit_plot(gap_data, bpm_data, model="gauss")
+
+    if gomax:
+        gm = xm
+    else:
+        gm = gcen
+
+    print("gap off by ", g0 - gm, " mm")
+    if detune:
+        gm=gm*0.996 
+        print("moving to detuned gap value, slightly (0.15 %) below max")
+    
+    if centre:
+        time.sleep(0.2)
+        if min(gap_data) <= gm <= max(gap_data): 
+            scans.umv(dev.id_gap, gm, relative=False)
+            print("moving to ", gm, " mm")
+        else:
+            print("Fit too far off, try using option gomax=1")
+            
+    return
+

pxii_bec/macros/calculator.py

@@ -0,0 +1,355 @@
+"""Utility functions for calculating energy, wavelength, and Bragg angle."""
+
+from dataclasses import dataclass
+import numpy as np
+# from pxii_parameters import (EnergyDefaults, CamConversion)
+
+
+@dataclass(frozen=True)
+class Constants:
+    """Constants used in energy calculations"""
+
+    # # Physical Constants from https://physics.nist.gov/cuu/Constants/index.html
+    ANGSTROM_CONVERSION = 1e10  # Convert meters to angstrom
+    PLANCK_CONST_EV = 4.135667696e-15  # eV/Hz
+    SPEED_OF_LIGHT = 299792458  # m/s
+
+    # d-spacings
+    d_spacing = {120: 3.13481, 298: 3.13562}
+
+
+def speed_of_light_ang():
+    """
+    Calculate the speed of light in angstroms per second.
+
+    Returns:
+        float: The speed of light converted to angstroms per second.
+    """
+    return Constants.SPEED_OF_LIGHT * Constants.ANGSTROM_CONVERSION
+
+
+def en_wav_factor():
+    """
+    Calculate the energy wavelength factor.
+
+    This function computes a constant factor used to calculate energy
+    values in relation to wavelength by combining Planck's constant,
+    in eV/Hz, and the speed of light in angstrom.
+
+    Returns:
+        float: The computed energy wavelength factor.
+    """
+    return Constants.PLANCK_CONST_EV * speed_of_light_ang()
+
+
+# Helper Functions
+def convert_to_degrees(angle_mrad: float) -> float:
+    """
+    Convert an angle from milliradians to degrees.
+
+    Args:
+        angle_mrad: The angle value in milliradians.
+
+    Returns:
+        The angle converted into degrees as a float.
+    """
+    return np.rad2deg(angle_mrad / 1000)
+
+
+def create_conversion_result(
+    energy_ev: float, wavelength: float, bragg_angle_mrad: float
+) -> dict:
+    """
+    Creates a dictionary containing converted values of energy and angles.
+
+    This function takes the energy in electron-volts, the wavelength,
+    and the Bragg angle in milliradians as input. It computes and
+    returns a dictionary containing the energy in both electron-volts
+    and kiloelectron-volts, the wavelength, the Bragg angle in milliradians,
+    and the Bragg angle converted to degrees.
+
+    Args:
+        energy_ev: Energy value in electron-volts.
+        wavelength: Wavelength value.
+        bragg_angle_mrad: Bragg angle in milliradians.
+
+    Returns:
+        dict: A dictionary containing the following keys:
+            - "energy_kev": Energy value in kiloelectron-volts.
+            - "energy_ev": Energy value in electron-volts.
+            - "wavelength": Wavelength value.
+            - "bragg_angle_mrad": Bragg angle in milliradians.
+            - "bragg_angle_deg": Bragg angle in degrees.
+    """
+    return {
+        "energy_kev": energy_ev / 1000,
+        "energy_ev": energy_ev,
+        "wavelength": wavelength,
+        "bragg_angle_mrad": float(bragg_angle_mrad),
+        "bragg_angle_deg": float(convert_to_degrees(bragg_angle_mrad)),
+    }
+
+
+def print_conversion_result(result: dict) -> None:
+    """
+    Prints the energy-related conversion results to the console.
+    """
+
+    line = (
+        f"energy: {result['energy_ev']:.6g} eV, energy: {result['energy_kev']:.6g} keV, "
+        f"wavelength: {result['wavelength']:.4g} Å, "
+        f"bragg angle: {result['bragg_angle_mrad']:.5g} mrad, {result['bragg_angle_deg']:.4g} deg"
+    )
+    print(line)
+
+
+# Conversion Functions
+def calculate_wavelength_from_angle(bragg_angle_mrad: float, temp=120) -> float:
+    """
+    calculate_wavelength_from_angle(bragg_angle_mrad: float) -> float
+
+    Arguments:
+        bragg_angle_mrad: The Bragg angle in milliradians, used to compute the
+                          sine value required for the wavelength calculation.
+
+    Returns:
+        The calculated wavelength as a float value.
+    """
+    d = Constants.d_spacing[temp]
+    return 2 * d * np.sin(bragg_angle_mrad / 1000)
+
+
+def calculate_energy_from_wavelength(wavelength: float) -> float:
+    """
+    Calculates the energy of a photon based on its wavelength.
+
+    Args:
+        wavelength: The wavelength of the photon in angstrom.
+
+    Returns:
+        The energy of the photon in eV.
+    """
+    return en_wav_factor() / wavelength
+
+
+def calculate_wavelength_from_energy(energy_ev: float) -> float:
+    """
+    Calculates the wavelength of a photon from its energy.
+
+    Arguments:
+        energy_ev: float
+            The energy of the photon in electronvolts (eV).
+
+    Returns:
+        float
+            The calculated wavelength of the photon in angstrom.
+    """
+    return en_wav_factor() / energy_ev
+
+
+def calculate_bragg_angle_from_wavelength(wavelength: float, temp=120) -> float:
+    """
+    Calculate the Bragg angle in milliradians for a given wavelength.
+
+    Args:
+        wavelength: The wavelength in angstrom.
+
+    Returns:
+        The Bragg angle in milliradians as a float.
+    """
+    d = Constants.d_spacing[temp]
+    angle_rad = np.arcsin(wavelength / (2 * d))
+    return angle_rad * 1000
+
+
+def convert_input_angle_to_mrad(bragg_angle: float) -> float:
+    """
+    Convert input angle into milliradians (mrad).
+
+    This function takes an angle as input and determines its likely unit,
+    converting it to milliradians (mrad) if necessary. If the input value
+    is less than 1, it is assumed to be in radians and is converted to
+    mrad. If the input value falls between predefined minimum and
+    maximum values for mrad, it is assumed to be in degrees and thus
+    converted to mrad using the degrees-to-radians conversion factor.
+
+    For input values that don't match these scenarios, it assumes
+    that the input is already in mrad and returns it unchanged.
+
+    Arguments:
+        bragg_angle (float): The input Bragg angle, which can be in
+        radians, degrees, or milliradians.
+
+    Returns:
+        float: The Bragg angle converted into milliradians (mrad).
+    """
+    if bragg_angle < 1:  # Likely the input angle is in radians
+        return bragg_angle * 1000
+    if 3 < bragg_angle < 25:  # Likely input angle is in degrees
+        return np.deg2rad(bragg_angle) * 1000
+    return bragg_angle  # Already in mrad
+
+
+# Core Functions
+def validate_energy(energy_ev):
+    """
+    Validates the energy value to ensure it falls within the acceptable range. The function
+    converts the provided energy from keV to eV if the input value is less than 1/1000 of the
+    maximum energy value. It then checks whether the energy is within the defined bounds.
+    If the energy value is outside the acceptable range, the function raises a ValueError.
+
+    Args:
+        energy_ev (float): The energy value in eV or keV to be validated. If this value is
+            smaller than 1/1000 of the maximum allowed energy (in eV), it will be multiplied
+            by 1000 to convert it from keV to eV.
+
+    Returns:
+        float: The validated energy value in eV that falls within the acceptable range.
+
+    Raises:
+        ValueError: If the energy value is outside the defined range of
+            [MIN_ENERGY_EV, MAX_ENERGY_EV].
+    """
+    if energy_ev < EnergyDefaults.max_energy_ev / 1000:  # Assuming the input is in keV.
+        energy_ev *= 1000
+    if not EnergyDefaults.min_energy_ev <= energy_ev <= EnergyDefaults.max_energy_ev:
+        raise ValueError(
+            f"Energy of {energy_ev} eV is outside the valid range "
+            f"({EnergyDefaults.min_energy_ev} eV to {EnergyDefaults.max_energy_ev} eV)"
+        )
+    return energy_ev
+
+
+def convert_from_bragg(
+    bragg_angle_mrad: float, temp=120, print_result: bool = False
+) -> dict:
+    """
+    Convert the Bragg angle to wavelength and energy, returning the result as a dictionary.
+
+    This function converts a given Bragg angle (in milliradians) into the corresponding
+    wavelength and energy values, and returns them in a dictionary format. The function
+    also supports optional printing of the calculated results.
+
+    Args:
+        bragg_angle_mrad (float): The Bragg angle in milliradians to be converted.
+        print_result (bool): Whether to print the conversion result. Defaults to False.
+
+    Returns:
+        dict: A dictionary containing the following keys:
+            - 'energy_ev': Energy in electronvolts.
+            - 'wavelength': Wavelength corresponding to the input angle.
+            - 'bragg_angle_mrad': Input Bragg angle in milliradians.
+    """
+    bragg_angle_mrad = convert_input_angle_to_mrad(bragg_angle_mrad)
+    wavelength = float(calculate_wavelength_from_angle(bragg_angle_mrad, temp=temp))
+    energy_ev = float(calculate_energy_from_wavelength(wavelength))
+    result = create_conversion_result(energy_ev, wavelength, bragg_angle_mrad)
+    if print_result:
+        print_conversion_result(result)
+    return result
+
+
+def convert_from_energy(energy_ev: float, temp=120, print_result: bool = False) -> dict:
+    """
+    Convert energy in electron volts (eV) to  wavelength and Bragg angle in milliradians
+    (mrad). This method validates the given energy, calculates corresponding properties,
+    and optionally prints the result.
+
+    Args:
+        energy_ev: Energy value in electron volts (float) to be converted.
+        print_result: Flag indicating whether to print the resulting
+            conversion details (bool). Defaults to False.
+
+    Returns:
+        A dictionary containing the following key-value pairs:
+            - "energy_ev" (float): Validated energy in eV.
+            - "wavelength" (float): Calculated wavelength in meters.
+            - "bragg_angle_mrad" (float): Calculated Bragg angle in mrad.
+    """
+    energy_ev = validate_energy(energy_ev)
+    wavelength = calculate_wavelength_from_energy(energy_ev)
+    bragg_angle_mrad = float(
+        calculate_bragg_angle_from_wavelength(wavelength, temp=temp)
+    )
+    result = create_conversion_result(energy_ev, wavelength, bragg_angle_mrad)
+    if print_result:
+        print_conversion_result(result)
+    return result
+
+
+def convert_from_wavelength(
+    wavelength: float,
+    temp: float = 120,
+    print_result: bool = False,
+) -> dict:
+    """
+    Convert a given wavelength value into corresponding energy, Bragg angle, and
+    generate a result dictionary.
+
+    The function processes a wavelength value, checks its validity against a
+    permitted range, calculates corresponding energy and Bragg angle, and
+    formats the results into a dictionary. Optionally, the function can print
+    the result.
+
+    Parameters:
+        wavelength: float
+            The input wavelength value in Angstroms to be converted. Should
+            fall within the permitted wavelength range.
+        print_result: bool
+            Optional flag indicating whether to print the conversion result.
+            Default is False.
+
+    Returns:
+        dict
+            A dictionary containing the energy (electron-volts), wavelength
+            (Angstroms), and Bragg angle (milliradians). If the wavelength is
+            outside of the permitted range, returns None.
+    """
+    energy_ev = calculate_energy_from_wavelength(wavelength)
+    bragg_angle_mrad = float(
+        calculate_bragg_angle_from_wavelength(wavelength, temp=temp)
+    )
+    result = create_conversion_result(energy_ev, wavelength, bragg_angle_mrad)
+    if print_result:
+        print_conversion_result(result)
+    return result
+
+
+def calc_perp_position(
+    energy_ev: float,
+    print_result: bool = False,
+) -> float:
+    """
+    Calculate the perpendicular motor position based on provided energy in electron-volts (eV).
+
+    This function computes the perpendicular motor position using the given energy value in
+    electron-volts. The calculation is based on the Bragg angle derived from the energy. An optional
+    parameter allows printing the result during execution.
+
+    Parameters:
+        energy_ev (float): The energy value in electron-volts used for the calculation.
+        print_result (bool): Flag to determine whether to print the computed perpendicular offset.
+            Default is False.
+
+    Returns:
+        float: The computed perpendicular position.
+
+    Raises:
+        None
+    """
+    result = convert_from_energy(energy_ev, print_result=False)
+    bragg_angle_rad = result["bragg_angle_mrad"] / 1000
+    perp_offset = float(EnergyDefaults.beam_offset / (2 * np.cos(bragg_angle_rad))) - 3
+    if print_result:
+        print(f"Perp = {perp_offset: .4f}")
+    return perp_offset
+
+
+def calc_scam_microns(pixels, zoom=1000):
+    """Convert pixels to microns for the sample camera"""
+    return float(pixels / (CamConversion.a * np.exp(CamConversion.b * zoom)))
+
+def calc_bsccam_microns(pixels):
+    """Convert pixels to microns for the BSC camera"""
+    return pixels*20
+

pxii_bec/macros/flux_di.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+# ### PYTHON pro to determine the flux from the current 
+# ### on a Si diode 
+# ### A Pauluhn, Dec 2012 
+# ### from IDL pro flux_diode.pro (2010) 
+
+
+## NOTE: for TRANSFER DIODE:
+## switch off ANY ambient light 
+## note that: 10 um Si, and 0 um Al  for transfer diode Hamamatsu 
+
+# ### import the standard 
+import os, sys, string, time 
+import re
+import numpy as np
+import math
+
+
+
+
+def fluxcalc(curr, en, t_air, t_si, t_al):
+
+    #------ parameters:
+    rsi = 2.33      # g/cm^3 density of Si
+    ral = 2.699     # g/cm^3 density of Al 
+    rair = 1.205e-3  # g/cm^3 density of air 
+
+    eps_si = 3.62 # eV , energy req for charge separation in Si (el/hole pair) 
+    e = 1.602e-19 # As , elem charge 
+
+    t_si = 0.0012  # thickness of diode in cm (==12 micrometres) ; [*10000]
+    t_al = 0.002   # thickness of diode in cm (==20 micrometres) 
+
+    #--------- calc----------------------------------------------------
+    #energy deposit in Silicon
+    #ratio of photoelectric cross section to density for Silicon
+    #alog10(A_Si)= 4.158 - 2.238*alog10(Ep) - 0.477*alog10(Ep)^2 + 0.0789*alog10(Ep)^3 
+
+    ps = np.poly1d([0.0789, - 0.477, - 2.238, 4.158])
+    #print ps
+    # ps.c 
+    # print ps(2.3) 
+
+    leval = math.log10(en) 
+    hlp_si = ps(leval) 
+    a_si = np.pow(10, hlp_si) 
+    #efact = np.exp(-A_Si*rsi*t_Si) 
+    #sifact =1 - efact
+    sifact = - np.expm1(-a_si*rsi*t_si) 
+    #print 'sifact = ', sifact
+
+    fl = 1000.*curr * eps_si/1.602/en/sifact *1.e13 # curr expected in A
+    #print 'fl = ', fl 
+
+    #Aluminium attenuation
+    #ratio of photoelectric cross section to density for Aluminium
+    pa = np.poly1d([ 0.0638, - 0.413, - 2.349, 4.106])
+    
+    hlp_al = pa(leval)
+    a_al = pow(10, hlp_al) 
+
+    # attenuation due to aluminium
+    alfact = np.exp(-a_al*ral*t_al)
+    #print 'alfact = ', alfact
+    # Air attenuation
+    # ratio of photoelectric cross section to density for air
+
+    pair = np.poly1d([0.928, - 2.348, - 1.026, 3.153])
+    hlp_air = pair(leval)
+    a_air = np.pow(10, hlp_air) 
+
+    #print('hlp_air, a_air, rair, t_air = ', hlp_air, a_air, rair , t_air)
+
+    # attenuation due to air 
+    airfact = np.exp(-a_air*rair*t_air/10.)
+    #airfact = 1.
+
+    
+    #print 'airfact = ', airfact
+    #print ' t_air = ', t_air 
+
+    # total flux from photocurrent 
+    fl = fl/alfact/airfact
+    #print 'fl = ', fl 
+
+    return fl 
+
+#
+# Main
+#
+
+def flux_x10sa():
+    curr = input('Please enter the current [in A] ')
+    curr = float(curr)
+    print('Current =  ', curr)
+    en = input('Please enter the energy [in keV] ')
+    en = float(en) 
+    print('Energy =  ', en)
+    ## d_det = input('Please enter the detector distance [in mm] ')
+    d_det = dev.det_z.user_readback.get()
+    d_off = 15 ## at X10SA   ### input('Please enter the OFFSET of the diode to det surface [in mm] ') 
+    
+    t_air = float(d_det) +  float(d_off)
+    print('Air path length (CARE ! add det distance and offset of diode) =  ', t_air) # include additional diode distance
+
+    #  t_si = input('Please enter the Si thickness of diode [in um] ') 
+    #  t_si = float(t_si) / 10000. # to cm 
+
+    t_si = 12 # um ## at X10SA
+    #  t_al = input('Please enter the Al thickness of diode [in um] ') 
+    #  t_al = float(t_al) / 10000. # to cm
+
+    t_al = 20 # um ## at X10SA
+    fl = fluxcalc(curr, en, t_air,t_si, t_al)
+
+    print( " Diode Current = %.6f  A " % curr)
+    print( " Energy = %.4f  keV "  % en )
+
+    print( " Thickness of active Si layer = %.4f\t  cm " % t_si) #*10000. 
+    print( " Thickness of Al layer in front of diode = %.4f\t  cm "  % t_al) #*10000. 
+    print( " Length of path of air in front of diode = %.4f\t  mm " % t_air) #*10000.
+
+    print (" Flux = %6.2e\t ph/s "  % fl)
+
+    return fl
+

pxii_bec/macros/mx_basics.py

@@ -0,0 +1,225 @@
+"""Get data from an h5 file or BEC history and perform fitting."""
+
+import numpy as np
+from lmfit.models import GaussianModel, LorentzianModel, VoigtModel, ConstantModel, LinearModel
+from scipy.ndimage import gaussian_filter1d
+import h5py
+import matplotlib.pyplot as plt
+
+
+def create_fit_parameters(
+    deriv: bool = False, model: str = "Voigt", baseline: str = "Linear", smoothing: None = None
+):
+    """Store the fit parameters in a dictionary."""
+    # map input model to lmfit model name
+    model_mappings = {
+        "Gaussian": GaussianModel,
+        "Lorentzian": LorentzianModel,
+        "Voigt": VoigtModel,
+        "Constant": ConstantModel,
+        "Linear": LinearModel,
+    }
+    return {
+        "deriv": deriv,
+        "model": model_mappings[model],
+        "baseline": model_mappings[baseline],
+        "smoothing": smoothing,
+    }
+
+
+def get_data_from_h5(signal_name: str = "lu_bpmsum"):
+    """Get data from an h5 file."""
+    with h5py.File("scan_676.h5", "r") as f:
+        entry = f["entry"]["collection"]
+        y_data = entry["devices"][signal_name][signal_name]["value"][:]
+        motor_data = entry["metadata"]["bec"]
+        motor_name = motor_data["scan_motors"][0].decode()
+        scan_number = motor_data["scan_number"][()]
+        x_data = entry["devices"][motor_name][motor_name]["value"][:]
+    return {
+        "x_data": x_data,
+        "y_data": y_data,
+        "signal_name": signal_name,
+        "motor_name": motor_name,
+        "scan_number": str(scan_number),
+    }
+
+
+def get_data_from_history(history_index: int, signal_name: str = "lu_bpmsum"):
+    """Read data from the BEC history and return the X and Y data as arrays."""
+    scan = bec.history[history_index]
+    md = scan.metadata["bec"]
+    motor_name = md["scan_motors"][0].decode()
+    scan_number = md["scan_number"]
+    x_data = scan.devices[motor_name][motor_name].read()["value"]
+    y_data = scan.devices[signal_name][signal_name].read()["value"]
+    return {
+        "signal_name": signal_name,
+        "x_data": x_data,
+        "y_data": y_data,
+        "motor_name": motor_name,
+        "scan_number": scan_number,
+    }
+
+
+def process_data(data, fit_params):
+    """
+    Process the signal data for fitting based on derivative or smoothing.
+    """
+    smoothing, deriv = fit_params["smoothing"], fit_params["deriv"]
+    signal_name = data["signal_name"]
+    y_data = data["y_data"]
+
+    if deriv:
+        if smoothing:
+            y_smooth = gaussian_filter1d(y_data, smoothing)
+            fitting_data = np.gradient(y_smooth)
+            signal_name = f"Derivative of smoothed {signal_name}"
+        else:
+            fitting_data = np.gradient(y_data)
+            signal_name = f"Derivative of {signal_name}"
+    elif smoothing and smoothing > 0.01:
+        fitting_data = gaussian_filter1d(y_data, smoothing)
+        signal_name = f"Smoothed {signal_name}"
+    else:
+        fitting_data = y_data
+
+    updated_data = {"y_to_fit": fitting_data, "signal_name": signal_name}
+    data.update(updated_data)
+    return data
+
+
+def fit(data, fit_params):
+    """Fit a signal to a model and return the fitting results."""
+    # Create the model
+    peak_model = fit_params["model"](prefix="peak_")
+    baseline_model = fit_params["baseline"](prefix="base_")
+    full_model = peak_model + baseline_model
+
+    # Prepare data
+    processed_data = process_data(data, fit_params)
+    params = full_model.make_params()
+    y_min = np.min(processed_data["y_to_fit"])
+
+    # Configure baseline parameters
+    if fit_params["baseline"] == ConstantModel:
+        params["base_c"].set(value=y_min)
+    elif fit_params["baseline"] == LinearModel:
+        params["base_intercept"].set(value=y_min)
+        params["base_slope"].set(value=0)
+
+    # Add peak-specific parameters
+    params.update(peak_model.guess(processed_data["y_to_fit"], x=processed_data["x_data"]))
+
+    # Perform the fitting
+    lmfit_result = full_model.fit(processed_data["y_to_fit"], params, x=processed_data["x_data"])
+
+    # Find the X that gives the max Y
+    max_index = np.argmax(processed_data["y_to_fit"])
+    x_max = processed_data["x_data"][max_index]
+
+    # Collect results
+    return {
+        "model": fit_params["model"].__name__,
+        "fwhm": lmfit_result.params["peak_fwhm"].value,
+        "centre": lmfit_result.best_values["peak_center"],
+        "height": lmfit_result.params["peak_height"].value,
+        "chi_sq": lmfit_result.chisqr,
+        "lmfit_result": lmfit_result,
+        "x_max": x_max,
+    }
+
+
+def plot_fitted_data(data, fit_result):
+    """Plot the original data and the fitted model."""
+    plt.plot(data["x_data"], data["y_to_fit"], label="Data")
+    plt.plot(
+        data["x_data"],
+        fit_result["lmfit_result"].best_fit,
+        "-",
+        label=f"FWHM = {fit_result['fwhm']:.3f},"
+        f"Centre = {fit_result['centre']:.3f}, "
+        f"Height = {fit_result['height']:.3f}",
+    )
+    plt.xlabel(data["motor_name"])
+    plt.ylabel(data["signal_name"])
+    plt.title(f"Scan {data['scan_number']}, fitted with {fit_result['model']}")
+    plt.grid(True)
+    plt.legend()
+    plt.show()
+
+
+def select_bec_window(dock_area_name="Fitting"):
+    """Check to see if the fitting results dock is already open and re-create it if not"""
+    open_docks = bec.gui.windows
+    if open_docks.get(dock_area_name) is None:
+        dock_area = bec.gui.new(dock_area_name)
+        wf = dock_area.new(bec.gui.available_widgets.Waveform, object_name="Plot")
+        text_box = dock_area.new(
+            bec.gui.available_widgets.TextBox, object_name="Results", where="bottom"
+        )
+    else:
+        wf = bec.gui.Fitting.Plot.Waveform
+        text_box = bec.gui.Fitting.Results.TextBox
+    return wf, text_box
+
+
+def plot_live_data_bec(motor_name, signal_name, window_name="Fitting"):
+    """
+    Plotting live data for motor and signal using BEC.
+
+    This function plots live data from a specified motor and signal.
+    It clears the current plot window, sets its title, labels the axes
+    with the provided motor and signal names, and initializes live plotting
+    on the given signal against the motor.
+
+    Args:
+        motor_name (str): The name of the motor to be used as the x-axis.
+        signal_name (str): The name of the signal to be used as the y-axis.
+
+    Returns:
+        None
+    """
+    wf, text_box = select_bec_window(window_name)
+    text_box.set_plain_text("Plotting live data")
+    wf.clear_all()
+    wf.title = "Scan: Live scan"
+    wf.x_label = motor_name
+    wf.y_label = signal_name
+    wf.plot(device_x=motor_name, device_y=signal_name)
+
+
+def plot_fitted_data_bec(data, fit_result):
+    """
+    Plot fitted data and display fitting parameters in the specified window.
+
+    This function selects a BEC window and plots the original data along with the
+    fitted function. Additionally, it displays the fitting results in a text
+    box within the same window for better visualization of the fit results.
+
+    Parameters:
+    data : dict
+        Dictionary containing the original dataset, where 'x_data' and 'y_to_fit'
+        hold the independent variable and the dependent variable, respectively,
+        'scan_number' represents the scan number, 'motor_name' and 'signal_name'
+        provide axis labels.
+    fit_result : dict
+        Dictionary containing the results of the fit, including parameters such
+        as 'centre', 'fwhm', 'height', and the fitted model stored under
+        'lmfit_result', with its 'best_fit' attribute representing the fitted data.
+    """
+    wf, text_box = select_bec_window()
+    fit_text = (
+        f"Fit parameters: Centre = {fit_result['centre']:.4f}, "
+        f"FWHM = {fit_result['fwhm']:.3f}, "
+        f"Height = {fit_result['height']:.4f}\n"
+        f"Model = {fit_result['model']}\n"
+        f"Chi sq = {fit_result['chi_sq']:.3g}"
+    )
+    text_box.set_plain_text(fit_text)
+    wf.clear_all()
+    wf.title = f"Scan: {data['scan_number']}"
+    wf.x_label = data["motor_name"]
+    wf.y_label = data["signal_name"]
+    wf.plot(x=data["x_data"], y=data["y_to_fit"], label="data")
+    wf.plot(x=data["x_data"], y=fit_result["lmfit_result"].best_fit, label="Fit to data")

pxii_bec/macros/mx_methods.py

@@ -0,0 +1,363 @@
+"""Use the methods in mx_basics to perform:
+1) a go_to_peak scan, that scans a motor, finds the peak position and moves to peak
+2) fits data from a bec history file
+"""
+
+from dataclasses import dataclass
+import numpy as np
+
+# from pxiii_parameters import FitDefaults, BPMScans, MirrorConfig
+
+# from mx_basics import (
+#     create_fit_parameters,
+#     get_data_from_history,
+#     fit,
+#     plot_fitted_data_bec,
+#     plot_live_data_bec,
+# )
+
+
+# Method functions
+def calculate_step_size(start: float, stop: float, steps: int) -> float:
+    """
+    Provides the function to calculate the step size for dividing a specified range
+    into a given number of steps.
+
+    Args:
+        start: The starting value of the range.
+        stop: The stopping value of the range.
+        steps: The number of steps to divide the range into. Must be at least 1.
+
+    Raises:
+        ValueError: If the steps value is less than 1.
+
+    Returns:
+        The calculated step size as a float, rounded to three decimal places.
+    """
+    if steps < 1:
+        raise ValueError("Number of steps must be at least 1.")
+    return round((stop - start) / steps, 3)
+
+
+def move_to_position(motor_device, motor_name: str, position: float, data: dict):
+    """
+    Function to move a specified motor device to a given position.
+
+    The function verifies if the requested position is within the scan range of the
+    motor device provided. If the position is outside the range, the motor is
+    moved to the center of its scan range, an error message is raised, and the
+    operation is halted. If the position is valid, the motor is moved to the
+    specified position.
+
+    Parameters:
+        motor_device: The motor device to be moved.
+        motor_name: str
+            The name of the motor as a string
+        position: float
+            The desired position to move the motor to. Position should be within
+            the scan range of the motor determined by the provided data.
+        data: dict
+            A dictionary containing "x_data", which is used to determine the
+            scan range of the motor.
+
+    Raises:
+        ValueError: Raised if the specified position is outside the valid scan
+        range determined by "x_data" in the data dictionary. The motor will
+        return to the center of its scan range in this case.
+    """
+
+    motor_min = np.min(data["x_data"])
+    motor_max = np.max(data["x_data"])
+    motor_centre = (motor_max + motor_min) / 2
+
+    if not motor_min <= position <= motor_max:
+        scans.umv(motor_device, motor_centre, relative=False)
+        msg = (
+            f"Position {position: .2f} is outside the scan range of "
+            f"{motor_min: .2f}to {motor_max: .2f}. "
+            f"Returning to centre of scan range {motor_centre: .3f}."
+        )
+        raise ValueError(msg)
+    motor_position = round(position, 4)
+    scans.umv(motor_device, motor_position, relative=False)
+    print(f"\n Moving {motor_name} to position {motor_position: .3f}")
+
+
+@dataclass(frozen=True)
+class FitDefaults:
+    """Default values for fitting routines"""
+
+    # Constants for default models, baselines, and parameters
+    MODEL = "Voigt"
+    BASELINE = "Linear"
+    SETTLE_TIME = 0.1
+    RELATIVE_MODE = True
+
+
+def go_to_peak(
+    motor_device,
+    signal_device,
+    start: float,
+    stop: float,
+    steps: int,
+    relative: bool = FitDefaults.RELATIVE_MODE,
+    plot: bool = True,
+    settle: float = FitDefaults.SETTLE_TIME,
+    confirm: bool = True,
+    gomax: bool = False,
+):
+    """
+    Go to the peak of a signal by scanning a motor within a specified range and
+    identifying the optimal position based on signal peak data.
+
+    Parameters:
+        motor_device: The motor device to be scanned.
+        signal_device: The signal device to monitor during the scan.
+        start (float): The starting position of the scan. Ignored if `relative` is True.
+        stop (float): The ending position of the scan. Ignored if `relative` is True.
+        steps (int): The number of steps to divide the scan range into.
+        relative (bool, optional): If True, interpret `start` and `stop` as relative to
+            the current motor position. Defaults to RELATIVE_MODE constant.
+        plot (bool, optional): If True, plot the scan data and the fitted results.
+            Defaults to True.
+        settle (float, optional): The time in seconds to wait after each step for the
+            signal to stabilize. Defaults to DEFAULT_SETTLE_TIME constant.
+        confirm (bool, optional): If True, ask for user confirmation before starting
+            the scan. Defaults to True.
+
+    Raises:
+        Exception: Raises exceptions potentially raised by dependent functions or
+            operations such as plotting, fitting, or motor movement.
+
+    Returns:
+        None
+    """
+    motor_name = motor_device.name
+    signal_name = signal_device.name
+    # wf.plot(x_name=motor_name, y_name=signal_name)
+    if plot:
+        plot_live_data_bec(motor_name, signal_name)
+
+    # Validate and calculate step size
+    step_size = calculate_step_size(start, stop, steps)
+
+    # Confirm the scan range
+    # current_motor_position = motor_device.user_readback.get()
+    current_motor_position = motor_device.read()[motor_name]["value"]
+    if confirm:
+        if relative:
+            scan_start = current_motor_position + start
+            scan_end = current_motor_position + stop
+            print(
+                f"\nScanning from {scan_start: .6g} to {scan_end: .6g} in "
+                f"{steps} steps of size {step_size}"
+            )
+            print(f"Relative mode = {relative}")
+        else:
+            print(
+                f"\nScanning from {start: .5g} to {stop: .5g} in {steps} steps of size {step_size}"
+            )
+            print(f"Relative mode = {relative}")
+        input("Press Enter to continue...")
+    # Perform the scan
+    scan_result = scans.line_scan(
+        motor_device, start, stop, steps=steps, relative=relative, settling_time=settle
+    )
+    motor_data = scan_result.scan.live_data[motor_name][motor_name].val
+    signal_data = scan_result.scan.live_data[signal_name][signal_name].val
+    scan_number = "Current"
+
+    data = {
+        "x_data": np.array(motor_data),
+        "y_data": np.array(signal_data),
+        "motor_name": motor_name,
+        "signal_name": signal_name,
+        "motor_device": motor_device,
+        "scan_number": scan_number,
+    }
+
+    # Define and fit model to scan data
+    fit_params = create_fit_parameters(False, FitDefaults.MODEL, FitDefaults.BASELINE)
+    fit_result = fit(data, fit_params)
+
+    # Plot the fitted data if plot = True
+    if plot:
+        plot_fitted_data_bec(data, fit_result)
+
+    # If gomax is set then move to the maximum value, rather than the fit centre
+    if gomax:
+        value = fit_result["x_max"]
+        print(f"Max position is at {value}")
+        move_to_position(data["motor_device"], data["motor_name"], fit_result["x_max"], data)
+    else:
+        # Safely move the motor to the peak position
+        move_to_position(data["motor_device"], data["motor_name"], fit_result["centre"], data)
+
+
+def fit_history(
+    history_index: int,
+    signal_name: str,
+    deriv: bool = False,
+    model: str = FitDefaults.MODEL,
+    move_to_peak: bool = False,
+):
+    """
+    Retrieve and analyze historical data by fitting a model, optionally moving to
+    a peak position.
+
+    Parameters:
+        history_index (int): Index of the historical data set to retrieve.
+        signal_name (str): Name of the signal to fit.
+        deriv (bool, optional): Whether to include the derivative in the fitting
+            procedure. Defaults to False.
+        model (str, optional): Name of the model to use for fitting. Defaults to
+            DEFAULT_MODEL.
+        move_to_peak (bool, optional): Whether to move the motor to the peak position
+            after fitting. Defaults to False.
+
+    Raises:
+        KeyError: If required keys are not found in the retrieved data dictionary.
+        ValueError: If the fitting process fails or produces invalid results.
+
+    Returns:
+        None
+    """
+    # Retrieve historical data
+    data = get_data_from_history(history_index, signal_name)
+
+    # Define fitting parameters
+    fit_params = create_fit_parameters(deriv, model, FitDefaults.BASELINE)
+
+    # Perform fit and plot the data
+    fit_result = fit(data, fit_params)
+    plot_fitted_data_bec(data, fit_result)
+
+    # Optionally move the motor to the peak position
+    if move_to_peak:
+        move_to_position(data["motor_device"], data["motor_name"], fit_result["centre"], data)
+
+
+def scan_bpm(bpmname):
+    """
+    Runs a grid scan of a BPM in x and y,  and plots each channel
+    as a heatmap.
+
+    Parameters:
+        bpmname: the name of the bpm to be scanned e.g. "fe"
+
+    """
+
+    # Open a dock area and set up the heatmaps
+    dock_area = bec.gui.new("XBPM_Scan")
+    wf5 = dock_area.new(bec.gui.available_widgets.Heatmap, object_name="Sum")
+    wf1 = dock_area.new(
+        bec.gui.available_widgets.Heatmap,
+        object_name="Ch1",
+        relative_to="Sum",
+        where="bottom",
+    )
+    wf3 = dock_area.new(
+        bec.gui.available_widgets.Heatmap,
+        object_name="Ch3",
+        relative_to="Ch1",
+        where="right",
+    )
+    wf4 = dock_area.new(
+        bec.gui.available_widgets.Heatmap,
+        object_name="Ch4",
+        relative_to="Ch3",
+        where="bottom",
+    )
+    wf2 = dock_area.new(
+        bec.gui.available_widgets.Heatmap,
+        object_name="Ch2",
+        relative_to="Ch1",
+        where="bottom",
+    )
+    wfscan = dock_area.new(bec.gui.available_widgets.ScanControl, object_name="ScanControl")
+
+    cfg = getattr(BPMScans, bpmname)
+
+    wf1.x_label = cfg["x_name"]
+    wf1.y_label = cfg["y_name"]
+    wf1.plot(
+        device_x=cfg["x_name"],
+        device_y=cfg["y_name"],
+        device_z=cfg["z1_name"],
+        color_map="plasma",
+    )
+
+    wf2.x_label = cfg["x_name"]
+    wf2.y_label = cfg["y_name"]
+    wf2.plot(
+        device_x=cfg["x_name"],
+        device_y=cfg["y_name"],
+        device_z=cfg["z2_name"],
+        color_map="plasma",
+    )
+
+    wf3.x_label = cfg["x_name"]
+    wf3.y_label = cfg["y_name"]
+    wf3.plot(
+        device_x=cfg["x_name"],
+        device_y=cfg["y_name"],
+        device_z=cfg["z3_name"],
+        color_map="plasma",
+    )
+
+    wf4.x_label = cfg["x_name"]
+    wf4.y_label = cfg["y_name"]
+    wf4.plot(
+        device_x=cfg["x_name"],
+        device_y=cfg["y_name"],
+        device_z=cfg["z4_name"],
+        color_map="plasma",
+    )
+
+    wf5.x_label = cfg["x_name"]
+    wf5.y_label = cfg["y_name"]
+    wf5.plot(
+        device_x=cfg["x_name"],
+        device_y=cfg["y_name"],
+        device_z=cfg["z5_name"],
+        color_map="plasma",
+    )
+    # Run the scan
+    x_mot = cfg["x_device"]
+    y_mot = cfg["y_device"]
+    # scans.grid_scan(x_mot, -0.5, 0.5, 20, y_mot, -0.5, 0.5, 20,
+    #                     exp_time=0.5, relative=False, snaked=True)
+
+
+def optimise_kb(mirror):
+    """
+    Runs a grid scan of a the upstream and downstream benders,
+    and plots a heatmap of the sample camera x or y sigma.
+
+    Parameters:
+        mirror: either "hfm" or :vfm"
+
+    """
+
+    # Open a dock area and set up the heatmaps
+    dock_area = bec.gui.new(mirror)
+    wf1 = dock_area.new(bec.gui.available_widgets.Heatmap, object_name="Heatmap")
+
+    wfscan = dock_area.new(bec.gui.available_widgets.ScanControl, object_name="ScanControl")
+
+    cfg = getattr(MirrorConfig, mirror)
+
+    wf1.x_label = cfg["bu_name"]
+    wf1.y_label = cfg["bd_name"]
+    wf1.plot(
+        device_x=cfg["bu_name"],
+        device_y=cfg["bd_name"],
+        device_z=cfg["z_name"],
+        color_map="plasma",
+    )
+
+    # Run the scan
+    x_mot = cfg["x_device"]
+    y_mot = cfg["y_device"]
+    # scans.grid_scan(x_mot, -0.02, 0.02, 11, y_mot, -0.02, 0.02, 11,
+    #                     exp_time=0.5, relative=True, snaked=True)

pxii_bec/macros/pxii_energy.py

@@ -0,0 +1,248 @@
+"""Script to change energy at PXII by setting gap, DCM motors and mirror stripe
+
+Moving DCM motors - implemented for Bragg, pitch and perp
+Gap - optional, can be switched off using move_gap=False
+Mirrors - change of mirror stripe is not yet implemented
+Plotting optional
+
+"""
+
+import numpy as np
+
+# from pxii_gap import set_gap
+# from mx_methods import go_to_peak
+# from pxii_parameters import EnergyDefaults, Calibration
+
+# from calculator import (
+#     calc_perp_position,
+#     validate_energy,
+#     convert_from_bragg,
+#     convert_from_energy,
+# )
+
+
+def get_current_energy():
+    """
+    Returns the energy in eV from the current bragg angle.
+    """
+    current_bragg_angle = dev.dcm_bragg.user_readback.get()
+    current_energy = convert_from_bragg(current_bragg_angle, print_result=False)["energy_ev"]
+    return current_energy
+
+
+# Functions below are common to all beamlines
+def calculate_energy_difference(current_energy, target_energy):
+    """
+    Calculate the absolute difference in energy between the current energy level
+    and the target energy level.
+    """
+    return abs(target_energy - current_energy)
+
+
+def get_mirror_stripe(energy_ev):
+    """
+    Determines the mirror stripe material based on the energy level provided
+     and the specified thresholds for silicon, rhodium, and platinum.
+
+    Args:
+        energy_ev (float): Energy level in electron volts, used to determine
+            the corresponding material type.
+
+    Returns:
+        str: A string indicating the material type corresponding to the provided
+            energy level. Possible values are "silicon", "rhodium", or "platinum".
+    """
+    if energy_ev <= EnergyDefaults.stripe_thresholds["silicon"]:
+        return "silicon"
+    if (
+        EnergyDefaults.stripe_thresholds["silicon"]
+        < energy_ev
+        <= EnergyDefaults.stripe_thresholds["rhodium"]
+    ):
+        return "rhodium"
+    return "platinum"
+
+
+def set_mirror_stripe(energy_ev):
+    """
+    Selects and sets the appropriate mirror stripe based on the given energy value
+    in electron volts (eV).
+
+    Args:
+        energy_ev (float): The energy value in electron volts
+
+    Returns:
+        None
+    """
+    selected_stripe = get_mirror_stripe(energy_ev)
+    print(f"Selected mirror stripe: {selected_stripe}")
+
+
+def mono_pitch_scan(plot=True):
+    """Scan the monochromator pitch and move to the peak."""
+    # Move to the calculated pitch value for the current energy
+    energy = get_current_energy()
+    pos = get_dcm_motors_positions(energy)
+    print(f"Setting DCM Pitch to default value of {pos['dcm_pitch']}")
+    scans.umv(EnergyDefaults.mono_pitch, pos["dcm_pitch"], relative=False)
+    # Go to peak using default parameters from EnergyDefaults
+    if plot:
+        print("Scanning monochromator pitch and moving to peak, with plotting.")
+        go_to_peak(
+            EnergyDefaults.mono_pitch,
+            EnergyDefaults.signals["sig1"],
+            -EnergyDefaults.pitch_scan["halfwidth"],
+            EnergyDefaults.pitch_scan["halfwidth"],
+            steps=EnergyDefaults.pitch_scan["steps"],
+            relative=True,
+            settle=0.01,
+            plot=True,
+            confirm=False,
+        )
+    else:
+        print("Scanning monochromator pitch and moving to peak, without plotting.")
+        go_to_peak(
+            EnergyDefaults.mono_pitch,
+            EnergyDefaults.signals["sig1"],
+            -EnergyDefaults.pitch_scan["halfwidth"],
+            EnergyDefaults.pitch_scan["halfwidth"],
+            steps=EnergyDefaults.pitch_scan["steps"],
+            relative=True,
+            settle=0.01,
+            plot=False,
+            confirm=False,
+        )
+
+
+# Specific functions - need to be edited for each beamline
+
+
+def move_gap_if_needed(energy_ev, move_gap=False):
+    """
+    Move the gap position based on energy if the move_gap flag is set.
+
+    Args:
+        energy_ev (float): The energy value in electron volts (eV) used for gap movement.
+        move_gap (bool): A flag indicating whether the gap position should be moved or not.
+
+    Returns:
+        None
+    """
+    if move_gap:
+        set_gap(energy_ev)
+    else:
+        print("Not moving gap.")
+
+
+# def get_roll(energy_ev):
+#     """Calculates the roll based on calibration data."""
+#     calib = Calibration()
+#     roll = np.poly1d(calib.roll)(energy_ev)
+#     if roll < 0:
+#         return 0
+#     if roll >= 9.95:
+#         return 9.95
+#     return float(roll)
+
+
+def get_dcm_motors_positions(energy_ev):
+    """
+    Pitch and roll are calculated based on fits of the measured calibration values.
+    Perp and Bragg are calculated based on the energy value via calculator.py.
+
+    Arguments:
+        energy_ev (float): The energy value in electron volts for which the
+                           DCM motor positions are to be calculated.
+
+    Returns:
+        dict: A dictionary containing the calculated DCM motor positions
+              including values retrieved from the lookup table, Bragg angle
+              in milliradians, and perpendicular position.
+    """
+    # dcm_motor_values = get_value_from_lut(energy_ev)
+    dcm_motor_values = {}
+    # pitch = float(np.poly1d(Calibration.pitch)(energy_ev))
+    pitch = -5.304
+    roll = EnergyDefaults.mono_roll_value
+    perp = calc_perp_position(energy_ev, print_result=False)
+    bragg_angle = convert_from_energy(energy_ev, print_result=False)["bragg_angle_mrad"]
+    dcm_motor_values.update(
+        {"bragg_angle": bragg_angle, "perp": perp, "dcm_pitch": pitch, "dcm_roll": roll}
+    )
+    return dcm_motor_values
+
+
+def move_dcm_motors(energy_ev):
+    """
+    Move the DCM bragg, pitch and perp motors to the required positions
+    for the given energy in eV.
+
+    """
+    dcm_pos = get_dcm_motors_positions(energy_ev)
+    print(
+        f"Moving DCM bragg: {dcm_pos['bragg_angle']: .5g} mrad, "
+        f"DCM perp: {dcm_pos['perp']: .3g} mm, "
+        f"DCM pitch fixed at -5.304 mrad, "
+        # f"DCM pitch: {dcm_pos['dcm_pitch']: .5g} mrad, "
+        f"DCM Roll: {dcm_pos['dcm_roll']:.5g} V"
+    )
+    # umv(EnergyDefaults.energy, dcm_pos["bragg_angle"])
+    # umv(EnergyDefaults.mono_perp, dcm_pos["perp"])
+    # umv(EnergyDefaults.mono_pitch, dcm_pos["dcm_pitch"])
+    # umv(EnergyDefaults.mono_roll, dcm_pos["dcm_roll"])
+    # print("\n***DCM Roll movement is currently disabled ***\n")
+    scans.umv(
+        EnergyDefaults.energy,
+        dcm_pos["bragg_angle"],
+        EnergyDefaults.mono_perp,
+        dcm_pos["perp"],
+        EnergyDefaults.mono_pitch,
+        dcm_pos["dcm_pitch"],
+        EnergyDefaults.mono_roll,
+        dcm_pos["dcm_roll"],
+        relative=False,
+    )
+
+
+def bl_energy(energy_ev, move_gap=False, mono_scan=True, plot=True):
+    """
+    Adjusts the beamline's energy to the specified energy in electron volts (eV).
+    The function validates the target energy, checks the current energy, and makes
+    adjustments only if the energy difference is significant enough. It performs
+    necessary operations including moving the gap, changing DCM motors, updating
+    the mirror stripe, and scanning to find the optimal DCM pitch.
+
+    Args:
+        energy_ev: Target energy in electron volts to which the beamline should be adjusted.
+        move_gap: Boolean flag indicating whether to adjust the gap before setting energy.
+        plot: Boolean flag indicating whether to plot the DCM pitch scan for finding the peak.
+
+    Returns:
+        None
+    """
+    energy_ev = validate_energy(energy_ev)  # Ensure energy is valid.
+
+    # Check current energy to avoid unnecessary adjustments.
+
+    current_energy = get_current_energy()
+    energy_diff = calculate_energy_difference(current_energy, energy_ev)
+
+    if energy_diff <= EnergyDefaults.min_energy_change:
+        print(f"Energy change of {energy_diff:.2f} eV is too small, not changing energy.")
+        return
+
+    # Step 1: Move the gap if needed.
+    move_gap_if_needed(energy_ev, move_gap)
+
+    # Step 2: Move and set the DCM motors.
+    move_dcm_motors(energy_ev)
+
+    # Step 3: Update the mirror stripe.
+    set_mirror_stripe(energy_ev)
+
+    # Step 4: Perform DCM pitch scan and move to peak.
+    if mono_scan:
+        if plot:
+            mono_pitch_scan(plot=True)
+        else:
+            mono_pitch_scan(plot=False)

pxii_bec/macros/pxii_parameters.py

@@ -0,0 +1,338 @@
+"""File to store beamline parameters and defaults"""
+
+from dataclasses import dataclass
+from typing import Callable
+import numpy as np
+import yaml
+
+
+
+
+@dataclass(frozen=True)
+class EnergyDefaults:
+    """Parameters for PXII energy changes"""
+
+    min_energy_change = 1
+    min_energy_ev = 4800
+    max_energy_ev = 30002
+    beam_offset = 6
+    signals = {"sig1": dev.lu_bpmsum, "sig2": dev.bsc_bpmsum, "sig3": dev.bcu_bpmsum}
+    energy = dev.dcm_bragg
+    mono_pitch = dev.dcm_pitch
+    mono_perp = dev.dcm_perp
+    mono_roll = dev.dcm_froll
+    mono_roll_value = 4.65
+    LUT_table = "luts/energy_lut.csv"
+    stripe_thresholds = {"silicon": 9000, "rhodium": 20000, "platinum": 40000}
+    pitch_scan = {"halfwidth": 0.075, "steps": 20}
+
+
+@dataclass(frozen=True)
+class Calibration:
+    """Calibration parameters for PXII optics"""
+
+    pitch = np.array([4.61823701e-14, -1.97330772e-09, 2.89694543e-05, -5.34468669e00])
+    roll = np.array([2.28291039e-03, -2.41928101e01])
+
+
+@dataclass(frozen=True)
+class Gap:
+    """Fit parameters to calculate gap from harmonics"""
+
+    harmonics = {
+        "H3": np.array([9.15e-04, 4.49e-01]),
+        "H5": np.array([5.19e-04, 7.149e-01]),
+        "H7": np.array([3.57694643e-04, 8.73775476e-01]),
+        "H9": np.array([2.76335714e-04, 8.98471905e-01]),
+        "H11": np.array([2.2225e-04, 9.582e-01]),
+        "H13": np.array([1.9e-4, 9.262e-01]),
+        "H15": np.array([1.67e-4, 8.83e-01]),
+    }
+    # Define harmonic ranges as a constant
+    harmonic_ranges = {
+        "H3": (4900, 7000),
+        "H5": (7000, 10000),
+        "H7": (10000, 13000),
+        "H9": (13000, 16000),
+        "H11": (16000, 19000),
+        "H13": (19000, 22000),
+        "H15": (22000, float("inf")),
+    }
+
+    @staticmethod
+    def get_harmonic_by_energy(energy_ev: float):
+        """
+        Determines the harmonic key based on the provided energy.
+
+        Args:
+            energy_ev (float): The energy value (eV).
+
+        Returns:
+            Optional[str]: The harmonic name (e.g., 'H3', 'H7')
+            if the range matches, None otherwise.
+        """
+        for harmonic, (low, high) in Gap.harmonic_ranges.items():
+            if low < energy_ev <= high:
+                return harmonic
+        return None
+
+    def get_harmonic_values(self, energy_ev: float):
+        """
+        Retrieves the harmonic array based on the energy value.
+
+        Args:
+            energy_ev (float): The energy value (eV).
+
+        Returns:
+            Optional[np.array]: The corresponding array of harmonic values
+            if the range matches, None otherwise.
+        """
+        harmonic = self.get_harmonic_by_energy(energy_ev)
+        return self.harmonics.get(harmonic) if harmonic else None
+
+
+@dataclass(frozen=True)
+class Harmonics:
+    """Anuschka's harmonics data for PXII U19 Undulator"""
+
+    min_gap_value = 4.5
+    map = {
+        3: np.array([1.07141725, -0.52834258]),
+        4: np.array([0.007111, -0.13678409, 1.52295567, -1.06882785]),
+        5: np.array([0.00315051, -0.0766359, 1.13107469, -0.86442062]),
+        6: np.array([0.00162862, -0.04452336, 0.82948259, -0.37329674]),
+        7: np.array([0.00080764, -0.02418882, 0.59212947, 0.15702886]),
+        8: np.array([1.16699242e-03, -4.68207543e-02, 9.43972396e-01, -1.94201019e00]),
+        9: np.array([0.00048419, -0.02010648, 0.55514114, -0.38180067]),
+        10: np.array([2.11583333e-04, -8.03503571e-03, 3.43141595e-01, 6.02318000e-01]),
+        11: np.array([0.00334392, -0.18405336, 3.58338994, -19.3640241]),
+        12: np.array([3.20520798e-05, 2.39253145e-03, 8.09198503e-02, 2.22897377e00]),
+        13: np.array([0.00278744, 0.07979874, 2.05143916]),
+    }
+    energy_ranges = {3: (0, 7), 5: (7, 10), 7: (10, 13), 9: (13, 16), 11: (16, 19), 13: (19, 22)}
+    high_energy = [(15, (23, 25)), (17, (25, 29)), (19, (29, float("inf")))]
+
+
+@dataclass(frozen=True)
+class CamConversion:
+    """Convert pixels to microns for sam cam"""
+
+    a = 0.5208
+    b = 0.002586
+
+
+@dataclass(frozen=True)
+class BPMScans:
+    """Define the names of the motors and bpm channels"""
+
+    fe = {
+        "x_name": dev.fe_bpm_x.name,
+        "y_name": dev.fe_bpm_y.name,
+        "z1_name": dev.fe_bpm1.name,
+        "z2_name": dev.fe_bpm2.name,
+        "z3_name": dev.fe_bpm3.name,
+        "z4_name": dev.fe_bpm3.name,
+        "z5_name": dev.fe_bpmsum.name,
+        "x_device": dev.fe_bpm_x,
+        "y_device": dev.fe_bpm_y,
+    }
+    lu = {
+        "x_name": dev.lu_bpm_x.name,
+        "y_name": dev.lu_bpm_y.name,
+        "z1_name": dev.lu_bpm1.name,
+        "z2_name": dev.lu_bpm2.name,
+        "z3_name": dev.lu_bpm3.name,
+        "z4_name": dev.lu_bpm4.name,
+        "z5_name": dev.lu_bpmsum.name,
+        "x_device": dev.lu_bpm_x,
+        "y_device": dev.lu_bpm_y,
+    }
+    bsc = {
+        "x_name": dev.bsc_bpm_x.name,
+        "y_name": dev.bsc_bpm_y.name,
+        "z1_name": dev.bsc_bpm1.name,
+        "z2_name": dev.bsc_bpm2.name,
+        "z3_name": dev.bsc_bpm3.name,
+        "z4_name": dev.bsc_bpm4.name,
+        "z5_name": dev.bsc_bpmsum.name,
+        "x_device": dev.bsc_bpm_x,
+        "y_device": dev.bsc_bpm_y,
+    }
+    bcu = {
+        "x_name": dev.bcu_bpm_x.name,
+        "y_name": dev.bcu_bpm_y.name,
+        "z1_name": dev.bcu_bpm1.name,
+        "z2_name": dev.bcu_bpm2.name,
+        "z3_name": dev.bcu_bpm3.name,
+        "z4_name": dev.bcu_bpm4.name,
+        "z5_name": dev.bcu_bpmsum.name,
+        "x_device": dev.bcu_bpm_x,
+        "y_device": dev.bcu_bpm_y,
+    }
+
+
+@dataclass(frozen=True)
+class MirrorConfig:
+    """Define the names of the mirror channels"""
+
+    hfm = {
+        "bu_name": dev.hfm_bu.name,
+        "bd_name": dev.hfm_bd.name,
+        "z_name": dev.samcam_xsig.name,
+        "x_device": dev.hfm_bu,
+        "y_device": dev.hfm_bd,
+    }
+    vfm = {
+        "bu_name": dev.vfm_bu.name,
+        "bd_name": dev.vfm_bd.name,
+        "z_name": dev.samcam_ysig.name,
+        "x_device": dev.vfm_bu,
+        "y_device": dev.vfm_bd,
+    }
+
+
+@dataclass
+class PositionedDevice:
+    """Class for devices with defined in and out positions"""
+
+    device_name: str
+    type: str
+    name: str
+    inpos: float
+    outpos: float
+    tol: float
+    mot: str
+    reader: Callable[[], float]
+
+    @property
+    def actual(self):
+        """Returns current motor position"""
+        return self.reader()
+
+    def checkin(self):
+        """Returns True if motor in in the 'in' position"""
+        return abs(self.actual - self.inpos) <= self.tol
+
+    def mvin(self):
+        """Moves motor to the 'in' position"""
+        scans.umv(self.mot, self.inpos, relative=False)
+
+    def mvout(self):
+        """Moves motor to the 'out' position"""
+        scans.umv(self.mot, self.outpos, relative=False)
+    
+    def status(self):
+        """ Check if device is in or out or moving"""
+        positions = ("in", "out", "moving", "undefined")
+        target_in = self.inpos
+        target_out = self.outpos
+        actual = self.actual
+        delta_in = actual - target_in
+        delta_out = actual - target_out
+        # Check if motor is moving
+        if "Signal" in self.type:
+            moving = 0
+        elif "Motor" in self.type:
+            d = getattr(dev, self.device_name)
+            moving = d.motor_is_moving.get()
+            if moving:
+                pos = positions[2]
+                return {"position": pos.upper(),
+                        "name": self.name,
+                        "moving": moving}
+
+        if abs(delta_in) > self.tol and abs(delta_out) > self.tol:
+            pos = positions[3]
+            return {"position": pos.upper(),
+                    "name": self.name,
+                    "actual": actual,
+                    "moving": moving}
+ 
+        elif abs(delta_in) <= self.tol:
+            target = self.inpos
+            pos = positions[0]
+            delta = delta_in
+        elif abs(delta_out) <= self.tol:
+            target = self.outpos
+            pos = positions[1]
+            delta = delta_out
+
+        return {
+            "name": self.name,
+            "position": pos.upper(),
+            "target": target,
+            "actual": actual,
+            "delta": delta,
+            "tol": self.tol,
+            "moving": moving,
+        }
+    
+    def report(self):
+        """ Print status of motor """
+        s = self.status()
+
+        if s['position'] == "UNDEFINED":
+            return (f"{s['name']:15s}: "
+                f"{s['position']} "
+                f"position {s['actual']:.3f}")
+        elif s['position'] == "MOVING":
+                return (f"{s['name']:15s}: "
+                f"{s['position']} ") 
+        else:
+            return (
+                f"{s['name']:15s}: "
+                f"[{s['position']}] "
+                f"actual = {s['actual']:.3f} "
+                f"target = {s['target']:.3f} "
+                f"delta = {s['delta']:.3f}"
+            )
+
+
+
+@dataclass(frozen=True)
+class PD:
+    """Class for positioned device positions"""
+
+
+def build_pd(yaml_file):
+    """Takes the in and out values from the yaml file
+    and adds them to the PD class
+    """
+    with open(yaml_file, encoding="utf-8") as f:
+        data = yaml.safe_load(f)
+    for device_name, cfg in data.items():
+        # Skip devices without userParameter
+        user = cfg.get("userParameter")
+        if not user:
+            continue
+        # Set tolerance
+        if "tol" not in user:
+            user["tol"] = 0.01
+
+        try:
+            dev_obj = getattr(dev, device_name)
+        except:
+            raise KeyError(f"Device {device_name} not found in device list")
+        desc = cfg.get("description")
+        type = cfg.get("deviceClass")
+        target = PositionedDevice(
+            device_name=device_name,
+            type = type,
+            name=desc,
+            inpos=user["in"],
+            outpos=user["out"],
+            tol=user["tol"],
+            mot=dev_obj,
+            reader=lambda d=dev_obj, n=device_name: d.read()[n]["value"],
+        )
+
+        setattr(PD, device_name, target)
+
+
+def init_positioned_devices():
+    """Initialises the positioned devices"""
+    file = (
+        "/sls/x10sa/config/bec/production/pxii_bec/pxii_bec/device_configs/pxii-autogenerated.yaml"
+    )
+    build_pd(file)
+    print("Defined positions for devices have been updated from pxii-autogenerated.yaml")

pyproject.toml

@@ -5,8 +5,8 @@ build-backend = "hatchling.build"
 [project]
 name = "pxii_bec"
 version = "0.0.0"
-description = "The PX-II plugin repository for BEC"
-requires-python = ">=3.10"
+description = "A plugin repository for BEC"
+requires-python = ">=3.11"
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Programming Language :: Python :: 3",
@@ -25,6 +25,7 @@ dev = [
     "pytest-random-order",
     "ophyd_devices",
     "bec_server",
+    "requests-mock",
 ]
 
 [project.entry-points."bec"]

tests/tests_devices/mock_smargopolo.py

@@ -0,0 +1,110 @@
+"""A mock smargopolo REST interface with mock motoers, for testing devices against"""
+
+import asyncio
+import random
+import time
+from contextlib import asynccontextmanager
+from typing import Iterable
+
+import uvicorn
+from fastapi import FastAPI, HTTPException, Query, Request
+from pydantic import BaseModel
+
+AXES = ["SHX", "SHY", "SHZ", "PHI", "CHI"]
+
+
+class Motor:
+    def __init__(self, velocity: float = 1.0):
+        self.position = 0.0
+        self.target = 0.0
+        self.velocity = velocity
+        self.moving = False
+        self._last_update = time.monotonic()
+
+    def update(self):
+        now = time.monotonic()
+        dt = now - self._last_update
+        self._last_update = now
+
+        if not self.moving:
+            return
+
+        jitter_factor = random.random() * 0.05 - 0.025  # +- 2.5% jitter in step
+        distance = self.target - self.position
+        direction = 1 if distance > 0 else -1
+        step = direction * self.velocity * dt
+
+        if abs(step) >= abs(distance):
+            self.position = self.target + (step * jitter_factor)
+            self.moving = False
+        else:
+            self.position += step * (1 + jitter_factor)
+
+
+motors: dict[str, Motor] = {
+    "SHX": Motor(velocity=3),
+    "SHY": Motor(velocity=2.5),
+    "SHZ": Motor(velocity=2),
+    "PHI": Motor(velocity=1.0),
+    "CHI": Motor(velocity=0.7),
+}
+
+
+class MoveRequest(BaseModel):
+    target: float
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    async def updater():
+        while True:
+            for motor in motors.values():
+                motor.update()
+            await asyncio.sleep(0.02)  # 50 Hz update loop
+
+    task = asyncio.create_task(updater())
+    yield
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+def validate_axes(axes: Iterable[str] | None) -> list[str]:
+    if axes is None:
+        return AXES
+    for a in axes:
+        if a not in AXES:
+            raise HTTPException(status_code=404, detail=f"Unknown axis: {a}")
+    return list(axes)
+
+
+@app.get("/readbackSCS")
+async def readback_scs(axis: list[str] | None = Query(None)):
+    selected_axes = validate_axes(axis)
+    return {ax: motors[ax].position for ax in selected_axes}
+
+
+@app.put("/targetSCS")
+async def target_scs(req: Request):
+    targets = {ax: float(t) for ax, t in req.query_params.items()}
+    if targets is None:
+        return {}
+
+    selected_axes = validate_axes(targets.keys())
+
+    for a in selected_axes:
+        motor = motors[a]
+        motor.update()
+        motor.target = targets[a]
+        motor.moving = True
+
+    return {"targets": targets, "message": "Move started"}
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000, reload=False)

tests/tests_devices/test_smargon.py

@@ -0,0 +1,50 @@
+from copy import copy
+from threading import RLock
+from unittest.mock import ANY, MagicMock, patch
+
+import pytest
+
+
+class MockServer:
+    def __init__(self) -> None:
+        self.lock = RLock()
+        self.mock_data = {"SHX": 1.0, "SHY": 1.0, "SHZ": 1.0, "PHI": 1.0, "CHI": 1.0}
+
+    def get(self, endpoint):
+        with self.lock:
+            return copy(self.mock_data)
+
+    def put(self, val: dict[str, float]):
+        with self.lock:
+            self.mock_data.update(val)
+
+
+@pytest.fixture
+def smargon():
+    mock_server = MockServer()
+    from pxii_bec.devices.smargopolo_smargon import Smargon
+
+    s = Smargon(name="smargon", prefix="http://test-smargopolo.psi.ch")
+    s.controller._rest_get = mock_server.get
+    s.controller._rest_put = mock_server.put
+    yield s
+    s.controller._stop_monitor_readback_event.set()
+
+
+class TestSmargon:
+    def test_smargon_read(self, smargon):
+        smargon.wait_for_connection()
+        reading = smargon.read()
+        assert dict(reading) == {
+            "smargon_x": {"value": 1.0, "timestamp": ANY},
+            "smargon_y": {"value": 1.0, "timestamp": ANY},
+            "smargon_z": {"value": 1.0, "timestamp": ANY},
+            "smargon_phi": {"value": 1.0, "timestamp": ANY},
+            "smargon_chi": {"value": 1.0, "timestamp": ANY},
+        }
+
+    def test_smargon_set_with_status(self, smargon):
+        smargon.wait_for_connection()
+        st = smargon.x.set(5.0)
+        st.wait(timeout=1)
+        assert smargon.x.get() == 5.0