spacr.workspace

Save and restore the interactive workspace associated with a run.

spacr.run_journal records pipeline inputs, settings, environment data, hashes, and logs. This module complements that record with interactive state: attached databases, generated montages, figure views, and the files those panels reference. The workspace is stored beside the run:

~/.spacr/runs/<run>/workspace.json
~/.spacr/runs/<run>/workspace/files/<digest>__<name>   # 'copy' mode only

Public API:

from spacr import workspace

workspace.register("volcano", lambda: the_panel)     # GUI, once
doc = workspace.collect(workspace.providers())
workspace.save(run_dir, doc, mode="reference")

doc = workspace.load(run_dir)
report = workspace.restore(workspace.providers(), doc)

Each registered provider supplies its own state through workspace_state() and restores it through apply_workspace_state(). The collector does not inspect widget internals. Existing plot_state and apply_plot_state methods are accepted for compatibility.

Every referenced file is recorded with its size, modification time, and SHA-256 digest. reference mode stores only this inventory; copy mode also carries files up to the configured per-file limit. A provider can mark an individual artifact for copying regardless of mode or size. Skipped files and copy failures remain visible in the document and restore report.

The module uses only the Python standard library so headless pipelines can record workspace state without importing pandas or Qt.

Attributes

MODES	Supported workspace persistence modes.
SCHEMA_VERSION	Schema version written to workspace.json.

Functions

check_files(→ List[Dict[str, Any]])	Check the current state of every file in a workspace document.
clear_providers(→ None)	Remove all providers and restore the built-in defaults.
collect(→ Dict[str, Any])	Build a workspace document without writing it to disk.
copy_limit_from_settings(→ float)	Return the non-negative per-file copy limit in megabytes.
default_copy_limit_mb(→ float)	Return the process-wide per-file copy limit in megabytes.
default_mode(→ str)	Return the process-wide workspace mode.
has_workspace(→ bool)	Return whether run_dir contains workspace.json.
hash_file(→ Optional[str])	A file's full SHA-256, or None if it cannot be read.
inventory(→ List[Dict[str, Any]])	Build metadata records for files referenced by workspace sections.
inventory_text(→ str)	Format a workspace document as a human-readable inventory.
load(→ Optional[Dict[str, Any]])	Read a workspace document from a run folder or document path.
mode_from_settings(→ str)	Return the workspace mode requested by a settings mapping.
providers(→ Dict[str, Callable[[], Any]])	Return a snapshot of the registered workspace providers.
register(→ None)	Register a named workspace-state provider.
report_text(→ str)	Format a restore report as human-readable text.
resolve_mode(→ str)	Normalize a workspace-saving mode.
restore(→ Dict[str, Any])	Restore each workspace section through its registered provider.
save(→ Optional[pathlib.Path])	Write a workspace document and optionally copy referenced files.
save_for_run(→ Optional[pathlib.Path])	Collect the registered workspace and write it beside a finished run.
section_states(→ Tuple[Dict[str, Any], List[Dict[str, ...)	Collect one JSON-compatible state section from each provider.
set_default_mode(→ str)	Set the process-wide workspace defaults.
unregister(→ bool)	Remove a provider and return whether it was registered.

Module Contents

spacr.workspace.check_files(doc: Mapping[str, Any], *, run_dir: Any = None) → List[Dict[str, Any]]

Check the current state of every file in a workspace document.

state is one of present (there, and the same bytes), changed (there, different digest), carried (gone from its original place but inside the bundle), or missing.

Parameters:

• doc – decoded workspace document.

• run_dir – bundle root used to resolve carried files.

Returns:

one state record per referenced file.

spacr.workspace.clear_providers() → None

Remove all providers and restore the built-in defaults.

This resets process state during application shutdown and test teardown.

spacr.workspace.collect(contributors: Mapping[str, Any], *, app_key: str = '', saved: str = '', hash_files: bool = True) → Dict[str, Any]

Build a workspace document without writing it to disk.

Parameters:

• contributors – named providers or state-bearing objects.

• app_key – application key associated with the workspace.

• saved – the timestamp to stamp, ISO-8601. Injected rather than read from the clock so a caller can produce a byte-identical document twice, which is what makes the writer testable.

• hash_files – compute file digests for the inventory when true.

Returns:

JSON-compatible workspace document.

spacr.workspace.copy_limit_from_settings(settings: Mapping[str, Any]) → float

Return the non-negative per-file copy limit in megabytes.

Parameters:

settings – run settings that may declare a workspace copy limit.

spacr.workspace.default_copy_limit_mb() → float

Return the process-wide per-file copy limit in megabytes.

spacr.workspace.default_mode() → str

Return the process-wide workspace mode.

spacr.workspace.has_workspace(run_dir: Any) → bool

Return whether run_dir contains workspace.json.

Parameters:

run_dir – run directory to inspect.

spacr.workspace.hash_file(path: pathlib.Path, chunk_size: int = 1 << 20) → str | None

A file’s full SHA-256, or None if it cannot be read.

Parameters:

path – regular file whose bytes are hashed.

spacr.workspace.inventory(sections: Mapping[str, Any], *, hash_files: bool = True) → List[Dict[str, Any]]

Build metadata records for files referenced by workspace sections.

String values that resolve to existing paths are discovered recursively. A section may also list files under the reserved carry key to request that their bytes be included in the bundle.

Parameters:

• sections – workspace state keyed by section name.

• hash_files – compute SHA-256 for regular files when true.

Returns:

deterministic file records ordered by source path.

spacr.workspace.inventory_text(doc: Mapping[str, Any], *, run_dir: Any = None) → str

Format a workspace document as a human-readable inventory.

Parameters:

doc – decoded workspace document to summarize.

spacr.workspace.load(run_dir: Any) → Dict[str, Any] | None

Read a workspace document from a run folder or document path.

Parameters:

run_dir – run directory or workspace document path to read.

Returns:

the decoded document, or None when it is absent or invalid.

spacr.workspace.mode_from_settings(settings: Mapping[str, Any]) → str

Return the workspace mode requested by a settings mapping.

Parameters:

settings – run settings that may declare save_workspace.

spacr.workspace.providers() → Dict[str, Callable[[], Any]]

Return a snapshot of the registered workspace providers.

spacr.workspace.register(name: str, provider: Callable[[], Any]) → None

Register a named workspace-state provider.

Parameters:

• name – stable section name written into workspace.json.

• provider – zero-argument callable returning either a mapping or an object with workspace_state(). Resolving the object at collection time avoids retaining a widget that has since been rebuilt or closed.

Raises:

• ValueError – if name is empty.

• TypeError – if provider is not callable.

spacr.workspace.report_text(report: Mapping[str, Any]) → str

Format a restore report as human-readable text.

Parameters:

report – workspace restoration report to summarize.

spacr.workspace.resolve_mode(value: Any) → str

Normalize a workspace-saving mode.

Parameters:

value – mode name, boolean, common yes/no alias, or None.

None and unrecognized values select DEFAULT_MODE; boolean and common yes/no aliases are accepted. Invalid values do not stop a run.

spacr.workspace.restore(contributors: Mapping[str, Any], doc: Mapping[str, Any], *, run_dir: Any = None) → Dict[str, Any]

Restore each workspace section through its registered provider.

Restoration is best-effort. Missing panels, unsupported state, exceptions, and provider refusals are reported under skipped rather than silently discarded.

Parameters:

• contributors – providers keyed by workspace section name.

• doc – decoded workspace document.

• run_dir – optional bundle root used to resolve carried files.

Returns:

report containing restored sections, skipped sections, and file states.

spacr.workspace.save(run_dir: Any, doc: Mapping[str, Any], *, mode: str = DEFAULT_MODE, copy_limit_mb: float = DEFAULT_COPY_LIMIT_MB) → pathlib.Path | None

Write a workspace document and optionally copy referenced files.

off writes nothing at all — not an empty document — so a run folder saved with the feature disabled is byte-for-byte what it was before this module existed.

Parameters:

• run_dir – destination run folder.

• doc – document returned by collect().

• mode – 'off', 'reference', or 'copy'.

• copy_limit_mb – maximum size of each automatically copied file in MB. Explicitly carried artifacts are not limited.

Returns:

path to workspace.json, or None in off mode.

spacr.workspace.save_for_run(run_dir: Any, settings: Mapping[str, Any] | None = None, *, app_key: str = '', contributors: Mapping[str, Any] | None = None) → pathlib.Path | None

Collect the registered workspace and write it beside a finished run.

Nothing is written when saving is disabled or no provider supplies state. Headless commands therefore do not gain an empty workspace document.

Parameters:

• run_dir – destination run folder.

• settings – run settings controlling mode and copy limit.

• app_key – application key associated with the saved state.

• contributors – optional providers to use instead of the registry.

Returns:

path to workspace.json, or None when nothing was saved.

spacr.workspace.section_states(contributors: Mapping[str, Any]) → Tuple[Dict[str, Any], List[Dict[str, str]]]

Collect one JSON-compatible state section from each provider.

Parameters:

contributors – {name: provider-or-object}. A callable value is called; anything else is used directly.

Returns:

(sections, problems). A provider that raises or returns no usable state is omitted and described in problems; other sections are still collected.

spacr.workspace.set_default_mode(mode: Any, copy_limit_mb: Any = None) → str

Set the process-wide workspace defaults.

An explicit save_workspace value in run settings overrides this mode.

Parameters:

• mode – 'off', 'reference', or 'copy'. Common boolean and yes/no aliases are accepted by resolve_mode().

• copy_limit_mb – optional non-negative per-file copy limit in MB. Invalid or negative values leave the current limit unchanged.

Returns:

the normalized default mode.

spacr.workspace.unregister(name: str) → bool

Remove a provider and return whether it was registered.

Parameters:

name – registered workspace section name to remove.

spacr.workspace.MODES = ('off', 'reference', 'copy')

Supported workspace persistence modes.

off

Do not write workspace metadata.

reference

Record referenced files and copy only artifacts explicitly marked for retention. This is the default.

copy

Also copy each recorded file within the configured per-file size limit.

spacr.workspace.SCHEMA_VERSION = 1

Schema version written to workspace.json.

Nested helpers

inventory.add(role: str, raw: str, carry: bool) → None

Add or promote one path in the captured inventory.

Parameters:

• role – source role recorded when the path is first seen.

• raw – path text to expand and inspect.

• carry – whether the file must be included in a bundle.

Returns:

None. Invalid and missing paths are ignored; repeated paths reuse their first record and are promoted when any occurrence requests carrying. File hashing follows the captured hash_files policy.

spacr/workspace.py:336