picurv_cli.storage.compatibility Namespace Reference

Functions
list	_capture_study_context (dict target)
	Embed small study control-plane files with an individually archived member.
dict	_capture_parameter_summary (str root)
	Capture the few case values that identify what a run actually solved.
list	_capture_workspace_assets (dict target)
	List the asset ids a workspace archive carries.
list	_capture_run_assets (str root)
	Record reusable asset references carried by run-local input snapshots.
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.
None	_restore_study_context (dict manifest, str case_destination)
	Recreate missing study control-plane files around a restored member.
list	_rebase_restored_text_paths (str root, list replacements)
	Rebase known generated text artifacts after an explicit relocated restore.

Function Documentation

_capture_study_context()

list picurv_cli.storage.compatibility._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 33 of file compatibility.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
 
 

_capture_parameter_summary()

dict picurv_cli.storage.compatibility._capture_parameter_summary ( str  root )

protected

Capture the few case values that identify what a run actually solved.

storage show has to answer "which run was this?" months later without restoring anything, and an id plus a label does not. These are read from the run's own configuration snapshot, so they describe the run rather than whatever the editable workspace says now.

Parameters

[in]

root

Artifact root directory.

Returns

Parameter summary, empty when no readable case snapshot exists.

Definition at line 61 of file compatibility.py.

def _capture_parameter_summary(root: str) -> dict:
    """!
    @brief Capture the few case values that identify what a run actually solved.
 
    @details `storage show` has to answer "which run was this?" months later without
             restoring anything, and an id plus a label does not. These are read from
             the run's own configuration snapshot, so they describe the run rather than
             whatever the editable workspace says now.
    @param[in] root Artifact root directory.
    @return Parameter summary, empty when no readable case snapshot exists.
    """
    for relative in (("config", "case.yml"), ("cases", "case_0000", "config", "case.yml")):
        case_path = os.path.join(root, *relative)
        if os.path.isfile(case_path):
            break
    else:
        return {}
    try:
        with open(case_path, "r", encoding="utf-8") as stream:
            case = yaml.safe_load(stream) or {}
    except (OSError, ValueError):
        return {}
    if not isinstance(case, dict):
        return {}
    properties = case.get("properties") or {}
    scaling = properties.get("scaling") or {}
    fluid = properties.get("fluid") or {}
    grid = case.get("grid") or {}
    run_control = case.get("run_control") or {}
    summary = {
        "title": case.get("title"),
        "grid_mode": grid.get("mode"),
        "grid_dimensions": None,
        "blocks": ((case.get("models") or {}).get("domain") or {}).get("blocks"),
        "start_step": run_control.get("start_step"),
        "total_steps": run_control.get("total_steps"),
        "dt_physical": run_control.get("dt_physical"),
        "length_ref": scaling.get("length_ref"),
        "velocity_ref": scaling.get("velocity_ref"),
        "density": fluid.get("density"),
        "viscosity": fluid.get("viscosity"),
        "reynolds": None,
    }
    programmatic = grid.get("programmatic_settings")
    if isinstance(programmatic, dict):
        dims = [programmatic.get(key) for key in ("im", "jm", "km")]
        if all(isinstance(value, int) for value in dims):
            summary["grid_dimensions"] = dims
    try:
        summary["reynolds"] = (
            float(fluid["density"]) * float(scaling["velocity_ref"])
            * float(scaling["length_ref"]) / float(fluid["viscosity"])
        )
    except (KeyError, TypeError, ValueError, ZeroDivisionError):
        summary["reynolds"] = None
    return {key: value for key, value in summary.items() if value is not None}
 
 

_capture_workspace_assets()

list picurv_cli.storage.compatibility._capture_workspace_assets ( dict  target )

protected

List the asset ids a workspace archive carries.

Parameters

[in]

target

Local artifact target.

Returns

Sorted asset ids, empty for non-workspace artifacts.

Definition at line 119 of file compatibility.py.

def _capture_workspace_assets(target: dict) -> list:
    """!
    @brief List the asset ids a workspace archive carries.
    @param[in] target Local artifact target.
    @return Sorted asset ids, empty for non-workspace artifacts.
    """
    if target["artifact_type"] != "workspace":
        return []
    objects_root = os.path.join(target["root_path"], "assets", "objects")
    if not os.path.isdir(objects_root):
        return []
    found = []
    for kind in sorted(os.listdir(objects_root)):
        kind_root = os.path.join(objects_root, kind)
        if os.path.isdir(kind_root):
            found.extend(
                name for name in sorted(os.listdir(kind_root))
                if os.path.isdir(os.path.join(kind_root, name))
            )
    return sorted(found)
 
 

_capture_run_assets()

list picurv_cli.storage.compatibility._capture_run_assets ( str  root )

protected

Record reusable asset references carried by run-local input snapshots.

Parameters

[in]

root

Run or study artifact root.

Returns

Portable asset references for the remote archive catalog.

Definition at line 141 of file compatibility.py.

def _capture_run_assets(root: str) -> list:
    """!
    @brief Record reusable asset references carried by run-local input snapshots.
    @param[in] root Run or study artifact root.
    @return Portable asset references for the remote archive catalog.
    """
    captured = []
    archive_root = Path(os.path.abspath(root))
    for runtime_root_text in _artifact_runtime_roots(root):
        runtime_root = Path(runtime_root_text)
        lock_path = runtime_root / "inputs" / "assets.lock.yml"
        if not lock_path.is_file():
            continue
        try:
            payload = yaml.safe_load(lock_path.read_text(encoding="utf-8")) or {}
        except (OSError, ValueError, TypeError):
            continue
        assets = payload.get("assets") if isinstance(payload, dict) else None
        if not isinstance(assets, dict):
            continue
        runtime_prefix = runtime_root.relative_to(archive_root).as_posix()
        for kind, reference in sorted(assets.items()):
            if not isinstance(reference, dict):
                continue
            files = reference.get("files") or reference.get("exposed") or []
            normalized_files = []
            for item in files:
                if not isinstance(item, dict) or not isinstance(item.get("path"), str):
                    continue
                run_path = item["path"]
                archived_path = (
                    f"{runtime_prefix}/{run_path}" if runtime_prefix != "." else run_path
                )
                normalized_files.append({
                    "path": run_path,
                    "archived_path": archived_path,
                    "bytes": item.get("bytes"),
                    "sha256": item.get("sha256"),
                })
            captured.append({
                "kind": kind,
                "asset_id": reference.get("asset_id"),
                "provider": reference.get("provider"),
                "provider_spec_sha256": reference.get("provider_spec_sha256"),
                "files": normalized_files,
            })
    return captured
 
 

_config_fingerprints()

dict picurv_cli.storage.compatibility._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 190 of file compatibility.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
 
 

_git_provenance()

dict picurv_cli.storage.compatibility._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 210 of file compatibility.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
 
 

_restore_study_context()

None picurv_cli.storage.compatibility._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 233 of file compatibility.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
 
 

_rebase_restored_text_paths()

list picurv_cli.storage.compatibility._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 257 of file compatibility.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