picurv_cli.storage.operations Namespace Reference

Functions
list	restore_cold_study_members (str study_path, case_ids, str profile_name=None, str storage_config=None)
	Restore named cold study members in place from their local markers.
dict	build_storage_plan (dict target, dict profile, str compression=None, str policy=None, keep_latest_checkpoint=None, int workers=None, retain=None, drop=None)
	Build the read-only plan consumed by protect and offload.
None	_render_plan (dict plan)
	Print a concise archive/offload plan.
dict	archive_artifact (dict target, dict profile, str label=None, tags=None, str compression=None, bool prune_local=False, str policy=None, keep_latest_checkpoint=None, int workers=None, str notes=None, retain=None, drop=None)
	Package, upload, verify, register, and optionally prune one artifact.
dict	_register_existing_archive (dict target, dict profile, dict manifest, dict inventory, dict plan, *bool prune_local, str label=None, tags=None, str notes=None)
	Point an artifact at an already-uploaded archive of its current content.
None	_prune_archived_payload (str root, dict inventory, dict policy)
	Remove only verified heavy payload while retaining control-plane files.
None	_download_archive_components (dict profile, dict manifest, set components, str destination, int workers=None)
	Download and safely merge selected semantic archive components.
dict	restore_missing_workspace_assets (str workspace_root, provider_spec_hashes)
	Recover immutable workspace assets from archived run-local input copies.
dict	restore_archive (dict profile, str archive_id, str destination=None, checkpoints=None, bool force=False, int workers=None, components=None)
	Download, verify, extract, and materialize an archive or selected checkpoints.
list	_expand_checkpoint_selection (explicit, ranges)
	Combine repeated --checkpoint values with START:END[:STRIDE] ranges.
str	_resolve_archive_id_from_args (args)
	Resolve an explicit archive ID or a local artifact marker.
None	_render_status (dict inventory)
	Render one artifact status row and safety details.
dict	report_storage_configuration (dict profile)
	State which configuration answered, and warn when it came from outside.
dict	load_reported_storage_profile (args)
	Load the storage profile a command will act on, and report its origin.
None	storage_setup_workflow (args)
	Create or update a non-secret workspace storage profile.
None	storage_status_workflow (args)
	Print local storage and lifecycle status for selected artifacts.
None	storage_plan_workflow (args)
	Render a read-only packaging and safety plan.
None	storage_archive_workflow (args, bool prune_local)
	Execute protect or offload for one or more explicit local targets.
None	storage_restore_workflow (args)
	Restore a remote archive by globally unique ID or local marker.
None	storage_prune_workflow (args)
	Remove local asset objects nothing local needs and storage has verified.
None	storage_verify_workflow (args)
	Verify all remote chunks for one archive.
None	storage_list_workflow (args)
	Search the remote manifest catalog without local artifact state.
None	storage_show_workflow (args)
	Show one complete remote archive manifest.
None	storage_workflow (args)
	Dispatch nested storage actions using existing PICurv workflow conventions.
argparse.ArgumentParser	add_storage_parser (subparsers)
	Attach the nested storage command parser to PICurv's top-level parser.

Function Documentation

restore_cold_study_members()

list picurv_cli.storage.operations.restore_cold_study_members	(	str	study_path,
			case_ids,
		str	profile_name = None,
		str	storage_config = None
	)

Restore named cold study members in place from their local markers.

Parameters

[in]	study_path	Study directory owning the members.
[in]	case_ids	Member ids to restore.
[in]	profile_name	Optional storage profile name.
[in]	storage_config	Optional explicit storage configuration path.

Returns

Restored member ids.

Exceptions

StorageError

when a member carries no usable archive reference.

Definition at line 109 of file operations.py.

                               storage_config: str = None) -> list:
    """!
    @brief Restore named cold study members in place from their local markers.
    @param[in] study_path Study directory owning the members.
    @param[in] case_ids Member ids to restore.
    @param[in] profile_name Optional storage profile name.
    @param[in] storage_config Optional explicit storage configuration path.
    @return Restored member ids.
    @throws StorageError when a member carries no usable archive reference.
    """
    # This runs inside a staging workflow rather than a storage command, so the
    # configuration answering it has had no chance to be named yet.
    profile = report_storage_configuration(load_storage_profile(profile_name, storage_config))
    restored = []
    for case_id in case_ids:
        member = os.path.join(study_path, "cases", case_id)
        state = read_storage_state(member) or {}
        archive_id = state.get("archive_id")
        if not archive_id:
            raise StorageError(
                f"{case_id} is cold but its marker names no archive; restore it by id."
            )
        restore_archive(profile, archive_id, destination=member, force=True)
        restored.append(case_id)
    return restored
 
 

Here is the call graph for this function:

build_storage_plan()

dict picurv_cli.storage.operations.build_storage_plan	(	dict	target,
		dict	profile,
		str	compression = None,
		str	policy = None,
			keep_latest_checkpoint = None,
		int	workers = None,
			retain = None,
			drop = None
	)

Build the read-only plan consumed by protect and offload.

Parameters

[in]	target	Value supplied through the target argument.
[in]	profile	Value supplied through the profile argument.
[in]	compression	Value supplied through the compression argument.
[in]	policy	Optional semantic retention policy override.
[in]	keep_latest_checkpoint	Optional newest-checkpoint retention override.
[in]	workers	Optional compression worker count override.
[in]	retain	Components to retain locally regardless of the policy preset.
[in]	drop	Components to prune locally regardless of the policy preset.

Returns

Result produced by this operation.

Definition at line 137 of file operations.py.

                       workers: int = None, retain=None, drop=None) -> dict:
    """!
    @brief Build the read-only plan consumed by protect and offload.
    @param[in] target Value supplied through the `target` argument.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] compression Value supplied through the `compression` argument.
    @param[in] policy Optional semantic retention policy override.
    @param[in] keep_latest_checkpoint Optional newest-checkpoint retention override.
    @param[in] workers Optional compression worker count override.
    @param[in] retain Components to retain locally regardless of the policy preset.
    @param[in] drop Components to prune locally regardless of the policy preset.
    @return Result produced by this operation.
    """
    inventory = inspect_artifact(target)
    selected_compression = _select_compression(compression, inventory["total_bytes"], profile)
    specs = _build_chunk_specs(inventory, profile["chunk_size_bytes"])
    retention = _resolve_offload_policy(profile, policy, keep_latest_checkpoint, retain, drop)
    latest_step = max(inventory["checkpoint_steps"], default=None)
    retained_bytes = sum(
        entry["size"] for entry in inventory["entries"]
        if entry["type"] != "directory" and _entry_retained_by_policy(entry, retention, latest_step)
    )
    worker_count = int(workers or profile.get("workers", DEFAULT_STORAGE_WORKERS))
    if worker_count <= 0:
        raise StorageError("Storage workers must be positive.")
    compressed_range = _compression_size_range(inventory["total_bytes"], selected_compression)
    return {
        "inventory": inventory,
        "compression": selected_compression,
        "chunk_count": len(specs),
        "workers": worker_count,
        "offload_policy": retention,
        "retained_local_bytes": retained_bytes,
        "pruned_local_bytes": max(0, inventory["total_bytes"] - retained_bytes),
        "estimated_stored_bytes": {"low": compressed_range[0], "high": compressed_range[1]},
        "chunks": [
            {
                "component": spec["component"],
                "file_count": len(spec["entries"]),
                "uncompressed_bytes": spec["uncompressed_bytes"],
            }
            for spec in specs
        ],
    }
 
 

Here is the caller graph for this function:

_render_plan()

None picurv_cli.storage.operations._render_plan ( dict  plan )

protected

Print a concise archive/offload plan.

Parameters

[in]

plan

Value supplied through the plan argument.

Definition at line 185 of file operations.py.

def _render_plan(plan: dict) -> None:
    """!
    @brief Print a concise archive/offload plan.
    @param[in] plan Value supplied through the `plan` argument.
    """
    inventory = plan["inventory"]
    target = inventory["target"]
    print(f"[INFO] Artifact type : {target['artifact_type']}")
    print(f"[INFO] Artifact path : {target['root_path']}")
    print(f"[INFO] Local size    : {_human_bytes(inventory['total_bytes'])}")
    print(f"[INFO] Files         : {inventory['file_count']}")
    print(f"[INFO] Checkpoints   : {len(inventory['checkpoint_steps'])}")
    print(f"[INFO] Compression   : {plan['compression']}")
    print(f"[INFO] CPU workers   : {plan['workers']}")
    print(f"[INFO] Archive chunks: {plan['chunk_count']}")
    estimate = plan["estimated_stored_bytes"]
    print(
        f"[INFO] Remote estimate: {_human_bytes(estimate['low'])}–"
        f"{_human_bytes(estimate['high'])} (content-dependent)"
    )
    policy = plan["offload_policy"]
    print(f"[INFO] Offload policy: {policy['name']}")
    # The preset name alone no longer says what is kept once --retain/--drop adjust it,
    # so the resolved component set is reported rather than left to be inferred.
    print(f"[INFO] Kept components: {', '.join(policy['retained_components'])}")
    if policy.get("explicit_retain"):
        print(f"[INFO] Retained by flag: {', '.join(policy['explicit_retain'])}")
    if policy.get("explicit_drop"):
        print(f"[INFO] Dropped by flag: {', '.join(policy['explicit_drop'])}")
    print(f"[INFO] Retained local: {_human_bytes(plan['retained_local_bytes'])}")
    print(f"[INFO] Pruned local  : {_human_bytes(plan['pruned_local_bytes'])}")
    if "checkpoints" in policy["retained_components"]:
        print(f"[INFO] Kept checkpoint: every committed step ({len(inventory['checkpoint_steps'])})")
    elif policy["keep_latest_checkpoint"]:
        latest = max(inventory["checkpoint_steps"], default=None)
        print(f"[INFO] Kept checkpoint: {latest if latest is not None else 'none available'}")
    if inventory["external_paths"]:
        print("[WARNING] External configured paths are recorded but are not followed automatically:")
        for item in inventory["external_paths"]:
            print(f"  - {item['source']}: {item['path']}")
    if inventory["dependencies"]:
        print("[WARNING] External run dependencies:")
        for item in inventory["dependencies"]:
            print(f"  - {item['kind']}: {item['path']}")
    unclassified = [
        entry for entry in inventory["entries"]
        if entry["type"] != "directory" and entry["component"] == UNCLASSIFIED_COMPONENT
    ]
    if unclassified:
        print(
            f"[WARNING] {len(unclassified)} file(s) are not part of any known component. "
            "They are archived, and retained locally by every policy, because storage "
            "does not delete files whose purpose it cannot state:"
        )
        for entry in unclassified[:10]:
            print(f"  - {entry['path']}")
        if len(unclassified) > 10:
            print(f"  - ... and {len(unclassified) - 10} more")
 
 

archive_artifact()

dict picurv_cli.storage.operations.archive_artifact	(	dict	target,
		dict	profile,
		str	label = None,
			tags = None,
		str	compression = None,
		bool	prune_local = False,
		str	policy = None,
			keep_latest_checkpoint = None,
		int	workers = None,
		str	notes = None,
			retain = None,
			drop = None
	)

Package, upload, verify, register, and optionally prune one artifact.

Parameters

[in]	target	Value supplied through the target argument.
[in]	profile	Value supplied through the profile argument.
[in]	label	Value supplied through the label argument.
[in]	tags	Value supplied through the tags argument.
[in]	compression	Value supplied through the compression argument.
[in]	prune_local	Value supplied through the prune_local argument.
[in]	policy	Optional semantic retention policy override.
[in]	keep_latest_checkpoint	Optional newest-checkpoint retention override.
[in]	workers	Optional compression worker count override.
[in]	notes	Optional free-text note recorded with the archive.
[in]	retain	Components to retain locally regardless of the policy preset.
[in]	drop	Components to prune locally regardless of the policy preset.

Returns

Result produced by this operation.

Definition at line 245 of file operations.py.

                     workers: int = None, notes: str = None, retain=None, drop=None) -> dict:
    """!
    @brief Package, upload, verify, register, and optionally prune one artifact.
    @param[in] target Value supplied through the `target` argument.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] label Value supplied through the `label` argument.
    @param[in] tags Value supplied through the `tags` argument.
    @param[in] compression Value supplied through the `compression` argument.
    @param[in] prune_local Value supplied through the `prune_local` argument.
    @param[in] policy Optional semantic retention policy override.
    @param[in] keep_latest_checkpoint Optional newest-checkpoint retention override.
    @param[in] workers Optional compression worker count override.
    @param[in] notes Optional free-text note recorded with the archive.
    @param[in] retain Components to retain locally regardless of the policy preset.
    @param[in] drop Components to prune locally regardless of the policy preset.
    @return Result produced by this operation.
    """
    with storage_operation_lock(target["root_path"], "offload" if prune_local else "protect"):
        plan = build_storage_plan(
            target, profile, compression, policy=policy,
            keep_latest_checkpoint=keep_latest_checkpoint, workers=workers,
            retain=retain, drop=drop,
        )
        inventory = plan["inventory"]
        _assert_archive_safe(inventory)
        fingerprint = _inventory_fingerprint(inventory)
        # An explicitly requested compression must be honoured; without one, any
        # current archive of the artifact will do (protect, then a plain offload).
        reusable = _find_reusable_archive(
            profile, target, fingerprint,
            compression=plan["compression"] if compression else None,
        )
        if reusable is not None:
            # `protect` then `offload` is the documented workflow for "back it up now,
            # free the space later". Re-packaging and re-uploading an unchanged artifact
            # would store a second full copy of it, which for a campaign-sized run is
            # the most expensive thing this command can do for no benefit.
            print(
                f"[INFO] Reusing verified archive {reusable['archive_id']}: "
                f"{target['root_path']} is unchanged since it was archived."
            )
            return _register_existing_archive(
                target, profile, reusable, inventory, plan, prune_local=prune_local,
                label=label, tags=tags, notes=notes,
            )
        archive_id = uuid.uuid4().hex
        specs = _build_chunk_specs(inventory, profile["chunk_size_bytes"])
        staging_parent = profile.get("staging_directory")
        if staging_parent:
            staging_parent = os.path.abspath(os.path.expanduser(str(staging_parent)))
            os.makedirs(staging_parent, exist_ok=True)
        # As many chunks as `plan["workers"]` can be staged (packaged, not yet
        # uploaded) at once; each is deleted right after its verified upload, so
        # staging never needs archive-sized space, only room for the largest
        # concurrent handful of uncompressed chunks.
        concurrent_chunks = sorted(
            (spec["uncompressed_bytes"] for spec in specs), reverse=True
        )[:plan["workers"]]
        _require_free_space(
            staging_parent or tempfile.gettempdir(), sum(concurrent_chunks),
            f"stage {target['root_path']}",
        )
        print(f"[INFO] Creating archive {archive_id} for {target['root_path']}")
        with tempfile.TemporaryDirectory(prefix="picurv-storage-", dir=staging_parent) as staging:
            def package_and_upload(index_spec):
                """!
                @brief Package and verify one component chunk.
                @param[in] index_spec Tuple of chunk index and inventory specification.
                @return Uploaded chunk manifest entry.
                """
                index, spec = index_spec
                component = _safe_component_name(spec["component"])
                filename = f"{index:05d}_{component}{_chunk_extension(plan['compression'])}"
                local_chunk = os.path.join(staging, filename)
                print(
                    f"[INFO] Packaging chunk {index + 1}/{len(specs)}: "
                    f"{spec['component']} ({_human_bytes(spec['uncompressed_bytes'])})"
                )
                compressor = _write_tar_chunk(
                    target["root_path"], spec, local_chunk, plan["compression"],
                    workers=plan["workers"],
                )
                try:
                    digest = _sha256_file(local_chunk)
                    # Content-addressed: an identical chunk anywhere on this remote is
                    # already stored, so a rerun after a failed upload skips what
                    # succeeded, and two archives of the same checkpoint share one copy.
                    if _remote_blob_present(profile, digest):
                        print(f"[INFO] Chunk {index + 1}/{len(specs)} already stored; skipping upload.")
                        verified = {
                            "sha256": digest, "stored_bytes": os.path.getsize(local_chunk),
                        }
                    else:
                        verified = _transport._upload_verified(local_chunk, _blob_remote(profile, digest))
                    return {
                        "name": filename,
                        "component": spec["component"],
                        "file_count": len(spec["entries"]),
                        "uncompressed_bytes": spec["uncompressed_bytes"],
                        "stored_bytes": verified["stored_bytes"],
                        "sha256": verified["sha256"],
                        "compressor": compressor,
                        "blob": True,
                    }
                finally:
                    try:
                        os.remove(local_chunk)
                    except FileNotFoundError:
                        pass
 
            # pigz/xz already use every requested CPU on one large chunk. The
            # Python fallback instead gains parallelism across independent chunks.
            native_parallel = (
                plan["compression"] in _PARALLEL_GZIP_VALUES and shutil.which("pigz")
            ) or (plan["compression"] == "maximum" and shutil.which("xz"))
            task_workers = 1 if native_parallel else min(plan["workers"], max(1, len(specs)))
            if task_workers == 1:
                chunks = [package_and_upload(item) for item in enumerate(specs)]
            else:
                with concurrent.futures.ThreadPoolExecutor(max_workers=task_workers) as executor:
                    chunks = list(executor.map(package_and_upload, enumerate(specs)))
            chunks.sort(key=lambda item: item["name"])
 
            manifest = {
                "storage_schema_version": STORAGE_SCHEMA_VERSION,
                "archive_id": archive_id,
                "created_at": _utc_now(),
                "artifact_type": target["artifact_type"],
                "run_id": target.get("run_id"),
                "study_id": target.get("study_id"),
                "case_id": target.get("case_id"),
                "workspace_id": target.get("workspace_id"),
                "label": label or os.path.basename(target["root_path"]),
                "notes": notes,
                "tags": _parse_tags(tags),
                # What was actually solved, so `storage show` answers "which run was
                # this?" without restoring anything.
                "parameters": _capture_parameter_summary(target["root_path"]),
                "original_path": target["original_path"],
                "original_study_path": target.get("study_path"),
                "profile": profile["name"],
                "remote": profile["remote"],
                "source_bytes": inventory["total_bytes"],
                "source_file_count": inventory["file_count"],
                "inventory_sha256": fingerprint,
                "compression": plan["compression"],
                "workers": plan["workers"],
                "offload_policy": plan["offload_policy"],
                "checkpoint_format_version": KNOWN_CHECKPOINT_VERSION,
                "checkpoint_steps": inventory["checkpoint_steps"],
                "chunks": chunks,
                "config_sha256": _config_fingerprints(target["root_path"]),
                "git": _git_provenance(target["root_path"]),
                "external_paths": inventory["external_paths"],
                "dependencies": inventory["dependencies"],
                "study_context": _capture_study_context(target),
                "run_assets": _capture_run_assets(target["root_path"]),
                "workspace_assets": _capture_workspace_assets(target),
                "capabilities": {
                    "restorable": True,
                    "continuable": bool(inventory["checkpoint_steps"]),
                    "reprocessable": bool(inventory["checkpoint_steps"]),
                    "exact_binary_reproduction": False,
                },
            }
            manifest_path = os.path.join(staging, REMOTE_MANIFEST_FILENAME)
            _atomic_write_json(manifest_path, manifest)
            _transport._upload_verified(manifest_path, _object_remote(profile, archive_id, REMOTE_MANIFEST_FILENAME))
            manifest_digest = _sha256_file(manifest_path)
            complete_path = os.path.join(staging, REMOTE_COMPLETE_FILENAME)
            with open(complete_path, "w", encoding="ascii") as stream:
                stream.write(manifest_digest + "\n")
            _transport._upload_verified(complete_path, _object_remote(profile, archive_id, REMOTE_COMPLETE_FILENAME))
 
        state = {
            "storage_schema_version": STORAGE_SCHEMA_VERSION,
            "archive_id": archive_id,
            "profile": profile["name"],
            "remote": profile["remote"],
            "label": manifest["label"],
            "archived_at": manifest["created_at"],
            "local_pruned": False,
            "restored_components": [],
            "retained_components": [],
        }
        _atomic_write_json(_state_path(target["root_path"]), state)
        if prune_local:
            _prune_archived_payload(target["root_path"], inventory, plan["offload_policy"])
            state["local_pruned"] = True
            state["pruned_at"] = _utc_now()
            state["offload_policy"] = plan["offload_policy"]
            latest_step = max(inventory["checkpoint_steps"], default=None)
            retained = list(plan["offload_policy"]["retained_components"])
            if plan["offload_policy"]["keep_latest_checkpoint"] and latest_step is not None:
                retained.append(f"checkpoint:{latest_step}")
            state["retained_components"] = retained
            _atomic_write_json(_state_path(target["root_path"]), state)
        print(
            f"[SUCCESS] {'Offloaded' if prune_local else 'Protected'} {target['root_path']} "
            f"as archive {archive_id}."
        )
        return manifest
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_register_existing_archive()

dict picurv_cli.storage.operations._register_existing_archive ( dict  target, dict  profile, dict  manifest, dict  inventory, dict  plan, *bool  prune_local, str   label = None,   tags = None, str   notes = None  )

protected

Point an artifact at an already-uploaded archive of its current content.

Used when nothing has changed since a previous protect or offload. The remote object is immutable and is not rewritten; only this artifact's local marker is updated, and a label or tags supplied now are recorded there so the second invocation is not silently less descriptive than the first.

Parameters

[in]	target	Local artifact target.
[in]	profile	Resolved storage profile.
[in]	manifest	Remote manifest being reused.
[in]	inventory	Current local inventory.
[in]	plan	Storage plan carrying the requested offload policy.
[in]	prune_local	Whether to prune the verified payload afterwards.
[in]	label	Optional label supplied to this invocation.
[in]	tags	Optional tags supplied to this invocation.
[in]	notes	Optional free-text note supplied to this invocation.

Returns

The reused remote manifest.

Definition at line 452 of file operations.py.

                               label: str = None, tags=None, notes: str = None) -> dict:
    """!
    @brief Point an artifact at an already-uploaded archive of its current content.
 
    @details Used when nothing has changed since a previous `protect` or `offload`. The
             remote object is immutable and is not rewritten; only this artifact's local
             marker is updated, and a label or tags supplied now are recorded there so
             the second invocation is not silently less descriptive than the first.
    @param[in] target Local artifact target.
    @param[in] profile Resolved storage profile.
    @param[in] manifest Remote manifest being reused.
    @param[in] inventory Current local inventory.
    @param[in] plan Storage plan carrying the requested offload policy.
    @param[in] prune_local Whether to prune the verified payload afterwards.
    @param[in] label Optional label supplied to this invocation.
    @param[in] tags Optional tags supplied to this invocation.
    @param[in] notes Optional free-text note supplied to this invocation.
    @return The reused remote manifest.
    """
    state = {
        "storage_schema_version": STORAGE_SCHEMA_VERSION,
        "archive_id": manifest["archive_id"],
        "profile": profile["name"],
        "remote": profile["remote"],
        "label": label or manifest.get("label") or os.path.basename(target["root_path"]),
        "archived_at": manifest.get("created_at"),
        "reused_existing_archive": True,
        "local_pruned": False,
        "restored_components": [],
        "retained_components": [],
    }
    parsed_tags = _parse_tags(tags)
    if parsed_tags:
        state["tags"] = parsed_tags
    if notes:
        state["notes"] = notes
    _atomic_write_json(_state_path(target["root_path"]), state)
    if prune_local:
        _prune_archived_payload(target["root_path"], inventory, plan["offload_policy"])
        state["local_pruned"] = True
        state["pruned_at"] = _utc_now()
        state["offload_policy"] = plan["offload_policy"]
        latest_step = max(inventory["checkpoint_steps"], default=None)
        retained = list(plan["offload_policy"]["retained_components"])
        if plan["offload_policy"]["keep_latest_checkpoint"] and latest_step is not None:
            retained.append(f"checkpoint:{latest_step}")
        state["retained_components"] = retained
        _atomic_write_json(_state_path(target["root_path"]), state)
    print(
        f"[SUCCESS] {'Offloaded' if prune_local else 'Protected'} {target['root_path']} "
        f"using existing archive {manifest['archive_id']}."
    )
    return manifest
 
 

_prune_archived_payload()

None picurv_cli.storage.operations._prune_archived_payload ( str  root, dict  inventory, dict  policy  )

protected

Remove only verified heavy payload while retaining control-plane files.

Parameters

[in]	root	Value supplied through the root argument.
[in]	inventory	Value supplied through the inventory argument.
[in]	policy	Normalized semantic retention policy.

Definition at line 509 of file operations.py.

def _prune_archived_payload(root: str, inventory: dict, policy: dict) -> None:
    """!
    @brief Remove only verified heavy payload while retaining control-plane files.
    @param[in] root Value supplied through the `root` argument.
    @param[in] inventory Value supplied through the `inventory` argument.
    @param[in] policy Normalized semantic retention policy.
    """
    latest_step = max(inventory["checkpoint_steps"], default=None)
    for entry in sorted(inventory["entries"], key=lambda item: item["path"], reverse=True):
        if entry["type"] == "directory" or _entry_retained_by_policy(entry, policy, latest_step):
            continue
        path = os.path.join(root, *entry["path"].split("/"))
        try:
            if os.path.islink(path) or os.path.isfile(path):
                os.remove(path)
        except FileNotFoundError:
            pass
    for entry in sorted(
        (item for item in inventory["entries"] if item["type"] == "directory"),
        key=lambda item: item["path"].count("/"), reverse=True,
    ):
        path = os.path.join(root, *entry["path"].split("/"))
        try:
            os.rmdir(path)
        except OSError:
            pass
 
 

_download_archive_components()

None picurv_cli.storage.operations._download_archive_components ( dict  profile, dict  manifest, set  components, str  destination, int   workers = None  )

protected

Download and safely merge selected semantic archive components.

Parameters

[in]	profile	Active storage profile.
[in]	manifest	Verified remote archive manifest.
[in]	components	Semantic components to retrieve.
[in]	destination	Directory receiving merged content.
[in]	workers	Optional parallel worker count.

Returns

None.

Definition at line 537 of file operations.py.

                                 destination: str, workers: int = None) -> None:
    """!
    @brief Download and safely merge selected semantic archive components.
    @param[in] profile Active storage profile.
    @param[in] manifest Verified remote archive manifest.
    @param[in] components Semantic components to retrieve.
    @param[in] destination Directory receiving merged content.
    @param[in] workers Optional parallel worker count.
    @return None.
    """
    chunks = [
        chunk for chunk in manifest.get("chunks", [])
        if str(chunk.get("component", "")) in components
    ]
    if not chunks:
        raise StorageError(
            f"Archive {manifest.get('archive_id')} has none of the requested components: "
            + ", ".join(sorted(components))
        )
    worker_count = int(workers or profile.get("workers", DEFAULT_STORAGE_WORKERS))
    if worker_count <= 0:
        raise StorageError("Restore workers must be positive.")
    os.makedirs(destination, exist_ok=True)
    # Each chunk is extracted in full before `_merge_tree` copies it into `destination`,
    # so both copies exist on disk at once.
    extracted_bytes = sum(chunk.get("uncompressed_bytes", 0) for chunk in chunks)
    _require_free_space(
        destination, 2 * extracted_bytes, f"restore archive {manifest.get('archive_id')} components"
    )
    with tempfile.TemporaryDirectory(prefix="picurv-component-restore-") as staging:
        def download_and_extract(index_chunk):
            """!
            @brief Download, verify, and extract one selected chunk.
            @param[in] index_chunk Tuple of chunk index and manifest entry.
            @return Chunk index and extraction directory.
            """
            index, chunk = index_chunk
            chunk_root = os.path.join(staging, f"extract-{index:05d}")
            os.makedirs(chunk_root)
            local_chunk = os.path.join(staging, chunk["name"])
            _transport._run_rclone([
                "copyto",
                _chunk_remote_path(profile, manifest["archive_id"], chunk),
                local_chunk,
            ])
            actual = _sha256_file(local_chunk)
            if actual != chunk.get("sha256"):
                raise StorageError(
                    f"Downloaded chunk checksum mismatch: {chunk['name']} "
                    f"(expected {chunk.get('sha256')}, got {actual})."
                )
            _extract_chunk(local_chunk, chunk_root)
            return index, chunk_root
 
        task_workers = min(worker_count, len(chunks))
        if task_workers == 1:
            extracted = [download_and_extract(item) for item in enumerate(chunks)]
        else:
            with concurrent.futures.ThreadPoolExecutor(max_workers=task_workers) as executor:
                extracted = list(executor.map(download_and_extract, enumerate(chunks)))
        for _, chunk_root in sorted(extracted):
            _merge_tree(chunk_root, destination)
 
 

restore_missing_workspace_assets()

dict picurv_cli.storage.operations.restore_missing_workspace_assets	(	str	workspace_root,
			provider_spec_hashes
	)

Recover immutable workspace assets from archived run-local input copies.

Parameters

[in]	workspace_root	Initialized workspace receiving recovered objects.
[in]	provider_spec_hashes	Required provider-specification hashes.

Returns

Provider-specification hash to recovered asset-reference mapping.

Definition at line 602 of file operations.py.

def restore_missing_workspace_assets(workspace_root: str, provider_spec_hashes) -> dict:
    """!
    @brief Recover immutable workspace assets from archived run-local input copies.
    @param[in] workspace_root Initialized workspace receiving recovered objects.
    @param[in] provider_spec_hashes Required provider-specification hashes.
    @return Provider-specification hash to recovered asset-reference mapping.
    """
    requested = {str(value) for value in provider_spec_hashes if value}
    if not requested:
        return {}
    workspace = Path(os.path.abspath(workspace_root))
    (workspace / "assets").mkdir(parents=True, exist_ok=True)
    profile = load_storage_profile(config_path=str(workspace / STORAGE_CONFIG_FILENAME))
    matches = {}
    for manifest in sorted(
        list_remote_manifests(profile), key=lambda item: item.get("created_at", ""), reverse=True
    ):
        for reference in manifest.get("run_assets", []):
            spec_hash = reference.get("provider_spec_sha256")
            if spec_hash in requested and spec_hash not in matches:
                matches[spec_hash] = (manifest, reference)
        if requested <= set(matches):
            break
    if not matches:
        return {}
 
    restored = {}
    by_archive = {}
    for spec_hash, (manifest, reference) in matches.items():
        group = by_archive.setdefault(manifest["archive_id"], {"manifest": manifest, "items": []})
        group["items"].append((spec_hash, reference))
    kind_directories = {
        "grid": "grids",
        "initial-condition": "initial_conditions",
        "inlet-profiles": "inlet_profiles",
    }
    for group in by_archive.values():
        with tempfile.TemporaryDirectory(prefix=".asset-restore-", dir=workspace / "assets") as temporary:
            _download_archive_components(profile, group["manifest"], {"inputs"}, temporary)
            for spec_hash, reference in group["items"]:
                kind = reference.get("kind")
                asset_id = reference.get("asset_id")
                if kind not in kind_directories or not asset_id:
                    continue
                object_relative = f"assets/objects/{kind_directories[kind]}/{asset_id}"
                object_root = workspace / object_relative
                if not object_root.is_dir():
                    object_root.parent.mkdir(parents=True, exist_ok=True)
                    object_staging = Path(tempfile.mkdtemp(
                        prefix=f".{asset_id[:12]}-", dir=object_root.parent
                    ))
                    try:
                        files = []
                        for item in reference.get("files", []):
                            source = Path(temporary, *item["archived_path"].split("/"))
                            if not source.is_file() or (
                                item.get("sha256") and _sha256_file(str(source)) != item["sha256"]
                            ):
                                raise StorageError(
                                    f"Archived asset payload is missing or corrupt: {item['archived_path']}"
                                )
                            destination = object_staging / "payload" / item["path"]
                            destination.parent.mkdir(parents=True, exist_ok=True)
                            shutil.copy2(source, destination)
                            files.append({
                                "path": item["path"],
                                "bytes": destination.stat().st_size,
                                "sha256": _sha256_file(str(destination)),
                            })
                        _atomic_write_json(str(object_staging / "asset.json"), {
                            "schema_version": 1,
                            "asset_id": asset_id,
                            "kind": kind,
                            "provider": reference.get("provider"),
                            "provider_spec_sha256": spec_hash,
                            "recovered_from_archive": group["manifest"]["archive_id"],
                            "files": files,
                        })
                        try:
                            os.replace(object_staging, object_root)
                        except OSError as exc:
                            if exc.errno != errno.ENOTEMPTY and not object_root.is_dir():
                                raise
                    finally:
                        if object_staging.is_dir():
                            shutil.rmtree(object_staging, ignore_errors=True)
                restored[spec_hash] = {
                    "asset_id": asset_id,
                    "kind": kind,
                    "provider": reference.get("provider"),
                    "provider_spec_sha256": spec_hash,
                    "object": object_relative,
                    "files": reference.get("files", []),
                }
                print(f"[INFO] Restored shared asset {kind}: {asset_id}")
    return restored
 
 

restore_archive()

dict picurv_cli.storage.operations.restore_archive	(	dict	profile,
		str	archive_id,
		str	destination = None,
			checkpoints = None,
		bool	force = False,
		int	workers = None,
			components = None
	)

Download, verify, extract, and materialize an archive or selected checkpoints.

Parameters

[in]	profile	Value supplied through the profile argument.
[in]	archive_id	Value supplied through the archive_id argument.
[in]	destination	Value supplied through the destination argument.
[in]	checkpoints	Value supplied through the checkpoints argument.
[in]	force	Value supplied through the force argument.
[in]	workers	Optional parallel download and extraction worker count.
[in]	components	Optional semantic components to restore.

Returns

Result produced by this operation.

Definition at line 700 of file operations.py.

                    workers: int = None, components=None) -> dict:
    """!
    @brief Download, verify, extract, and materialize an archive or selected checkpoints.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] archive_id Value supplied through the `archive_id` argument.
    @param[in] destination Value supplied through the `destination` argument.
    @param[in] checkpoints Value supplied through the `checkpoints` argument.
    @param[in] force Value supplied through the `force` argument.
    @param[in] workers Optional parallel download and extraction worker count.
    @param[in] components Optional semantic components to restore.
    @return Result produced by this operation.
    """
    manifest = _load_remote_manifest(profile, archive_id)
    original = os.path.abspath(manifest["original_path"])
    destination_abs = os.path.abspath(destination or original)
    selected_steps = {int(step) for step in (checkpoints or [])}
    selected_components = {str(component) for component in (components or [])}
    chunks = []
    for chunk in manifest.get("chunks", []):
        component = str(chunk.get("component", ""))
        if selected_steps or selected_components:
            if component in ALWAYS_RESTORED_COMPONENTS:
                chunks.append(chunk)
            elif component.startswith("checkpoint:") and int(component.split(":", 1)[1]) in selected_steps:
                chunks.append(chunk)
            elif component in selected_components:
                chunks.append(chunk)
        else:
            chunks.append(chunk)
    if selected_steps:
        available = set(manifest.get("checkpoint_steps", []))
        missing = selected_steps - available
        if missing:
            raise StorageError(f"Archive {archive_id} does not contain checkpoint step(s): {sorted(missing)}")
    existing_state = read_storage_state(destination_abs) if os.path.isdir(destination_abs) else None
    if os.path.exists(destination_abs) and not force:
        if not existing_state or existing_state.get("archive_id") != archive_id:
            raise StorageError(
                f"Restore destination already exists and is not the matching cold artifact: {destination_abs}. "
                "Choose --to or use --force after verifying the destination."
            )
        # Components this artifact already holds locally - restored earlier or never
        # pruned - are not downloaded again; --force fetches everything.
        already_local = (set(existing_state.get("restored_components") or [])
                         | set(existing_state.get("retained_components") or []))
        def held_locally(component: str) -> bool:
            """!
            @brief Whether a chunk's component is already present in the destination.
            @param[in] component Chunk component name.
            @return True for a restored or retained component; a retained `checkpoints`
                    entry covers every committed step.
            """
            return component in already_local or (
                component.startswith("checkpoint:") and "checkpoints" in already_local)
 
        skipped = [chunk["component"] for chunk in chunks
                   if chunk.get("component") not in ALWAYS_RESTORED_COMPONENTS
                   and held_locally(str(chunk.get("component", "")))]
        if skipped:
            chunks = [chunk for chunk in chunks if chunk.get("component") not in skipped]
            print(f"[INFO] Already local, not downloaded again: {', '.join(skipped)}")
    parent = os.path.dirname(destination_abs)
    os.makedirs(parent, exist_ok=True)
    # Every selected chunk is extracted in full before `_merge_tree` copies it into
    # `materialized`, so both copies exist on disk at once ahead of the final move.
    extracted_bytes = sum(chunk.get("uncompressed_bytes", 0) for chunk in chunks)
    _require_free_space(parent, 2 * extracted_bytes, f"restore archive {archive_id}")
    temporary = tempfile.mkdtemp(prefix=f".picurv-restore-{archive_id[:8]}-", dir=parent)
    materialized = os.path.join(temporary, "materialized")
    os.makedirs(materialized)
    try:
        newly_restored = (
            [f"checkpoint:{step}" for step in sorted(selected_steps)]
            + sorted(selected_components)
        )
        prior_restored = (existing_state or {}).get("restored_components") or []
        worker_count = int(workers or profile.get("workers", DEFAULT_STORAGE_WORKERS))
        if worker_count <= 0:
            raise StorageError("Restore workers must be positive.")
 
        def download_and_extract(index_chunk):
            """!
            @brief Download, verify, and extract one archive chunk.
            @param[in] index_chunk Tuple of chunk index and manifest entry.
            @return Chunk index and extraction directory.
            """
            index, chunk = index_chunk
            chunk_root = os.path.join(temporary, f"extract-{index:05d}")
            os.makedirs(chunk_root)
            local_chunk = os.path.join(temporary, chunk["name"])
            print(f"[INFO] Restoring chunk {index + 1}/{len(chunks)}: {chunk['component']}")
            _transport._run_rclone([
                "copyto", _chunk_remote_path(profile, archive_id, chunk), local_chunk,
            ])
            actual = _sha256_file(local_chunk)
            if actual != chunk.get("sha256"):
                raise StorageError(
                    f"Downloaded chunk checksum mismatch: {chunk['name']} "
                    f"(expected {chunk.get('sha256')}, got {actual})."
                )
            _extract_chunk(local_chunk, chunk_root)
            os.remove(local_chunk)
            return index, chunk_root
 
        task_workers = min(worker_count, max(1, len(chunks)))
        if task_workers == 1:
            extracted = [download_and_extract(item) for item in enumerate(chunks)]
        else:
            with concurrent.futures.ThreadPoolExecutor(max_workers=task_workers) as executor:
                extracted = list(executor.map(download_and_extract, enumerate(chunks)))
        for _, chunk_root in sorted(extracted):
            _merge_tree(chunk_root, materialized)
 
        if os.path.isdir(destination_abs):
            _merge_tree(materialized, destination_abs)
        else:
            os.replace(materialized, destination_abs)
        _restore_study_context(manifest, destination_abs)
 
        replacements = [(original, destination_abs)]
        original_study = manifest.get("original_study_path")
        if original_study and manifest.get("artifact_type") == "study-case":
            replacements.append((os.path.abspath(original_study), str(Path(destination_abs).parent.parent)))
        rebased = _rebase_restored_text_paths(destination_abs, replacements)
        state = {
            "storage_schema_version": STORAGE_SCHEMA_VERSION,
            "archive_id": archive_id,
            "profile": profile["name"],
            "remote": profile["remote"],
            "label": manifest.get("label"),
            "archived_at": manifest.get("created_at"),
            "restored_at": _utc_now(),
            "local_pruned": bool(selected_steps or selected_components),
            "restored_components": sorted(set(prior_restored) | set(newly_restored)),
            "retained_components": (existing_state or {}).get("retained_components") or [],
            "relocated_from": original if original != destination_abs else None,
            "rebased_files": len(rebased),
        }
        _atomic_write_json(_state_path(destination_abs), state)
    except Exception:
        if os.path.isdir(temporary):
            shutil.rmtree(temporary, ignore_errors=True)
        raise
    if os.path.isdir(temporary):
        shutil.rmtree(temporary, ignore_errors=True)
    print(f"[SUCCESS] Restored archive {archive_id} to {destination_abs}.")
    if rebased:
        print(f"[INFO] Rebased {len(rebased)} generated text artifact(s) to the restored path.")
    return manifest
 
 

Here is the caller graph for this function:

_expand_checkpoint_selection()

list picurv_cli.storage.operations._expand_checkpoint_selection (   explicit,   ranges  )

protected

Combine repeated --checkpoint values with START:END[:STRIDE] ranges.

Parameters

[in]	explicit	Individually named steps, or None.
[in]	ranges	Range expressions, or None.

Returns

Sorted unique step list, or None when nothing was selected.

Definition at line 853 of file operations.py.

def _expand_checkpoint_selection(explicit, ranges) -> list:
    """!
    @brief Combine repeated `--checkpoint` values with `START:END[:STRIDE]` ranges.
    @param[in] explicit Individually named steps, or None.
    @param[in] ranges Range expressions, or None.
    @return Sorted unique step list, or None when nothing was selected.
    """
    selected = set(explicit or ())
    for expression in ranges or ():
        parts = str(expression).split(":")
        if len(parts) not in (2, 3):
            raise StorageError(
                f"--checkpoints expects START:END or START:END:STRIDE, got {expression!r}."
            )
        try:
            numbers = [int(part) for part in parts]
        except ValueError:
            raise StorageError(
                f"--checkpoints expects integer steps, got {expression!r}."
            ) from None
        start, end = numbers[0], numbers[1]
        stride = numbers[2] if len(numbers) == 3 else 1
        if stride <= 0:
            raise StorageError(f"--checkpoints stride must be positive, got {expression!r}.")
        if end < start:
            raise StorageError(f"--checkpoints end precedes start in {expression!r}.")
        selected.update(range(start, end + 1, stride))
    return sorted(selected) or None
 
 

_resolve_archive_id_from_args()

str picurv_cli.storage.operations._resolve_archive_id_from_args (   args )

protected

Resolve an explicit archive ID or a local artifact marker.

Parameters

[in]

args

Value supplied through the args argument.

Returns

Result produced by this operation.

Definition at line 883 of file operations.py.

def _resolve_archive_id_from_args(args) -> str:
    """!
    @brief Resolve an explicit archive ID or a local artifact marker.
    @param[in] args Value supplied through the `args` argument.
    @return Result produced by this operation.
    """
    archive_id = getattr(args, "archive_id", None)
    if archive_id:
        return archive_id
    workspace_id = getattr(args, "workspace_id", None)
    if workspace_id:
        return resolve_workspace_archive_id(
            load_storage_profile(
                getattr(args, "profile", None), getattr(args, "storage_config", None)
            ),
            workspace_id,
        )
    targets = resolve_local_storage_targets(
        getattr(args, "run_dir", None), getattr(args, "study_dir", None), getattr(args, "case_ids", None),
        workspace=getattr(args, "workspace", None),
    )
    if len(targets) != 1:
        raise StorageError("Restore/verify by local marker requires exactly one target.")
    state = read_storage_state(targets[0]["root_path"])
    if not state or not state.get("archive_id"):
        raise StorageError(f"No storage marker with an archive ID exists under {targets[0]['root_path']}.")
    return state["archive_id"]
 
 

_render_status()

None picurv_cli.storage.operations._render_status ( dict  inventory )

protected

Render one artifact status row and safety details.

Parameters

[in]

inventory

Value supplied through the inventory argument.

Definition at line 912 of file operations.py.

def _render_status(inventory: dict) -> None:
    """!
    @brief Render one artifact status row and safety details.
    @param[in] inventory Value supplied through the `inventory` argument.
    """
    target = inventory["target"]
    storage = inventory["storage"]
    activity = "BUSY" if inventory["active_locks"] or inventory["slurm"]["active"] else storage["state"]
    identity = (
        target.get("case_id") or target.get("run_id") or target.get("study_id")
        or target.get("workspace_id") or os.path.basename(target["root_path"])
    )
    print(
        f"{identity:<36} {target['artifact_type']:<12} {activity:<10} "
        f"{_human_bytes(inventory['total_bytes']):>12}  {storage.get('label') or ''}"
    )
 
 

report_storage_configuration()

dict picurv_cli.storage.operations.report_storage_configuration

(

dict

profile

)

State which configuration answered, and warn when it came from outside.

Every command below acts on the remote this file names, and offload prunes local payload once that remote has verified the upload. Discovery searches upward past the workspace boundary, so the file that answered is not always the one the user thinks they are standing in; naming it here is what keeps an inherited configuration a choice rather than an accident.

Parameters

[in]

profile

Resolved storage profile, carrying its config_path.

Returns

The profile, unchanged, so callers can wrap the load.

Definition at line 930 of file operations.py.

def report_storage_configuration(profile: dict) -> dict:
    """!
    @brief State which configuration answered, and warn when it came from outside.
 
    @details Every command below acts on the remote this file names, and `offload`
             prunes local payload once that remote has verified the upload. Discovery
             searches upward past the workspace boundary, so the file that answered is
             not always the one the user thinks they are standing in; naming it here is
             what keeps an inherited configuration a choice rather than an accident.
    @param[in] profile Resolved storage profile, carrying its `config_path`.
    @return The profile, unchanged, so callers can wrap the load.
    """
    config_path = profile.get("config_path")
    origin = storage_config_origin(config_path)
    print(f"[INFO] Storage config : {config_path}")
    print(f"[INFO] Storage profile: {profile['name']} -> {profile['remote']}")
    if origin == "shared":
        print(
            f"[WARNING] That configuration is outside this workspace; it is shared with "
            f"everything below {os.path.dirname(os.path.abspath(config_path))}. Pass "
            "--storage-config to choose another, or run 'picurv storage setup' here to "
            "give this workspace its own.",
            file=sys.stderr,
        )
    return profile
 
 

Here is the caller graph for this function:

load_reported_storage_profile()

dict picurv_cli.storage.operations.load_reported_storage_profile

(

args

)

Load the storage profile a command will act on, and report its origin.

Parameters

[in]

args

Parsed storage command arguments.

Returns

Resolved storage profile.

Definition at line 957 of file operations.py.

def load_reported_storage_profile(args) -> dict:
    """!
    @brief Load the storage profile a command will act on, and report its origin.
    @param[in] args Parsed storage command arguments.
    @return Resolved storage profile.
    """
    return report_storage_configuration(
        load_storage_profile(getattr(args, "profile", None),
                             getattr(args, "storage_config", None))
    )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_setup_workflow()

None picurv_cli.storage.operations.storage_setup_workflow

(

args

)

Create or update a non-secret workspace storage profile.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 969 of file operations.py.

def storage_setup_workflow(args) -> None:
    """!
    @brief Create or update a non-secret workspace storage profile.
    @param[in] args Value supplied through the `args` argument.
    """
    explicit_config = getattr(args, "storage_config", None)
    config_path = resolve_storage_config_path(explicit_config, require=False)
    workspace_root = storage_workspace_root()
    # Discovery walks past the workspace boundary, which is right for reading a shared
    # configuration and wrong for writing one: `setup` inside a fresh workspace would
    # otherwise rewrite the remote of the campaign directory above it, silently
    # re-pointing every other workspace under it. Configure the workspace being stood
    # in; editing a shared file stays available, but has to be asked for.
    if not explicit_config and workspace_root and storage_config_origin(config_path, workspace_root) == "shared":
        inherited = config_path
        config_path = os.path.join(workspace_root, STORAGE_CONFIG_FILENAME)
        print(
            f"[INFO] A shared configuration exists at {inherited}, but it belongs to a "
            "directory above this workspace."
        )
        print(
            f"       Writing this workspace's own configuration instead. To edit the "
            f"shared one, pass --storage-config {inherited}."
        )
    payload = {}
    if os.path.isfile(config_path):
        with open(config_path, "r", encoding="utf-8") as stream:
            payload = yaml.safe_load(stream) or {}
    profiles = payload.setdefault("profiles", {})
    profile_name = args.profile or DEFAULT_PROFILE_NAME
    profile = {
        "remote": args.remote.rstrip("/"),
        "compression": args.compression,
        "chunk_size_gib": args.chunk_size_gib,
        "workers": getattr(args, "workers", DEFAULT_STORAGE_WORKERS),
        "offload_policy": getattr(args, "offload_policy", "metadata-only"),
        "keep_latest_checkpoint": bool(getattr(args, "keep_latest_checkpoint", False)),
    }
    if args.staging_directory:
        profile["staging_directory"] = os.path.abspath(os.path.expanduser(args.staging_directory))
    payload["default_profile"] = profile_name
    profiles[profile_name] = profile
    print(f"[INFO] Storage config : {config_path}")
    print(f"[INFO] Profile        : {profile_name}")
    print(f"[INFO] Remote         : {profile['remote']}")
    if args.dry_run:
        print("[INFO] Dry-run only. No configuration or remote directories were changed.")
        return
    _transport._run_rclone(["mkdir", _remote_join(profile["remote"], REMOTE_OBJECTS_DIRECTORY)])
    os.makedirs(os.path.dirname(config_path), exist_ok=True)
    temporary = f"{config_path}.tmp.{os.getpid()}"
    with open(temporary, "w", encoding="utf-8") as stream:
        yaml.safe_dump(payload, stream, sort_keys=False)
    os.replace(temporary, config_path)
    print("[SUCCESS] Storage profile configured and remote access verified.")
 
 

Here is the caller graph for this function:

storage_status_workflow()

None picurv_cli.storage.operations.storage_status_workflow

(

args

)

Print local storage and lifecycle status for selected artifacts.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1026 of file operations.py.

def storage_status_workflow(args) -> None:
    """!
    @brief Print local storage and lifecycle status for selected artifacts.
    @param[in] args Value supplied through the `args` argument.
    """
    if args.study_dir and not args.case_ids:
        study_root = os.path.abspath(args.study_dir)
        if getattr(args, "completed", False):
            case_ids = _select_completed_case_ids(study_root)
        else:
            case_ids = [
                path.name for path in sorted((Path(study_root) / "cases").glob("case_*")) if path.is_dir()
            ]
        targets = resolve_local_storage_targets(None, study_root, case_ids) if case_ids else resolve_local_storage_targets(None, study_root)
    else:
        targets = resolve_local_storage_targets(
        args.run_dir, args.study_dir, args.case_ids,
        workspace=getattr(args, "workspace", None),
        include_inputs=getattr(args, "include_inputs", False),
        completed=getattr(args, "completed", False),
    )
    inventories = [inspect_artifact(target) for target in targets]
    if args.output_format == "json":
        serializable = []
        for item in inventories:
            copy_item = dict(item)
            copy_item.pop("entries", None)
            serializable.append(copy_item)
        print(json.dumps(serializable, indent=2, sort_keys=True))
        return
    print(f"{'ARTIFACT':<36} {'TYPE':<12} {'STATE':<10} {'LOCAL SIZE':>12}  LABEL")
    for inventory in inventories:
        _render_status(inventory)
 
 

Here is the caller graph for this function:

storage_plan_workflow()

None picurv_cli.storage.operations.storage_plan_workflow

(

args

)

Render a read-only packaging and safety plan.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1061 of file operations.py.

def storage_plan_workflow(args) -> None:
    """!
    @brief Render a read-only packaging and safety plan.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    targets = resolve_local_storage_targets(
        args.run_dir, args.study_dir, args.case_ids,
        workspace=getattr(args, "workspace", None),
        include_inputs=getattr(args, "include_inputs", False),
        completed=getattr(args, "completed", False),
    )
    for index, target in enumerate(targets):
        if index:
            print()
        plan = build_storage_plan(
            target, profile, args.compression,
            policy=getattr(args, "policy", None),
            keep_latest_checkpoint=getattr(args, "keep_latest_checkpoint", None),
            workers=getattr(args, "workers", None),
            retain=getattr(args, "retain", None), drop=getattr(args, "drop", None),
        )
        _render_plan(plan)
        _assert_archive_safe(plan["inventory"])
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_archive_workflow()

None picurv_cli.storage.operations.storage_archive_workflow	(		args,
		bool	prune_local
	)

Execute protect or offload for one or more explicit local targets.

Parameters

[in]	args	Value supplied through the args argument.
[in]	prune_local	Value supplied through the prune_local argument.

Definition at line 1087 of file operations.py.

def storage_archive_workflow(args, prune_local: bool) -> None:
    """!
    @brief Execute protect or offload for one or more explicit local targets.
    @param[in] args Value supplied through the `args` argument.
    @param[in] prune_local Value supplied through the `prune_local` argument.
    """
    profile = load_reported_storage_profile(args)
    targets = resolve_local_storage_targets(
        args.run_dir, args.study_dir, args.case_ids,
        workspace=getattr(args, "workspace", None),
        include_inputs=getattr(args, "include_inputs", False),
        completed=getattr(args, "completed", False),
    )
    for target in targets:
        if args.dry_run:
            _render_plan(build_storage_plan(
                target, profile, args.compression,
                policy=getattr(args, "policy", None),
                keep_latest_checkpoint=getattr(args, "keep_latest_checkpoint", None),
                workers=getattr(args, "workers", None),
                retain=getattr(args, "retain", None), drop=getattr(args, "drop", None),
            ))
            print("[INFO] Dry-run only. No files were packaged, uploaded, or pruned.")
            continue
        archive_artifact(
            target,
            profile,
            label=args.label,
            tags=args.tags,
            compression=args.compression,
            prune_local=prune_local,
            policy=getattr(args, "policy", None),
            keep_latest_checkpoint=getattr(args, "keep_latest_checkpoint", None),
            workers=getattr(args, "workers", None),
            notes=getattr(args, "notes", None),
            retain=getattr(args, "retain", None),
            drop=getattr(args, "drop", None),
        )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_restore_workflow()

None picurv_cli.storage.operations.storage_restore_workflow

(

args

)

Restore a remote archive by globally unique ID or local marker.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1127 of file operations.py.

def storage_restore_workflow(args) -> None:
    """!
    @brief Restore a remote archive by globally unique ID or local marker.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    archive_id = _resolve_archive_id_from_args(args)
    restore_archive(
        profile,
        archive_id,
        destination=args.destination,
        checkpoints=_expand_checkpoint_selection(
            args.checkpoints, getattr(args, "checkpoint_ranges", None)
        ),
        force=args.force,
        workers=getattr(args, "workers", None),
        components=getattr(args, "components", None),
    )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_prune_workflow()

None picurv_cli.storage.operations.storage_prune_workflow

(

args

)

Remove local asset objects nothing local needs and storage has verified.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1147 of file operations.py.

def storage_prune_workflow(args) -> None:
    """!
    @brief Remove local asset objects nothing local needs and storage has verified.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    workspace_root = os.path.abspath(args.workspace) if args.workspace else _find_upwards(
        os.getcwd(), WORKSPACE_CONFIG_FILENAME
    )
    if workspace_root and os.path.isfile(workspace_root):
        workspace_root = os.path.dirname(workspace_root)
    if not workspace_root:
        raise StorageError(
            "No initialized workspace found. Pass --workspace, or run inside one."
        )
    decisions = prune_unused_workspace_assets(workspace_root, profile, dry_run=args.dry_run)
    if not decisions:
        print("[INFO] The workspace asset store holds no published objects.")
        return
    for decision in decisions:
        print(
            f"{decision['asset_id'][:16]}  {decision['kind']}\n"
            f"  referenced by remote runs       {decision['cold_runs']}\n"
            f"  referenced by active local runs {decision['active_local_runs']}\n"
            f"  remote protection               {decision['remote_protection']}\n"
            f"  local removal                   {decision['local_removal']}"
        )
    removed = [item for item in decisions if item.get("removed")]
    failed = [item for item in decisions if item.get("error")]
    if args.dry_run:
        safe = [item for item in decisions if item["local_removal"] == "safe"]
        print(f"[INFO] Dry-run only. {len(safe)} object(s) would be removed.")
    else:
        for item in failed:
            print(f"[ERROR] Could not remove {item['object']}: {item['error']}", file=sys.stderr)
        print(f"[SUCCESS] Removed {len(removed)} local asset object(s).")
        if failed:
            raise StorageError(f"{len(failed)} asset object(s) could not be removed; see above.")
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_verify_workflow()

None picurv_cli.storage.operations.storage_verify_workflow

(

args

)

Verify all remote chunks for one archive.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1187 of file operations.py.

def storage_verify_workflow(args) -> None:
    """!
    @brief Verify all remote chunks for one archive.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    archive_id = _resolve_archive_id_from_args(args)
    manifest = verify_remote_archive(profile, archive_id)
    print(
        f"[SUCCESS] Archive {archive_id} is complete; verified {len(manifest.get('chunks', []))} chunk(s)."
    )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_list_workflow()

None picurv_cli.storage.operations.storage_list_workflow

(

args

)

Search the remote manifest catalog without local artifact state.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1200 of file operations.py.

def storage_list_workflow(args) -> None:
    """!
    @brief Search the remote manifest catalog without local artifact state.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    manifests = list_remote_manifests(profile)
    workspace_label = getattr(args, "workspace_label", None)
    if workspace_label:
        manifests = [
            item for item in manifests
            if str(item.get("workspace_id") or "") == workspace_label
        ]
    query = str(args.search or "").lower()
    if query:
        manifests = [
            item for item in manifests
            if query in " ".join(
                str(item.get(key, "")) for key in (
                    "archive_id", "label", "notes", "run_id", "study_id", "case_id",
                    "workspace_id", "tags",
                )
            ).lower()
        ]
    if args.output_format == "json":
        print(json.dumps(manifests, indent=2, sort_keys=True))
        return
    print(f"{'ARCHIVE ID':<34} {'TYPE':<12} {'IDENTITY':<32} LABEL")
    for item in manifests:
        identity = (
            item.get("case_id") or item.get("run_id") or item.get("study_id")
            or item.get("workspace_id") or "-"
        )
        print(f"{item['archive_id']:<34} {item.get('artifact_type', '-'):<12} {identity:<32} {item.get('label', '')}")
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_show_workflow()

None picurv_cli.storage.operations.storage_show_workflow

(

args

)

Show one complete remote archive manifest.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1236 of file operations.py.

def storage_show_workflow(args) -> None:
    """!
    @brief Show one complete remote archive manifest.
    @param[in] args Value supplied through the `args` argument.
    """
    profile = load_reported_storage_profile(args)
    manifest = _load_remote_manifest(profile, args.archive_id)
    print(json.dumps(manifest, indent=2, sort_keys=True))
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_workflow()

None picurv_cli.storage.operations.storage_workflow

(

args

)

Dispatch nested storage actions using existing PICurv workflow conventions.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1246 of file operations.py.

def storage_workflow(args) -> None:
    """!
    @brief Dispatch nested storage actions using existing PICurv workflow conventions.
    @param[in] args Value supplied through the `args` argument.
    """
    try:
        action = args.storage_action
        if action == "setup":
            storage_setup_workflow(args)
        elif action == "status":
            storage_status_workflow(args)
        elif action == "plan":
            storage_plan_workflow(args)
        elif action == "protect":
            storage_archive_workflow(args, prune_local=False)
        elif action == "offload":
            storage_archive_workflow(args, prune_local=True)
        elif action == "restore":
            storage_restore_workflow(args)
        elif action == "prune":
            storage_prune_workflow(args)
        elif action == "verify":
            storage_verify_workflow(args)
        elif action == "list":
            storage_list_workflow(args)
        elif action == "show":
            storage_show_workflow(args)
        else:
            raise StorageError(f"Unsupported storage action: {action}")
    except StorageError as exc:
        print(f"[FATAL] {exc}", file=sys.stderr)
        raise SystemExit(1)
 
 

Here is the call graph for this function:

add_storage_parser()

argparse.ArgumentParser picurv_cli.storage.operations.add_storage_parser

(

subparsers

)

Attach the nested storage command parser to PICurv's top-level parser.

Parameters

[in]

subparsers

Value supplied through the subparsers argument.

Returns

Result produced by this operation.

Definition at line 1280 of file operations.py.

def add_storage_parser(subparsers) -> argparse.ArgumentParser:
    """!
    @brief Attach the nested storage command parser to PICurv's top-level parser.
    @param[in] subparsers Value supplied through the `subparsers` argument.
    @return Result produced by this operation.
    """
    parser = subparsers.add_parser(
        "storage",
        help="Protect, offload, inspect, verify, and restore run/study artifacts.",
        formatter_class=argparse.RawTextHelpFormatter,
        description=(
            "Manage PICurv run and study data through a configured rclone remote.\n"
            "Remote archives are checksum-verified before local payload can be pruned.\n\n"
            "Examples:\n"
            "  picurv storage setup --remote labstore:picurv-data\n"
            "  picurv storage status --run-dir runs/my_run\n"
            "  picurv storage protect --run-dir runs/my_run --label 'baseline'\n"
            "  picurv storage offload --study-dir studies/my_study --case-id case_0003\n"
            "  picurv storage list --search '64-grid'\n"
            "  picurv storage restore --archive-id <id>"
        ),
        epilog="Use `picurv storage <action> --help` for action-specific controls.",
    )
    actions = parser.add_subparsers(dest="storage_action", required=True, help="Storage action")
 
    setup = actions.add_parser("setup", help="Configure a non-secret rclone storage profile.")
    setup.add_argument("--remote", required=True, help="Rclone remote and base path, such as labstore:picurv-data.")
    setup.add_argument("--profile", default=DEFAULT_PROFILE_NAME, help="Profile name (default: archive).")
    setup.add_argument("--storage-config", help=f"Storage YAML path (default: ./{STORAGE_CONFIG_FILENAME}).")
    setup.add_argument("--compression", choices=list(STORAGE_COMPRESSION_POLICIES), default="auto")
    setup.add_argument("--chunk-size-gib", type=float, default=DEFAULT_CHUNK_SIZE_GIB)
    setup.add_argument("--workers", type=int, default=DEFAULT_STORAGE_WORKERS,
                       help="CPU workers for compression and restoration.")
    setup.add_argument("--offload-policy", choices=list(STORAGE_OFFLOAD_POLICIES), default="metadata-only")
    setup.add_argument("--keep-latest-checkpoint", action="store_true",
                       help="Retain the newest committed checkpoint after offload.")
    setup.add_argument("--staging-directory", help="Optional local directory for one archive chunk at a time.")
    setup.add_argument("--dry-run", action="store_true")
 
    def add_profile_options(action_parser):
        """!
        @brief Attach shared storage-profile selectors to one action parser.
        @param[in] action_parser Value supplied through the `action_parser` argument.
        """
        action_parser.add_argument("--profile", help="Configured storage profile name.")
        action_parser.add_argument("--storage-config", help="Explicit storage YAML path.")
 
    def add_local_target(action_parser, require=True, allow_workspace=True):
        """!
        @brief Attach the standard run/study/workspace target selectors to one parser.
        @param[in] action_parser Value supplied through the `action_parser` argument.
        @param[in] require Value supplied through the `require` argument.
        @param[in] allow_workspace Whether a whole workspace is a valid target here.
        """
        group = action_parser.add_mutually_exclusive_group(required=require)
        group.add_argument("--run-dir", help="Standalone run directory.")
        group.add_argument("--study-dir", help="Sweep study directory.")
        if allow_workspace:
            group.add_argument(
                "--workspace",
                help="Workspace root: its configuration, catalog, and assets, not its "
                     "runs and studies, which are their own artifacts.",
            )
        action_parser.add_argument(
            "--case-id", dest="case_ids", action="append",
            help="One numbered study member, such as case_0003; repeat to select several.",
        )
        action_parser.add_argument(
            "--completed", action="store_true",
            help="With --study-dir, select every finished member and skip the rest.",
        )
 
    def add_retention_options(parser):
        """!
        @brief Add the per-component local-retention overrides to one action parser.
        @param[in] parser Action parser being configured.
        @return None.
        """
        parser.add_argument(
            "--retain", action="append", metavar="COMPONENT",
            help="Keep this component local regardless of --policy; repeatable and\n"
                 "comma-separated. One of: " + ", ".join(STORAGE_RETENTION_COMPONENTS) + ".",
        )
        parser.add_argument(
            "--drop", action="append", metavar="COMPONENT",
            help="Prune this component locally regardless of --policy; same names as --retain.",
        )
 
    status = actions.add_parser("status", help="Show local, protected, cold, and busy artifact state.")
    add_local_target(status)
    status.add_argument("--format", dest="output_format", choices=list(CLI_OUTPUT_FORMATS), default="text")
 
    plan = actions.add_parser("plan", help="Show packaging, dependencies, and safety checks without writing.")
    add_local_target(plan)
    add_profile_options(plan)
    plan.add_argument("--compression", choices=list(STORAGE_COMPRESSION_POLICIES))
    plan.add_argument("--policy", choices=list(STORAGE_OFFLOAD_POLICIES))
    add_retention_options(plan)
    plan.add_argument("--workers", type=int)
    plan_checkpoint = plan.add_mutually_exclusive_group()
    plan_checkpoint.add_argument("--keep-latest-checkpoint", dest="keep_latest_checkpoint", action="store_true")
    plan_checkpoint.add_argument("--drop-all-checkpoints", dest="keep_latest_checkpoint", action="store_false")
    plan.set_defaults(keep_latest_checkpoint=None)
 
    for name, help_text in (
        ("protect", "Upload and verify an archive while retaining all local files."),
        ("offload", "Upload and verify an archive, then prune heavy local payload."),
    ):
        action_parser = actions.add_parser(name, help=help_text)
        add_local_target(action_parser)
        add_profile_options(action_parser)
        action_parser.add_argument(
            "--include-inputs", action="store_true",
            help="With --workspace, also archive user-supplied files under inputs/.",
        )
        action_parser.add_argument("--label", help="Human-readable searchable label.")
        action_parser.add_argument(
            "--notes", help="Free-text note recorded with the archive and shown by `show`."
        )
        action_parser.add_argument("--tag", dest="tags", action="append", help="Repeatable KEY=VALUE catalog tag.")
        action_parser.add_argument("--compression", choices=list(STORAGE_COMPRESSION_POLICIES))
        action_parser.add_argument("--policy", choices=list(STORAGE_OFFLOAD_POLICIES))
        add_retention_options(action_parser)
        action_parser.add_argument("--workers", type=int)
        checkpoint_group = action_parser.add_mutually_exclusive_group()
        checkpoint_group.add_argument("--keep-latest-checkpoint", dest="keep_latest_checkpoint", action="store_true")
        checkpoint_group.add_argument("--drop-all-checkpoints", dest="keep_latest_checkpoint", action="store_false")
        action_parser.set_defaults(keep_latest_checkpoint=None)
        action_parser.add_argument("--dry-run", action="store_true")
 
    restore = actions.add_parser("restore", help="Restore a complete archive or selected checkpoints.")
    restore_source = restore.add_mutually_exclusive_group(required=True)
    restore_source.add_argument("--archive-id", help="Globally unique remote archive ID.")
    restore_source.add_argument(
        "--workspace-id", help="Restore a workspace archive by its recorded workspace identity."
    )
    restore_source.add_argument("--run-dir", help="Cold run containing a local storage marker.")
    restore_source.add_argument("--study-dir", help="Cold study containing a local storage marker.")
    restore.add_argument("--case-id", dest="case_ids", action="append")
    add_profile_options(restore)
    restore.add_argument("--to", dest="destination", help="Optional alternate restore destination.")
    restore.add_argument(
        "--checkpoint", dest="checkpoints", action="append", type=int,
        help="One committed step; repeat to select several.",
    )
    restore.add_argument(
        "--checkpoints", dest="checkpoint_ranges", action="append",
        help="An inclusive step range as START:END or START:END:STRIDE; repeatable.",
    )
    restore.add_argument(
        "--component", dest="components", action="append",
        choices=STORAGE_RESTORE_COMPONENTS,
        help="Restore one semantic component; repeat as needed.",
    )
    restore.add_argument("--force", action="store_true", help="Allow merge into a non-matching existing destination.")
    restore.add_argument("--workers", type=int, help="Parallel download/extraction workers.")
 
    prune = actions.add_parser(
        "prune", help="Remove verified local asset objects that nothing local still needs."
    )
    prune.add_argument("--workspace", help="Workspace root; defaults to discovery from the cwd.")
    prune.add_argument(
        "--assets", action="store_true", required=True,
        help="Select the workspace asset store. Required: prune removes nothing else.",
    )
    prune.add_argument(
        "--unused-locally", action="store_true", required=True,
        help="Confirm that only objects with no active local run are removed.",
    )
    prune.add_argument("--dry-run", action="store_true", help="Report the decision only.")
    add_profile_options(prune)
 
    verify = actions.add_parser("verify", help="Verify a remote archive completion marker and chunk checksums.")
    verify_source = verify.add_mutually_exclusive_group(required=True)
    verify_source.add_argument("--archive-id")
    verify_source.add_argument("--run-dir")
    verify_source.add_argument("--study-dir")
    verify_source.add_argument("--workspace", help="Workspace whose own archive to verify.")
    verify.add_argument("--case-id", dest="case_ids", action="append")
    add_profile_options(verify)
 
    list_parser = actions.add_parser("list", help="List/search remote archives without local directories.")
    add_profile_options(list_parser)
    list_parser.add_argument("--search", help="Case-insensitive search across IDs, labels, identities, and tags.")
    list_parser.add_argument(
        "--workspace-label", help="Show only archives belonging to this workspace identity."
    )
    list_parser.add_argument("--format", dest="output_format", choices=list(CLI_OUTPUT_FORMATS), default="text")
 
    show = actions.add_parser("show", help="Print the complete manifest for one archive.")
    show.add_argument("--archive-id", required=True)
    add_profile_options(show)
    return parser