picurv_cli.storage Namespace Reference

Data Structures
class	StorageError
	User-facing storage workflow failure. More...

Functions
str	_utc_now ()
	Return a stable UTC timestamp for storage metadata.
str	_human_bytes (int value)
	Format a byte count for concise command output.
str	_sha256_file (str path)
	Calculate SHA-256 without loading a potentially large file into memory.
None	_atomic_write_json (str path, dict payload)
	Atomically replace a JSON state or manifest file.
	_read_json (str path)
	Read a JSON mapping when present, otherwise return None.
	_find_upwards (str start, str filename)
	Find the nearest named file at or above a filesystem anchor.
str	resolve_storage_config_path (str explicit_path=None, bool require=True)
	Resolve an explicit or nearest workspace storage configuration.
dict	load_storage_profile (str profile_name=None, str config_path=None)
	Load and validate one non-secret rclone storage profile.
str	_remote_join (str remote, *str parts)
	Join path components without corrupting rclone remote syntax.
str	_object_remote (dict profile, str archive_id, *str parts)
	Return the remote path for one immutable archive object.
subprocess.CompletedProcess	_run_rclone (list arguments, bool check=True)
	Invoke rclone through the same argv-based subprocess boundary as other PICurv tools.
str	_remote_sha256 (str remote_path)
	Ask rclone to calculate or retrieve the SHA-256 of one remote object.
dict	_upload_verified (str local_path, str remote_path)
	Upload one file, then verify its remote SHA-256.
bytes	_read_remote_bytes (str remote_path)
	Read a small remote catalog object through rclone.
dict	_load_remote_manifest (dict profile, str archive_id, bool require_complete=True)
	Fetch and validate one versioned remote storage manifest.
str	_state_path (str root_path)
	Return the local storage state marker path for an artifact root.
	read_storage_state (str root_path)
	Read the nearest applicable storage marker for a run, study, or study member.
bool	is_artifact_cold (str root_path)
	Return whether a local artifact marker says payload data was pruned.
list	cold_study_members (str study_path)
	Return numbered study members whose local payload was pruned.
None	require_storage_payload_local (str root_path, str operation, int checkpoint=None, checkpoints=None)
	Reject a workflow that requires payload currently held in cold storage.
dict	storage_state_summary (str root_path)
	Return a compact storage status for summarize and status commands.
str	_validate_case_id (str case_id)
	Validate a canonical numbered study-member identifier.
list	resolve_local_storage_targets (str run_dir=None, str study_dir=None, case_ids=None)
	Resolve explicit run/study selectors into concrete artifact descriptions.
bool	_path_is_within (str root, str candidate)
	Return whether an absolute candidate remains within a root directory.
str	_resolve_configured_path (str root, str value)
	Resolve runtime-directory syntax using the run directory as working directory.
list	_artifact_runtime_roots (str root)
	Return run-like roots contained by a standalone run or whole study.
list	_discover_external_paths (str root)
	Report configured data paths that escape the archived directory boundary.
list	_discover_dependencies (str root)
	Discover absolute restart/source paths embedded in generated controls.
list	_walk_archive_entries (str root)
	Enumerate archive entries without following symlinks.
	_checkpoint_component (str relative_path)
	Return a checkpoint component token for a path inside a committed step bundle.
str	_classify_component (str relative_path)
	Classify one artifact path for packaging and local retention.
list	_checkpoint_steps (list entries)
	Return committed checkpoint steps represented in an inventory.
list	_find_incomplete_checkpoints (list entries)
	Return incomplete checkpoint paths that make archival unsafe.
bool	_lock_owner_active (str metadata_path)
	Conservatively determine whether a solver/post/storage owner marker is active.
set	_collect_job_ids (payload)
	Recursively collect submitted Slurm job IDs from scheduler metadata.
dict	_slurm_activity (str root)
	Query live Slurm state for every job recorded below an artifact scheduler directory.
dict	inspect_artifact (dict target, bool query_scheduler=True)
	Build a read-only inventory and lifecycle assessment for one artifact.
None	_assert_archive_safe (dict inventory)
	Refuse to package a changing or scheduler-ambiguous artifact.
	storage_operation_lock (str root_path, str operation)
	Hold an exclusive local storage-operation marker for one artifact.
	runtime_stage_lock (str root_path, str stage)
	Mark a locally executed solver/post stage as active for storage safety.
str	_select_compression (str requested, int total_bytes, dict profile)
	Resolve automatic or configured compression policy.
str	_chunk_extension (str compression)
	Return the archive suffix for a compression policy.
list	_build_chunk_specs (dict inventory, int chunk_size_bytes)
	Group archive entries into independently transferable component chunks.
str	_safe_component_name (str component)
	Convert a component token into a portable archive filename fragment.
None	_write_tar_chunk (str root, dict spec, str destination, str compression)
	Package explicitly inventoried entries without following symlinks.
list	_capture_study_context (dict target)
	Embed small study control-plane files with an individually archived member.
dict	_config_fingerprints (str root)
	Hash small canonical YAML inputs stored under an artifact.
dict	_git_provenance (str root)
	Record best-effort current source revision and dirty state.
dict	_parse_tags (raw_tags)
	Parse repeatable KEY=VALUE tags into deterministic metadata.
dict	build_storage_plan (dict target, dict profile, str compression=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)
	Package, upload, verify, register, and optionally prune one artifact.
None	_prune_archived_payload (str root, dict inventory)
	Remove only verified heavy payload while retaining control-plane files.
list	list_remote_manifests (dict profile)
	Enumerate completed archive manifests from the remote catalog.
dict	verify_remote_archive (dict profile, str archive_id)
	Verify the completion marker and every stored chunk checksum.
None	_validate_tar_members (tarfile.TarFile archive)
	Reject archive members that could escape the restore destination.
None	_extract_chunk (str path, str destination)
	Safely extract one verified tar chunk into a staging tree.
None	_restore_study_context (dict manifest, str case_destination)
	Recreate missing study control-plane files around a restored member.
None	_merge_tree (str source, str destination)
	Merge a verified restore tree into a known cold artifact skeleton.
list	_rebase_restored_text_paths (str root, list replacements)
	Rebase known generated text artifacts after an explicit relocated restore.
dict	restore_archive (dict profile, str archive_id, str destination=None, checkpoints=None, bool force=False)
	Download, verify, extract, and materialize an archive or selected checkpoints.
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.
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_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.

Variables
str	STORAGE_CONFIG_FILENAME = ".picurv-storage.yml"
str	STORAGE_STATE_FILENAME = ".picurv-storage.json"
str	STORAGE_LOCK_FILENAME = ".picurv-storage.lock.json"
int	STORAGE_SCHEMA_VERSION = 1
str	REMOTE_OBJECTS_DIRECTORY = "objects"
str	REMOTE_MANIFEST_FILENAME = "manifest.json"
str	REMOTE_COMPLETE_FILENAME = "COMPLETE"
str	DEFAULT_PROFILE_NAME = "archive"
float	DEFAULT_CHUNK_SIZE_GIB = 8.0
int	AUTO_NO_COMPRESSION_BYTES = 256 * 1024 * 1024
int	AUTO_MAXIMUM_COMPRESSION_BYTES = 20 * 1024 * 1024 * 1024
	CHECKPOINT_DIRECTORY_PATTERN = re.compile(r"^step_(\d{12})$")
	INCOMPLETE_CHECKPOINT_PATTERN = re.compile(r"^\.step_\d{12}\.incomplete\.")
	ARCHIVE_ID_PATTERN = re.compile(r"^[0-9a-f]{32}$")
int	KNOWN_CHECKPOINT_VERSION = 1

Function Documentation

_utc_now()

str picurv_cli.storage._utc_now ( )

protected

Return a stable UTC timestamp for storage metadata.

Returns

Result produced by this operation.

Definition at line 54 of file storage.py.

def _utc_now() -> str:
    """!
    @brief Return a stable UTC timestamp for storage metadata.
    @return Result produced by this operation.
    """
    return datetime.datetime.now(datetime.timezone.utc).isoformat()
 
 

Here is the caller graph for this function:

_human_bytes()

str picurv_cli.storage._human_bytes ( int  value )

protected

Format a byte count for concise command output.

Parameters

[in]

value

Value supplied through the value argument.

Returns

Result produced by this operation.

Definition at line 62 of file storage.py.

def _human_bytes(value: int) -> str:
    """!
    @brief Format a byte count for concise command output.
    @param[in] value Value supplied through the `value` argument.
    @return Result produced by this operation.
    """
    size = float(value)
    for suffix in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"):
        if size < 1024.0 or suffix == "PiB":
            return f"{size:.1f} {suffix}" if suffix != "B" else f"{int(size)} B"
        size /= 1024.0
    return f"{int(value)} B"
 
 

_sha256_file()

str picurv_cli.storage._sha256_file ( str  path )

protected

Calculate SHA-256 without loading a potentially large file into memory.

Parameters

[in]

path

Value supplied through the path argument.

Returns

Result produced by this operation.

Definition at line 76 of file storage.py.

def _sha256_file(path: str) -> str:
    """!
    @brief Calculate SHA-256 without loading a potentially large file into memory.
    @param[in] path Value supplied through the `path` argument.
    @return Result produced by this operation.
    """
    digest = hashlib.sha256()
    with open(path, "rb") as stream:
        while True:
            block = stream.read(8 * 1024 * 1024)
            if not block:
                break
            digest.update(block)
    return digest.hexdigest()
 
 

Here is the caller graph for this function:

_atomic_write_json()

None picurv_cli.storage._atomic_write_json ( str  path, dict  payload  )

protected

Atomically replace a JSON state or manifest file.

Parameters

[in]	path	Value supplied through the path argument.
[in]	payload	Value supplied through the payload argument.

Definition at line 92 of file storage.py.

def _atomic_write_json(path: str, payload: dict) -> None:
    """!
    @brief Atomically replace a JSON state or manifest file.
    @param[in] path Value supplied through the `path` argument.
    @param[in] payload Value supplied through the `payload` argument.
    """
    path_abs = os.path.abspath(path)
    os.makedirs(os.path.dirname(path_abs), exist_ok=True)
    temporary = f"{path_abs}.tmp.{os.getpid()}"
    with open(temporary, "w", encoding="utf-8") as stream:
        json.dump(payload, stream, indent=2, sort_keys=True)
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(temporary, path_abs)
 
 

Here is the caller graph for this function:

_read_json()

picurv_cli.storage._read_json ( str  path )

protected

Read a JSON mapping when present, otherwise return None.

Parameters

[in]

path

Value supplied through the path argument.

Returns

Result produced by this operation.

Definition at line 109 of file storage.py.

def _read_json(path: str):
    """!
    @brief Read a JSON mapping when present, otherwise return None.
    @param[in] path Value supplied through the `path` argument.
    @return Result produced by this operation.
    """
    try:
        with open(path, "r", encoding="utf-8") as stream:
            payload = json.load(stream)
    except (OSError, ValueError):
        return None
    return payload if isinstance(payload, dict) else None
 
 

Here is the caller graph for this function:

_find_upwards()

picurv_cli.storage._find_upwards ( str  start, str  filename  )

protected

Find the nearest named file at or above a filesystem anchor.

Parameters

[in]	start	Value supplied through the start argument.
[in]	filename	Value supplied through the filename argument.

Returns

Result produced by this operation.

Definition at line 123 of file storage.py.

def _find_upwards(start: str, filename: str):
    """!
    @brief Find the nearest named file at or above a filesystem anchor.
    @param[in] start Value supplied through the `start` argument.
    @param[in] filename Value supplied through the `filename` argument.
    @return Result produced by this operation.
    """
    current = os.path.abspath(start)
    if os.path.isfile(current):
        current = os.path.dirname(current)
    while True:
        candidate = os.path.join(current, filename)
        if os.path.isfile(candidate):
            return candidate
        parent = os.path.dirname(current)
        if parent == current:
            return None
        current = parent
 
 

Here is the caller graph for this function:

resolve_storage_config_path()

str picurv_cli.storage.resolve_storage_config_path	(	str	explicit_path = None,
		bool	require = True
	)

Resolve an explicit or nearest workspace storage configuration.

Parameters

[in]	explicit_path	Optional user-selected YAML path.
[in]	require	Whether a missing configuration is an error.

Returns

Result produced by this operation.

Definition at line 143 of file storage.py.

def resolve_storage_config_path(explicit_path: str = None, require: bool = True) -> str:
    """!
    @brief Resolve an explicit or nearest workspace storage configuration.
    @param[in] explicit_path Optional user-selected YAML path.
    @param[in] require Whether a missing configuration is an error.
    @return Result produced by this operation.
    """
    if explicit_path:
        result = os.path.abspath(explicit_path)
    else:
        result = _find_upwards(os.getcwd(), STORAGE_CONFIG_FILENAME)
    if result and os.path.isfile(result):
        return result
    if require:
        raise StorageError(
            "No PICurv storage configuration was found. Run "
            "'picurv storage setup --remote <rclone-remote:path>' first or pass --storage-config."
        )
    return os.path.abspath(explicit_path or os.path.join(os.getcwd(), STORAGE_CONFIG_FILENAME))
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

load_storage_profile()

dict picurv_cli.storage.load_storage_profile	(	str	profile_name = None,
		str	config_path = None
	)

Load and validate one non-secret rclone storage profile.

Parameters

[in]	profile_name	Value supplied through the profile_name argument.
[in]	config_path	Value supplied through the config_path argument.

Returns

Result produced by this operation.

Definition at line 164 of file storage.py.

def load_storage_profile(profile_name: str = None, config_path: str = None) -> dict:
    """!
    @brief Load and validate one non-secret rclone storage profile.
    @param[in] profile_name Value supplied through the `profile_name` argument.
    @param[in] config_path Value supplied through the `config_path` argument.
    @return Result produced by this operation.
    """
    resolved_config = resolve_storage_config_path(config_path)
    with open(resolved_config, "r", encoding="utf-8") as stream:
        payload = yaml.safe_load(stream) or {}
    profiles = payload.get("profiles")
    if not isinstance(profiles, dict):
        raise StorageError(f"Storage config has no 'profiles' mapping: {resolved_config}")
    selected = profile_name or payload.get("default_profile") or DEFAULT_PROFILE_NAME
    profile = profiles.get(selected)
    if not isinstance(profile, dict):
        raise StorageError(f"Storage profile '{selected}' does not exist in {resolved_config}.")
    remote = profile.get("remote")
    if not isinstance(remote, str) or not remote.strip():
        raise StorageError(f"Storage profile '{selected}' requires a non-empty remote.")
    result = dict(profile)
    result["name"] = selected
    result["remote"] = remote.rstrip("/")
    result["config_path"] = resolved_config
    try:
        chunk_size_gib = float(result.get("chunk_size_gib", DEFAULT_CHUNK_SIZE_GIB))
    except (TypeError, ValueError) as exc:
        raise StorageError(f"Storage profile '{selected}' chunk_size_gib must be numeric.") from exc
    if chunk_size_gib <= 0.0:
        raise StorageError(f"Storage profile '{selected}' chunk_size_gib must be positive.")
    result["chunk_size_bytes"] = int(chunk_size_gib * 1024 ** 3)
    return result
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_remote_join()

str picurv_cli.storage._remote_join ( str  remote, *str  parts  )

protected

Join path components without corrupting rclone remote syntax.

Parameters

[in]	remote	Value supplied through the remote argument.
[in]	parts	Value supplied through the parts argument.

Returns

Result produced by this operation.

Definition at line 198 of file storage.py.

def _remote_join(remote: str, *parts: str) -> str:
    """!
    @brief Join path components without corrupting rclone remote syntax.
    @param[in] remote Value supplied through the `remote` argument.
    @param[in] parts Value supplied through the `parts` argument.
    @return Result produced by this operation.
    """
    clean_parts = [str(part).strip("/") for part in parts if str(part).strip("/")]
    suffix = "/".join(clean_parts)
    if not suffix:
        return remote
    if remote.endswith(":"):
        return remote + suffix
    return remote.rstrip("/") + "/" + suffix
 
 

Here is the caller graph for this function:

_object_remote()

str picurv_cli.storage._object_remote ( dict  profile, str  archive_id, *str  parts  )

protected

Return the remote path for one immutable archive object.

Parameters

[in]	profile	Value supplied through the profile argument.
[in]	archive_id	Value supplied through the archive_id argument.
[in]	parts	Value supplied through the parts argument.

Returns

Result produced by this operation.

Definition at line 214 of file storage.py.

def _object_remote(profile: dict, archive_id: str, *parts: str) -> str:
    """!
    @brief Return the remote path for one immutable archive object.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] archive_id Value supplied through the `archive_id` argument.
    @param[in] parts Value supplied through the `parts` argument.
    @return Result produced by this operation.
    """
    return _remote_join(profile["remote"], REMOTE_OBJECTS_DIRECTORY, archive_id, *parts)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_run_rclone()

subprocess.CompletedProcess picurv_cli.storage._run_rclone ( list  arguments, bool   check = True  )

protected

Invoke rclone through the same argv-based subprocess boundary as other PICurv tools.

Parameters

[in]	arguments	Value supplied through the arguments argument.
[in]	check	Value supplied through the check argument.

Returns

Result produced by this operation.

Definition at line 225 of file storage.py.

def _run_rclone(arguments: list, check: bool = True) -> subprocess.CompletedProcess:
    """!
    @brief Invoke rclone through the same argv-based subprocess boundary as other PICurv tools.
    @param[in] arguments Value supplied through the `arguments` argument.
    @param[in] check Value supplied through the `check` argument.
    @return Result produced by this operation.
    """
    executable = shutil.which("rclone")
    if not executable:
        raise StorageError("rclone was not found on PATH. Install/configure rclone before using PICurv storage.")
    result = subprocess.run(
        [executable] + [str(item) for item in arguments],
        text=True,
        capture_output=True,
        check=False,
    )
    if check and result.returncode != 0:
        detail = (result.stderr or result.stdout or "unknown rclone error").strip()
        raise StorageError(f"rclone {' '.join(str(item) for item in arguments[:2])} failed: {detail}")
    return result
 
 

Here is the caller graph for this function:

_remote_sha256()

str picurv_cli.storage._remote_sha256 ( str  remote_path )

protected

Ask rclone to calculate or retrieve the SHA-256 of one remote object.

Parameters

[in]

remote_path

Value supplied through the remote_path argument.

Returns

Result produced by this operation.

Definition at line 247 of file storage.py.

def _remote_sha256(remote_path: str) -> str:
    """!
    @brief Ask rclone to calculate or retrieve the SHA-256 of one remote object.
    @param[in] remote_path Value supplied through the `remote_path` argument.
    @return Result produced by this operation.
    """
    result = _run_rclone(["hashsum", "SHA-256", remote_path])
    for line in result.stdout.splitlines():
        token = line.strip().split(None, 1)[0] if line.strip() else ""
        if re.fullmatch(r"[0-9a-fA-F]{64}", token):
            return token.lower()
    raise StorageError(f"rclone did not return a SHA-256 for {remote_path}.")
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_upload_verified()

dict picurv_cli.storage._upload_verified ( str  local_path, str  remote_path  )

protected

Upload one file, then verify its remote SHA-256.

Parameters

[in]	local_path	Value supplied through the local_path argument.
[in]	remote_path	Value supplied through the remote_path argument.

Returns

Result produced by this operation.

Definition at line 261 of file storage.py.

def _upload_verified(local_path: str, remote_path: str) -> dict:
    """!
    @brief Upload one file, then verify its remote SHA-256.
    @param[in] local_path Value supplied through the `local_path` argument.
    @param[in] remote_path Value supplied through the `remote_path` argument.
    @return Result produced by this operation.
    """
    local_digest = _sha256_file(local_path)
    _run_rclone(["copyto", local_path, remote_path])
    remote_digest = _remote_sha256(remote_path)
    if remote_digest != local_digest:
        raise StorageError(
            f"Remote checksum mismatch after upload: {remote_path} "
            f"(local {local_digest}, remote {remote_digest})."
        )
    return {"sha256": local_digest, "stored_bytes": os.path.getsize(local_path)}
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_read_remote_bytes()

bytes picurv_cli.storage._read_remote_bytes ( str  remote_path )

protected

Read a small remote catalog object through rclone.

Parameters

[in]

remote_path

Value supplied through the remote_path argument.

Returns

Result produced by this operation.

Definition at line 279 of file storage.py.

def _read_remote_bytes(remote_path: str) -> bytes:
    """!
    @brief Read a small remote catalog object through rclone.
    @param[in] remote_path Value supplied through the `remote_path` argument.
    @return Result produced by this operation.
    """
    executable = shutil.which("rclone")
    if not executable:
        raise StorageError("rclone was not found on PATH.")
    result = subprocess.run(
        [executable, "cat", remote_path], capture_output=True, check=False
    )
    if result.returncode != 0:
        detail = (result.stderr or result.stdout or b"unknown rclone error").decode("utf-8", "replace").strip()
        raise StorageError(f"Unable to read remote object {remote_path}: {detail}")
    return result.stdout
 
 

Here is the caller graph for this function:

_load_remote_manifest()

dict picurv_cli.storage._load_remote_manifest ( dict  profile, str  archive_id, bool   require_complete = True  )

protected

Fetch and validate one versioned remote storage manifest.

Parameters

[in]	profile	Value supplied through the profile argument.
[in]	archive_id	Value supplied through the archive_id argument.
[in]	require_complete	Value supplied through the require_complete argument.

Returns

Result produced by this operation.

Definition at line 297 of file storage.py.

def _load_remote_manifest(profile: dict, archive_id: str, require_complete: bool = True) -> dict:
    """!
    @brief Fetch and validate one versioned remote storage manifest.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] archive_id Value supplied through the `archive_id` argument.
    @param[in] require_complete Value supplied through the `require_complete` argument.
    @return Result produced by this operation.
    """
    if not ARCHIVE_ID_PATTERN.fullmatch(str(archive_id)):
        raise StorageError(f"Invalid archive ID: {archive_id!r}.")
    manifest_bytes = _read_remote_bytes(_object_remote(profile, archive_id, REMOTE_MANIFEST_FILENAME))
    if require_complete:
        complete = _read_remote_bytes(_object_remote(profile, archive_id, REMOTE_COMPLETE_FILENAME))
        recorded = complete.decode("ascii", "replace").strip().lower()
        actual = hashlib.sha256(manifest_bytes).hexdigest()
        if recorded != actual:
            raise StorageError(f"Archive {archive_id} has no valid completion marker.")
    try:
        manifest = json.loads(manifest_bytes.decode("utf-8"))
    except (UnicodeDecodeError, ValueError) as exc:
        raise StorageError(f"Archive {archive_id} has an invalid manifest.") from exc
    if not isinstance(manifest, dict) or manifest.get("storage_schema_version") != STORAGE_SCHEMA_VERSION:
        raise StorageError(
            f"Archive {archive_id} uses unsupported storage schema "
            f"{manifest.get('storage_schema_version') if isinstance(manifest, dict) else 'unknown'}."
        )
    return manifest
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_state_path()

str picurv_cli.storage._state_path ( str  root_path )

protected

Return the local storage state marker path for an artifact root.

Parameters

[in]

root_path

Value supplied through the root_path argument.

Returns

Result produced by this operation.

Definition at line 326 of file storage.py.

def _state_path(root_path: str) -> str:
    """!
    @brief Return the local storage state marker path for an artifact root.
    @param[in] root_path Value supplied through the `root_path` argument.
    @return Result produced by this operation.
    """
    return os.path.join(os.path.abspath(root_path), STORAGE_STATE_FILENAME)
 
 

Here is the caller graph for this function:

read_storage_state()

picurv_cli.storage.read_storage_state

(

str

root_path

)

Read the nearest applicable storage marker for a run, study, or study member.

Parameters

[in]

root_path

Value supplied through the root_path argument.

Returns

Result produced by this operation.

Definition at line 335 of file storage.py.

def read_storage_state(root_path: str):
    """!
    @brief Read the nearest applicable storage marker for a run, study, or study member.
    @param[in] root_path Value supplied through the `root_path` argument.
    @return Result produced by this operation.
    """
    root = Path(os.path.abspath(root_path))
    state = _read_json(_state_path(str(root)))
    if state:
        return state
    # A whole-study archive owns every numbered member.  Let ordinary run
    # workflows see that parent state without treating unrelated ancestors as
    # storage owners.
    if root.parent.name == "cases":
        return _read_json(_state_path(str(root.parent.parent)))
    return None
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

is_artifact_cold()

bool picurv_cli.storage.is_artifact_cold

(

str

root_path

)

Return whether a local artifact marker says payload data was pruned.

Parameters

[in]

root_path

Value supplied through the root_path argument.

Returns

Result produced by this operation.

Definition at line 353 of file storage.py.

def is_artifact_cold(root_path: str) -> bool:
    """!
    @brief Return whether a local artifact marker says payload data was pruned.
    @param[in] root_path Value supplied through the `root_path` argument.
    @return Result produced by this operation.
    """
    state = read_storage_state(root_path)
    return bool(state and state.get("local_pruned"))
 
 

Here is the call graph for this function:

cold_study_members()

list picurv_cli.storage.cold_study_members

(

str

study_path

)

Return numbered study members whose local payload was pruned.

Parameters

[in]

study_path

Value supplied through the study_path argument.

Returns

Result produced by this operation.

Definition at line 363 of file storage.py.

def cold_study_members(study_path: str) -> list:
    """!
    @brief Return numbered study members whose local payload was pruned.
    @param[in] study_path Value supplied through the `study_path` argument.
    @return Result produced by this operation.
    """
    cases_dir = Path(os.path.abspath(study_path)) / "cases"
    if not cases_dir.is_dir():
        return []
    return [
        child.name for child in sorted(cases_dir.iterdir())
        if child.is_dir() and is_artifact_cold(str(child))
    ]
 
 

require_storage_payload_local()

None picurv_cli.storage.require_storage_payload_local	(	str	root_path,
		str	operation,
		int	checkpoint = None,
			checkpoints = None
	)

Reject a workflow that requires payload currently held in cold storage.

Parameters

[in]	root_path	Run or study-member directory checked by an existing workflow.
[in]	operation	Human-readable consuming operation.
[in]	checkpoint	Optional single required checkpoint step.
[in]	checkpoints	Optional iterable of every required checkpoint step.

Definition at line 378 of file storage.py.

) -> None:
    """!
    @brief Reject a workflow that requires payload currently held in cold storage.
    @param[in] root_path Run or study-member directory checked by an existing workflow.
    @param[in] operation Human-readable consuming operation.
    @param[in] checkpoint Optional single required checkpoint step.
    @param[in] checkpoints Optional iterable of every required checkpoint step.
    """
    state = read_storage_state(root_path)
    if not state or not state.get("local_pruned"):
        return
    required_steps = set()
    if checkpoint is not None:
        required_steps.add(int(checkpoint))
    if checkpoints is not None:
        required_steps.update(int(step) for step in checkpoints)
    restored = set(state.get("restored_components") or [])
    missing_steps = sorted(
        step for step in required_steps if f"checkpoint:{step}" not in restored
    )
    if required_steps and not missing_steps:
        return
    archive_id = state.get("archive_id", "<archive-id>")
    if missing_steps and len(missing_steps) <= 8:
        suffix = "".join(f" --checkpoint {step}" for step in missing_steps)
    else:
        # A full restore is clearer than printing hundreds of repeatable selectors.
        suffix = ""
    raise StorageError(
        f"{operation} requires payload archived from {os.path.abspath(root_path)}. Restore it first with:\n"
        f"  picurv storage restore --archive-id {archive_id}{suffix}"
    )
 
 

Here is the call graph for this function:

storage_state_summary()

dict picurv_cli.storage.storage_state_summary

(

str

root_path

)

Return a compact storage status for summarize and status commands.

Parameters

[in]

root_path

Value supplied through the root_path argument.

Returns

Result produced by this operation.

Definition at line 417 of file storage.py.

def storage_state_summary(root_path: str) -> dict:
    """!
    @brief Return a compact storage status for summarize and status commands.
    @param[in] root_path Value supplied through the `root_path` argument.
    @return Result produced by this operation.
    """
    state = read_storage_state(root_path)
    if not state:
        return {"state": "LOCAL", "archive_id": None, "label": None}
    if state.get("local_pruned"):
        status = "PARTIAL" if state.get("restored_components") else "COLD"
    else:
        status = "PROTECTED"
    return {
        "state": status,
        "archive_id": state.get("archive_id"),
        "label": state.get("label"),
        "profile": state.get("profile"),
        "remote": state.get("remote"),
    }
 
 

Here is the call graph for this function:

_validate_case_id()

str picurv_cli.storage._validate_case_id ( str  case_id )

protected

Validate a canonical numbered study-member identifier.

Parameters

[in]

case_id

Value supplied through the case_id argument.

Returns

Result produced by this operation.

Definition at line 439 of file storage.py.

def _validate_case_id(case_id: str) -> str:
    """!
    @brief Validate a canonical numbered study-member identifier.
    @param[in] case_id Value supplied through the `case_id` argument.
    @return Result produced by this operation.
    """
    if not re.fullmatch(r"case_\d+", str(case_id or "")):
        raise StorageError(f"Invalid study case ID {case_id!r}; expected a value such as case_0003.")
    return str(case_id)
 
 

Here is the caller graph for this function:

resolve_local_storage_targets()

list picurv_cli.storage.resolve_local_storage_targets	(	str	run_dir = None,
		str	study_dir = None,
			case_ids = None
	)

Resolve explicit run/study selectors into concrete artifact descriptions.

Parameters

[in]	run_dir	Value supplied through the run_dir argument.
[in]	study_dir	Value supplied through the study_dir argument.
[in]	case_ids	Value supplied through the case_ids argument.

Returns

Result produced by this operation.

Definition at line 450 of file storage.py.

def resolve_local_storage_targets(run_dir: str = None, study_dir: str = None, case_ids=None) -> list:
    """!
    @brief Resolve explicit run/study selectors into concrete artifact descriptions.
    @param[in] run_dir Value supplied through the `run_dir` argument.
    @param[in] study_dir Value supplied through the `study_dir` argument.
    @param[in] case_ids Value supplied through the `case_ids` argument.
    @return Result produced by this operation.
    """
    if bool(run_dir) == bool(study_dir):
        raise StorageError("Select exactly one of --run-dir or --study-dir.")
    if run_dir:
        if case_ids:
            raise StorageError("--case-id is valid only with --study-dir.")
        root = os.path.abspath(run_dir)
        if not os.path.isdir(root):
            raise StorageError(f"Run directory not found: {root}")
        if not os.path.isdir(os.path.join(root, "config")):
            raise StorageError(f"Directory does not look like a PICurv run (missing config/): {root}")
        return [{
            "artifact_type": "run",
            "root_path": root,
            "original_path": root,
            "run_id": os.path.basename(root),
            "study_id": None,
            "case_id": None,
        }]
 
    study_root = os.path.abspath(study_dir)
    if not os.path.isdir(study_root):
        raise StorageError(f"Study directory not found: {study_root}")
    if not os.path.isdir(os.path.join(study_root, "cases")):
        raise StorageError(f"Directory does not look like a PICurv study (missing cases/): {study_root}")
    if case_ids:
        targets = []
        for raw_case_id in case_ids:
            case_id = _validate_case_id(raw_case_id)
            case_root = os.path.join(study_root, "cases", case_id)
            if not os.path.isdir(case_root):
                raise StorageError(f"Study member not found: {case_root}")
            targets.append({
                "artifact_type": "study-case",
                "root_path": case_root,
                "original_path": case_root,
                "run_id": case_id,
                "study_id": os.path.basename(study_root),
                "case_id": case_id,
                "study_path": study_root,
            })
        return targets
    return [{
        "artifact_type": "study",
        "root_path": study_root,
        "original_path": study_root,
        "run_id": None,
        "study_id": os.path.basename(study_root),
        "case_id": None,
    }]
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_path_is_within()

bool picurv_cli.storage._path_is_within ( str  root, str  candidate  )

protected

Return whether an absolute candidate remains within a root directory.

Parameters

[in]	root	Value supplied through the root argument.
[in]	candidate	Value supplied through the candidate argument.

Returns

Result produced by this operation.

Definition at line 509 of file storage.py.

def _path_is_within(root: str, candidate: str) -> bool:
    """!
    @brief Return whether an absolute candidate remains within a root directory.
    @param[in] root Value supplied through the `root` argument.
    @param[in] candidate Value supplied through the `candidate` argument.
    @return Result produced by this operation.
    """
    try:
        return os.path.commonpath([os.path.abspath(root), os.path.abspath(candidate)]) == os.path.abspath(root)
    except ValueError:
        return False
 
 

Here is the caller graph for this function:

_resolve_configured_path()

str picurv_cli.storage._resolve_configured_path ( str  root, str  value  )

protected

Resolve runtime-directory syntax using the run directory as working directory.

Parameters

[in]	root	Value supplied through the root argument.
[in]	value	Value supplied through the value argument.

Returns

Result produced by this operation.

Definition at line 522 of file storage.py.

def _resolve_configured_path(root: str, value: str) -> str:
    """!
    @brief Resolve runtime-directory syntax using the run directory as working directory.
    @param[in] root Value supplied through the `root` argument.
    @param[in] value Value supplied through the `value` argument.
    @return Result produced by this operation.
    """
    return os.path.abspath(value if os.path.isabs(value) else os.path.join(root, value))
 
 

Here is the caller graph for this function:

_artifact_runtime_roots()

list picurv_cli.storage._artifact_runtime_roots ( str  root )

protected

Return run-like roots contained by a standalone run or whole study.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 532 of file storage.py.

def _artifact_runtime_roots(root: str) -> list:
    """!
    @brief Return run-like roots contained by a standalone run or whole study.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    root_path = Path(os.path.abspath(root))
    case_root = root_path / "cases"
    if case_root.is_dir():
        return [
            str(path) for path in sorted(case_root.glob("case_*"))
            if path.is_dir()
        ]
    return [str(root_path)]
 
 

Here is the caller graph for this function:

_discover_external_paths()

list picurv_cli.storage._discover_external_paths ( str  root )

protected

Report configured data paths that escape the archived directory boundary.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 548 of file storage.py.

def _discover_external_paths(root: str) -> list:
    """!
    @brief Report configured data paths that escape the archived directory boundary.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    archive_root = os.path.abspath(root)
    external = []
    for runtime_root in _artifact_runtime_roots(root):
        config_dir = os.path.join(runtime_root, "config")
        source_prefix = os.path.relpath(runtime_root, archive_root).replace(os.sep, "/")
        source_prefix = "" if source_prefix == "." else source_prefix + ":"
        monitor_path = os.path.join(config_dir, "monitor.yml")
        if os.path.isfile(monitor_path):
            try:
                with open(monitor_path, "r", encoding="utf-8") as stream:
                    monitor = yaml.safe_load(stream) or {}
                directories = ((monitor.get("io") or {}).get("directories") or {})
                if isinstance(directories, dict):
                    for key, value in directories.items():
                        if isinstance(value, str) and value.strip():
                            resolved = _resolve_configured_path(runtime_root, value.strip())
                            if not _path_is_within(archive_root, resolved):
                                external.append({
                                    "source": f"{source_prefix}monitor.io.directories.{key}",
                                    "path": resolved,
                                })
            except (OSError, ValueError, TypeError):
                pass
        post_path = os.path.join(config_dir, "post.yml")
        if not os.path.isfile(post_path):
            continue
        try:
            with open(post_path, "r", encoding="utf-8") as stream:
                post = yaml.safe_load(stream) or {}
            values = [
                ("post.io.output_directory", (post.get("io") or {}).get("output_directory")),
                ("post.source_data.directory", (post.get("source_data") or {}).get("directory")),
            ]
            for source, value in values:
                if not isinstance(value, str) or not value.strip() or value == "<solver_output_dir>":
                    continue
                resolved = _resolve_configured_path(runtime_root, value.strip())
                if not _path_is_within(archive_root, resolved):
                    external.append({"source": f"{source_prefix}{source}", "path": resolved})
        except (OSError, ValueError, TypeError):
            pass
    return external
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_discover_dependencies()

list picurv_cli.storage._discover_dependencies ( str  root )

protected

Discover absolute restart/source paths embedded in generated controls.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 598 of file storage.py.

def _discover_dependencies(root: str) -> list:
    """!
    @brief Discover absolute restart/source paths embedded in generated controls.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    dependencies = []
    archive_root = os.path.abspath(root)
    controls = []
    for runtime_root in _artifact_runtime_roots(root):
        controls.extend(sorted(Path(runtime_root).glob("config/*.control")))
    for control in controls:
        try:
            lines = control.read_text(encoding="utf-8", errors="replace").splitlines()
        except OSError:
            continue
        for line in lines:
            stripped = line.strip()
            if not stripped.startswith("-restart_dir "):
                continue
            try:
                tokens = __import__("shlex").split(stripped)
            except ValueError:
                continue
            if len(tokens) >= 2 and os.path.isabs(tokens[1]) and not _path_is_within(archive_root, tokens[1]):
                dependencies.append({"kind": "restart", "path": os.path.abspath(tokens[1])})
    return dependencies
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_walk_archive_entries()

list picurv_cli.storage._walk_archive_entries ( str  root )

protected

Enumerate archive entries without following symlinks.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 627 of file storage.py.

def _walk_archive_entries(root: str) -> list:
    """!
    @brief Enumerate archive entries without following symlinks.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    root_abs = os.path.abspath(root)
    entries = []
 
    def visit(directory: str):
        """!
        @brief Recursively inventory directory entries without following symlinks.
        @param[in] directory Value supplied through the `directory` argument.
        """
        try:
            children = sorted(os.scandir(directory), key=lambda item: item.name)
        except OSError as exc:
            raise StorageError(f"Unable to inventory {directory}: {exc}") from exc
        for child in children:
            if child.name in {STORAGE_STATE_FILENAME, STORAGE_LOCK_FILENAME}:
                continue
            rel = os.path.relpath(child.path, root_abs).replace(os.sep, "/")
            try:
                stat_result = child.stat(follow_symlinks=False)
            except OSError as exc:
                raise StorageError(f"Unable to stat {child.path}: {exc}") from exc
            if child.is_symlink():
                entry_type = "symlink"
                size = 0
            elif child.is_dir(follow_symlinks=False):
                entry_type = "directory"
                size = 0
            elif child.is_file(follow_symlinks=False):
                entry_type = "file"
                size = int(stat_result.st_size)
            else:
                raise StorageError(f"Unsupported filesystem entry in artifact: {child.path}")
            entries.append({
                "path": rel,
                "type": entry_type,
                "size": size,
                "mode": int(stat_result.st_mode & 0o7777),
                "mtime_ns": int(stat_result.st_mtime_ns),
            })
            if entry_type == "directory":
                visit(child.path)
 
    visit(root_abs)
    return entries
 
 

Here is the caller graph for this function:

_checkpoint_component()

picurv_cli.storage._checkpoint_component ( str  relative_path )

protected

Return a checkpoint component token for a path inside a committed step bundle.

Parameters

[in]

relative_path

Value supplied through the relative_path argument.

Returns

Result produced by this operation.

Definition at line 678 of file storage.py.

def _checkpoint_component(relative_path: str):
    """!
    @brief Return a checkpoint component token for a path inside a committed step bundle.
    @param[in] relative_path Value supplied through the `relative_path` argument.
    @return Result produced by this operation.
    """
    parts = relative_path.split("/")
    for index, part in enumerate(parts):
        match = CHECKPOINT_DIRECTORY_PATTERN.fullmatch(part)
        if match and index > 0 and parts[index - 1] == "checkpoints":
            return f"checkpoint:{int(match.group(1))}"
    return None
 
 

Here is the caller graph for this function:

_classify_component()

str picurv_cli.storage._classify_component ( str  relative_path )

protected

Classify one artifact path for packaging and local retention.

Parameters

[in]

relative_path

Value supplied through the relative_path argument.

Returns

Result produced by this operation.

Definition at line 692 of file storage.py.

def _classify_component(relative_path: str) -> str:
    """!
    @brief Classify one artifact path for packaging and local retention.
    @param[in] relative_path Value supplied through the `relative_path` argument.
    @return Result produced by this operation.
    """
    checkpoint = _checkpoint_component(relative_path)
    if checkpoint:
        return checkpoint
    parts = relative_path.split("/")
    first = parts[0]
    base = os.path.basename(relative_path)
    if len(parts) >= 3 and parts[0] == "cases" and re.fullmatch(r"case_\d+", parts[1]):
        first = parts[2]
    if first in {"config", "scheduler"} or base in {
        "manifest.json", "study_manifest.json", "study.yml", "cluster.yml"
    }:
        return "metadata"
    if first == "results":
        return "results"
    if first == "logs":
        return "logs"
    return "data"
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_checkpoint_steps()

list picurv_cli.storage._checkpoint_steps ( list  entries )

protected

Return committed checkpoint steps represented in an inventory.

Parameters

[in]

entries

Value supplied through the entries argument.

Returns

Result produced by this operation.

Definition at line 717 of file storage.py.

def _checkpoint_steps(entries: list) -> list:
    """!
    @brief Return committed checkpoint steps represented in an inventory.
    @param[in] entries Value supplied through the `entries` argument.
    @return Result produced by this operation.
    """
    candidates = {}
    for entry in entries:
        component = _checkpoint_component(entry["path"])
        if component:
            step = int(component.split(":", 1)[1])
            candidates.setdefault(step, set()).add(os.path.basename(entry["path"]))
    return sorted(step for step, names in candidates.items() if {"checkpoint.meta", "COMMITTED"} <= names)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_find_incomplete_checkpoints()

list picurv_cli.storage._find_incomplete_checkpoints ( list  entries )

protected

Return incomplete checkpoint paths that make archival unsafe.

Parameters

[in]

entries

Value supplied through the entries argument.

Returns

Result produced by this operation.

Definition at line 732 of file storage.py.

def _find_incomplete_checkpoints(entries: list) -> list:
    """!
    @brief Return incomplete checkpoint paths that make archival unsafe.
    @param[in] entries Value supplied through the `entries` argument.
    @return Result produced by this operation.
    """
    return [
        entry["path"] for entry in entries
        if any(INCOMPLETE_CHECKPOINT_PATTERN.match(part) for part in entry["path"].split("/"))
    ]
 
 

Here is the caller graph for this function:

_lock_owner_active()

bool picurv_cli.storage._lock_owner_active ( str  metadata_path )

protected

Conservatively determine whether a solver/post/storage owner marker is active.

Parameters

[in]

metadata_path

Value supplied through the metadata_path argument.

Returns

Result produced by this operation.

Definition at line 744 of file storage.py.

def _lock_owner_active(metadata_path: str) -> bool:
    """!
    @brief Conservatively determine whether a solver/post/storage owner marker is active.
    @param[in] metadata_path Value supplied through the `metadata_path` argument.
    @return Result produced by this operation.
    """
    owner = _read_json(metadata_path)
    if not owner:
        return True
    host = owner.get("host")
    pid = owner.get("pid")
    if host and host != socket.gethostname():
        return True
    try:
        pid = int(pid)
    except (TypeError, ValueError):
        return True
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except (PermissionError, OSError):
        return True
    return True
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_collect_job_ids()

set picurv_cli.storage._collect_job_ids (   payload )

protected

Recursively collect submitted Slurm job IDs from scheduler metadata.

Parameters

[in]

payload

Value supplied through the payload argument.

Returns

Result produced by this operation.

Definition at line 770 of file storage.py.

def _collect_job_ids(payload) -> set:
    """!
    @brief Recursively collect submitted Slurm job IDs from scheduler metadata.
    @param[in] payload Value supplied through the `payload` argument.
    @return Result produced by this operation.
    """
    result = set()
    if isinstance(payload, dict):
        if payload.get("submitted") and payload.get("job_id") is not None:
            result.add(str(payload["job_id"]).strip())
        for value in payload.values():
            result.update(_collect_job_ids(value))
    elif isinstance(payload, list):
        for value in payload:
            result.update(_collect_job_ids(value))
    return {item for item in result if item}
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_slurm_activity()

dict picurv_cli.storage._slurm_activity ( str  root )

protected

Query live Slurm state for every job recorded below an artifact scheduler directory.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 788 of file storage.py.

def _slurm_activity(root: str) -> dict:
    """!
    @brief Query live Slurm state for every job recorded below an artifact scheduler directory.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    job_ids = set()
    scheduler_dirs = [Path(root) / "scheduler"]
    if (Path(root) / "cases").is_dir():
        scheduler_dirs.extend((Path(root) / "cases").glob("*/scheduler"))
    for scheduler in scheduler_dirs:
        if not scheduler.is_dir():
            continue
        for path in scheduler.glob("submission*.json"):
            job_ids.update(_collect_job_ids(_read_json(str(path))))
    if not job_ids:
        return {"job_ids": [], "active": [], "unknown": False}
    squeue = shutil.which("squeue")
    if not squeue:
        return {"job_ids": sorted(job_ids), "active": [], "unknown": True}
    result = subprocess.run(
        [squeue, "-h", "-j", ",".join(sorted(job_ids)), "-o", "%i|%T"],
        text=True,
        capture_output=True,
        check=False,
    )
    if result.returncode != 0:
        return {"job_ids": sorted(job_ids), "active": [], "unknown": True}
    active = []
    for line in result.stdout.splitlines():
        if not line.strip():
            continue
        job_id, _, state = line.partition("|")
        active.append({"job_id": job_id.strip(), "state": state.strip() or "UNKNOWN"})
    return {"job_ids": sorted(job_ids), "active": active, "unknown": False}
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

inspect_artifact()

dict picurv_cli.storage.inspect_artifact	(	dict	target,
		bool	query_scheduler = True
	)

Build a read-only inventory and lifecycle assessment for one artifact.

Parameters

[in]	target	Value supplied through the target argument.
[in]	query_scheduler	Value supplied through the query_scheduler argument.

Returns

Result produced by this operation.

Definition at line 825 of file storage.py.

def inspect_artifact(target: dict, query_scheduler: bool = True) -> dict:
    """!
    @brief Build a read-only inventory and lifecycle assessment for one artifact.
    @param[in] target Value supplied through the `target` argument.
    @param[in] query_scheduler Value supplied through the `query_scheduler` argument.
    @return Result produced by this operation.
    """
    root = target["root_path"]
    entries = _walk_archive_entries(root)
    for entry in entries:
        entry["component"] = _classify_component(entry["path"])
    lock_paths = []
    for name in ("post.lock.json", "solver.lock.json"):
        for candidate in Path(root).glob(f"**/scheduler/{name}"):
            if _lock_owner_active(str(candidate)):
                lock_paths.append(os.path.relpath(candidate, root))
    slurm = _slurm_activity(root) if query_scheduler else {"job_ids": [], "active": [], "unknown": False}
    state = storage_state_summary(root)
    return {
        "target": dict(target),
        "entries": entries,
        "file_count": sum(entry["type"] in {"file", "symlink"} for entry in entries),
        "total_bytes": sum(entry["size"] for entry in entries),
        "checkpoint_steps": _checkpoint_steps(entries),
        "incomplete_checkpoints": _find_incomplete_checkpoints(entries),
        "active_locks": sorted(lock_paths),
        "slurm": slurm,
        "external_paths": _discover_external_paths(root),
        "dependencies": _discover_dependencies(root),
        "storage": state,
    }
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_assert_archive_safe()

None picurv_cli.storage._assert_archive_safe ( dict  inventory )

protected

Refuse to package a changing or scheduler-ambiguous artifact.

Parameters

[in]

inventory

Value supplied through the inventory argument.

Definition at line 858 of file storage.py.

def _assert_archive_safe(inventory: dict) -> None:
    """!
    @brief Refuse to package a changing or scheduler-ambiguous artifact.
    @param[in] inventory Value supplied through the `inventory` argument.
    """
    problems = []
    if inventory["incomplete_checkpoints"]:
        problems.append("incomplete checkpoint(s): " + ", ".join(inventory["incomplete_checkpoints"][:3]))
    if inventory["active_locks"]:
        problems.append("active runtime lock(s): " + ", ".join(inventory["active_locks"]))
    if inventory["slurm"]["active"]:
        problems.append(
            "active Slurm job(s): " + ", ".join(
                f"{item['job_id']} ({item['state']})" for item in inventory["slurm"]["active"]
            )
        )
    if inventory["slurm"]["unknown"]:
        problems.append(
            "recorded Slurm job IDs could not be checked because squeue is unavailable or failed"
        )
    if problems:
        raise StorageError("Artifact is not safe to archive/offload: " + "; ".join(problems) + ".")
 
 
@contextlib.contextmanager

Here is the caller graph for this function:

storage_operation_lock()

picurv_cli.storage.storage_operation_lock	(	str	root_path,
		str	operation
	)

Hold an exclusive local storage-operation marker for one artifact.

Parameters

[in]	root_path	Value supplied through the root_path argument.
[in]	operation	Value supplied through the operation argument.

Definition at line 883 of file storage.py.

def storage_operation_lock(root_path: str, operation: str):
    """!
    @brief Hold an exclusive local storage-operation marker for one artifact.
    @param[in] root_path Value supplied through the `root_path` argument.
    @param[in] operation Value supplied through the `operation` argument.
    """
    root = os.path.abspath(root_path)
    lock_path = os.path.join(root, STORAGE_LOCK_FILENAME)
    if os.path.exists(lock_path):
        if _lock_owner_active(lock_path):
            raise StorageError(f"Another storage operation owns {lock_path}.")
        try:
            os.remove(lock_path)
        except OSError as exc:
            raise StorageError(f"Unable to remove stale storage lock {lock_path}: {exc}") from exc
    payload = {"operation": operation, "pid": os.getpid(), "host": socket.gethostname(), "started_at": _utc_now()}
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    try:
        descriptor = os.open(lock_path, flags, 0o600)
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            json.dump(payload, stream, indent=2, sort_keys=True)
            stream.write("\n")
        yield
    finally:
        try:
            os.remove(lock_path)
        except FileNotFoundError:
            pass
 
 
@contextlib.contextmanager

Here is the call graph for this function:

Here is the caller graph for this function:

runtime_stage_lock()

picurv_cli.storage.runtime_stage_lock	(	str	root_path,
		str	stage
	)

Mark a locally executed solver/post stage as active for storage safety.

Parameters

[in]	root_path	Run directory used as the runtime working directory.
[in]	stage	Runtime stage label; storage currently uses this for solver execution.

Definition at line 914 of file storage.py.

def runtime_stage_lock(root_path: str, stage: str):
    """!
    @brief Mark a locally executed solver/post stage as active for storage safety.
    @param[in] root_path Run directory used as the runtime working directory.
    @param[in] stage Runtime stage label; storage currently uses this for solver execution.
    """
    scheduler = os.path.join(os.path.abspath(root_path), "scheduler")
    os.makedirs(scheduler, exist_ok=True)
    lock_path = os.path.join(scheduler, f"{stage}.lock.json")
    if os.path.exists(lock_path):
        if _lock_owner_active(lock_path):
            raise StorageError(f"A {stage} runtime stage already owns {lock_path}.")
        os.remove(lock_path)
    payload = {"stage": stage, "pid": os.getpid(), "host": socket.gethostname(), "started_at": _utc_now()}
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    descriptor = os.open(lock_path, flags, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
        json.dump(payload, stream, indent=2, sort_keys=True)
        stream.write("\n")
    try:
        yield
    finally:
        try:
            os.remove(lock_path)
        except FileNotFoundError:
            pass
 
 

Here is the call graph for this function:

_select_compression()

str picurv_cli.storage._select_compression ( str  requested, int  total_bytes, dict  profile  )

protected

Resolve automatic or configured compression policy.

Parameters

[in]	requested	Value supplied through the requested argument.
[in]	total_bytes	Value supplied through the total_bytes argument.
[in]	profile	Value supplied through the profile argument.

Returns

Result produced by this operation.

Definition at line 942 of file storage.py.

def _select_compression(requested: str, total_bytes: int, profile: dict) -> str:
    """!
    @brief Resolve automatic or configured compression policy.
    @param[in] requested Value supplied through the `requested` argument.
    @param[in] total_bytes Value supplied through the `total_bytes` argument.
    @param[in] profile Value supplied through the `profile` argument.
    @return Result produced by this operation.
    """
    selected = requested or profile.get("compression", "auto")
    selected = str(selected).strip().lower()
    if selected not in {"auto", "none", "fast", "balanced", "maximum"}:
        raise StorageError("Compression must be one of: auto, none, fast, balanced, maximum.")
    if selected != "auto":
        return selected
    if total_bytes < AUTO_NO_COMPRESSION_BYTES:
        return "none"
    if total_bytes >= AUTO_MAXIMUM_COMPRESSION_BYTES:
        return "maximum"
    return "balanced"
 
 

Here is the caller graph for this function:

_chunk_extension()

str picurv_cli.storage._chunk_extension ( str  compression )

protected

Return the archive suffix for a compression policy.

Parameters

[in]

compression

Value supplied through the compression argument.

Returns

Result produced by this operation.

Definition at line 963 of file storage.py.

def _chunk_extension(compression: str) -> str:
    """!
    @brief Return the archive suffix for a compression policy.
    @param[in] compression Value supplied through the `compression` argument.
    @return Result produced by this operation.
    """
    return {"none": ".tar", "fast": ".tar.gz", "balanced": ".tar.gz", "maximum": ".tar.xz"}[compression]
 
 

_build_chunk_specs()

list picurv_cli.storage._build_chunk_specs ( dict  inventory, int  chunk_size_bytes  )

protected

Group archive entries into independently transferable component chunks.

Parameters

[in]	inventory	Value supplied through the inventory argument.
[in]	chunk_size_bytes	Value supplied through the chunk_size_bytes argument.

Returns

Result produced by this operation.

Definition at line 972 of file storage.py.

def _build_chunk_specs(inventory: dict, chunk_size_bytes: int) -> list:
    """!
    @brief Group archive entries into independently transferable component chunks.
    @param[in] inventory Value supplied through the `inventory` argument.
    @param[in] chunk_size_bytes Value supplied through the `chunk_size_bytes` argument.
    @return Result produced by this operation.
    """
    groups = {}
    directories = []
    for entry in inventory["entries"]:
        if entry["type"] == "directory":
            directories.append(entry["path"])
            continue
        groups.setdefault(entry["component"], []).append(entry)
    specs = []
    if directories:
        specs.append({"component": "metadata", "entries": directories, "uncompressed_bytes": 0})
    component_order = sorted(groups, key=lambda name: (not name.startswith("checkpoint:"), name))
    for component in component_order:
        current = []
        current_bytes = 0
        for entry in groups[component]:
            entry_size = max(1, int(entry["size"]))
            if current and current_bytes + entry_size > chunk_size_bytes:
                specs.append({"component": component, "entries": current, "uncompressed_bytes": current_bytes})
                current = []
                current_bytes = 0
            current.append(entry["path"])
            current_bytes += int(entry["size"])
        if current:
            specs.append({"component": component, "entries": current, "uncompressed_bytes": current_bytes})
    return specs
 
 

Here is the caller graph for this function:

_safe_component_name()

str picurv_cli.storage._safe_component_name ( str  component )

protected

Convert a component token into a portable archive filename fragment.

Parameters

[in]

component

Value supplied through the component argument.

Returns

Result produced by this operation.

Definition at line 1006 of file storage.py.

def _safe_component_name(component: str) -> str:
    """!
    @brief Convert a component token into a portable archive filename fragment.
    @param[in] component Value supplied through the `component` argument.
    @return Result produced by this operation.
    """
    return re.sub(r"[^A-Za-z0-9_.-]+", "-", component).strip("-") or "data"
 
 

Here is the caller graph for this function:

_write_tar_chunk()

None picurv_cli.storage._write_tar_chunk ( str  root, dict  spec, str  destination, str  compression  )

protected

Package explicitly inventoried entries without following symlinks.

Parameters

[in]	root	Value supplied through the root argument.
[in]	spec	Value supplied through the spec argument.
[in]	destination	Value supplied through the destination argument.
[in]	compression	Value supplied through the compression argument.

Definition at line 1015 of file storage.py.

def _write_tar_chunk(root: str, spec: dict, destination: str, compression: str) -> None:
    """!
    @brief Package explicitly inventoried entries without following symlinks.
    @param[in] root Value supplied through the `root` argument.
    @param[in] spec Value supplied through the `spec` argument.
    @param[in] destination Value supplied through the `destination` argument.
    @param[in] compression Value supplied through the `compression` argument.
    """
    kwargs = {}
    if compression == "none":
        mode = "w"
    elif compression == "fast":
        mode = "w:gz"
        kwargs["compresslevel"] = 1
    elif compression == "balanced":
        mode = "w:gz"
        kwargs["compresslevel"] = 6
    else:
        mode = "w:xz"
        kwargs["preset"] = 9
    with tarfile.open(destination, mode, dereference=False, **kwargs) as archive:
        for relative in spec["entries"]:
            source = os.path.join(root, *relative.split("/"))
            if not os.path.lexists(source):
                raise StorageError(f"Artifact changed during packaging; entry disappeared: {source}")
            archive.add(source, arcname=relative, recursive=False)
 
 

Here is the caller graph for this function:

_capture_study_context()

list picurv_cli.storage._capture_study_context ( dict  target )

protected

Embed small study control-plane files with an individually archived member.

Parameters

[in]

target

Value supplied through the target argument.

Returns

Result produced by this operation.

Definition at line 1043 of file storage.py.

def _capture_study_context(target: dict) -> list:
    """!
    @brief Embed small study control-plane files with an individually archived member.
    @param[in] target Value supplied through the `target` argument.
    @return Result produced by this operation.
    """
    if target["artifact_type"] != "study-case":
        return []
    study_root = Path(target["study_path"])
    candidates = [
        study_root / "study.yml",
        study_root / "cluster.yml",
        study_root / "study_manifest.json",
        study_root / "scheduler" / "case_index.tsv",
        study_root / "scheduler" / "submission.json",
    ]
    captured = []
    for path in candidates:
        if not path.is_file() or path.stat().st_size > 5 * 1024 * 1024:
            continue
        captured.append({
            "path": path.relative_to(study_root).as_posix(),
            "mode": int(path.stat().st_mode & 0o7777),
            "content_base64": base64.b64encode(path.read_bytes()).decode("ascii"),
        })
    return captured
 
 

Here is the caller graph for this function:

_config_fingerprints()

dict picurv_cli.storage._config_fingerprints ( str  root )

protected

Hash small canonical YAML inputs stored under an artifact.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 1071 of file storage.py.

def _config_fingerprints(root: str) -> dict:
    """!
    @brief Hash small canonical YAML inputs stored under an artifact.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    result = {}
    root_path = Path(root)
    candidates = set(root_path.glob("config/*.yml"))
    candidates.update(root_path.glob("base_configs/*.yml"))
    candidates.update(root_path.glob("cases/case_*/config/*.yml"))
    for path in sorted(candidates):
        result[path.relative_to(root).as_posix()] = _sha256_file(str(path))
    for name in ("study.yml", "cluster.yml"):
        path = root_path / name
        if path.is_file():
            result[name] = _sha256_file(str(path))
    return result
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_git_provenance()

dict picurv_cli.storage._git_provenance ( str  root )

protected

Record best-effort current source revision and dirty state.

Parameters

[in]

root

Value supplied through the root argument.

Returns

Result produced by this operation.

Definition at line 1091 of file storage.py.

def _git_provenance(root: str) -> dict:
    """!
    @brief Record best-effort current source revision and dirty state.
    @param[in] root Value supplied through the `root` argument.
    @return Result produced by this operation.
    """
    result = {"commit": None, "dirty": None}
    try:
        commit = subprocess.run(
            ["git", "rev-parse", "HEAD"], cwd=root, text=True, capture_output=True, check=False
        )
        status = subprocess.run(
            ["git", "status", "--porcelain"], cwd=root, text=True, capture_output=True, check=False
        )
        if commit.returncode == 0:
            result["commit"] = commit.stdout.strip()
        if status.returncode == 0:
            result["dirty"] = bool(status.stdout.strip())
    except OSError:
        pass
    return result
 
 

Here is the caller graph for this function:

_parse_tags()

dict picurv_cli.storage._parse_tags (   raw_tags )

protected

Parse repeatable KEY=VALUE tags into deterministic metadata.

Parameters

[in]

raw_tags

Value supplied through the raw_tags argument.

Returns

Result produced by this operation.

Definition at line 1114 of file storage.py.

def _parse_tags(raw_tags) -> dict:
    """!
    @brief Parse repeatable KEY=VALUE tags into deterministic metadata.
    @param[in] raw_tags Value supplied through the `raw_tags` argument.
    @return Result produced by this operation.
    """
    tags = {}
    for item in raw_tags or []:
        key, separator, value = str(item).partition("=")
        if not separator or not key.strip() or not value.strip():
            raise StorageError(f"Invalid tag {item!r}; expected KEY=VALUE.")
        tags[key.strip()] = value.strip()
    return tags
 
 

Here is the caller graph for this function:

build_storage_plan()

dict picurv_cli.storage.build_storage_plan	(	dict	target,
		dict	profile,
		str	compression = 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.

Returns

Result produced by this operation.

Definition at line 1129 of file storage.py.

def build_storage_plan(target: dict, profile: dict, compression: str = 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.
    @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"])
    return {
        "inventory": inventory,
        "compression": selected_compression,
        "chunk_count": len(specs),
        "chunks": [
            {
                "component": spec["component"],
                "file_count": len(spec["entries"]),
                "uncompressed_bytes": spec["uncompressed_bytes"],
            }
            for spec in specs
        ],
    }
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_render_plan()

None picurv_cli.storage._render_plan ( dict  plan )

protected

Print a concise archive/offload plan.

Parameters

[in]

plan

Value supplied through the plan argument.

Definition at line 1155 of file storage.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] Archive chunks: {plan['chunk_count']}")
    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']}")
 
 

Here is the caller graph for this function:

archive_artifact()

dict picurv_cli.storage.archive_artifact	(	dict	target,
		dict	profile,
		str	label = None,
			tags = None,
		str	compression = None,
		bool	prune_local = False
	)

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.

Returns

Result produced by this operation.

Definition at line 1179 of file storage.py.

                     compression: str = None, prune_local: bool = False) -> 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.
    @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)
        inventory = plan["inventory"]
        _assert_archive_safe(inventory)
        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)
        print(f"[INFO] Creating archive {archive_id} for {target['root_path']}")
        with tempfile.TemporaryDirectory(prefix="picurv-storage-", dir=staging_parent) as staging:
            chunks = []
            for index, spec in enumerate(specs):
                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'])})"
                )
                _write_tar_chunk(target["root_path"], spec, local_chunk, plan["compression"])
                remote_path = _object_remote(profile, archive_id, "chunks", filename)
                verified = _upload_verified(local_chunk, remote_path)
                chunks.append({
                    "name": filename,
                    "component": spec["component"],
                    "file_count": len(spec["entries"]),
                    "uncompressed_bytes": spec["uncompressed_bytes"],
                    "stored_bytes": verified["stored_bytes"],
                    "sha256": verified["sha256"],
                })
 
            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"),
                "label": label or os.path.basename(target["root_path"]),
                "tags": _parse_tags(tags),
                "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"],
                "compression": plan["compression"],
                "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),
                "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)
            _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")
            _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": [],
        }
        _atomic_write_json(_state_path(target["root_path"]), state)
        if prune_local:
            _prune_archived_payload(target["root_path"], inventory)
            state["local_pruned"] = True
            state["pruned_at"] = _utc_now()
            _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:

_prune_archived_payload()

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

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.

Definition at line 1288 of file storage.py.

def _prune_archived_payload(root: str, inventory: 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.
    """
    removable = {"data"}
    removable.update(f"checkpoint:{step}" for step in inventory["checkpoint_steps"])
    for entry in sorted(inventory["entries"], key=lambda item: item["path"], reverse=True):
        if entry["component"] not in removable or entry["type"] == "directory":
            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
 
 

Here is the caller graph for this function:

list_remote_manifests()

list picurv_cli.storage.list_remote_manifests

(

dict

profile

)

Enumerate completed archive manifests from the remote catalog.

Parameters

[in]

profile

Value supplied through the profile argument.

Returns

Result produced by this operation.

Definition at line 1316 of file storage.py.

def list_remote_manifests(profile: dict) -> list:
    """!
    @brief Enumerate completed archive manifests from the remote catalog.
    @param[in] profile Value supplied through the `profile` argument.
    @return Result produced by this operation.
    """
    root = _remote_join(profile["remote"], REMOTE_OBJECTS_DIRECTORY)
    result = _run_rclone([
        "lsf", root, "--recursive", "--files-only", "--include", f"*/{REMOTE_MANIFEST_FILENAME}"
    ])
    manifests = []
    for relative in sorted(line.strip() for line in result.stdout.splitlines() if line.strip()):
        archive_id = relative.split("/", 1)[0]
        if not ARCHIVE_ID_PATTERN.fullmatch(archive_id):
            continue
        try:
            manifests.append(_load_remote_manifest(profile, archive_id))
        except StorageError:
            continue
    return manifests
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

verify_remote_archive()

dict picurv_cli.storage.verify_remote_archive	(	dict	profile,
		str	archive_id
	)

Verify the completion marker and every stored chunk checksum.

Parameters

[in]	profile	Value supplied through the profile argument.
[in]	archive_id	Value supplied through the archive_id argument.

Returns

Result produced by this operation.

Definition at line 1338 of file storage.py.

def verify_remote_archive(profile: dict, archive_id: str) -> dict:
    """!
    @brief Verify the completion marker and every stored chunk checksum.
    @param[in] profile Value supplied through the `profile` argument.
    @param[in] archive_id Value supplied through the `archive_id` argument.
    @return Result produced by this operation.
    """
    manifest = _load_remote_manifest(profile, archive_id)
    for chunk in manifest.get("chunks", []):
        remote = _object_remote(profile, archive_id, "chunks", chunk["name"])
        actual = _remote_sha256(remote)
        if actual != chunk.get("sha256"):
            raise StorageError(
                f"Archive {archive_id} chunk checksum mismatch: {chunk['name']} "
                f"(expected {chunk.get('sha256')}, got {actual})."
            )
    return manifest
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_validate_tar_members()

None picurv_cli.storage._validate_tar_members ( tarfile.TarFile  archive )

protected

Reject archive members that could escape the restore destination.

Parameters

[in]

archive

Value supplied through the archive argument.

Definition at line 1357 of file storage.py.

def _validate_tar_members(archive: tarfile.TarFile) -> None:
    """!
    @brief Reject archive members that could escape the restore destination.
    @param[in] archive Value supplied through the `archive` argument.
    """
    link_paths = set()
    members = archive.getmembers()
    for member in members:
        normalized = os.path.normpath(member.name.replace("\\", "/"))
        if normalized.startswith("../") or normalized == ".." or os.path.isabs(normalized):
            raise StorageError(f"Unsafe archive member path: {member.name}")
        for link_path in link_paths:
            if normalized == link_path or normalized.startswith(link_path.rstrip("/") + "/"):
                raise StorageError(f"Archive member traverses an earlier symlink: {member.name}")
        if member.issym() or member.islnk():
            if member.islnk():
                link_target = os.path.normpath(member.linkname.replace("\\", "/"))
                if link_target.startswith("../") or link_target == ".." or os.path.isabs(link_target):
                    raise StorageError(f"Unsafe archive hardlink target: {member.linkname}")
            link_paths.add(normalized)
 
 

Here is the caller graph for this function:

_extract_chunk()

None picurv_cli.storage._extract_chunk ( str  path, str  destination  )

protected

Safely extract one verified tar chunk into a staging tree.

Parameters

[in]	path	Value supplied through the path argument.
[in]	destination	Value supplied through the destination argument.

Definition at line 1379 of file storage.py.

def _extract_chunk(path: str, destination: str) -> None:
    """!
    @brief Safely extract one verified tar chunk into a staging tree.
    @param[in] path Value supplied through the `path` argument.
    @param[in] destination Value supplied through the `destination` argument.
    """
    with tarfile.open(path, "r:*") as archive:
        _validate_tar_members(archive)
        try:
            archive.extractall(destination, filter="data")
        except TypeError:
            archive.extractall(destination)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_restore_study_context()

None picurv_cli.storage._restore_study_context ( dict  manifest, str  case_destination  )

protected

Recreate missing study control-plane files around a restored member.

Parameters

[in]	manifest	Value supplied through the manifest argument.
[in]	case_destination	Value supplied through the case_destination argument.

Definition at line 1393 of file storage.py.

def _restore_study_context(manifest: dict, case_destination: str) -> None:
    """!
    @brief Recreate missing study control-plane files around a restored member.
    @param[in] manifest Value supplied through the `manifest` argument.
    @param[in] case_destination Value supplied through the `case_destination` argument.
    """
    if manifest.get("artifact_type") != "study-case":
        return
    study_root = Path(case_destination).parent.parent
    for item in manifest.get("study_context", []):
        relative = item.get("path")
        if not isinstance(relative, str):
            continue
        destination = study_root.joinpath(*relative.split("/"))
        if destination.exists():
            continue
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(base64.b64decode(item.get("content_base64", "")))
        try:
            destination.chmod(int(item.get("mode", 0o644)))
        except OSError:
            pass
 
 

Here is the caller graph for this function:

_merge_tree()

None picurv_cli.storage._merge_tree ( str  source, str  destination  )

protected

Merge a verified restore tree into a known cold artifact skeleton.

Parameters

[in]	source	Value supplied through the source argument.
[in]	destination	Value supplied through the destination argument.

Definition at line 1417 of file storage.py.

def _merge_tree(source: str, destination: str) -> None:
    """!
    @brief Merge a verified restore tree into a known cold artifact skeleton.
    @param[in] source Value supplied through the `source` argument.
    @param[in] destination Value supplied through the `destination` argument.
    """
    os.makedirs(destination, exist_ok=True)
    for entry in os.scandir(source):
        target = os.path.join(destination, entry.name)
        if entry.is_symlink():
            if os.path.lexists(target):
                if os.path.isdir(target) and not os.path.islink(target):
                    shutil.rmtree(target)
                else:
                    os.remove(target)
            os.symlink(os.readlink(entry.path), target)
        elif entry.is_dir(follow_symlinks=False):
            _merge_tree(entry.path, target)
        else:
            os.makedirs(os.path.dirname(target), exist_ok=True)
            shutil.copy2(entry.path, target)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_rebase_restored_text_paths()

list picurv_cli.storage._rebase_restored_text_paths ( str  root, list  replacements  )

protected

Rebase known generated text artifacts after an explicit relocated restore.

Parameters

[in]	root	Value supplied through the root argument.
[in]	replacements	Value supplied through the replacements argument.

Returns

Result produced by this operation.

Definition at line 1440 of file storage.py.

def _rebase_restored_text_paths(root: str, replacements: list) -> list:
    """!
    @brief Rebase known generated text artifacts after an explicit relocated restore.
    @param[in] root Value supplied through the `root` argument.
    @param[in] replacements Value supplied through the `replacements` argument.
    @return Result produced by this operation.
    """
    changed = []
    allowed_suffixes = {".control", ".run", ".sbatch", ".json", ".tsv", ".yml", ".yaml"}
    for path in Path(root).rglob("*"):
        if not path.is_file() or path.is_symlink() or path.stat().st_size > 32 * 1024 * 1024:
            continue
        if path.suffix.lower() not in allowed_suffixes:
            continue
        try:
            content = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        updated = content
        for old, new in replacements:
            if old and new and old != new:
                updated = updated.replace(old, new)
        if updated != content:
            path.write_text(updated, encoding="utf-8")
            changed.append(str(path))
    return changed
 
 

Here is the caller graph for this function:

restore_archive()

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

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.

Returns

Result produced by this operation.

Definition at line 1468 of file storage.py.

                    checkpoints=None, force: bool = False) -> 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.
    @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 [])}
    chunks = []
    for chunk in manifest.get("chunks", []):
        component = str(chunk.get("component", ""))
        if selected_steps:
            if component == "metadata":
                chunks.append(chunk)
            elif component.startswith("checkpoint:") and int(component.split(":", 1)[1]) in selected_steps:
                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."
            )
    parent = os.path.dirname(destination_abs)
    os.makedirs(parent, exist_ok=True)
    temporary = tempfile.mkdtemp(prefix=f".picurv-restore-{archive_id[:8]}-", dir=parent)
    materialized = os.path.join(temporary, "materialized")
    os.makedirs(materialized)
    try:
        for index, chunk in enumerate(chunks):
            local_chunk = os.path.join(temporary, chunk["name"])
            remote_chunk = _object_remote(profile, archive_id, "chunks", chunk["name"])
            print(f"[INFO] Restoring chunk {index + 1}/{len(chunks)}: {chunk['component']}")
            _run_rclone(["copyto", remote_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, 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),
            "restored_components": [f"checkpoint:{step}" for step in sorted(selected_steps)],
            "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 call graph for this function:

Here is the caller graph for this function:

_resolve_archive_id_from_args()

str picurv_cli.storage._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 1561 of file storage.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
    targets = resolve_local_storage_targets(
        getattr(args, "run_dir", None), getattr(args, "study_dir", None), getattr(args, "case_ids", 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"]
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_render_status()

None picurv_cli.storage._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 1581 of file storage.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")
    print(
        f"{identity:<36} {target['artifact_type']:<12} {activity:<10} "
        f"{_human_bytes(inventory['total_bytes']):>12}  {storage.get('label') or ''}"
    )
 
 

Here is the caller graph for this function:

storage_setup_workflow()

None picurv_cli.storage.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 1596 of file storage.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.
    """
    config_path = resolve_storage_config_path(getattr(args, "storage_config", None), require=False)
    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,
    }
    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
    _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 call graph for this function:

Here is the caller graph for this function:

storage_status_workflow()

None picurv_cli.storage.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 1632 of file storage.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)
        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)
    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 call graph for this function:

Here is the caller graph for this function:

storage_plan_workflow()

None picurv_cli.storage.storage_plan_workflow

(

args

)

Render a read-only packaging and safety plan.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1659 of file storage.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_storage_profile(args.profile, args.storage_config)
    targets = resolve_local_storage_targets(args.run_dir, args.study_dir, args.case_ids)
    for index, target in enumerate(targets):
        if index:
            print()
        plan = build_storage_plan(target, profile, args.compression)
        _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.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 1674 of file storage.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_storage_profile(args.profile, args.storage_config)
    targets = resolve_local_storage_targets(args.run_dir, args.study_dir, args.case_ids)
    for target in targets:
        if args.dry_run:
            _render_plan(build_storage_plan(target, profile, args.compression))
            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,
        )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_restore_workflow()

None picurv_cli.storage.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 1697 of file storage.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_storage_profile(args.profile, args.storage_config)
    archive_id = _resolve_archive_id_from_args(args)
    restore_archive(
        profile,
        archive_id,
        destination=args.destination,
        checkpoints=args.checkpoints,
        force=args.force,
    )
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

storage_verify_workflow()

None picurv_cli.storage.storage_verify_workflow

(

args

)

Verify all remote chunks for one archive.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1713 of file storage.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_storage_profile(args.profile, args.storage_config)
    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.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 1726 of file storage.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_storage_profile(args.profile, args.storage_config)
    manifests = list_remote_manifests(profile)
    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", "run_id", "study_id", "case_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 "-"
        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.storage_show_workflow

(

args

)

Show one complete remote archive manifest.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1750 of file storage.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_storage_profile(args.profile, args.storage_config)
    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.storage_workflow

(

args

)

Dispatch nested storage actions using existing PICurv workflow conventions.

Parameters

[in]

args

Value supplied through the args argument.

Definition at line 1760 of file storage.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 == "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.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 1792 of file storage.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=["auto", "none", "fast", "balanced", "maximum"], default="auto")
    setup.add_argument("--chunk-size-gib", type=float, default=DEFAULT_CHUNK_SIZE_GIB)
    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):
        """!
        @brief Attach the standard run-or-study target selectors to one action parser.
        @param[in] action_parser Value supplied through the `action_parser` argument.
        @param[in] require Value supplied through the `require` argument.
        """
        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.")
        action_parser.add_argument(
            "--case-id", dest="case_ids", action="append",
            help="One numbered study member, such as case_0003; repeat to select several.",
        )
 
    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=["text", "json"], 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=["auto", "none", "fast", "balanced", "maximum"])
 
    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("--label", help="Human-readable searchable label.")
        action_parser.add_argument("--tag", dest="tags", action="append", help="Repeatable KEY=VALUE catalog tag.")
        action_parser.add_argument("--compression", choices=["auto", "none", "fast", "balanced", "maximum"])
        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("--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)
    restore.add_argument("--force", action="store_true", help="Allow merge into a non-matching existing destination.")
 
    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.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("--format", dest="output_format", choices=["text", "json"], 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

Variable Documentation

STORAGE_CONFIG_FILENAME

str picurv_cli.storage.STORAGE_CONFIG_FILENAME = ".picurv-storage.yml"

Definition at line 33 of file storage.py.

STORAGE_STATE_FILENAME

str picurv_cli.storage.STORAGE_STATE_FILENAME = ".picurv-storage.json"

Definition at line 34 of file storage.py.

STORAGE_LOCK_FILENAME

str picurv_cli.storage.STORAGE_LOCK_FILENAME = ".picurv-storage.lock.json"

Definition at line 35 of file storage.py.

STORAGE_SCHEMA_VERSION

int picurv_cli.storage.STORAGE_SCHEMA_VERSION = 1

Definition at line 36 of file storage.py.

REMOTE_OBJECTS_DIRECTORY

str picurv_cli.storage.REMOTE_OBJECTS_DIRECTORY = "objects"

Definition at line 37 of file storage.py.

REMOTE_MANIFEST_FILENAME

str picurv_cli.storage.REMOTE_MANIFEST_FILENAME = "manifest.json"

Definition at line 38 of file storage.py.

REMOTE_COMPLETE_FILENAME

str picurv_cli.storage.REMOTE_COMPLETE_FILENAME = "COMPLETE"

Definition at line 39 of file storage.py.

DEFAULT_PROFILE_NAME

str picurv_cli.storage.DEFAULT_PROFILE_NAME = "archive"

Definition at line 40 of file storage.py.

DEFAULT_CHUNK_SIZE_GIB

float picurv_cli.storage.DEFAULT_CHUNK_SIZE_GIB = 8.0

Definition at line 41 of file storage.py.

AUTO_NO_COMPRESSION_BYTES

int picurv_cli.storage.AUTO_NO_COMPRESSION_BYTES = 256 * 1024 * 1024

Definition at line 42 of file storage.py.

AUTO_MAXIMUM_COMPRESSION_BYTES

int picurv_cli.storage.AUTO_MAXIMUM_COMPRESSION_BYTES = 20 * 1024 * 1024 * 1024

Definition at line 43 of file storage.py.

CHECKPOINT_DIRECTORY_PATTERN

picurv_cli.storage.CHECKPOINT_DIRECTORY_PATTERN = re.compile(r"^step_(\d{12})$")

Definition at line 44 of file storage.py.

INCOMPLETE_CHECKPOINT_PATTERN

picurv_cli.storage.INCOMPLETE_CHECKPOINT_PATTERN = re.compile(r"^\.step_\d{12}\.incomplete\.")

Definition at line 45 of file storage.py.

ARCHIVE_ID_PATTERN

picurv_cli.storage.ARCHIVE_ID_PATTERN = re.compile(r"^[0-9a-f]{32}$")

Definition at line 46 of file storage.py.

KNOWN_CHECKPOINT_VERSION

int picurv_cli.storage.KNOWN_CHECKPOINT_VERSION = 1

Definition at line 47 of file storage.py.