OneScience-Group/MatPL

1#!/usr/bin/env python32"""Validate the standardized OneScience MatPL dataset package."""3 4from __future__ import annotations5 6import argparse7import hashlib8import sys9from pathlib import Path10 11import numpy as np12 13 14EXPECTED_TOP_LEVEL = ["AuAg", "Cu", "HfO2", "LiSiC"]15 16 17def sha256_file(path: Path) -> str:18    digest = hashlib.sha256()19    with path.open("rb") as handle:20        for chunk in iter(lambda: handle.read(1024 * 1024), b""):21            digest.update(chunk)22    return digest.hexdigest()23 24 25def require_file(path: Path) -> None:26    if not path.is_file():27        raise FileNotFoundError(f"missing file: {path}")28 29 30def require_dir(path: Path) -> None:31    if not path.is_dir():32        raise FileNotFoundError(f"missing directory: {path}")33 34 35def check_movement(path: Path) -> None:36    require_file(path)37    with path.open(encoding="utf-8", errors="replace") as handle:38        text = handle.read(4096)39    if not text.strip():40        raise ValueError(f"empty MOVEMENT file: {path}")41    if "Iteration" not in text and "atoms" not in text.lower() and "lattice" not in text.lower():42        raise ValueError(f"unexpected MOVEMENT header: {path}")43 44 45def check_xyz(path: Path) -> None:46    require_file(path)47    with path.open(encoding="utf-8", errors="replace") as handle:48        first = handle.readline().strip()49        second = handle.readline()50    atoms = int(first)51    if atoms <= 0:52        raise ValueError(f"invalid xyz atom count in {path}")53    if not second:54        raise ValueError(f"missing xyz comment line in {path}")55 56 57def check_npy(path: Path) -> None:58    require_file(path)59    arr = np.load(path, allow_pickle=False)60    if arr.size == 0:61        raise ValueError(f"empty npy array: {path}")62 63 64def read_checksum_manifest(path: Path) -> list[tuple[str, str]]:65    require_file(path)66    entries: list[tuple[str, str]] = []67    with path.open(encoding="utf-8") as handle:68        for line_no, raw in enumerate(handle, start=1):69            line = raw.strip()70            if not line:71                continue72            parts = line.split(None, 1)73            if len(parts) != 2:74                raise ValueError(f"invalid checksum line {line_no}: {raw!r}")75            entries.append((parts[0], parts[1]))76    return entries77 78 79def validate_checksums(package_root: Path, manifest_path: Path) -> int:80    entries = read_checksum_manifest(manifest_path)81    for expected_hash, rel_path in entries:82        target = package_root / rel_path83        require_file(target)84        if sha256_file(target) != expected_hash:85            raise ValueError(f"checksum mismatch: {rel_path}")86    return len(entries)87 88 89def main() -> int:90    parser = argparse.ArgumentParser(description=__doc__)91    parser.add_argument("--dataset-root", default="data/MatPL")92    parser.add_argument("--checksum-manifest", default="metadata/sha256_manifest.txt")93    parser.add_argument("--skip-checksum", action="store_true")94    args = parser.parse_args()95 96    package_root = Path.cwd()97    dataset_root = Path(args.dataset_root)98    for name in EXPECTED_TOP_LEVEL:99        require_dir(dataset_root / name)100 101    check_movement(dataset_root / "Cu/pwdata/0_300_MOVEMENT")102    check_movement(dataset_root / "Cu/pwdata/1_500_MOVEMENT")103    check_movement(dataset_root / "Cu/pwdata/valid_movement")104    check_xyz(dataset_root / "AuAg/AuAg-5762.xyz")105 106    npy_files = sorted(dataset_root.glob("**/*.npy"))107    if len(npy_files) < 100:108        raise ValueError(f"too few npy files: {len(npy_files)}")109    for sample in npy_files[:8]:110        check_npy(sample)111 112    checksum_count = 0113    if not args.skip_checksum:114        checksum_count = validate_checksums(package_root, Path(args.checksum_manifest))115 116    print("MatPL dataset validation passed")117    print(f"top-level systems: {', '.join(EXPECTED_TOP_LEVEL)}")118    print(f"npy files: {len(npy_files)}")119    print(f"checksum entries verified: {checksum_count}")120    return 0121 122 123if __name__ == "__main__":124    try:125        raise SystemExit(main())126    except Exception as exc:127        print(f"MatPL dataset validation failed: {exc}", file=sys.stderr)128        raise SystemExit(1)129