Canonical cache-key computation and path conventions for MarsRecon.
All hashing logic lives here. Other modules import from this module to
ensure identical cache locations are computed regardless of which pipeline
component constructs the path.
Cache directory layout relative to :
.cache/
litdata// # LitData streaming chunks (train/ val/ test/)
litdata_raw// # build_litdata_raw output (train/ val/ test/)
manifests/.parquet # Adapter manifest parquet files
sampler_splits/.json # Train/val/test split assignments
spatial/ # GeoPackage spatial indexes
spatial_cache[].gpkg
spatial_cache_*.gpkg # Legacy root-level location (not written by new code)
Usage
from cache import (
compute_hash, litdata_cache_key, litdata_cache_root,
manifest_cache_dir, sampler_split_cache_dir, spatial_cache_dir,
write_manifest, validate_cache, run_id_hash,
)
CacheManifest
dataclass
CacheManifest(cache_hash: str, config_snapshot: dict, created_at: str, git_commit: str, python_version: str, platform: str, key_packages: dict[str, str], file_checksums: dict[str, str] = dict())
Reproducibility metadata written alongside every cache entry.
compute_hash
compute_hash(data: dict[str, Any], length: int = HASH_LENGTH) -> str
SHA-256 of a JSON-serialised dict with sorted keys.
Stable across Python restarts and dict insertion orders. This is the
single hashing primitive for the entire project — never call hashlib
directly in other modules.
Source code in src/cache.py
| def compute_hash(data: dict[str, Any], length: int = HASH_LENGTH) -> str:
"""SHA-256 of a JSON-serialised dict with sorted keys.
Stable across Python restarts and dict insertion orders. This is the
single hashing primitive for the entire project — never call hashlib
directly in other modules.
"""
raw = json.dumps(data, sort_keys=True, default=str)
return hashlib.sha256(raw.encode()).hexdigest()[:length]
|
file_sha256
file_sha256(path: Path) -> str
SHA-256 checksum of a file, read in streaming 1 MiB chunks.
Source code in src/cache.py
| def file_sha256(path: Path) -> str:
"""SHA-256 checksum of a file, read in streaming 1 MiB chunks."""
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1 << 20), b""):
h.update(chunk)
return h.hexdigest()
|
litdata_cache_key
litdata_cache_key(config) -> str
Deterministic hash over all fields that affect LitData chunk content.
Always includes clip — this fixes a bug in the old litdata_datamodule.py
where clip was omitted, causing it to look for a different directory than
build_litdata.py produced.
Source code in src/cache.py
| def litdata_cache_key(config) -> str:
"""Deterministic hash over all fields that affect LitData chunk content.
Always includes clip — this fixes a bug in the old litdata_datamodule.py
where clip was omitted, causing it to look for a different directory than
build_litdata.py produced.
"""
from omegaconf import OmegaConf
key_parts: dict[str, Any] = {
"hirise": OmegaConf.to_container(config.data.hirise, resolve=True),
"sampler": OmegaConf.to_container(config.data.sampler, resolve=True),
"resolution": config.data.get("resolution", 512),
"dtm_normalization": config.data.get("dtm_normalization", "relative"),
"clip": config.data.get("clip", False),
}
return compute_hash(key_parts)
|
litdata_cache_root
litdata_cache_root(config) -> Path
Canonical root directory for the LitData streaming cache.
Source code in src/cache.py
| def litdata_cache_root(config) -> Path:
"""Canonical root directory for the LitData streaming cache."""
return Path(config.data.hirise.root) / ".cache" / "litdata" / litdata_cache_key(config)
|
litdata_tmp_dir
litdata_tmp_dir(config, split: str) -> Path
Temporary extraction directory for LitData build (auto-deleted on success).
Source code in src/cache.py
| def litdata_tmp_dir(config, split: str) -> Path:
"""Temporary extraction directory for LitData build (auto-deleted on success)."""
return litdata_cache_root(config).parent / f"_tmp_{litdata_cache_key(config)}_{split}"
|
run_id_hash
run_id_hash(config, length: int = 8) -> str
Short identifier for training-run logs — NOT a cache key.
Uses 8 hex chars (32-bit) to avoid collisions across a handful of
concurrent runs without the verbosity of a full 16-char hash.
Source code in src/cache.py
| def run_id_hash(config, length: int = 8) -> str:
"""Short identifier for training-run logs — NOT a cache key.
Uses 8 hex chars (32-bit) to avoid collisions across a handful of
concurrent runs without the verbosity of a full 16-char hash.
"""
from omegaconf import OmegaConf
raw = json.dumps(
OmegaConf.to_container(config, resolve=True),
sort_keys=True, default=str,
)
return hashlib.sha256(raw.encode()).hexdigest()[:length]
|
write_manifest
write_manifest(cache_path: Path, cache_hash: str, config_snapshot: dict, checksum_files: list[Path] | None = None) -> Path
Write a _MANIFEST.json sidecar alongside cache_path.
For a directory: writes /_MANIFEST.json
For a file: writes /_MANIFEST.json
Returns the path of the written manifest.
Source code in src/cache.py
| def write_manifest(
cache_path: Path,
cache_hash: str,
config_snapshot: dict,
checksum_files: list[Path] | None = None,
) -> Path:
"""Write a _MANIFEST.json sidecar alongside cache_path.
For a directory: writes <cache_path>/_MANIFEST.json
For a file: writes <cache_path.parent>/<stem>_MANIFEST.json
Returns the path of the written manifest.
"""
if checksum_files is None:
checksum_files = _select_checksum_files(cache_path)
checksums: dict[str, str] = {}
for f in checksum_files:
try:
rel = str(f.relative_to(cache_path.parent if cache_path.is_file() else cache_path))
except ValueError:
rel = f.name
try:
checksums[rel] = file_sha256(f)
except OSError as e:
logger.warning("Could not checksum %s: %s", f, e)
manifest = CacheManifest(
cache_hash=cache_hash,
config_snapshot=config_snapshot,
created_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
git_commit=_git_commit(),
python_version=sys.version,
platform=platform.platform(),
key_packages=_key_package_versions(),
file_checksums=checksums,
)
if cache_path.is_dir():
manifest_path = cache_path / "_MANIFEST.json"
else:
manifest_path = cache_path.parent / f"{cache_path.stem}_MANIFEST.json"
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest_path.write_text(json.dumps(asdict(manifest), indent=2))
logger.info("Wrote cache manifest to %s", manifest_path)
return manifest_path
|
validate_cache
validate_cache(cache_path: Path, config_snapshot: dict | None = None, verify_checksums: bool = False) -> tuple[bool, list[str]]
Validate a cache entry against its manifest.
Parameters:
| Name |
Type |
Description |
Default |
cache_path
|
Path
|
Directory or file that was cached.
|
required
|
config_snapshot
|
dict | None
|
If provided, verify the manifest's config matches.
|
None
|
verify_checksums
|
bool
|
If True, recompute SHA-256 for recorded files
(slow for large caches — off by default).
|
False
|
Returns:
| Type |
Description |
tuple[bool, list[str]]
|
(is_valid, issues) — issues is an empty list on success.
|
Source code in src/cache.py
| def validate_cache(
cache_path: Path,
config_snapshot: dict | None = None,
verify_checksums: bool = False,
) -> tuple[bool, list[str]]:
"""Validate a cache entry against its manifest.
Args:
cache_path: Directory or file that was cached.
config_snapshot: If provided, verify the manifest's config matches.
verify_checksums: If True, recompute SHA-256 for recorded files
(slow for large caches — off by default).
Returns:
(is_valid, issues) — issues is an empty list on success.
"""
issues: list[str] = []
mp = _manifest_path(cache_path)
if not mp.exists():
issues.append(f"Manifest not found: {mp}")
return False, issues
try:
data = json.loads(mp.read_text())
except json.JSONDecodeError as e:
issues.append(f"Manifest JSON invalid: {e}")
return False, issues
stored_hash = data.get("cache_hash", "")
if config_snapshot is not None:
expected_hash = compute_hash(config_snapshot)
if stored_hash != expected_hash:
issues.append(
f"Hash mismatch: manifest has {stored_hash!r}, "
f"current config produces {expected_hash!r}"
)
elif not stored_hash:
issues.append("Manifest missing cache_hash field")
if verify_checksums:
base = cache_path if cache_path.is_dir() else cache_path.parent
for rel, expected in data.get("file_checksums", {}).items():
fpath = base / rel
if not fpath.exists():
issues.append(f"Checksummed file missing: {fpath}")
continue
actual = file_sha256(fpath)
if actual != expected:
issues.append(
f"Checksum mismatch for {rel}: "
f"expected {expected[:16]}…, got {actual[:16]}…"
)
return len(issues) == 0, issues
|
main
CLI entry point: python -m cache validate [--checksums]
Source code in src/cache.py
| def main() -> None:
"""CLI entry point: python -m cache validate <path> [--checksums]"""
parser = argparse.ArgumentParser(
prog="python -m cache",
description="MarsRecon cache utility",
)
sub = parser.add_subparsers(dest="command", required=True)
val_parser = sub.add_parser("validate", help="Validate a cache entry against its manifest")
val_parser.add_argument("path", help="Cache directory or file to validate")
val_parser.add_argument(
"--checksums", action="store_true",
help="Recompute SHA-256 for recorded files (slow for large caches)",
)
args = parser.parse_args()
sys.exit(_cli_validate(args))
|