How to choose where acquisition files go¶

Give a device the session's path provider, choose the folder the session writes under, and name the files after the plan that made them. Where a device writes explains the layout.

Prerequisites¶

You need a device that writes its own files and asks an ophyd-async PathProvider where to put them, as the file-writing detectors of ophyd-async do. The blocks below are parts of one script, and the whole script is at the end.

Take the path provider¶

Name a constructor parameter path_provider:

class MyCamera(StandardReadable):
    def __init__(self, path_provider: PathProvider, name: str = "") -> None:
        self.path_provider = path_provider
        super().__init__(name=name)

The session passes its SessionPathProvider to every device with that parameter, by keyword, so a device that takes it by position only is left out. You can't give path_provider yourself in a declaration or a session file, because the session refuses it. A device without the parameter chooses its own paths.

Each file then goes to

<base_dir>/<session>/<YYYY-MM-DD>/<datakey>/<plan>_<counter>

and the folder is created when the device asks for the path. The session also writes its log files under <base_dir>/logs/<session>.

Choose the folder¶

Set base_dir in the storage section of the session file:

session: my-lab
storage:
  base_dir: "D:/experiments/2026-09"

In a session class, put it in config:

config: ClassVar[dict[str, Any]] = {
    "session": "my-lab",
    "storage": {"base_dir": "~/experiments"},
}

Give an absolute folder, or one starting with ~, which the session expands. It accepts a relative folder here and refuses it only when a device asks for its first path. Without base_dir, the root is the redsun folder in your platform's user data folder.

The <session> folder is the session name, or the name of the session class when you give none. Each run of characters other than letters, digits, ., - and _ becomes _, so my lab/run writes into my_lab_run.

Name the files after the plan¶

Give the component that runs plans two signals: one it emits with the plan's name when a plan starts, and one it emits when the plan ends:

class MyController:
    sig_started = Signal(str)
    sig_finished = Signal()

    def __init__(
        self, name: str, *, devices: DeviceMapping, paths: SessionPathProvider
    ) -> None:
        self.name = name
        self.devices = devices
        self.paths = paths
        self.engine = RunEngine()

    @slot
    def snap(self) -> None:
        camera = self.devices["camera"]
        assert isinstance(camera, MyCamera)
        self.sig_started.emit("snap")
        future = self.engine(bp.count([camera]))
        future.add_done_callback(lambda _: self.sig_finished.emit())

In wire, link them to set_plan and reset_plan:

yield self.ctrl.sig_started, self.path_provider.set_plan
yield self.ctrl.sig_finished, self.path_provider.reset_plan

In a session file the provider is the component path_provider:

wiring:
  ctrl.sig_started: path_provider.set_plan
  ctrl.sig_finished: path_provider.reset_plan

Without these links, every file is named unknown_<counter>.

To change the width of the counter, five digits by default, set storage.max_digits. SessionPathProvider describes how the counter is kept.

Let the user change the folder¶

Give a view a signal carrying the folder the user chose:

class FolderView(QWidget):
    placement: Placement = Dock("left")
    sig_snap = Signal()
    sig_directory_chosen = Signal(str)

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        snap = QPushButton("Snap")
        snap.clicked.connect(lambda: self.sig_snap.emit())
        choose = QPushButton("Choose folder")
        choose.clicked.connect(self.choose_folder)
        layout = QVBoxLayout(self)
        layout.addWidget(snap)
        layout.addWidget(choose)

    def choose_folder(self) -> None:
        folder = QFileDialog.getExistingDirectory(self, "Folder for the files")
        if folder:
            self.sig_directory_chosen.emit(folder)

In wire, link it to set_base_dir:

yield self.folder_view.sig_directory_chosen, self.path_provider.set_base_dir

The next path is under the new folder, and the session's log files move there too. set_base_dir raises RuntimeError between set_plan and reset_plan. It also raises it in a session that keeps a catalog, because the catalog fixes the folder when it starts.

Find the folder from code¶

A component gets the provider by asking for it by type, as MyController above does with paths: SessionPathProvider, and paths.session_dir is <base_dir>/<session>. Outside a session, session_directory gives the same folder under the default root:

from redsun.path_provider import session_directory

session_directory("my-lab")

The example in full¶

The whole script

"""The session of the guide "How to choose where acquisition files go"."""

from __future__ import annotations

from collections.abc import Iterator  # noqa: TC003
from typing import Any, ClassVar

import bluesky.plans as bp
from ophyd_async.core import PathProvider, StandardReadable
from psygnal import Signal
from qtpy.QtWidgets import QFileDialog, QPushButton, QVBoxLayout, QWidget

from redsun import (
    AsDevice,
    AsPresenter,
    AsView,
    DeviceMapping,
    Link,
    Placement,
    slot,
)
from redsun.engine import RunEngine
from redsun.path_provider import SessionPathProvider  # noqa: TC001
from redsun.qt import Dock, QtSession


class MyCamera(StandardReadable):
    def __init__(self, path_provider: PathProvider, name: str = "") -> None:
        self.path_provider = path_provider
        super().__init__(name=name)


class MyController:
    sig_started = Signal(str)
    sig_finished = Signal()

    def __init__(
        self, name: str, *, devices: DeviceMapping, paths: SessionPathProvider
    ) -> None:
        self.name = name
        self.devices = devices
        self.paths = paths
        self.engine = RunEngine()

    @slot
    def snap(self) -> None:
        camera = self.devices["camera"]
        assert isinstance(camera, MyCamera)
        self.sig_started.emit("snap")
        future = self.engine(bp.count([camera]))
        future.add_done_callback(lambda _: self.sig_finished.emit())


class FolderView(QWidget):
    placement: Placement = Dock("left")
    sig_snap = Signal()
    sig_directory_chosen = Signal(str)

    def __init__(self, name: str, parent: QWidget) -> None:
        super().__init__(parent)
        self.name = name
        snap = QPushButton("Snap")
        snap.clicked.connect(lambda: self.sig_snap.emit())
        choose = QPushButton("Choose folder")
        choose.clicked.connect(self.choose_folder)
        layout = QVBoxLayout(self)
        layout.addWidget(snap)
        layout.addWidget(choose)

    def choose_folder(self) -> None:
        folder = QFileDialog.getExistingDirectory(self, "Folder for the files")
        if folder:
            self.sig_directory_chosen.emit(folder)


class MyApp(QtSession):
    config: ClassVar[dict[str, Any]] = {
        "session": "my-lab",
        "storage": {"base_dir": "~/experiments"},
    }
    camera: AsDevice[MyCamera]
    ctrl: AsPresenter[MyController]
    folder_view: AsView[FolderView]

    def wire(self) -> Iterator[Link]:
        yield self.folder_view.sig_snap, self.ctrl.snap
        yield self.ctrl.sig_started, self.path_provider.set_plan
        yield self.ctrl.sig_finished, self.path_provider.reset_plan
        yield self.folder_view.sig_directory_chosen, self.path_provider.set_base_dir


if __name__ == "__main__":
    MyApp().run()