i15-1: Add centring result to data collection metadata

pyproject.toml

@@ -29,6 +29,7 @@ requires-python = ">=3.11"
 
 [dependency-groups]
 dev = [
+    "heliotrapi>=0.3.3",
     "httpx2>=2.13.1",
     "myst-parser",
     "podman-compose>=1.5.0",

src/daq_queuing_service/plugins/converter.py

@@ -5,6 +5,7 @@
 
 from daq_queuing_service.external_interaction.blueapi.blueapi_call import (
     BlueapiCall,
+    BlueapiCallResponse,
 )
 from daq_queuing_service.task_queue.task import (
     Experiment,
@@ -99,7 +100,7 @@ def construct_blueapi_calls(
         return call_list
 
     def pause_on_complete_blueapi_call(
-        self, call: BlueapiCall, result: TaskResult
+        self, call: BlueapiCallResponse, result: TaskResult
     ) -> bool:
         """Decide whether to pause the queue after a blueapi call completes successfully
 
@@ -113,7 +114,7 @@ def pause_on_complete_blueapi_call(
         return False
 
     def pause_on_fail_blueapi_call(
-        self, call: BlueapiCall, errors: list[str | TaskError] | None
+        self, call: BlueapiCallResponse, errors: list[str | TaskError] | None
     ) -> bool:
         """Decide whether to pause the queue after a blueapi call fails
 
@@ -126,6 +127,24 @@ def pause_on_fail_blueapi_call(
         """
         return True
 
+    def process_running_task(
+        self, completed_plans: list[BlueapiCallResponse], plans_to_do: list[BlueapiCall]
+    ) -> list[BlueapiCall]:
+        """An opportunity to process the currently running task, by amending the
+        blueapi calls yet to be made. Get's called right after a blueapi call completes.
+
+        Args:
+            completed_plans (list[BlueapiCallResponse]): Completed blueapi calls
+                associated with the currently running task.
+            plans_to_do (list[BlueapiCall]): Future blueapi calls associated with the
+                currently running task.
+
+        Returns:
+            list[BlueapiCall]: An amended list of future blueapi calls for the running
+                task.
+        """
+        return plans_to_do
+
     def _construct_blueapi_task_request(
         self,
         experiment: Experiment | TaskRequest,

src/daq_queuing_service/plugins/i15_1/i15_1_converter.py

@@ -8,6 +8,7 @@
 
 from daq_queuing_service.external_interaction.blueapi.blueapi_call import (
     BlueapiCall,
+    BlueapiCallResponse,
 )
 from daq_queuing_service.external_interaction.tiled.tiled import get_tiled_client
 from daq_queuing_service.log import (
@@ -127,7 +128,7 @@ def construct_blueapi_calls(
         return call_list
 
     def pause_on_fail_blueapi_call(
-        self, call: BlueapiCall, errors: list[str | TaskError] | None
+        self, call: BlueapiCallResponse, errors: list[str | TaskError] | None
     ) -> bool:
 
         for error in errors or []:
@@ -140,6 +141,24 @@ def pause_on_fail_blueapi_call(
                     return False
         return True
 
+    def process_running_task(
+        self, completed_plans: list[BlueapiCallResponse], plans_to_do: list[BlueapiCall]
+    ):
+        if completed_plans[-1].task_request.name == "centre_sample":
+            centring_plan = completed_plans[-1]
+            assert len(centring_plan.tiled_ids) <= 1
+            tiled_id = (
+                centring_plan.tiled_ids[0] if len(centring_plan.tiled_ids) else None
+            )
+            for plan in plans_to_do:
+                if centring_plan.result and plan.task_request.name == "data_collection":
+                    plan.task_request.params["metadata"]["centring_scan"] = {
+                        "result": centring_plan.result,
+                        "tiled_id": tiled_id,
+                    }
+
+        return super().process_running_task(completed_plans, plans_to_do)
+
     def _construct_blueapi_tasks_from_experiment(
         self, experiment: Experiment, task_id: str
     ) -> list[TaskRequest]:

src/daq_queuing_service/task_queue/queue.py

@@ -215,8 +215,7 @@ def _sync(self):
             # and for in progress tasks to inform conversion
             call
             for call in self._call_queue
-            if call.parent_task_id
-            and call.parent_task_id in self._queue
+            if call.parent_task_id in self._queue
             and self._tasks[call.parent_task_id].status == Status.IN_PROGRESS
             and call.status not in (CallStatus.SUCCESS, CallStatus.ERROR)
         ]
@@ -356,7 +355,7 @@ async def complete_call(self, call: BlueapiCall, result: TaskResult):
         with add_session_to_logger(call.task_request.instrument_session):
             tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call)
             should_pause = self._converter.pause_on_complete_blueapi_call(
-                call.model_copy(), result
+                call.to_response(), result
             )
             LOGGER.info(f"Plan {call} completed successfully. Result: {result}")
 
@@ -366,6 +365,7 @@ async def complete_call(self, call: BlueapiCall, result: TaskResult):
             self._check_call_valid_to_be_returned(call)
             call.succeed(result)
             self._call_history.append(call)
+            self._process_running_task(self._tasks[call.parent_task_id])
             if should_pause:
                 self._pause_queue(reason=PauseReason.PLUGIN_REQUESTED)
 
@@ -381,7 +381,7 @@ async def fail_call(
         """
         with add_session_to_logger(call.task_request.instrument_session):
             should_pause = self._converter.pause_on_fail_blueapi_call(
-                call.model_copy(), errors
+                call.to_response(), errors
             )
             tiled_ids, scan_ids = get_tiled_and_scan_ids(self._tiled_client, call)
             LOGGER.error(f"Plan {call} has failed with the following errors: {errors}")
@@ -400,6 +400,23 @@ async def fail_call(
             call.fail(errors)
             self._call_history.append(call)
 
+    def _process_running_task(self, running_task: Task):
+        done_calls = [
+            call
+            for call in running_task.blueapi_calls
+            if call.status in [CallStatus.SUCCESS, CallStatus.ERROR]
+        ]
+        next_calls = [
+            call
+            for call in running_task.blueapi_calls
+            if call.status not in [CallStatus.SUCCESS, CallStatus.ERROR]
+        ]
+        next_calls = self._converter.process_running_task(
+            [call.to_response() for call in done_calls],
+            next_calls,
+        )
+        running_task.blueapi_calls = done_calls + next_calls
+
     async def get_task_by_id(self, task_id: str) -> TaskWithPosition:
         """Returns a task based on it's task ID
 

src/daq_queuing_service/task_queue/task.py

@@ -77,7 +77,8 @@ def status(self) -> Status:
         if self._cancelled:
             return Status.CANCELLED
         if self.blueapi_calls and all(
-            call.status == CallStatus.SUCCESS for call in self.blueapi_calls
+            call.status in (CallStatus.SUCCESS, CallStatus.SKIPPED)
+            for call in self.blueapi_calls
         ):
             return Status.COMPLETE
         if any(call.status == CallStatus.ERROR for call in self.blueapi_calls):

tests/unit_tests/plugins/i15_1/conftest.py

@@ -1,5 +1,7 @@
+import datetime
 from typing import Any
 from unittest.mock import patch
+from uuid import UUID
 
 import pytest
 from daq_config_server.client import ConfigClient
@@ -11,6 +13,7 @@
     StandardsPuck,
 )
 from daq_config_server.testing import MockServerResponse
+from heliotrapi.client import AnalysisResult
 
 from daq_queuing_service.plugins.i15_1.i15_1_converter import (
     QUEUE_PARAMS_CONFIG_PATH,
@@ -92,3 +95,37 @@ def mock_config_client(mock_config: dict[str, Any]):
         return_value=config_client,
     ):
         yield config_client
+
+
+@pytest.fixture
+def centring_result() -> AnalysisResult:
+    return AnalysisResult(
+        request_id=UUID("7d71e99e-f6f6-4e93-a1e4-ac91f8804006"),
+        status="completed",
+        analysis_name="sample_alignment_i15_1",
+        inputs={
+            "filepath": "/dls/i15-1/data/2026/cm44163-4/i15-1-100259.nxs",
+            "beamline": "i15-1",
+        },
+        result={
+            "centre": -1.7327164569369127,
+            "peaks": [
+                {
+                    "amplitude": 684.5096633867339,
+                    "centre": -12.500470871238504,
+                    "fwhm": 16.126336919254836,
+                    "normalised": False,
+                },
+                {
+                    "amplitude": 14080.218114003073,
+                    "centre": -1.7327164569369127,
+                    "fwhm": 7.5161854303657645,
+                    "normalised": False,
+                },
+            ],
+            "scores": [11038.633456260188, 105829.53024484203],
+        },
+        created_at=datetime.datetime(2026, 10, 12, 22, 22, 45, 370672),
+        finished_at=datetime.datetime(2026, 10, 12, 22, 22, 47, 577230),
+        version="0.3.4.dev0+gfbb23e084.d20260929",
+    )

tests/unit_tests/plugins/i15_1/test_i15_1_converter.py

@@ -6,14 +6,15 @@
 
 import pytest
 from blueapi.service.model import TaskRequest
-from blueapi.worker.event import TaskError
+from blueapi.worker.event import TaskError, TaskResult
 from daq_config_server.models.i15_1.queue_parameters import QueueParameters
 from daq_config_server.models.i15_1.standards_puck import (
     STANDARD_CAPILLARY,
     STANDARD_SAMPLE,
     StandardsPin,
     StandardsPuck,
 )
+from heliotrapi.client import AnalysisResult
 
 from daq_queuing_service.broadcaster import Broadcaster, serialise
 from daq_queuing_service.plugins.i15_1.auxiliary import (
@@ -964,3 +965,62 @@ def test_no_wait_for_beam_queued_if_turned_off_in_queue_params(
         [TaskWithPosition.from_task(i15_1_tasks[0])], [], []
     )
     assert not any(call.task_request.name == "wait_for_beam" for call in blueapi_calls)
+
+
+async def test_centring_info_added_to_data_collection_metadata(
+    queue_with_i15_1_plugin: TaskQueue, centring_result: AnalysisResult
+):
+    while True:
+        call = await queue_with_i15_1_plugin.get_next_call_once_available()
+        if (
+            call.task_request.name == "data_collection"
+            and call.task_request.params["metadata"].get("sample") is not None
+        ):
+            break
+        call.put_in_progress()
+        result = (
+            centring_result.model_dump(mode="json")
+            if call.task_request.name == "centre_sample"
+            else None
+        )
+        await queue_with_i15_1_plugin.complete_call(
+            call, TaskResult(outcome="success", result=result, type="AnalysisResult")
+        )
+
+    assert call.task_request.params["metadata"]["centring_scan"] == {
+        "result": TaskResult(
+            outcome="success",
+            result={
+                "request_id": "7d71e99e-f6f6-4e93-a1e4-ac91f8804006",
+                "status": "completed",
+                "analysis_name": "sample_alignment_i15_1",
+                "inputs": {
+                    "filepath": "/dls/i15-1/data/2026/cm44163-4/i15-1-100259.nxs",
+                    "beamline": "i15-1",
+                },
+                "result": {
+                    "centre": -1.7327164569369127,
+                    "peaks": [
+                        {
+                            "amplitude": 684.5096633867339,
+                            "centre": -12.500470871238504,
+                            "fwhm": 16.126336919254836,
+                            "normalised": False,
+                        },
+                        {
+                            "amplitude": 14080.218114003073,
+                            "centre": -1.7327164569369127,
+                            "fwhm": 7.5161854303657645,
+                            "normalised": False,
+                        },
+                    ],
+                    "scores": [11038.633456260188, 105829.53024484203],
+                },
+                "created_at": "2026-10-12T22:22:45.370672",
+                "finished_at": "2026-10-12T22:22:47.577230",
+                "version": "0.3.4.dev0+gfbb23e084.d20260929",
+            },
+            type="AnalysisResult",
+        ),
+        "tiled_id": None,
+    }

tests/unit_tests/test_queue.py

@@ -1446,6 +1446,27 @@ async def test_fail_call_gets_md_from_tiled_and_adds_to_call_object(
     assert call.tiled_ids == ["tiled_id_1"]
 
 
+async def test_complete_call_amends_future_calls_based_on_plugin_processing(
+    task_queue_in_progress: TaskQueue, converter: Converter
+):
+    future_call = BlueapiCall(
+        task_request=TaskRequest(
+            name="test process_running task", params={}, instrument_session=""
+        ),
+        parent_task_id="",
+    )
+
+    converter.process_running_task = MagicMock(return_value=[future_call])
+    await task_queue_in_progress.complete_call(
+        task_queue_in_progress._call_queue[0],
+        result=TaskResult(outcome="success", result=None, type=""),
+    )
+
+    current_task = task_queue_in_progress._get_running_task()
+    assert current_task
+    assert current_task.blueapi_calls[1] == future_call.to_response()
+
+
 def test__add_runtime_estimates_to_tasks_gets_runtime_for_all_queued_tasks(
     task_queue_in_progress: TaskQueue, mock_runtime_estimator_client: MagicMock
 ):