src.neurodags.outputs¶

Where outputs go and when a unit is done: output layout and .done manifests.

This module is the single place that maps a (source file, derivative) unit to files on disk (spec sections 4 and 6). Every reader and writer (execution, dry-run, status, dataframe collection, the TUI) goes through it, so all of them agree on one location and one definition of “done”.

Layout (spec 4.1)¶

For a source file discovered at path, rel is its path relative to the dataset’s source_root (POSIX separators, from neurodags.iterators). The reference base of the source is:

<derivatives_path>[/<derivatives_subdir>]/<dirname(rel)>/<basename(rel)>

or <dirname(path)>/<basename> when the dataset has no derivatives_path (outputs next to the source). The output prefix of a unit is <reference base>@<derivative name>; artifact files are prefix + suffix and markers prefix + ".done" | ".error" | ".skip". The source basename is kept verbatim (it may contain @, e.g. a chained source sub-1.fif@Clean.fif); the last @ of an output name separates source and derivative. Nothing here depends on which other files were discovered, on limits, indices or previous runs. See output_reference_base().

Completion (spec 6)¶

A unit is done when <prefix>.done is a valid manifest: it parses, its naming_version and derivative match, every file it lists exists, and every file still matches its recorded fingerprint (size and mtime_ns, compared exactly) when the manifest has fingerprints (see manifest_artifacts()). The manifest is written last and atomically, from the paths the artifact writers reported (writer_paths()), never from a glob. Files that match the prefix but have no valid manifest make the unit incomplete.

Atomic publication¶

Artifact writers never write at the final names. Each (re)computation writes into a hidden staging folder next to the outputs (make_staging_dir(), .<prefix basename>.tmp~<token>/, using the final basenames), then moves every file into place with os.replace (publish_staged()): continuation parts first, each main file last. A reader therefore sees either a complete old file or a complete new one, never a partially written one. A staging folder left by a killed process is ignored everywhere (discovery, status, dataframes, the legacy guard) and reported by status (UnitState.temps).

Attributes¶

`MANIFEST_VERSION`
`NAMING_VERSION`
`UnitStatus`

Exceptions¶

LegacyOutputsError

Outputs exist only in the pre-0.4.0 layout; neurodags migrate converts them.

Classes¶

UnitState

Completion state of one (source, derivative) unit; see unit_state().

Functions¶

`output_reference_base`(rel, source_path, derivatives_base) → str	Return the reference base of one source file (spec 4.1).
`dataset_derivatives_base`(dataset_config[, mount_point]) → str \| None	Return the resolved `derivatives_path[/derivatives_subdir]` of a dataset, or `None`.
`entry_reference_base`(entry, dataset_config[, mount_point]) → str	Absolute reference base of an enumerated input (`entry.rel`, `entry.path`).
`output_prefix`(reference_base, name) → str	Return `<reference_base>@<name>`, the prefix of every file of one unit.
`marker_path`(prefix, marker) → str	Return `prefix + "." + marker` for `done`, `error` or `skip`.
`check_output_collisions`(units) → None	Refuse two sources that map to the same reference base (spec 4.3).
`check_legacy_chained_outputs`(datasets, mount_point, names) → None	Refuse unmigrated 0.3.x outputs of sources whose basename contained `@`.
`writer_paths`(target, returned) → list[str]	Normalize what an artifact writer returned into the list of files it wrote.
`canonical_reports`(staging, paths) → list[str]	The files a writer reported, spelled as the directory entries they are in staging.
`file_identity`(path) → tuple[int, int]	`(size, mtime_ns)` of the entry path (a link itself, not followed).
`check_reports_unchanged`(staging, staged, identities) → None	Refuse files that were removed or changed after the writer that reported them returned.
`check_artifact_ownership`(written) → None	Refuse a file claimed by two artifacts of one unit.
`check_unreported_split_parts`(staging, written) → None	Refuse an MNE split recording of which a writer reported only some files.
`is_staging_name`(basename) → bool	Whether basename is a staging folder (`.<prefix basename>.tmp~<token>`, the
`looks_like_staging`(basename) → bool	Whether basename is a hidden folder named like a staging folder (`.…tmp~…`):
`staging_dirs`(prefix) → list[str]	Staging folders of one unit on disk (live ones of a running job, or left by a killed one).
`make_staging_dir`(prefix) → str	Create a fresh hidden staging folder for one (re)computation of a unit.
`remove_staging_dir`(path) → None	Delete a staging folder (after publishing or after a failure).
`publish_staged`(staging, prefix, staged) → dict[str, list[str]]	Move the files of every artifact from staging into place with `os.replace`.
`plan_publication`(staging, prefix, staged) → tuple[dict[str, list[str]], list[tuple[str, str]]]	The checks of `publish_staged()` (same errors) without moving anything: the final
`move_staged`(ordered[, moved]) → None	Move staged files into place with `os.replace`, in the given order; each target
`build_manifest`(prefix, name, *, dataset, rel, artifacts) → dict[str, Any]	Build the `.done` manifest of one unit (spec 6.1).
`file_fingerprint`(path) → tuple[int, int]	`(size, mtime_ns)` of a file, as recorded in a manifest's `fingerprints`.
`write_manifest`(prefix, manifest) → str	Publish `prefix + ".done"` atomically (temp file in the same folder + `os.replace`).
`read_manifest`(prefix) → dict[str, Any] \| None	Return the parsed `prefix + ".done"`, or `None` when absent or not a JSON object.
`manifest_artifacts`(prefix, name) → dict[str, list[str]] \| None	Cache hit check (spec 6.2): the unit's files per suffix, or `None`.
`unit_state`(prefix, name, *[, scan]) → UnitState	Classify one unit (spec 6.2): `skipped`, `done`, `error`, `incomplete`, `missing`.
`remove_markers`(prefix, *markers) → None	Delete the given markers (`done`, `error`, `skip`) of a unit if present.
`publish_marker`(prefix, marker, text) → str	Write a `.error` or `.skip` marker and remove the unit's other markers.

Module Contents¶

src.neurodags.outputs.MANIFEST_VERSION = 1¶

src.neurodags.outputs.NAMING_VERSION = 1¶

src.neurodags.outputs.UnitStatus¶

src.neurodags.outputs.output_reference_base(rel: str, source_path: str, derivatives_base: str | None) → str¶

Return the reference base of one source file (spec 4.1).

This is the one function that decides where a source file’s outputs go; append "@" + name (output_prefix()) for a unit’s prefix.

Parameters:

rel (str) – Path of the source relative to its dataset’s source_root, with / separators (neurodags.iterators.InputEntry.rel).
source_path (str) – The source file’s path on the active mount; used only when derivatives_base is None (outputs next to the source).
derivatives_base (str or None) – derivatives_path joined with the optional derivatives_subdir (dataset_derivatives_base()), or None.

Returns:

<derivatives_base>/<dirname(rel)>/<basename(rel)> or <dirname(source_path)>/<basename(rel)>, with the basename verbatim.

Return type:

str

Raises:

ValueError – If rel is empty, absolute or contains .././empty components.

src.neurodags.outputs.dataset_derivatives_base(dataset_config: Any, mount_point: Any = None) → str | None¶: Return the resolved derivatives_path[/derivatives_subdir] of a dataset, or None.

src.neurodags.outputs.entry_reference_base(entry: Any, dataset_config: Any, mount_point: Any = None) → str¶

Absolute reference base of an enumerated input (entry.rel, entry.path).

Absolute against the current working directory, so workers (which may run with a stale cwd) receive an already resolved path (#35).

src.neurodags.outputs.output_prefix(reference_base: str | os.PathLike, name: str) → str¶

Return <reference_base>@<name>, the prefix of every file of one unit.

The name is validated (neurodags.naming.output_prefix_basename()); the reference base’s basename is kept verbatim.

src.neurodags.outputs.marker_path(prefix: str, marker: str) → str¶: Return prefix + "." + marker for done, error or skip.

src.neurodags.outputs.check_output_collisions(units: collections.abc.Iterable[tuple[str, str]]) → None¶

Refuse two sources that map to the same reference base (spec 4.3).

Parameters:: units (iterable of (label, reference_base)) – One pair per enumerated source; label describes the source in the message (e.g. "study: sub-01/a@b.vhdr").
Raises:: DatasetConfigError – Listing every group of sources sharing a reference base (they would read and overwrite each other’s outputs).

exception src.neurodags.outputs.LegacyOutputsError(problems: collections.abc.Sequence[str])¶

Outputs exist only in the pre-0.4.0 layout; neurodags migrate converts them.

Initialize self. See help(type(self)) for accurate signature.

problems¶

src.neurodags.outputs.check_legacy_chained_outputs(datasets: collections.abc.Mapping[str, Any], mount_point: Any, names: collections.abc.Iterable[str], entries: collections.abc.Iterable[Any] = ()) → None¶

Refuse unmigrated 0.3.x outputs of sources whose basename contained @.

Before 0.4.0 every @ of a source basename was written as & (a chained source x@A.fif gave x&A.fif@B.nc); 0.4.0 keeps it verbatim (x@A.fif@B.nc). Such units would silently be recomputed, so this stops with a pointer to neurodags migrate.

One scan per active output tree: every derivatives_path (whole, so pilot-era flat files and folders from before a derivatives_subdir count; code/ and .neurodags/ skipped) and, for datasets writing next to their sources, the folders of the enumerated entries. A file is legacy when its name parses (last @) as an output of one of names (or its 0.3.x CamelCase token), its source part contains &, and nothing shows it belongs to a 0.4.0 unit:

no .done in the scanned trees has the same source part, in its own spelling or with @ written & (0.3.x never wrote .done; this also covers duplicates left by migrate), and
no source file of that exact spelling exists where the dataset’s source_root would hold it (a real source named with &, whatever is selected now).

Downstream outputs (x@Clean.fif@Fit.nc@Power.nc) never count: their source part holds no &.

Parameters:

datasets (mapping) – Dataset configurations (skip: true ones are ignored).
mount_point (str or None) – Active mount.
names (iterable of str) – Derivative names of the pipeline.
entries (iterable) – Enumerated inputs (dataset, path), for datasets without derivatives_path.

Raises:

LegacyOutputsError – Listing the files (a neurodags.utils.DatasetConfigError).

src.neurodags.outputs.writer_paths(target: str, returned: Any) → list[str]¶

Normalize what an artifact writer returned into the list of files it wrote.

Artifact.writer(path) returns the paths it wrote, or None meaning exactly target. A single path is accepted as a one-element list. Any other return value (for example the character count of open(path, "w").write(...)) is treated like None. Every reported file must exist afterwards. During a run, target lies in the unit’s staging folder (make_staging_dir()); publish_staged() then moves the reported files into place.

Returns:: Reported paths, relative ones taken relative to the current working directory like target itself, without duplicates, in the order reported.
Return type:: list of str
Raises:: RuntimeError – If a reported file does not exist (the writer did not write what it reported).

src.neurodags.outputs.canonical_reports(staging: str, paths: collections.abc.Sequence[str]) → list[str]¶

The files a writer reported, spelled as the directory entries they are in staging.

Each path is resolved (os.path.realpath(): folder links, .., //) and re-spelled below staging with every component exactly as the folder lists it, so the split-recording check (check_unreported_split_parts()) and publication see the same names. Hard links stay separate names. Duplicates are dropped, keeping the first.

Raises:: RuntimeError – If a reported file resolves outside staging, is itself a symbolic link (publishing its target would rename the file), or names an entry with other letter case than the folder lists (a case-insensitive filesystem).

src.neurodags.outputs.file_identity(path: str) → tuple[int, int]¶

(size, mtime_ns) of the entry path (a link itself, not followed).

Inode and device numbers are left out: some filesystems (FAT, CIFS noserverino, FUSE without use_ino) may renumber an unchanged file during a run.

src.neurodags.outputs.check_reports_unchanged(staging: str, staged: collections.abc.Mapping[str, collections.abc.Sequence[str]], identities: collections.abc.Mapping[str, collections.abc.Sequence[tuple[int, int]]]) → None¶

Refuse files that were removed or changed after the writer that reported them returned.

A later artifact’s writer can remove, move, replace (by a link, a hard link or a copy) or rewrite a file an earlier artifact reported; the .done would then certify a file its writer did not write (for example the first part of an MNE split recording replaced by its last part). Changes are detected by size and modification time, not by content: a same-size rewrite that keeps the modification time (restored, or within one tick of a coarse filesystem clock) is not seen.

Parameters:

staging (str) – The unit’s staging folder.
staged (mapping of suffix to paths) – Each artifact’s files, as canonical_reports() returned them after its writer.
identities (mapping of suffix to identities) – file_identity() of each of those files, taken right after its writer.

Raises:

RuntimeError – If a file is gone, is now a link, changed size or modification time, or is no longer reached under the same spelling inside staging.

src.neurodags.outputs.check_artifact_ownership(written: collections.abc.Mapping[str, collections.abc.Sequence[str]]) → None¶

Refuse a file claimed by two artifacts of one unit.

Two artifacts reporting the same file (for example .EO.fif, whose MNE split part is <prefix>.EO-1.fif, and an artifact .EO-1.fif) mean one overwrote the other, so the unit must not be certified done.

Parameters:: written (mapping of suffix to paths) – The files each artifact reported (or is about to write).
Raises:: RuntimeError – Naming every file claimed by more than one artifact.

src.neurodags.outputs.check_unreported_split_parts(staging: str, written: collections.abc.Mapping[str, collections.abc.Sequence[str]]) → None¶

Refuse an MNE split recording of which a writer reported only some files.

A writer that saves an MNE object larger than its split size and returns None (def w(path): raw.save(path)) reports only the first file; a writer that slices the list MNE returns reports only some. The files it left out would be discarded with the staging folder and the .done would certify a truncated recording.

The FIF files in staging are joined into recordings by the rules discovery uses (neurodags.iterators.find_split_continuations()), each one wherever it applies, so a file can be joined by both: continuations X-1.fif, X-2.fif, … (X.fif-1.gz, …) with each other and with their entry X.fif (X.fif.gz), and BIDS names that differ only in _split-NN_. The unit prefix is part of the name, so a source named ..._split-01_meg.fif makes the BIDS rule apply to its outputs too, and the neuromag rule then keeps the recording together. A recording with both reported and unreported files fails. written must hold the spellings canonical_reports() returns, the ones publication moves.

Parameters:

staging (str) – The unit’s staging folder.
written (mapping of suffix to paths) – The files each artifact reported.

Raises:

RuntimeError – Naming the unreported files and the artifacts their recording belongs to.

src.neurodags.outputs.is_staging_name(basename: str) → bool¶: Whether basename is a staging folder (.<prefix basename>.tmp~<token>, the name make_staging_dir() gives: an output prefix and a 16-hex-digit token).

src.neurodags.outputs.looks_like_staging(basename: str) → bool¶: Whether basename is a hidden folder named like a staging folder (.…tmp~…): never scanned as outputs. is_staging_name() decides what counts as one.

src.neurodags.outputs.staging_dirs(prefix: str) → list[str]¶: Staging folders of one unit on disk (live ones of a running job, or left by a killed one).

src.neurodags.outputs.make_staging_dir(prefix: str) → str¶

Create a fresh hidden staging folder for one (re)computation of a unit.

The folder is <dir>/.<prefix basename>.tmp~<token>: next to the outputs, so moving its files into place is a same-filesystem os.replace, and unique per computation, so concurrent computations of the same unit never write into each other’s files.

src.neurodags.outputs.remove_staging_dir(path: str) → None¶: Delete a staging folder (after publishing or after a failure).

src.neurodags.outputs.publish_staged(staging: str, prefix: str, staged: collections.abc.Mapping[str, collections.abc.Sequence[str]]) → dict[str, list[str]]¶

Move the files of every artifact from staging into place with os.replace.

Each staged file keeps its path relative to staging (normally just the final basename, e.g. x@Clean.fif and its MNE split part x@Clean-1.fif). Continuation files (everything after the first file an artifact reported) are moved first and each main file last, so a reader that finds a main file finds the parts it links to (MNE links split parts by basename, so moving them keeps them linked).

Parameters:

staging (str) – The staging folder (make_staging_dir()).
prefix (str) – The unit’s prefix; files land in its folder.
staged (mapping of suffix to paths) – The files each artifact writer reported, all inside staging (main file first).

Returns:

The final paths per suffix, in the reported order.

Return type:

dict of str to list of str

Raises:

RuntimeError – If a writer reported a file outside staging (writers must write into the folder of the path they are given; a file written elsewhere cannot be published atomically), or a file whose name is not owned by the unit (its first path component must start with the prefix basename, see _unit_namespace_problem()). Nothing is moved then.

src.neurodags.outputs.plan_publication(staging: str, prefix: str, staged: collections.abc.Mapping[str, collections.abc.Sequence[str]]) → tuple[dict[str, list[str]], list[tuple[str, str]]]¶: The checks of publish_staged() (same errors) without moving anything: the final paths per suffix and the (staged, final) moves in publication order (continuation files first, main files last), for move_staged().

src.neurodags.outputs.move_staged(ordered: collections.abc.Sequence[tuple[str, str]], moved: list[str] | None = None) → None¶: Move staged files into place with os.replace, in the given order; each target is appended to moved once it is in place.

src.neurodags.outputs.build_manifest(prefix: str, name: str, *, dataset: str | None, rel: str | None, artifacts: collections.abc.Mapping[str, collections.abc.Sequence[str]], companions: collections.abc.Sequence[str] = (), origin: str = 'run', fingerprints: collections.abc.Mapping[str, tuple[int, int]] | None = None) → dict[str, Any]¶

Build the .done manifest of one unit (spec 6.1).

Every listed file gets a fingerprint [size, mtime_ns] (fingerprints in the manifest, keyed by the listed name). A later change of a file (another process replacing it, a partial overwrite) makes the manifest invalid; see unit_state().

Parameters:

prefix (str) – The unit’s prefix (output_prefix()); the manifest lives at prefix + ".done" and lists files relative to its directory.
name (str) – Derivative name.
dataset (str or None) – The source: dataset key and path relative to source_root (original spelling, including any @).
rel (str or None) – The source: dataset key and path relative to source_root (original spelling, including any @).
artifacts (mapping of suffix to paths) – Every file each artifact consists of, as reported by its writer (main file first).
companions (sequence of str) – Other files belonging to the unit that no artifact suffix covers.
origin ({"run", "migrate"}) – Who wrote the manifest.
fingerprints (mapping of path to (size, mtime_ns), optional) – Fingerprints of the listed files, keyed by the paths given in artifacts / companions. A run passes those of the staged files, taken before they are moved into place (os.replace keeps size and mtime), so a file another process replaced in between does not match. Missing entries are taken from the files on disk now (what migrate certifies).

src.neurodags.outputs.file_fingerprint(path: str) → tuple[int, int]¶: (size, mtime_ns) of a file, as recorded in a manifest’s fingerprints.

src.neurodags.outputs.write_manifest(prefix: str, manifest: collections.abc.Mapping[str, Any]) → str¶

Publish prefix + ".done" atomically (temp file in the same folder + os.replace).

A crash before the replace leaves no .done (only a hidden temp file), so the unit is recomputed. Returns the manifest path.

src.neurodags.outputs.read_manifest(prefix: str) → dict[str, Any] | None¶: Return the parsed prefix + ".done", or None when absent or not a JSON object.

src.neurodags.outputs.manifest_artifacts(prefix: str, name: str) → dict[str, list[str]] | None¶

Cache hit check (spec 6.2): the unit’s files per suffix, or None.

The unit is done when prefix + ".done" parses, its naming_version and derivative match, every listed file exists and matches its fingerprint (when the manifest records fingerprints; a file modified after publication fails). Files matching the prefix but not listed are ignored.

Returns:: {suffix: [paths]} (the main file first, then e.g. MNE split parts), with paths joined onto the manifest’s directory; None when there is no valid manifest.
Return type:: dict of str to list of str, or None

class src.neurodags.outputs.UnitState¶

Completion state of one (source, derivative) unit; see unit_state().

state: UnitStatus¶

prefix: str¶

artifacts: dict[str, list[str]]¶

origin: str | None = None¶

problem: str | None = None¶

unlisted: tuple[str, ...] = ()¶

error_path: str | None = None¶

skip_path: str | None = None¶

temps: tuple[str, ...] = ()¶

src.neurodags.outputs.unit_state(prefix: str, name: str, *, scan: bool = True) → UnitState¶

Classify one unit (spec 6.2): skipped, done, error, incomplete, missing.

Precedence follows what a run does: a .skip marker wins (the unit is not recomputed), then a valid .done, then .error. Without any of those, files matching the prefix (or an invalid .done) make it incomplete; otherwise it is missing. With scan False the directory is not globbed (unlisted and temps stay empty and leftover files without a manifest count as missing). Staging folders (temps) are reported but never change the state.

src.neurodags.outputs.remove_markers(prefix: str, *markers: str) → None¶: Delete the given markers (done, error, skip) of a unit if present.

src.neurodags.outputs.publish_marker(prefix: str, marker: str, text: str) → str¶

Write a .error or .skip marker and remove the unit’s other markers.

A unit carries at most one of .done/.error/.skip. Returns the marker path.