From abed6a7de6e90e6a3ca4d4228e65cd7af76507ff Mon Sep 17 00:00:00 2001 From: David Perl Date: Tue, 3 Nov 2026 10:42:13 +0100 Subject: [PATCH] fix: add mx_bec dep --- .gitea/workflows/ci.yml | 34 +++++++- pyproject.toml | 2 +- tests/tests_devices/mock_smargopolo.py | 110 ------------------------- tests/tests_devices/test_aerotech.py | 72 ++++++++-------- tests/tests_devices/test_smargon.py | 57 +++++++------ 5 files changed, 105 insertions(+), 170 deletions(-) delete mode 100644 tests/tests_devices/mock_smargopolo.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 4cf457f..508accf 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -24,6 +24,11 @@ on: required: false type: string default: "main" + MX_BEC_REPO_BRANCH: + description: "Branch of the mx_bec core repo to install" + required: false + type: string + default: "main" PYTHON_VERSION: description: "Python version to use" required: false @@ -46,6 +51,29 @@ jobs: with: python-version: "${{ inputs.PYTHON_VERSION || '3.12' }}" + # Use the same-named branch of mx_bec if it exists (e.g. a feature developed in both repos), otherwise main. + - name: Resolve mx_bec branch + id: mx_bec + shell: bash + run: | + ref="${{ inputs.MX_BEC_REPO_BRANCH || github.head_ref || github.ref_name }}" + auth="$(printf 'x-access-token:%s' '${{ github.token }}' | base64 -w0)" + if git -c "http.extraheader=Authorization: Basic ${auth}" ls-remote --exit-code --heads \ + "${{ github.server_url }}/bec/mx_bec.git" "refs/heads/${ref}" >/dev/null; then + echo "Using mx_bec branch '${ref}'" + else + echo "mx_bec has no branch '${ref}', using main" + ref=main + fi + echo "ref=${ref}" >> "$GITHUB_OUTPUT" + + - name: Checkout MX BEC Repository + uses: actions/checkout@v4 + with: + repository: bec/mx_bec + ref: "${{ steps.mx_bec.outputs.ref }}" + path: ./mx_bec + - name: Checkout BEC Plugin Repository uses: actions/checkout@v4 with: @@ -94,8 +122,10 @@ jobs: 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[dev] + uv pip install --system -e ./bec_widgets[dev] + # MX-internal dependencies (aarecommon, aare_config, jfjoch_client, ...) come from gitea + uv pip install --system --extra-index-url https://gitea.psi.ch/api/packages/mx/pypi/simple/ -e ./mx_bec[dev] + uv pip install --system --extra-index-url https://gitea.psi.ch/api/packages/mx/pypi/simple/ -e ./pxii_bec[dev] - name: Run Tests with Coverage id: coverage diff --git a/pyproject.toml b/pyproject.toml index eaec104..b5d968d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ classifiers = [ "Programming Language :: Python :: 3", "Topic :: Scientific/Engineering", ] -dependencies = ["pandas~=2.0", "matplotlib"] +dependencies = ["mx_bec", "pandas~=2.0", "matplotlib"] [project.optional-dependencies] dev = [ diff --git a/tests/tests_devices/mock_smargopolo.py b/tests/tests_devices/mock_smargopolo.py deleted file mode 100644 index 466063e..0000000 --- a/tests/tests_devices/mock_smargopolo.py +++ /dev/null @@ -1,110 +0,0 @@ -"""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) diff --git a/tests/tests_devices/test_aerotech.py b/tests/tests_devices/test_aerotech.py index 3cf76f5..90475eb 100644 --- a/tests/tests_devices/test_aerotech.py +++ b/tests/tests_devices/test_aerotech.py @@ -1,58 +1,64 @@ +"""Smoke tests of the Aerotech device as re-exported by pxii_bec; the device itself is tested in +mx_bec.""" + from copy import copy from threading import RLock from unittest.mock import ANY import pytest +from mx_bec.devices.aerotech import Aerotech as MxAerotech + +from pxii_bec.devices import Aerotech + +AXES = ("x", "y", "z", "u") class MockServer: + """Just enough of the AareScan REST API for reading and moving axes.""" + def __init__(self) -> None: self.lock = RLock() - self.mock_data = { - "x": {"pos": 1.0}, - "y": {"pos": 1.0}, - "z": {"pos": 1.0}, - "u": {"pos": 1.0}, - "vel_u_deg_s": {"pos": 1.0}, - } + self.mock_data = {axis: {"pos": 1.0} for axis in AXES} def get(self, endpoint): with self.lock: return copy(self.mock_data) - def put(self, params: dict | None = None, body: dict | None = None): + def post(self, params: dict | None = None, body: dict | None = None, endpoint=None): with self.lock: assert body is not None - for k, v in body.items(): - self.mock_data[k]["pos"] = v + for axis, value in body.items(): + if axis in self.mock_data: # the body also carries the velocity parameter + self.mock_data[axis]["pos"] = value @pytest.fixture def aerotech(): mock_server = MockServer() - from pxii_bec.devices import Aerotech - - s = Aerotech(name="aerotech", prefix="http://test-aerotech.psi.ch") - s.controller._rest_get = mock_server.get - s.controller._rest_post = mock_server.put - yield s - s.controller._stop_monitor_readback_event.set() + device = Aerotech(name="aerotech", prefix="http://test-aerotech.psi.ch") + device.controller._rest_get = mock_server.get + device.controller._rest_post = mock_server.post + device.controller.validate = lambda endpoint, body: (True, "") + yield device + device.controller._stop_monitor_readback_event.set() + device.destroy() -class TestAerotech: - def test_aerotech_read(self, aerotech): - aerotech.wait_for_connection() - reading = aerotech.read() - assert dict(reading) == { - "aerotech_x": {"value": 1.0, "timestamp": ANY}, - "aerotech_y": {"value": 1.0, "timestamp": ANY}, - "aerotech_z": {"value": 1.0, "timestamp": ANY}, - "aerotech_u": {"value": 1.0, "timestamp": ANY}, - "aerotech_vel_u_deg_s": {"value": 1.0, "timestamp": ANY}, - } +def test_is_the_mx_bec_device(): + assert Aerotech is MxAerotech - def test_aerotech_set_with_status(self, aerotech): - aerotech.wait_for_connection() - st = aerotech.x.set(5.0) - st.wait(timeout=1) - assert aerotech.x.get() == 5.0 + +def test_aerotech_read(aerotech): + aerotech.wait_for_connection() + reading = dict(aerotech.read()) + # besides the axes, read() reports the valid/message state of the scan parameter groups + assert {k: v for k, v in reading.items() if k.removeprefix("aerotech_") in AXES} == { + f"aerotech_{axis}": {"value": 1.0, "timestamp": ANY} for axis in AXES + } + assert reading["aerotech_gridscan_config_valid"]["value"] is True + + +def test_aerotech_set_with_status(aerotech): + aerotech.wait_for_connection() + aerotech.x.set(5.0).wait(timeout=1) + assert aerotech.x.get() == 5.0 diff --git a/tests/tests_devices/test_smargon.py b/tests/tests_devices/test_smargon.py index fc14015..1b8f7bc 100644 --- a/tests/tests_devices/test_smargon.py +++ b/tests/tests_devices/test_smargon.py @@ -1,14 +1,24 @@ +"""Smoke tests of the Smargon device as re-exported by pxii_bec; the device itself is tested in +mx_bec against a mock smargopolo server.""" + from copy import copy from threading import RLock from unittest.mock import ANY import pytest +from mx_bec.devices.smargopolo_smargon import Smargon as MxSmargon + +from pxii_bec.devices import Smargon + +AXES = ("SHX", "SHY", "SHZ", "PHI", "CHI") class MockServer: + """Just enough of the smargopolo REST API for reading and moving axes.""" + 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} + self.mock_data = {axis: 1.0 for axis in AXES} def get(self, endpoint): with self.lock: @@ -23,29 +33,28 @@ class MockServer: @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() + device = Smargon(name="smargon", prefix="http://test-smargopolo.psi.ch") + device.controller._rest_get = mock_server.get + device.controller._rest_put = mock_server.put + yield device + device.controller._stop_monitor_readback_event.set() + device.destroy() -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_is_the_mx_bec_device(): + assert Smargon is MxSmargon - 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 + +def test_smargon_read(smargon): + smargon.wait_for_connection() + reading = smargon.read() + assert dict(reading) == { + f"smargon_{axis}": {"value": 1.0, "timestamp": ANY} + for axis in ("x", "y", "z", "phi", "chi") + } + + +def test_smargon_set_with_status(smargon): + smargon.wait_for_connection() + smargon.x.set(5.0).wait(timeout=1) + assert smargon.x.get() == 5.0