Skip to content

redsun.path_provider

Where a session's acquisition files go.

Paths are <base_dir>/<session>/<YYYY-MM-DD>/<datakey>/<plan>_<counter>, base_dir defaulting to the user data directory. The container builds one provider per session and passes it to every device taking a path_provider keyword.

PATH_PROVIDER_PORT module-attribute 

PATH_PROVIDER_PORT: Final = 'path_provider'

Name the session's path provider is wired under.

SessionPathProvider 

SessionPathProvider(
    *,
    base_dir: Path | None = None,
    session: str = "unknown-session",
    max_digits: int = 5,
    now: Callable[[], datetime] | None = None,
)

Bases: PathProvider

Path provider for one session.

Paths have the layout:

base_dir / session_name / YYYY-MM-DD / datakey / <label>_<plan>_<counter>{.ext}

The label, set with set_label, is left out with its _ while empty.

The data key is the one the caller asks with, so each detector writes into a directory of its own and owns its store. A call without one leaves that level out.

The date is read on each request, not at construction. The counter belongs to (session, plan, datakey), starts one past the highest number already on disk, and only increases: date directories group files but do not reset the counter.

Parameters:

Name Type Description Default
base_dir Path | None

Base directory, with ~ expanded. Defaults to the user data directory, as session_directory gives it.

 None
session str

Session name, fixed for the provider's lifetime.

 'unknown-session'
max_digits int

Zero-padding width of the counter, as in 00001 for 5.

 5
now Callable[[], datetime] | None

Clock giving the date directory, for tests. Defaults to datetime.now.

 None

sig_base_dir_changed class-attribute instance-attribute 

sig_base_dir_changed = Signal(Path)

Emitted with the new root once set_base_dir accepted it.

base_dir property 

base_dir: Path

Root the session's files go under.

label property 

label: str

Label leading the file names of the next plan; empty for none.

session_dir property 

session_dir: Path

Directory inside base_dir holding the session's files.

Named after the session, each run of characters other than letters, digits, ., - and _ replaced by _ and outer dots removed.

set_base_dir 

set_base_dir(base_dir: str | Path) -> None

Change the base directory from the next request on, rescanning counters.

Raises:

Type Description
RuntimeError

If lock_base_dir was called, or a plan is running. A session not wiring set_plan and reset_plan cannot tell a plan is running.

lock_base_dir 

lock_base_dir(reason: str) -> None

Refuse every later set_base_dir, giving reason in the error.

set_label 

set_label(label: str) -> None

Lead the file names of every later plan with label.

Each run of characters other than letters, digits, ., - and _ becomes _, and a label of blanks only is no label. A plan already running keeps its names.

set_plan 

set_plan(plan: str) -> None

Change the active plan name, led by the label if there is one.

reset_plan 

reset_plan() -> None

Set the plan name back to its placeholder, rescanning counters.

A filename requested by the plan and never written returns to the pool, so the next plan continues from what is on disk.

__call__ 

__call__(datakey_name: str | None = None) -> PathInfo

Return the PathInfo of the next file for datakey_name.

Each call returns a new path, since it increments that data key's counter for the active plan. The folder of the path is created if it does not exist.

PlanFilenameProvider 

PlanFilenameProvider(*, max_digits: int = 5)

Bases: FilenameProvider

Filenames with a counter per plan.

Filenames are <plan_name>_<counter>, zero-padded to max_digits. Each data key counts on its own; its directory tells two detectors apart.

Parameters:

Name Type Description Default
max_digits int

Zero-padding width of the counter, as in 00001 for 5.

 5

plan property 

plan: str

The current plan name.

max_digits property 

max_digits: int

Zero-padding width of the counter.

set_plan 

set_plan(plan: str) -> None

Change the current plan name.

reset 

reset(counters: Mapping[tuple[str, str], int]) -> None

Replace every counter, as after a base directory change.

bump 

bump(plan: str, datakey_name: str, next_count: int) -> None

Raise the counter for a plan and data key to at least next_count.

__call__ 

__call__(datakey_name: str | None = None) -> str

Return the next, never repeated, filename for the plan and datakey_name.

session_directory 

session_directory(session: str) -> Path

Return session's directory under the default root.

Named as SessionPathProvider.session_dir names it.