Update repo with template version v1.2.8

.copier-answers.yml

@@ -2,7 +2,7 @@
 # It is needed to track the repo template version, and editing may break things.
 # This file will be overwritten by copier on template updates.
 
-_commit: v1.2.2
+_commit: v1.2.8
 _src_path: https://github.com/bec-project/plugin_copier_template.git
 make_commit: false
 project_name: debye_bec

.gitea/workflows/ci.yml

@@ -1,4 +1,4 @@
-name: CI for Debye_bec
+name: CI for debye_bec
 on:
   push:
   pull_request:
@@ -28,7 +28,7 @@ on:
         description: "Python version to use"
         required: false
         type: string
-        default: "3.11"
+        default: "3.12"
 
 permissions:
   pull-requests: write
@@ -39,36 +39,64 @@ jobs:
     env:
       QTWEBENGINE_DISABLE_SANDBOX: 1
       QT_QPA_PLATFORM: "offscreen"
+
     steps:
-      - name: Checkout shared action repository
+      - 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/bec_shared_actions
-          ref: main
-          path: bec_shared_actions
+          repository: bec/debye_bec
+          ref: "${{ inputs.BEC_PLUGIN_REPO_BRANCH || github.head_ref || github.sha }}"
+          path: ./debye_bec
 
-      - name: Checkout Plugin and BEC Repositories
-        uses: bec/bec_shared_actions/checkout_repositories@main
-        # uses: ./bec_shared_actions
+      - 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:
-          BEC_WIDGETS_BRANCH:   "${{ github.event.inputs.BEC_WIDGETS_BRANCH || 'main' }}"
-          BEC_CORE_BRANCH:      "${{ github.event.inputs.BEC_CORE_BRANCH || 'main' }}"
-          OPHYD_DEVICES_BRANCH: "${{ github.event.inputs.OPHYD_DEVICES_BRANCH || 'main' }}"
-          BEC_PLUGIN_REPO_BRANCH: "${{ github.event_name == 'workflow_dispatch' && (github.event.inputs.BEC_PLUGIN_REPO_BRANCH || 'main') || github.head_ref || github.ref_name }}"
-          PLUGIN_REPO_NAME: "${{ github.event.repository.name }}"
-          
-      - name: Install external APT packages
-        uses: bec/bec_shared_actions/install_apt_packages@main
+          repository: bec/bec
+          ref: "${{ inputs.BEC_CORE_BRANCH || 'main' }}"
+          path: ./bec
+
+      - name: Checkout Ophyd Devices
+        uses: actions/checkout@v4
         with:
-          APT_PACKAGES: "libgl1 libegl1 x11-utils libxkbcommon-x11-0 libdbus-1-3 xvfb libnss3 libxdamage1 libasound2t64 libatomic1 libxcursor1"
+          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
-        uses: bec/bec_shared_actions/install_python_env@main
-        with:
-          PYTHON_VERSION: "${{ github.event.inputs.PYTHON_VERSION || '3.11' }}"
-          PLUGIN_REPO_NAME: "${{ github.event.repository.name }}"
+        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 ./debye_bec
 
-      - name: Run pytest for Plugin
-        uses: bec/bec_shared_actions/run_pytest_for_plugin@main
-        with:
-          PLUGIN_REPO_NAME: "${{ github.event.repository.name }}"
+      - name: Run Pytest with Coverage
+        id: coverage
+        run: pytest --random-order --cov=./debye_bec --cov-config=./debye_bec/pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail ./debye_bec/tests/ || test $? -eq 5

.gitea/workflows/create_update_pr.yml

@@ -0,0 +1,62 @@
+name: Create template upgrade PR for debye_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: debye_bec
-    target: debye_bec
-    branch: $CHILD_PIPELINE_BRANCH
-  project: bec/awi_utils

debye_bec/devices/ionization_chambers/ionization_chamber.py

@@ -9,8 +9,7 @@ from ophyd import Component as Cpt
 from ophyd import Device
 from ophyd import DynamicDeviceComponent as Dcpt
 from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
-from ophyd.status import DeviceStatus, SubscriptionStatus
-from ophyd_devices import CompareStatus, TransitionStatus
+from ophyd_devices import CompareStatus, DeviceStatus, SubscriptionStatus, TransitionStatus
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 from typeguard import typechecked
 

debye_bec/devices/mo1_bragg/mo1_bragg.py

@@ -9,16 +9,15 @@ used to ensure that the action is executed completely. This is believed
 to allow for a more stable execution of the action."""
 
 import time
-from typing import Any, Literal
+from typing import Literal
 
 from bec_lib.devicemanager import ScanInfo
 from bec_lib.logger import bec_logger
 from ophyd import Component as Cpt
-from ophyd import DeviceStatus, Signal, StatusBase
-from ophyd.status import SubscriptionStatus, WaitTimeoutError
+from ophyd import DeviceStatus, StatusBase
+from ophyd.status import WaitTimeoutError
 from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
-from ophyd_devices.utils.errors import DeviceStopError
 from pydantic import BaseModel, Field
 from typeguard import typechecked
 
@@ -281,7 +280,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
             self.scan_control.scan_status,
             transitions=[ScanControlScanStatus.READY, ScanControlScanStatus.RUNNING],
             strict=True,
-            raise_states=[ScanControlScanStatus.PARAMETER_WRONG],
+            failure_states=[ScanControlScanStatus.PARAMETER_WRONG],
         )
         self.cancel_on_stop(status)
         start_func(1)

debye_bec/devices/nidaq/nidaq.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
-import time
-from typing import TYPE_CHECKING, Literal, cast
+from typing import TYPE_CHECKING, Literal
 
 from bec_lib.logger import bec_logger
 from ophyd import Component as Cpt
 from ophyd import Device, DeviceStatus, EpicsSignal, EpicsSignalRO, Kind, StatusBase
-from ophyd.status import SubscriptionStatus, WaitTimeoutError
+from ophyd.status import WaitTimeoutError
 from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 from ophyd_devices.sim.sim_signals import SetableSignal

debye_bec/devices/pilatus/pilatus.py

@@ -17,12 +17,10 @@ from ophyd.areadetector.cam import ADBase, PilatusDetectorCam
 from ophyd.areadetector.plugins import HDF5Plugin_V22 as HDF5Plugin
 from ophyd.areadetector.plugins import ImagePlugin_V22 as ImagePlugin
 from ophyd.status import WaitTimeoutError
-from ophyd_devices import CompareStatus, DeviceStatus, FileEventSignal, PreviewSignal
+from ophyd_devices import AndStatus, CompareStatus, DeviceStatus, FileEventSignal, PreviewSignal
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 from pydantic import BaseModel, Field
 
-from debye_bec.devices.pilatus.utils import AndStatusWithList
-
 if TYPE_CHECKING:  # pragma: no cover
     from bec_lib.devicemanager import ScanInfo
     from bec_lib.messages import DevicePreviewMessage, ScanStatusMessage
@@ -360,14 +358,12 @@ class Pilatus(PSIDeviceBase, ADBase):
     #             f"Live Mode on detector {self.name} did not stop: {content} after 10s."
     #         )
 
-    def check_detector_stop_running_acquisition(self) -> AndStatusWithList:
+    def check_detector_stop_running_acquisition(self) -> AndStatus:
         """Check if the detector is still running an acquisition."""
         status_acquire = CompareStatus(self.cam.acquire, ACQUIREMODE.DONE.value)
         status_writing = CompareStatus(self.hdf.capture, ACQUIREMODE.DONE.value)
         status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.UNARMED.value)
-        status = AndStatusWithList(
-            device=self, status_list=[status_acquire, status_writing, status_cam_server]
-        )
+        status = status_acquire & status_writing & status_cam_server
         return status
 
     def _calculate_trigger(self, scan_msg: ScanStatusMessage):
@@ -562,9 +558,7 @@ class Pilatus(PSIDeviceBase, ADBase):
             status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.ACQUIRING.value)
             status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.ACQUIRING.value)
             status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.ARMED.value)
-            status = AndStatusWithList(
-                device=self, status_list=[status_hdf, status_cam, status_cam_server]
-            )
+            status = status_hdf & status_cam & status_cam_server
             self.cam.acquire.put(1)
             self.hdf.capture.put(1)
             self.cancel_on_stop(status)
@@ -593,15 +587,15 @@ class Pilatus(PSIDeviceBase, ADBase):
             scan_msg.scan_name in self.xas_xrd_scan_names or scan_msg.scan_type == "step"
         ):  # TODO how to deal with fly scans?
             if status.success:
-                status.device.file_event.put(
-                    file_path=status.device._full_path,  # pylint: disable:protected-access
+                self.file_event.put(
+                    file_path=self._full_path,
                     done=True,
                     successful=True,
                     hinted_h5_entries={"data": "/entry/data/data"},
                 )
             else:
-                status.device.file_event.put(
-                    file_path=status.device._full_path,  # pylint: disable:protected-access
+                self.file_event.put(
+                    file_path=self._full_path,
                     done=True,
                     successful=False,
                     hinted_h5_entries={"data": "/entry/data/data"},
@@ -622,15 +616,12 @@ class Pilatus(PSIDeviceBase, ADBase):
                 # For long scans, it can be that the mono will execute one cycle more,
                 # meaning a few more XRD triggers will be sent
                 status_img_written = CompareStatus(
-                    self.hdf.num_captured, self.n_images, operation=">="
+                    self.hdf.num_captured, self.n_images, operation_success=">="
                 )
             else:
                 status_img_written = CompareStatus(self.hdf.num_captured, self.n_images)
             status_img_written = CompareStatus(self.hdf.num_captured, self.n_images)
-            status = AndStatusWithList(
-                device=self,
-                status_list=[status_hdf, status_cam, status_img_written, status_cam_server],
-            )
+            status = status_hdf & status_cam & status_img_written & status_cam_server
             status.add_callback(self._complete_callback)  # Callback that writing was successful
             self.cancel_on_stop(status)
             return status

debye_bec/devices/pilatus/utils.py

@@ -1,243 +0,0 @@
-"""Temporary utility module for Status Object implementations."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from ophyd import Device, DeviceStatus, StatusBase
-
-
-class AndStatusWithList(DeviceStatus):
-    """
-    Custom implementation of the AndStatus that combines the
-    option to add multiple statuses as a list, and in addition
-    allows for adding the Device as an object to access its
-    methods.
-
-    Args"""
-
-    def __init__(
-        self,
-        device: Device,
-        status_list: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus],
-        **kwargs,
-    ):
-        self.all_statuses = status_list if isinstance(status_list, list) else [status_list]
-        super().__init__(device=device, **kwargs)
-        self._trace_attributes["all"] = [st._trace_attributes for st in self.all_statuses]
-
-        def inner(status):
-            with self._lock:
-                if self._externally_initiated_completion:
-                    return
-                if self.done:  # Return if status is already done.. It must be resolved already
-                    return
-
-                for st in self.all_statuses:
-                    with st._lock:
-                        if st.done and not st.success:
-                            self.set_exception(st.exception())  # st._exception
-                            return
-
-                if all(st.done for st in self.all_statuses) and all(
-                    st.success for st in self.all_statuses
-                ):
-                    self.set_finished()
-
-        for st in self.all_statuses:
-            with st._lock:
-                st.add_callback(inner)
-
-    # TODO improve __repr__ and __str__
-    def __repr__(self):
-        return "<AndStatusWithList({self.all_statuses!r})>".format(self=self)
-
-    def __str__(self):
-        return "<AndStatusWithList(done={self.done}, success={self.success})>".format(self=self)
-
-    def __contains__(self, status: StatusBase | DeviceStatus) -> bool:
-        for child in self.all_statuses:
-            if child == status:
-                return True
-            if isinstance(child, AndStatusWithList):
-                if status in child:
-                    return True
-
-        return False
-
-    # # TODO Check if this actually works....
-    # def set_exception(self, exc):
-    #     # Propagate the exception to all sub-statuses that are not done yet.
-    #
-    # with self._lock:
-    #     if self._externally_initiated_completion:
-    #         return
-    #     if self.done:  # Return if status is already done.. It must be resolved already
-    #         return
-    #     super().set_exception(exc)
-    #     for st in self.all_statuses:
-    #         with st._lock:
-    #             if not st.done:
-    #                 st.set_exception(exc)
-
-    def _run_callbacks(self):
-        """
-        Set the Event and run the callbacks.
-        """
-        if self.timeout is None:
-            timeout = None
-        else:
-            timeout = self.timeout + self.settle_time
-        if not self._settled_event.wait(timeout):
-            self.log.warning("%r has timed out", self)
-            with self._externally_initiated_completion_lock:
-                if self._exception is None:
-                    exc = TimeoutError(
-                        f"AndStatus from device {self.device.name} failed to complete in specified timeout of {self.timeout + self.settle_time}."
-                    )
-                    self._exception = exc
-        # Mark this as "settled".
-        try:
-            self._settled()
-        except Exception:
-            self.log.exception("%r encountered error during _settled()", self)
-        with self._lock:
-            self._event.set()
-        if self._exception is not None:
-            try:
-                self._handle_failure()
-            except Exception:
-                self.log.exception("%r encountered an error during _handle_failure()", self)
-        for cb in self._callbacks:
-            try:
-                cb(self)
-            except Exception:
-                self.log.exception(
-                    "An error was raised on a background thread while "
-                    "running the callback %r(%r).",
-                    cb,
-                    self,
-                )
-        self._callbacks.clear()
-
-
-class AndStatus(StatusBase):
-    """Custom AndStatus for TimePix detector."""
-
-    def __init__(
-        self,
-        left: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus] | None,
-        name: str | Device | None = None,
-        right: StatusBase | DeviceStatus | list[StatusBase | DeviceStatus] | None = None,
-        **kwargs,
-    ):
-        self.left = left if isinstance(left, list) else [left]
-        if right is not None:
-            self.right = right if isinstance(right, list) else [right]
-        else:
-            self.right = []
-        self.all_statuses = self.left + self.right
-        if name is None:
-            name = "unname_status"
-        elif isinstance(name, Device):
-            name = name.name
-        else:
-            name = name
-        self.name = name
-        super().__init__(**kwargs)
-        self._trace_attributes["left"] = [st._trace_attributes for st in self.left]
-        self._trace_attributes["right"] = [st._trace_attributes for st in self.right]
-
-        def inner(status):
-            with self._lock:
-                if self._externally_initiated_completion:
-                    return
-                if self.done:  # Return if status is already done.. It must be resolved already
-                    return
-
-                for st in self.all_statuses:
-                    with st._lock:
-                        if st.done and not st.success:
-                            self.set_exception(st.exception())  # st._exception
-                            return
-
-                if all(st.done for st in self.all_statuses) and all(
-                    st.success for st in self.all_statuses
-                ):
-                    self.set_finished()
-
-        for st in self.all_statuses:
-            with st._lock:
-                st.add_callback(inner)
-
-    def __repr__(self):
-        return "({self.left!r} & {self.right!r})".format(self=self)
-
-    def __str__(self):
-        return "{0}(done={1.done}, " "success={1.success})" "".format(self.__class__.__name__, self)
-
-    def __contains__(self, status: StatusBase) -> bool:
-        for child in [self.left, self.right]:
-            if child == status:
-                return True
-            if isinstance(child, AndStatus):
-                if status in child:
-                    return True
-
-        return False
-
-    def _run_callbacks(self):
-        """
-        Set the Event and run the callbacks.
-        """
-        if self.timeout is None:
-            timeout = None
-        else:
-            timeout = self.timeout + self.settle_time
-        if not self._settled_event.wait(timeout):
-            # We have timed out. It's possible that set_finished() has already
-            # been called but we got here before the settle_time timer expired.
-            # And it's possible that in this space be between the above
-            # statement timing out grabbing the lock just below,
-            # set_exception(exc) has been called. Both of these possibilties
-            # are accounted for.
-            self.log.warning("%r has timed out", self)
-            with self._externally_initiated_completion_lock:
-                # Set the exception and mark the Status as done, unless
-                # set_exception(exc) was called externally before we grabbed
-                # the lock.
-                if self._exception is None:
-                    exc = TimeoutError(
-                        f"Status with name {self.name} failed to complete in specified timeout of {self.timeout + self.settle_time}."
-                    )
-                    self._exception = exc
-        # Mark this as "settled".
-        try:
-            self._settled()
-        except Exception:
-            # No alternative but to log this. We can't supersede set_exception,
-            # and we have to continue and run the callbacks.
-            self.log.exception("%r encountered error during _settled()", self)
-        # Now we know whether or not we have succeed or failed, either by
-        # timeout above or by set_exception(exc), so we can set the Event that
-        # will mark this Status as done.
-        with self._lock:
-            self._event.set()
-        if self._exception is not None:
-            try:
-                self._handle_failure()
-            except Exception:
-                self.log.exception("%r encountered an error during _handle_failure()", self)
-        # The callbacks have access to self, from which they can distinguish
-        # success or failure.
-        for cb in self._callbacks:
-            try:
-                cb(self)
-            except Exception:
-                self.log.exception(
-                    "An error was raised on a background thread while "
-                    "running the callback %r(%r).",
-                    cb,
-                    self,
-                )
-        self._callbacks.clear()

debye_bec/devices/pilatus_curtain.py

@@ -10,8 +10,6 @@ from ophyd import EpicsSignal, EpicsSignalRO
 from ophyd_devices import CompareStatus, DeviceStatus
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 
-from debye_bec.devices.pilatus.utils import AndStatusWithList
-
 if TYPE_CHECKING:
     from bec_lib.devicemanager import ScanInfo
 
@@ -83,8 +81,8 @@ class PilatusCurtain(PSIDeviceBase):
             self.open_cover.put(1)
             # TODO timeout ok?
             status_open = CompareStatus(self.cover_is_open, COVER.OPEN, timeout=5)
-            status_error = CompareStatus(self.cover_error, COVER.ERROR, operation="!=")
-            status = AndStatusWithList(device=self, status_list=[status_open, status_error])
+            status_error = CompareStatus(self.cover_error, COVER.ERROR, operation_success="!=")
+            status = status_open & status_error
             return status
         else:
             return None
@@ -95,8 +93,8 @@ class PilatusCurtain(PSIDeviceBase):
             self.close_cover.put(1)
             # TODO timeout ok?
             status_close = CompareStatus(self.cover_is_closed, COVER.CLOSED, timeout=5)
-            status_error = CompareStatus(self.cover_error, COVER.ERROR, operation="!=")
-            status = AndStatusWithList(device=self, status_list=[status_close, status_error])
+            status_error = CompareStatus(self.cover_error, COVER.ERROR, operation_success="!=")
+            status = status_close & status_error
             return status
         else:
             return None

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "hatchling.build"
 name = "debye_bec"
 version = "0.0.0"
 description = "A plugin repository for BEC"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Programming Language :: Python :: 3",

tests/tests_devices/test_pilatus.py

@@ -1,10 +1,8 @@
 # pylint: skip-file
-import os
 import threading
 from typing import TYPE_CHECKING, Generator
 from unittest import mock
 
-import numpy as np
 import ophyd
 import pytest
 from bec_lib.messages import ScanStatusMessage
@@ -177,7 +175,7 @@ def test_pilatus_on_trigger_cancel_on_stop(pilatus):
         status.wait(timeout=5)
 
 
-def test_pilatus_on_complete(pilatus):
+def test_pilatus_on_complete(pilatus: Pilatus):
     """Test the on_complete logic of the Pilatus detector."""
 
     if pilatus.scan_info.msg.scan_name.startswith("xas"):