From e8663c8e1ec8fc8b85a0d9778dfd481be9db9447 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:40:51 +0000 Subject: [PATCH 1/6] feat: subdirectory for numtracker --- docs/how-to/1.4 integrate-with-numtracker.md | 21 +++++++++ docs/reference/openapi.yaml | 9 +++- src/blueapi/config.py | 2 +- src/blueapi/core/context.py | 4 +- src/blueapi/service/interface.py | 2 + src/blueapi/service/model.py | 6 +++ src/blueapi/utils/numtracker.py | 35 ++++++++++----- src/blueapi/worker/task.py | 2 + tests/conftest.py | 16 ++++--- tests/unit_tests/service/test_interface.py | 45 +++++++++++++++++++- 10 files changed, 121 insertions(+), 21 deletions(-) diff --git a/docs/how-to/1.4 integrate-with-numtracker.md b/docs/how-to/1.4 integrate-with-numtracker.md index 9499ea0c25..fac3b11ee5 100644 --- a/docs/how-to/1.4 integrate-with-numtracker.md +++ b/docs/how-to/1.4 integrate-with-numtracker.md @@ -24,3 +24,24 @@ Numtracker will work for any ophyd-async [`StandardDetector`](https://blueskypro Blueapi uses an internal [`PathProvider`](https://blueskyproject.io/ophyd-async/main/_api/ophyd_async/ophyd_async.core.html#ophyd_async.core.PathProvider) when integrated with numtracker, it does not support custom providers in from dodal modules, if your beamline definition in dodal is prefixed with a call to `set_path_provider`, it should be removed before attempting to use numtracker: Numtracker should work with out-of-box plans that take data via runs. Opening a new run will make the `RunEngine` call numtracker and request a fresh data area. Some plans from before numtracker have a decorator that coordinates where detectors write their data: [`@attach_data_session_metadata_decorator`](https://github.com/DiamondLightSource/dodal/blob/10a9a124931901d7666659c6dbe77215d22a8bfd/src/dodal/plan_stubs/data_session.py#L60). This decorator becomes a no-op with numtracker and can be safely removed from plans. + +## Writing to a Subdirectory + +A task can ask numtracker to place its scans in a subdirectory of the instrument session's data directory by setting the optional `subdirectory` field when it is submitted to `POST /tasks`. It applies to every run in that task; tasks without it are not affected. + +```json +{ + "name": "count", + "params": {"detectors": ["det"]}, + "instrument_session": "cm12345-1", + "subdirectory": "sample1/calibration" +} +``` + +```{note} +The subdirectory only has an effect if your instrument's numtracker configuration uses the `{subdirectory}` placeholder, e.g. a scan template of `{subdirectory}/{instrument}-{scan_number}`. If it does not, the subdirectory is ignored. See the [numtracker documentation](https://github.com/DiamondLightSource/numtracker) for configuring templates. + +It changes where the scan file goes, not where detectors write: detector files are still written to the root of the data session directory using blueapi's `detector_file_template`. + +Scan numbers are allocated per instrument, not per subdirectory, so they keep incrementing across subdirectories. +``` diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index c17734bbe2..605ce06cb4 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -301,6 +301,13 @@ components: description: Values for parameters to plan, if any title: Params type: object + subdirectory: + anyOf: + - type: string + - type: 'null' + description: Subdirectory of the instrument session's data directory to + write this task's scans to, if numtracker is configured to use one + title: Subdirectory required: - name - instrument_session @@ -459,7 +466,7 @@ info: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html title: BlueAPI Control - version: 1.6.0 + version: 1.7.0 openapi: 3.1.0 paths: /api/v1/devices: diff --git a/src/blueapi/config.py b/src/blueapi/config.py index f65dd65f0a..cb63259b69 100644 --- a/src/blueapi/config.py +++ b/src/blueapi/config.py @@ -335,7 +335,7 @@ class ApplicationConfig(BlueapiBaseModel): """ #: API version to publish in OpenAPI schema - REST_API_VERSION: ClassVar[str] = "1.6.0" + REST_API_VERSION: ClassVar[str] = "1.7.0" LICENSE_INFO: ClassVar[dict[str, str]] = { "name": "Apache 2.0", diff --git a/src/blueapi/core/context.py b/src/blueapi/core/context.py index f8a96b0d6a..ebc761feed 100644 --- a/src/blueapi/core/context.py +++ b/src/blueapi/core/context.py @@ -157,7 +157,9 @@ def __post_init__(self, configuration: ApplicationConfig | None): async def _update_scan_num(md: dict[str, Any]) -> int: scan = await numtracker.create_scan( - md["instrument_session"], md["instrument"] + md["instrument_session"], + md["instrument"], + md.get("data_session_subdirectory"), ) md["data_session_directory"] = str(scan.scan.directory.path) md["detector_file_template"] = nt_conf.detector_file_template diff --git a/src/blueapi/service/interface.py b/src/blueapi/service/interface.py index 7df4745570..24b980d60e 100644 --- a/src/blueapi/service/interface.py +++ b/src/blueapi/service/interface.py @@ -211,6 +211,8 @@ def submit_task( metadata = {} metadata["instrument_session"] = task_request.instrument_session + if task_request.subdirectory is not None: + metadata["data_session_subdirectory"] = task_request.subdirectory if context().tiled_conf is not None: md = config().env.metadata # We raise an InvalidConfigError on setting tiled_conf if this isn't set diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index 1b690d5c04..3957ed9231 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -75,6 +75,12 @@ class TaskRequest(BlueapiBaseModel): instrument_session: str = Field( description="Instrument session associated with this task", ) + subdirectory: str | None = Field( + description="Subdirectory of the instrument session's data directory to " + "write this task's scans to, if numtracker is configured to use one", + default=None, + exclude_if=lambda subdirectory: subdirectory is None, + ) class DeviceRequest(BlueapiBaseModel): diff --git a/src/blueapi/utils/numtracker.py b/src/blueapi/utils/numtracker.py index 0fde7c699e..234c059f48 100644 --- a/src/blueapi/utils/numtracker.py +++ b/src/blueapi/utils/numtracker.py @@ -63,7 +63,7 @@ def set_headers(self, headers: Mapping[str, str]) -> None: self._headers = headers async def create_scan( - self, instrument_session: str, instrument: str + self, instrument_session: str, instrument: str, sub: str | None = None ) -> NumtrackerScanMutationResponse: """ Create a new scan with numtracker. @@ -73,25 +73,38 @@ async def create_scan( e.g. cm12345-1 instrument: The instrument to write data on e.g. i22 + + sub: Optional subdirectory of the session directory to write the + scan to e.g. sample1/calibration """ query = { - "query": dedent(f""" - mutation{{ + "query": dedent(""" + mutation( + $instrument: String!, + $instrumentSession: String!, + $sub: Subdirectory + ){ scan( - instrument: "{instrument}", - instrumentSession: "{instrument_session}" - ) {{ - directory{{ + instrument: $instrument, + instrumentSession: $instrumentSession, + sub: $sub + ) { + directory{ instrumentSession instrument path - }} + } scanFile scanNumber - }} - }} - """) + } + } + """), + "variables": { + "instrument": instrument, + "instrumentSession": instrument_session, + "sub": sub, + }, } async with httpx.AsyncClient() as client: diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index d4cf0e809a..2bdf768033 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -108,6 +108,8 @@ def do_task(self, ctx: BlueskyContext) -> None: ) plan = ctx.plan_functions[self.name] prepared_args, prepared_kwargs = self.prepare_params(ctx) + # RunEngine.md persists between tasks, subdirectory must not carry over + ctx.run_engine.md.pop("data_session_subdirectory", None) ctx.run_engine.md.update(self.metadata) result = ctx.run_engine(plan(*prepared_args, **prepared_kwargs)) if isinstance(result, tuple): # pragma: no cover diff --git a/tests/conftest.py b/tests/conftest.py index 6f517b3bf9..a0f5440c88 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -344,13 +344,18 @@ def mock_jwks_fetch(json_web_keyset: JWK): @pytest.fixture -def nt_query() -> dict[str, str]: +def nt_query() -> dict[str, Any]: return { "query": dedent(""" - mutation{ + mutation( + $instrument: String!, + $instrumentSession: String!, + $sub: Subdirectory + ){ scan( - instrument: "p46", - instrumentSession: "ab123" + instrument: $instrument, + instrumentSession: $instrumentSession, + sub: $sub ) { directory{ instrumentSession @@ -361,7 +366,8 @@ def nt_query() -> dict[str, str]: scanNumber } } - """) + """), + "variables": {"instrument": "p46", "instrumentSession": "ab123", "sub": None}, } diff --git a/tests/unit_tests/service/test_interface.py b/tests/unit_tests/service/test_interface.py index db4e687600..689a53a4fb 100644 --- a/tests/unit_tests/service/test_interface.py +++ b/tests/unit_tests/service/test_interface.py @@ -6,6 +6,7 @@ from typing import Any from unittest.mock import ANY, MagicMock, Mock, patch +import bluesky.plans as bp import pytest from bluesky.protocols import Stoppable from bluesky.utils import MsgGenerator @@ -42,7 +43,7 @@ WorkerTask, ) from blueapi.utils.invalid_config_error import InvalidConfigError -from blueapi.utils.numtracker import NumtrackerClient +from blueapi.utils.numtracker import NumtrackerClient, NumtrackerScanMutationResponse from blueapi.worker.event import ( TaskResult, TaskStatus, @@ -799,7 +800,47 @@ async def test_numtracker_create_scan_called_with_arguments_from_metadata( scan_id = ctx.run_engine.scan_id_source(ctx.run_engine.md) assert isawaitable(scan_id) and await scan_id - mock_create_scan.assert_called_once_with("ab123", "p46") + mock_create_scan.assert_called_once_with("ab123", "p46", None) + + +@patch("blueapi.service.interface.worker") +def test_numtracker_subdirectory_is_per_task(mock_worker, nt_response): + conf = ApplicationConfig( + numtracker=NumtrackerConfig( + url=HttpUrl("https://numtracker-example.com/graphql") + ), + env=EnvironmentConfig(metadata=MetadataConfig(instrument="p46")), + ) + interface.set_config(conf) + ctx = interface.context() + + def two_runs() -> MsgGenerator: + yield from bp.count([]) + yield from bp.count([]) + + ctx.register_plan(two_runs) + + with patch.object( + NumtrackerClient, + "create_scan", + return_value=NumtrackerScanMutationResponse.model_validate(nt_response["data"]), + ) as mock_create_scan: + for subdirectory in ("sample1", None): + interface.submit_task( + TaskRequest( + name="two_runs", + instrument_session="ab123", + subdirectory=subdirectory, + ) + ) + mock_worker.return_value.submit_task.call_args.args[0].do_task(ctx) + + assert [c.args for c in mock_create_scan.call_args_list] == [ + ("ab123", "p46", "sample1"), + ("ab123", "p46", "sample1"), + ("ab123", "p46", None), + ("ab123", "p46", None), + ] async def test_update_scan_num_side_effect_sets_data_session_directory_in_re_md( From bdeac43802b1b8fcf38f9efae2f3256b14471836 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh Date: Tue, 10 Nov 2026 20:09:24 +0000 Subject: [PATCH 2/6] Apply suggestion from @tpoliaw Co-authored-by: Peter Holloway --- src/blueapi/service/model.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/blueapi/service/model.py b/src/blueapi/service/model.py index 3957ed9231..d4f9ecdad5 100644 --- a/src/blueapi/service/model.py +++ b/src/blueapi/service/model.py @@ -75,7 +75,7 @@ class TaskRequest(BlueapiBaseModel): instrument_session: str = Field( description="Instrument session associated with this task", ) - subdirectory: str | None = Field( + subdirectory: str | SkipJsonSchema[None] = Field( description="Subdirectory of the instrument session's data directory to " "write this task's scans to, if numtracker is configured to use one", default=None, From 1c62d6994313ee02d8b25177f074f529463c727e Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:44:49 +0000 Subject: [PATCH 3/6] Add to client and cli --- docs/reference/openapi.yaml | 4 +- src/blueapi/cli/cli.py | 17 +++++- src/blueapi/client/client.py | 21 +++++++ src/blueapi/client/plan_cache.py | 1 + src/blueapi/client/protocols.py | 3 + src/blueapi/service/numtracker.py | 45 +++++++------- tests/conftest.py | 23 +------- tests/system_tests/services/startup.py | 19 +++--- tests/system_tests/test_blueapi_system.py | 69 ++++++++++++++++++---- tests/unit_tests/client/test_client.py | 14 ++++- tests/unit_tests/client/test_plan_cache.py | 2 + 11 files changed, 152 insertions(+), 66 deletions(-) diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index 605ce06cb4..b30016b50e 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -302,12 +302,10 @@ components: title: Params type: object subdirectory: - anyOf: - - type: string - - type: 'null' description: Subdirectory of the instrument session's data directory to write this task's scans to, if numtracker is configured to use one title: Subdirectory + type: string required: - name - instrument_session diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index d78a88d012..e2b504665e 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -405,6 +405,16 @@ def on_event( the session must be valid and active and you must be a member of it."""), required=True, ) +@click.option( + "-sub", + "--subdirectory", + type=str, + help=textwrap.dedent(""" + Subdirectory of the instrument session's data directory to write + data to, e.g. a per-sample directory. Passed to numtracker along with + the instrument session."""), + required=False, +) @click.pass_obj @check_connection def run_plan( @@ -414,6 +424,7 @@ def run_plan( foreground: bool, ws: bool | None, instrument_session: str, + subdirectory: str | None, args: list[Any], parameters: TaskParameters, ) -> None: @@ -426,7 +437,11 @@ def run_plan( client = cast(BlueapiClient, obj["client"]) task = TaskRequest( - name=name, args=args, params=parameters, instrument_session=instrument_session + name=name, + args=args, + params=parameters, + instrument_session=instrument_session, + subdirectory=subdirectory, ) try: diff --git a/src/blueapi/client/client.py b/src/blueapi/client/client.py index 2105ac64e3..85e298caa2 100644 --- a/src/blueapi/client/client.py +++ b/src/blueapi/client/client.py @@ -67,6 +67,7 @@ class BlueapiClient(ClientProtocol): _rest: BlueapiRestClient _events: EventBusClient | None _instrument_session: str | None = None + _subdirectory: str | None = None _callbacks: dict[int, OnAnyEvent] _callback_id: itertools.count @@ -137,11 +138,31 @@ def instrument_session(self) -> str: def instrument_session(self, session: str): log.debug("Setting instrument_session to %s", session) self._instrument_session = session + # A subdirectory belongs to a session, don't carry it over to a new one + if self._subdirectory is not None: + log.info( + "Clearing subdirectory %s as instrument_session changed", + self._subdirectory, + ) + self._subdirectory = None def with_instrument_session(self, session: str) -> Self: self.instrument_session = session return self + @property + def subdirectory(self) -> str | None: + return self._subdirectory + + @subdirectory.setter + def subdirectory(self, subdirectory: str | None): + log.debug("Setting subdirectory to %s", subdirectory) + self._subdirectory = subdirectory + + def with_subdirectory(self, subdirectory: str | None) -> Self: + self.subdirectory = subdirectory + return self + @start_as_current_span(TRACER) @deprecated("plans property") def get_plans(self) -> PlanResponse: diff --git a/src/blueapi/client/plan_cache.py b/src/blueapi/client/plan_cache.py index acf5b27ee8..b30b9b5e38 100644 --- a/src/blueapi/client/plan_cache.py +++ b/src/blueapi/client/plan_cache.py @@ -80,6 +80,7 @@ def __call__(self, *args, **kwargs) -> Any: args=list(args), params=kwargs, instrument_session=self._client.instrument_session, + subdirectory=self._client.subdirectory, ) try: match self._client.run_task(req): diff --git a/src/blueapi/client/protocols.py b/src/blueapi/client/protocols.py index dbb923af6c..aac98fb78a 100644 --- a/src/blueapi/client/protocols.py +++ b/src/blueapi/client/protocols.py @@ -20,6 +20,9 @@ def run_task( @property def instrument_session(self) -> str: ... + @property + def subdirectory(self) -> str | None: ... + class RestClientProtocol(Protocol): def get_devices(self) -> DeviceResponse: ... diff --git a/src/blueapi/service/numtracker.py b/src/blueapi/service/numtracker.py index 234c059f48..dfbea5bd65 100644 --- a/src/blueapi/service/numtracker.py +++ b/src/blueapi/service/numtracker.py @@ -11,6 +11,29 @@ LOGGER = logging.getLogger(__name__) +CREATE_SCAN_MUTATION = """ + mutation( + $instrument: String!, + $instrumentSession: String!, + $sub: Subdirectory + ){ + scan( + instrument: $instrument, + instrumentSession: $instrumentSession, + sub: $sub + ) { + directory{ + instrumentSession + instrument + path + } + scanFile + scanNumber + } + } + """ + + class DirectoryPath(BlueapiBaseModel): """ Directory location and associated metadata @@ -79,27 +102,7 @@ async def create_scan( """ query = { - "query": dedent(""" - mutation( - $instrument: String!, - $instrumentSession: String!, - $sub: Subdirectory - ){ - scan( - instrument: $instrument, - instrumentSession: $instrumentSession, - sub: $sub - ) { - directory{ - instrumentSession - instrument - path - } - scanFile - scanNumber - } - } - """), + "query": dedent(CREATE_SCAN_MUTATION), "variables": { "instrument": instrument, "instrumentSession": instrument_session, diff --git a/tests/conftest.py b/tests/conftest.py index bb38f1098d..6e7882b2d4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -21,6 +21,7 @@ from blueapi.config import ApplicationConfig, OIDCConfig from blueapi.service.model import Cache +from blueapi.service.numtracker import CREATE_SCAN_MUTATION @pytest.fixture(scope="function") @@ -343,27 +344,7 @@ def mock_jwks_fetch(json_web_keyset: JWK): @pytest.fixture def nt_query() -> dict[str, Any]: return { - "query": dedent(""" - mutation( - $instrument: String!, - $instrumentSession: String!, - $sub: Subdirectory - ){ - scan( - instrument: $instrument, - instrumentSession: $instrumentSession, - sub: $sub - ) { - directory{ - instrumentSession - instrument - path - } - scanFile - scanNumber - } - } - """), + "query": dedent(CREATE_SCAN_MUTATION), "variables": {"instrument": "p46", "instrumentSession": "ab123", "sub": None}, } diff --git a/tests/system_tests/services/startup.py b/tests/system_tests/services/startup.py index 2bfa722fcb..1c2b485a6b 100644 --- a/tests/system_tests/services/startup.py +++ b/tests/system_tests/services/startup.py @@ -277,15 +277,20 @@ def configure_numtracker(): response = requests.post( NT_URL, json={ - "query": """mutation { - configure(instrument: "adsim", - config: {directory: "/tmp/", - scan: "{instrument}-{scan_number}", - detector: "{instrument}-{scan_number}-{detector}", - scanNumber: 43}) { + "query": """mutation($instrument: String!, $config: ConfigurationUpdates!) { + configure(instrument: $instrument, config: $config) { scanTemplate } - }""" + }""", + "variables": { + "instrument": "adsim", + "config": { + "directory": "/tmp/", + "scan": "{instrument}-{scan_number}", + "detector": "{instrument}-{scan_number}-{detector}", + "scanNumber": 43, + }, + }, }, headers={"authorization": "Bearer " + access_token}, ) diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index f393c7ab8c..41d7800baf 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -249,31 +249,54 @@ def clean_existing_tasks(rest_client: BlueapiRestClient): yield -@pytest.fixture(autouse=True, scope="module") -def reset_numtracker(): +NUMTRACKER_CONFIGURE_MUTATION = """ +mutation($instrument: String!, $config: ConfigurationUpdates!) { + configure(instrument: $instrument, config: $config) { + scanTemplate + } +} +""" +SCAN_TEMPLATE = "{instrument}-{scan_number}" + + +def configure_numtracker(config: dict[str, str | int]): + """Update the adsim numtracker config, fields left out are unchanged""" server_config = load_config(Path(_DATA_PATH, "config.yaml")) nt_url = server_config.numtracker.url # type: ignore - if numtracker is None we should fail response = requests.post( str(nt_url), json={ - "query": f"""mutation {{ - configure(instrument: "adsim", - config: {{directory: "/tmp/", - scan: "{{instrument}}-{{scan_number}}", - detector: "{{instrument}}-{{scan_number}}-{{detector}}", - scanNumber: {CURRENT_NUMTRACKER_NUM}}}) {{ - scanTemplate - }} - }}""" + "query": NUMTRACKER_CONFIGURE_MUTATION, + "variables": {"instrument": "adsim", "config": config}, }, headers={"authorization": "Bearer " + get_access_token(User.admin)}, ) response.raise_for_status() if response.json().get("errors") is not None: raise Exception(response.json()) + + +@pytest.fixture(autouse=True, scope="module") +def reset_numtracker(): + configure_numtracker( + { + "directory": "/tmp/", + "scan": SCAN_TEMPLATE, + "detector": "{instrument}-{scan_number}-{detector}", + "scanNumber": CURRENT_NUMTRACKER_NUM, + } + ) yield +@pytest.fixture +def numtracker_with_subdirectory(): + """Numtracker only uses the subdirectory if the scan template asks for it""" + configure_numtracker({"scan": "{subdirectory}/" + SCAN_TEMPLATE}) + yield + configure_numtracker({"scan": SCAN_TEMPLATE}) + + def test_cannot_access_endpoints_without_authentication( client_without_auth: BlueapiClient, blueapi_rest_client_get_methods: list[str] ): @@ -655,6 +678,30 @@ def on_event(event: AnyEvent) -> None: assert "det" in start_metadata["detectors"] +@pytest.mark.usefixtures("numtracker_with_subdirectory") +def test_plan_runs_with_subdirectory(client_with_stomp: BlueapiClient, scan_id: int): + start = Queue(maxsize=1) + + def on_event(event: AnyEvent) -> None: + if isinstance(event, DataEvent) and event.name == "start": + start.put_nowait(event.doc) + + cb_id = client_with_stomp.add_callback(on_event) + try: + client_with_stomp.with_instrument_session( + VALID_INSTRUMENT_SESSION[User.alice] + ).with_subdirectory("sample1") + client_with_stomp.plans.count([client_with_stomp.devices.det], 3) + finally: + client_with_stomp.remove_callback(cb_id) + + start_doc = start.get_nowait() + assert start_doc["scan_id"] == scan_id + assert start_doc["data_session_subdirectory"] == "sample1" + assert start_doc["data_session_directory"] == "/tmp" + assert start_doc["scan_file"] == f"sample1/adsim-{scan_id}" + + @pytest.mark.parametrize( "task", [ diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index e7e31f9302..0acc945848 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -497,7 +497,7 @@ def callback(on_event: Callable[[AnyEvent, MessageContext], None]): def test_scripting_interface_returns_result(): - client = Mock(spec=BlueapiClient, instrument_session="cm12345-1") + client = Mock(spec=BlueapiClient, instrument_session="cm12345-1", subdirectory=None) client.run_task.return_value = TaskStatus( task_id="foobar", task_complete=True, @@ -513,7 +513,7 @@ def test_scripting_interface_returns_result(): def test_scripting_interface_raises_exceptions(): - client = Mock(spec=BlueapiClient, instrument_session="cm12345-1") + client = Mock(spec=BlueapiClient, instrument_session="cm12345-1", subdirectory=None) client.run_task.return_value = TaskStatus( task_id="foobar", task_complete=True, @@ -675,6 +675,16 @@ def test_fluent_instrument_session_setter(client): client2 = client.with_instrument_session("cm12345-3") assert client is client2 assert client.instrument_session == "cm12345-3" + assert client.subdirectory is None + + +def test_instrument_session_with_subdirectory(client): + client2 = client.with_instrument_session("cm12345-3").with_subdirectory("sample1") + assert client is client2 + assert client.subdirectory == "sample1" + # Changing session drops the old subdirectory + client.instrument_session = "cm12345-4" + assert client.subdirectory is None def test_adding_removing_callback(client): diff --git a/tests/unit_tests/client/test_plan_cache.py b/tests/unit_tests/client/test_plan_cache.py index 6b36246932..cc14ea4b3c 100644 --- a/tests/unit_tests/client/test_plan_cache.py +++ b/tests/unit_tests/client/test_plan_cache.py @@ -145,6 +145,7 @@ def test_plan_empty_fallback_help_text(client): def test_plan_param_forwarding(args, kwargs): client = Mock() client.instrument_session = "cm12345-1" + client.subdirectory = None plan = Plan("foobar", FULL_PLAN, client) plan(*args, **kwargs) @@ -158,6 +159,7 @@ def test_plan_param_forwarding(args, kwargs): def test_plan_invalid_param_mapping(): client = Mock() client.instrument_session = "cm12345-1" + client.subdirectory = None client.run_task.side_effect = InvalidParametersError( [ ParameterError( From 2a4d6506a902ded8f122a1366bdec221ae492277 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Wed, 11 Nov 2026 10:57:34 +0000 Subject: [PATCH 4/6] From 048d758cb74816dfb5c4db5f7c40c2bbdb2a8bb3 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Thu, 12 Nov 2026 08:55:20 +0000 Subject: [PATCH 5/6] Add review changes --- src/blueapi/cli/cli.py | 2 +- src/blueapi/service/numtracker.py | 45 ++++++++++++++++--------------- 2 files changed, 24 insertions(+), 23 deletions(-) diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index e2b504665e..133433d605 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -406,7 +406,7 @@ def on_event( required=True, ) @click.option( - "-sub", + "-s", "--subdirectory", type=str, help=textwrap.dedent(""" diff --git a/src/blueapi/service/numtracker.py b/src/blueapi/service/numtracker.py index dfbea5bd65..79ea689226 100644 --- a/src/blueapi/service/numtracker.py +++ b/src/blueapi/service/numtracker.py @@ -1,7 +1,6 @@ import logging from collections.abc import Mapping from pathlib import Path -from textwrap import dedent import httpx from pydantic import Field, HttpUrl @@ -12,26 +11,28 @@ CREATE_SCAN_MUTATION = """ - mutation( - $instrument: String!, - $instrumentSession: String!, - $sub: Subdirectory - ){ - scan( - instrument: $instrument, - instrumentSession: $instrumentSession, - sub: $sub - ) { - directory{ - instrumentSession - instrument - path - } - scanFile - scanNumber - } - } - """ +mutation( + $instrument: String!, + $instrumentSession: String!, + $sub: Subdirectory +){ + scan( + instrument: $instrument, + instrumentSession: $instrumentSession, + sub: $sub + ) { + directory{ + instrumentSession + instrument + path + } + scanFile + scanNumber + } +} +""" +# Collapse whitespace so the query is sent compactly; safe as it has no string literals +CREATE_SCAN_MUTATION = " ".join(CREATE_SCAN_MUTATION.split()) class DirectoryPath(BlueapiBaseModel): @@ -102,7 +103,7 @@ async def create_scan( """ query = { - "query": dedent(CREATE_SCAN_MUTATION), + "query": CREATE_SCAN_MUTATION, "variables": { "instrument": instrument, "instrumentSession": instrument_session, From e5db6bf0072aeb9bc2fdb2a712b5cd9bee767e91 Mon Sep 17 00:00:00 2001 From: Zoheb Shaikh <26975142+ZohebShaikh@users.noreply.github.com> Date: Fri, 13 Nov 2026 19:30:43 +0000 Subject: [PATCH 6/6] wip: try to figure out how to use numtracker correctly --- src/blueapi/utils/path_provider.py | 24 ++++++++++++----- tests/system_tests/test_blueapi_system.py | 6 +++++ tests/unit_tests/utils/test_path_provider.py | 27 ++++++++++++++++++++ 3 files changed, 50 insertions(+), 7 deletions(-) diff --git a/src/blueapi/utils/path_provider.py b/src/blueapi/utils/path_provider.py index f16d68d1ec..7edab7932f 100644 --- a/src/blueapi/utils/path_provider.py +++ b/src/blueapi/utils/path_provider.py @@ -38,22 +38,32 @@ def __call__(self, device_name: str | None = None) -> PathInfo: example: "detector_file_template": "custom-{device_name}-{scan_id}". If you do not provide a data_session_directory it will default to "/tmp". + + If the start document has a scan_file in a subdirectory of the data session + directory (e.g. "sample1/ixx-1234"), detector files are written alongside it + in that subdirectory, which is created if it does not exist. """ if not self._docs: raise BlueskyRunStructureError( "Start document not found. This call must be made inside a run." ) else: - template = self._docs[-1].get("detector_file_template") + doc = self._docs[-1] + template = doc.get("detector_file_template") if not template: raise ValueError("detector_file_template must be set in metadata") - sub_path = template.format_map( - self._docs[-1] | {"device_name": device_name} - ) - data_session_directory = Path( - self._docs[-1].get("data_session_directory", "/tmp") + sub_path = template.format_map(doc | {"device_name": device_name}) + data_session_directory = Path(doc.get("data_session_directory", "/tmp")) + # numtracker's scan_file is relative to data_session_directory, write + # detector files in the same directory as it ("." if there is no + # scan_file or it has no subdirectory) + scan_file_directory = Path(doc.get("scan_file", "")).parent + return PathInfo( + directory_path=data_session_directory / scan_file_directory, + filename=sub_path, + # Negative depth creates only the directories scan_file adds + create_dir_depth=-len(scan_file_directory.parts), ) - return PathInfo(directory_path=data_session_directory, filename=sub_path) class BlueskyRunStructureError(Exception): diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 41d7800baf..58b8b8dc47 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -681,10 +681,13 @@ def on_event(event: AnyEvent) -> None: @pytest.mark.usefixtures("numtracker_with_subdirectory") def test_plan_runs_with_subdirectory(client_with_stomp: BlueapiClient, scan_id: int): start = Queue(maxsize=1) + stream_resources = Queue() def on_event(event: AnyEvent) -> None: if isinstance(event, DataEvent) and event.name == "start": start.put_nowait(event.doc) + if isinstance(event, DataEvent) and event.name == "stream_resource": + stream_resources.put_nowait(event.doc) cb_id = client_with_stomp.add_callback(on_event) try: @@ -700,6 +703,9 @@ def on_event(event: AnyEvent) -> None: assert start_doc["data_session_subdirectory"] == "sample1" assert start_doc["data_session_directory"] == "/tmp" assert start_doc["scan_file"] == f"sample1/adsim-{scan_id}" + # Detector files are written alongside the scan file, in the subdirectory + resource = stream_resources.get_nowait() + assert resource["uri"].endswith(f"/tmp/sample1/adsim-{scan_id}-det.h5") @pytest.mark.parametrize( diff --git a/tests/unit_tests/utils/test_path_provider.py b/tests/unit_tests/utils/test_path_provider.py index cd5934e167..0e01f2bbb0 100644 --- a/tests/unit_tests/utils/test_path_provider.py +++ b/tests/unit_tests/utils/test_path_provider.py @@ -52,6 +52,33 @@ def test_start_document_path_provider_with_default_template_returns_correct_path ) +@pytest.mark.parametrize( + "scan_file,directory,create_dir_depth", + [ + ("p01-22", "/p01/ab123", 0), + ("sample1/p01-22", "/p01/ab123/sample1", -1), + ("sample1/calibration/p01-22", "/p01/ab123/sample1/calibration", -2), + ], +) +def test_start_document_path_provider_writes_alongside_scan_file( + start_doc_default_template: dict, + scan_file: str, + directory: str, + create_dir_depth: int, +): + pp = StartDocumentPathProvider() + pp.run_start( + name="start", + start_document=start_doc_default_template | {"scan_file": scan_file}, # type: ignore + ) + + assert pp("det") == PathInfo( + directory_path=PosixPath(directory), + filename="p01-22-det", + create_dir_depth=create_dir_depth, + ) + + @pytest.fixture def start_doc_custom_template() -> dict: return {