Skip to content

Resources and writable workspaces

The installed library contains its Python modules and typing information. Application sources, notebooks and scientific inputs are separate downloads. These utilities accept explicit directories, URLs and catalogues; they contain no built-in case inventory.

download_resource acquires a file with a declared SHA256 and byte limit. It reuses a matching local copy and preserves that copy when acquisition fails. workspace_from_archive applies the same identity check to a ZIP, validates its paths and expanded size, then extracts regular files into a writable directory. Extraction preserves existing local files and does not execute code. Importing companion sources is an explicit step in the consuming notebook.

A workspace may register an external .pymhm-resources.json, or callers can pass a mapping to register_resources. For example:

from pymhm.io.workspace import register_resources, resource_file

register_resources(
    "run",
    {
        "resources": {
            "data/coefficients.csv": {
                "url": coefficient_url,
                "sha256": coefficient_sha256,
                "size_bytes": coefficient_size_bytes,
            },
        },
    },
)
coefficients = resource_file("data/coefficients.csv", directory="run")

Registration describes inputs without downloading them. resource_file checks immutable inputs against the catalogue and acquires only the requested file. materialize_resources copies selected inputs and declared attribution files into the workspace, preserving user edits. local_resource supports existing mutable local files. Unavailable original inputs retain explicit reasons and are not replaced with different data.

source_identity hashes the actual source files used by an application. A logical src/pymhm/... path identifies the installed library's corresponding module; application sources resolve in the declared workspace. Missing Git or lockfile metadata remain absent from provenance records.

For the notebook workflow and download catalogue, see the notebook guide and data downloads.

Bounded acquisition of immutable scientific resources with literal byte checks.

download_resource(url: str, destination: str | Path, *, sha256: str, maximum_bytes: int = 512 * 1024 ** 2, timeout: float = 60.0) -> Path

Acquire immutable bytes with a size limit and an expected SHA256 digest.

maximum_bytes limits the complete payload in bytes, independently of HTTP headers. timeout is the positive finite network timeout in seconds. An existing file is reused only if its checksum matches; an unexpected existing file is preserved and raises ValueError. Failed, oversized or corrupt downloads cannot replace the destination. This function performs a network request only when explicitly called for a missing destination. PYMHM_RESOURCE_ORIGIN can select an HTTP mirror while retaining the source path and query; checksum and size verification remain unchanged.

Source code in src/pymhm/io/resources.py
def download_resource(
    url: str,
    destination: str | Path,
    *,
    sha256: str,
    maximum_bytes: int = 512 * 1024**2,
    timeout: float = 60.0,
) -> Path:
    """Acquire immutable bytes with a size limit and an expected SHA256 digest.

    ``maximum_bytes`` limits the complete payload in bytes, independently of
    HTTP headers. ``timeout`` is the positive finite network timeout in seconds.
    An existing file is reused only if its checksum matches; an unexpected
    existing file is preserved and raises ``ValueError``. Failed, oversized or
    corrupt downloads cannot replace the destination. This function performs
    a network request only when explicitly called for a missing destination.
    PYMHM_RESOURCE_ORIGIN can select an HTTP mirror while retaining the source
    path and query; checksum and size verification remain unchanged.
    """
    limit = positive_int(maximum_bytes, "maximum_bytes")
    if (
        not isinstance(sha256, str)
        or len(sha256) != 64
        or any(character not in "0123456789abcdef" for character in sha256)
    ):
        raise ValueError("sha256 must be a lowercase 64-character hexadecimal digest")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be positive and finite")
    target = Path(destination).expanduser()
    if target.exists():
        if file_digest(target) != sha256:
            raise ValueError(f"SHA256 mismatch for existing resource: {target}")
        return target
    target.parent.mkdir(parents=True, exist_ok=True)
    origin = os.environ.get("PYMHM_RESOURCE_ORIGIN")
    if origin:
        mirror, original = urlsplit(origin), urlsplit(url)
        url = urlunsplit((mirror.scheme, mirror.netloc, original.path, original.query, ""))
    with tempfile.TemporaryDirectory(prefix=".pymhm-resource-", dir=target.parent) as temporary:
        pending = Path(temporary) / target.name
        digest = hashlib.sha256()
        total = 0
        with urlopen(url, timeout=timeout) as response, pending.open("wb") as stream:
            while chunk := response.read(min(1024**2, limit + 1)):
                total += len(chunk)
                if total > limit:
                    raise ValueError(f"resource exceeds the {limit}-byte acquisition limit")
                digest.update(chunk)
                stream.write(chunk)
        if digest.hexdigest() != sha256:
            raise ValueError(f"SHA256 mismatch for downloaded resource: {url}")
        if target.exists():
            if file_digest(target) != sha256:
                raise ValueError(f"SHA256 mismatch for concurrently created resource: {target}")
        else:
            pending.replace(target)
    return target

Generic writable workspaces, external catalogues and literal source identities.

register_resources(directory: str | Path, manifest: Mapping[str, Any] | str | Path | None = None) -> None

Register external resources for one explicit directory without acquiring them.

A catalogue maps resources to records with sha256 and, for downloads, url and size_bytes. Records can declare related_resources such as attribution notices. An unavailable mapping provides reasons for absent original data. Pass a mapping or JSON path; the default is .pymhm-resources.json in the directory. The library contains no application catalogue or dataset.

Source code in src/pymhm/io/workspace.py
def register_resources(
    directory: str | Path, manifest: Mapping[str, Any] | str | Path | None = None
) -> None:
    """Register external resources for one explicit directory without acquiring them.

    A catalogue maps resources to records with sha256 and, for downloads, url
    and size_bytes. Records can declare related_resources such as attribution
    notices. An unavailable mapping provides reasons for absent original data.
    Pass a mapping or JSON path; the default is .pymhm-resources.json in the
    directory. The library contains no application catalogue or dataset.
    """
    root = Path(directory).expanduser().resolve()
    if isinstance(manifest, Mapping):
        payload = dict(manifest)
    else:
        path = root / RESOURCE_MANIFEST if manifest is None else Path(manifest)
        payload = json.loads(path.read_text(encoding="utf-8"))
    for key in ("resources", "unavailable"):
        entries = payload.setdefault(key, {})
        if not isinstance(entries, dict):
            raise ValueError(f"resource catalogue {key} must be a mapping")
        for name, record in entries.items():
            if _relative_name(name) != name or not isinstance(record, dict):
                raise ValueError("resource records require normalized names and mappings")
            if key == "resources":
                digest = record.get("sha256")
                if (
                    not isinstance(digest, str)
                    or len(digest) != 64
                    or any(c not in "0123456789abcdef" for c in digest)
                ):
                    raise ValueError("resource records require a lowercase SHA256 digest")
                if "url" in record:
                    if not isinstance(record["url"], str) or not record["url"]:
                        raise ValueError("resource URL must be a nonempty string")
                    positive_int(record.get("size_bytes"), "resource size bytes")
                for related in record.get("related_resources", []):
                    _relative_name(related)
            elif not isinstance(record.get("reason"), str):
                raise ValueError("unavailable resources require an explicit reason")
    if set(payload["resources"]) & set(payload["unavailable"]):
        raise ValueError("available and unavailable names must be disjoint")
    _CATALOGUES[root] = payload

case_workspace(directory: str | Path | None = None) -> Path

Return a writable directory without copying anything from the installation.

An explicit directory takes precedence over PYMHM_WORKSPACE and the current directory. A local .pymhm-resources.json is registered once when present. No resources are downloaded, and the process working directory is unchanged. Spawn workers can inherit the same environment and working directory.

Source code in src/pymhm/io/workspace.py
def case_workspace(directory: str | Path | None = None) -> Path:
    """Return a writable directory without copying anything from the installation.

    An explicit directory takes precedence over PYMHM_WORKSPACE and the current
    directory. A local .pymhm-resources.json is registered once when present.
    No resources are downloaded, and the process working directory is unchanged.
    Spawn workers can inherit the same environment and working directory.
    """
    root = (
        Path(directory if directory is not None else os.environ.get("PYMHM_WORKSPACE", Path.cwd()))
        .expanduser()
        .resolve()
    )
    root.mkdir(parents=True, exist_ok=True)
    if root not in _CATALOGUES and (root / RESOURCE_MANIFEST).is_file():
        register_resources(root)
    return root

resource_file(name: str | Path, *, directory: str | Path | None = None, package: str | None = None, cache: str | Path | None = None) -> Path

Resolve a workspace, explicitly catalogued or package-owned immutable input.

Without package, names are relative to the workspace. Missing catalogued inputs are acquired with their declared SHA256 and byte limit. An explicit package resolves that package's own relative resource, including zipped installations. Existing declared immutable files are verified. For mutable local inputs use local_resource; no PyMHM case assets are distributed.

Source code in src/pymhm/io/workspace.py
def resource_file(
    name: str | Path,
    *,
    directory: str | Path | None = None,
    package: str | None = None,
    cache: str | Path | None = None,
) -> Path:
    """Resolve a workspace, explicitly catalogued or package-owned immutable input.

    Without package, names are relative to the workspace. Missing catalogued
    inputs are acquired with their declared SHA256 and byte limit. An explicit
    package resolves that package's own relative resource, including zipped
    installations. Existing declared immutable files are verified. For mutable
    local inputs use local_resource; no PyMHM case assets are distributed.
    """
    relative = _relative_name(name)
    if package is not None:
        candidate = resources.files(package).joinpath(relative)
        if not candidate.is_file():
            raise FileNotFoundError(f"Package {package!r} does not supply resource {relative!r}")
        if isinstance(candidate, Path):
            return candidate
        payload = candidate.read_bytes()
        digest = hashlib.sha256(payload).hexdigest()
        target = _cache_directory(cache) / "resources" / digest / PurePosixPath(relative).name
        target.parent.mkdir(parents=True, exist_ok=True)
        if not target.is_file() or file_digest(target) != digest:
            target.write_bytes(payload)
        return target
    root = case_workspace(directory)
    candidate = root / relative
    catalogue = _catalogue(root)
    record = catalogue["resources"].get(relative)
    if candidate.is_file():
        if record is not None and file_digest(candidate) != record["sha256"]:
            raise ValueError(f"Declared resource differs from its catalogue: {relative}")
        return _remember_resource(candidate, relative)
    if record is not None and "url" in record:
        target = _cache_directory(cache) / "resources" / record["sha256"] / Path(relative).name
        return _remember_resource(
            download_resource(
                record["url"], target, sha256=record["sha256"], maximum_bytes=record["size_bytes"]
            ),
            relative,
        )
    if relative in catalogue["unavailable"]:
        raise FileNotFoundError(
            f"Resource {relative}: {catalogue['unavailable'][relative]['reason']}"
        )
    raise FileNotFoundError(f"Missing local or externally catalogued resource: {relative}")

materialize_resources(names: Iterable[str | Path], directory: str | Path, *, package: str | None = None) -> list[Path]

Copy requested resources and attribution files, preserving existing user files.

Destination symlinks cannot direct writes outside the directory. Only requested names and their explicitly declared related resources are acquired.

Source code in src/pymhm/io/workspace.py
def materialize_resources(
    names: Iterable[str | Path], directory: str | Path, *, package: str | None = None
) -> list[Path]:
    """Copy requested resources and attribution files, preserving existing user files.

    Destination symlinks cannot direct writes outside the directory. Only
    requested names and their explicitly declared related resources are acquired.
    """
    root = case_workspace(directory)
    pending = list(names)
    if package is None:
        for name in list(pending):
            pending.extend(
                _catalogue(root)["resources"].get(str(name), {}).get("related_resources", [])
            )
    copied = []
    for name in dict.fromkeys(pending):
        relative = _relative_name(name)
        destination = root / relative
        if not destination.resolve().is_relative_to(root):
            raise ValueError("resource destination escapes the working directory")
        if destination.exists():
            continue
        source = resource_file(relative, directory=root, package=package)
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(source, destination)
        copied.append(destination)
    return copied

ensure_resource(name: str | Path, directory: str | Path) -> Path

Acquire one declared input and its attribution files only when absent.

Source code in src/pymhm/io/workspace.py
def ensure_resource(name: str | Path, directory: str | Path) -> Path:
    """Acquire one declared input and its attribution files only when absent."""
    relative = _relative_name(name)
    materialize_resources([relative], directory)
    return Path(directory).expanduser().resolve() / relative

local_resource(path: str | Path, *, root: str | Path | None = None) -> Path

Acquire absent declared inputs while preserving existing local files.

Source code in src/pymhm/io/workspace.py
def local_resource(path: str | Path, *, root: str | Path | None = None) -> Path:
    """Acquire absent declared inputs while preserving existing local files."""
    value = Path(path)
    if value.exists():
        return value
    absolute = value.resolve()
    roots = (
        [case_workspace(root)]
        if root is not None
        else sorted(_CATALOGUES, key=lambda p: len(p.parts), reverse=True)
    )
    for directory in roots:
        if absolute.is_relative_to(directory):
            relative = absolute.relative_to(directory).as_posix()
            catalogue = _catalogue(directory)
            if relative in catalogue["resources"] or relative in catalogue["unavailable"]:
                return ensure_resource(relative, directory)
    return value

resource_glob(directory: str | Path, pattern: str, *, recursive: bool = False, root: str | Path | None = None) -> list[Path]

List sorted unique local and available external inputs matching a glob.

Wildcards match components; ** matches zero or more directories. recursive=True searches descendants as Path.rglob. Unavailable originals are never substituted. Acquisition checkpoint search must use filesystem globbing so published observations cannot silently become resumed results.

Source code in src/pymhm/io/workspace.py
def resource_glob(
    directory: str | Path,
    pattern: str,
    *,
    recursive: bool = False,
    root: str | Path | None = None,
) -> list[Path]:
    """List sorted unique local and available external inputs matching a glob.

    Wildcards match components; ** matches zero or more directories.
    recursive=True searches descendants as Path.rglob. Unavailable originals
    are never substituted. Acquisition checkpoint search must use filesystem
    globbing so published observations cannot silently become resumed results.
    """
    folder = Path(directory).expanduser().resolve()
    components = PurePosixPath(_relative_name(pattern)).parts
    if recursive:
        components = ("**", *components)
    found = set(folder.rglob(pattern) if recursive else folder.glob(pattern))
    roots = (
        [case_workspace(root)]
        if root is not None
        else sorted(_CATALOGUES, key=lambda p: len(p.parts), reverse=True)
    )
    for workspace in roots:
        if folder.is_relative_to(workspace):
            prefix = folder.relative_to(workspace).parts
            for name in _catalogue(workspace)["resources"]:
                parts = PurePosixPath(name).parts
                if parts[: len(prefix)] == prefix and _matches_glob(
                    parts[len(prefix) :], components
                ):
                    found.add(ensure_resource(name, workspace))
            break
    return sorted(found)

read_resource_text(path: str | Path, encoding: str | None = None, errors: str | None = None) -> str

Read local or declared external text with ordinary Path decoding rules.

Source code in src/pymhm/io/workspace.py
def read_resource_text(
    path: str | Path, encoding: str | None = None, errors: str | None = None
) -> str:
    """Read local or declared external text with ordinary Path decoding rules."""
    return local_resource(path).read_text(encoding=encoding, errors=errors)

read_resource_bytes(path: str | Path) -> bytes

Read literal bytes of a local or externally catalogued workspace input.

Source code in src/pymhm/io/workspace.py
def read_resource_bytes(path: str | Path) -> bytes:
    """Read literal bytes of a local or externally catalogued workspace input."""
    return local_resource(path).read_bytes()

workspace_from_archive(url: str, *, sha256: str, directory: str | Path | None = None, maximum_bytes: int = 64 * 1024 ** 2, maximum_extracted_bytes: int = 256 * 1024 ** 2, timeout: float = 60.0) -> Path

Acquire a verified ZIP and extract regular files into a writable workspace.

Download and expanded byte limits are independent. Absolute paths, traversal, duplicate names, symlinks, encryption and destination escapes are rejected before extraction. Existing local files are preserved. Extraction does not import or execute any code. An external catalogue named .pymhm-resources.json is registered after extraction when present.

Source code in src/pymhm/io/workspace.py
def workspace_from_archive(
    url: str,
    *,
    sha256: str,
    directory: str | Path | None = None,
    maximum_bytes: int = 64 * 1024**2,
    maximum_extracted_bytes: int = 256 * 1024**2,
    timeout: float = 60.0,
) -> Path:
    """Acquire a verified ZIP and extract regular files into a writable workspace.

    Download and expanded byte limits are independent. Absolute paths,
    traversal, duplicate names, symlinks, encryption and destination escapes
    are rejected before extraction. Existing local files are preserved.
    Extraction does not import or execute any code. An external catalogue
    named .pymhm-resources.json is registered after extraction when present.
    """
    limit = positive_int(maximum_extracted_bytes, "maximum extracted bytes")
    archive = download_resource(
        url,
        _cache_directory(None) / "archives" / sha256 / "resources.zip",
        sha256=sha256,
        maximum_bytes=maximum_bytes,
        timeout=timeout,
    )
    root = case_workspace(directory)
    with zipfile.ZipFile(archive) as zipped:
        members = zipped.infolist()
        names: set[str] = set()
        total = 0
        for member in members:
            name = _relative_name(member.filename)
            target = root / name
            if name in names or not target.resolve().is_relative_to(root):
                raise ValueError("archive has duplicate names or escaping destinations")
            names.add(name)
            mode = member.external_attr >> 16
            if member.flag_bits & 1 or (mode & 0o170000) not in (0, 0o100000, 0o040000):
                raise ValueError("archive supports only unencrypted regular files and directories")
            total += member.file_size
            if total > limit:
                raise ValueError("archive exceeds the maximum extracted byte limit")
        with tempfile.TemporaryDirectory(prefix=".pymhm-extract-", dir=root) as temporary:
            stage = Path(temporary)
            for member in members:
                target = root / _relative_name(member.filename)
                if member.is_dir() or target.exists():
                    continue
                pending = stage / _relative_name(member.filename)
                pending.parent.mkdir(parents=True, exist_ok=True)
                with zipped.open(member) as stream, pending.open("wb") as output:
                    shutil.copyfileobj(stream, output)
            for pending in sorted(stage.rglob("*")):
                if pending.is_file():
                    target = root / pending.relative_to(stage)
                    target.parent.mkdir(parents=True, exist_ok=True)
                    pending.replace(target)
    if (root / RESOURCE_MANIFEST).is_file():
        register_resources(root)
    return root

source_file(name: str | Path, *, root: str | Path | None = None) -> Path

Resolve a real workspace source or an importable package's logical owner.

src//... identifies that installed package. Other relative names first identify a workspace file, then an importable package with that name. Absolute paths retain their literal meaning. No missing development metadata or source identity is manufactured.

Source code in src/pymhm/io/workspace.py
def source_file(name: str | Path, *, root: str | Path | None = None) -> Path:
    """Resolve a real workspace source or an importable package's logical owner.

    src/<package>/... identifies that installed package. Other relative names
    first identify a workspace file, then an importable package with that name.
    Absolute paths retain their literal meaning. No missing development
    metadata or source identity is manufactured.
    """
    path = Path(name)
    if path.is_absolute():
        return path
    relative = _relative_name(name)
    base = Path(root) if root is not None else case_workspace()
    candidate = base / relative
    if not relative.startswith("src/") and candidate.exists():
        return candidate
    parts = PurePosixPath(relative).parts
    package, suffix = (
        (parts[1], parts[2:]) if parts[0] == "src" and len(parts) > 2 else (parts[0], parts[1:])
    )
    if suffix and path.suffix == ".py":
        try:
            directory = _package_directory(package)
        except ModuleNotFoundError:
            return candidate
        prefix = f"src/{package}" if parts[0] == "src" else package
        _SOURCE_ROOTS[directory] = prefix
        return directory.joinpath(*suffix)
    if relative in _catalogue(base.resolve())["resources"]:
        return resource_file(relative, directory=base)
    return candidate

source_label(path: str | Path, root: str | Path) -> str

Label literal files by their resource, source-owner or workspace path.

Source code in src/pymhm/io/workspace.py
def source_label(path: str | Path, root: str | Path) -> str:
    """Label literal files by their resource, source-owner or workspace path."""
    value = Path(path).resolve()
    if value in _RESOURCE_LABELS:
        return _RESOURCE_LABELS[value]
    _SOURCE_ROOTS.setdefault(_package_directory("pymhm"), "src/pymhm")
    for base, prefix in _SOURCE_ROOTS.items():
        if value.is_relative_to(base):
            return (PurePosixPath(prefix) / value.relative_to(base).as_posix()).as_posix()
    base = Path(root).resolve()
    return value.relative_to(base).as_posix() if value.is_relative_to(base) else value.as_posix()

source_identity(root: str | Path, files: Iterable[str | Path]) -> dict[str, str]

Fingerprint required source bytes with stable logical labels and SHA256.

Source code in src/pymhm/io/workspace.py
def source_identity(root: str | Path, files: Iterable[str | Path]) -> dict[str, str]:
    """Fingerprint required source bytes with stable logical labels and SHA256."""
    identities = {}
    for file in files:
        path = source_file(file, root=root)
        identities[source_label(path, root)] = file_digest(path)
    return dict(sorted(identities.items()))

Source fingerprints for fresh acquisitions, independent of archived result manifests.

current_source_manifest(entries: Mapping[str, str], *, packages: Iterable[str] = ('pymhm',)) -> dict[str, str]

Include every imported-package owner when producing a fresh source manifest.

New labels describe actual current paths and bytes, including reference basis, local elimination and execution owners. Existing acquisition manifests must be validated separately and must never be relabeled through this producer.

Source code in src/pymhm/io/provenance.py
def current_source_manifest(
    entries: Mapping[str, str], *, packages: Iterable[str] = ("pymhm",)
) -> dict[str, str]:
    """Include every imported-package owner when producing a fresh source manifest.

    New labels describe actual current paths and bytes, including reference basis,
    local elimination and execution owners. Existing acquisition manifests must be
    validated separately and must never be relabeled through this producer.
    """
    result = dict(entries)
    for name in packages:
        module = importlib.import_module(name)
        if module.__file__ is None:
            raise ValueError("source manifests require a package with identifiable source files")
        package = Path(module.__file__).resolve().parent
        prefix = "src/pymhm" if name == "pymhm" else name.replace(".", "/")
        result.update(
            {
                f"{prefix}/{path.relative_to(package).as_posix()}": file_digest(path)
                for path in sorted(package.rglob("*.py"))
            }
        )
    return result

file_digest(path: Path) -> str

Hash a source or archive in bounded memory, including large field files.

Source code in src/pymhm/io/provenance.py
def file_digest(path: Path) -> str:
    """Hash a source or archive in bounded memory, including large field files."""
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()

optional_file_digest(path: str | Path) -> str | None

Return SHA256 for an existing file, or None for unavailable metadata.

Use this for optional installation/development metadata such as a Pixi lockfile. Required scientific inputs must continue to use file_digest and its explicit filesystem errors. No placeholder digest is manufactured.

Source code in src/pymhm/io/provenance.py
def optional_file_digest(path: str | Path) -> str | None:
    """Return SHA256 for an existing file, or ``None`` for unavailable metadata.

    Use this for optional installation/development metadata such as a Pixi
    lockfile. Required scientific inputs must continue to use ``file_digest``
    and its explicit filesystem errors. No placeholder digest is manufactured.
    """
    source = Path(path)
    return file_digest(source) if source.is_file() else None

workspace_revision(directory: str | Path) -> str | None

Return the selected repository's actual HEAD, or None outside a repository.

A pip installation does not carry a Git checkout. This optional observation describes the workspace, not the identity of the installed numerical sources; source manifests retain the latter independently through their literal bytes.

Source code in src/pymhm/io/provenance.py
def workspace_revision(directory: str | Path) -> str | None:
    """Return the selected repository's actual HEAD, or ``None`` outside a repository.

    A pip installation does not carry a Git checkout. This optional observation
    describes the workspace, not the identity of the installed numerical sources;
    source manifests retain the latter independently through their literal bytes.
    """
    return _git_output(directory, ["rev-parse", "HEAD"])

workspace_git_dirty(directory: str | Path) -> bool | None

Report tracked/untracked workspace changes, or None without repository metadata.

Source code in src/pymhm/io/provenance.py
def workspace_git_dirty(directory: str | Path) -> bool | None:
    """Report tracked/untracked workspace changes, or ``None`` without repository metadata."""
    output = _git_output(directory, ["status", "--porcelain"])
    return None if output is None else bool(output)

Source modules

ECLIPSE property I/O and reservoir data with explicit physical units.

ECLIPSE values are ordered with I/x varying fastest. Permeability in the source files is in millidarcies and geometry is in feet. Neither coefficients nor coordinates are silently nondimensionalized. The OPM porosity file replaces zero porosity by 1e-7; this reader preserves those published values.

ReservoirData dataclass

Cellwise principal permeability and porosity with explicit physical units.

permeability has shape (nx, ny, nz, 3) and porosity has shape (nx, ny, nz). Components are Kx, Ky, Kz. Horizontal slices preserve the x/y ordering and original geometry, with no interpolation or upscaling.

__post_init__() -> None

Validate material ranges and store immutable independent arrays.

layer(number: int) -> tuple[CartesianCellField, CartesianCellField]

Return planar tensor K and scalar porosity for a one-based layer.

to_si() -> ReservoirData

Convert an mD/ft reservoir to m²/m; return SI input unchanged.

read_eclipse_properties(path: str | Path) -> dict[str, Array]

Read slash-terminated numeric property blocks, including count*value.

This deliberately reads property include files, not a complete simulator deck. Keywords must be unique, repetitions must specify a finite value, and malformed or unterminated blocks raise instead of truncating data. Returned flat arrays retain ECLIPSE I-fastest ordering.