CLN: Add types gridprop-import-xtgcpprop

mypy.ini

@@ -87,9 +87,6 @@ ignore_errors = True
 [mypy-xtgeo.grid3d._gridprop_import_grdecl]
 ignore_errors = True
 
-[mypy-xtgeo.grid3d._gridprop_import_xtgcpprop]
-ignore_errors = True
-
 [mypy-xtgeo.grid3d._gridprop_op1]
 ignore_errors = True
 

src/xtgeo/common/sys.py

@@ -202,7 +202,6 @@ def __init__(
                 f"a str, pathlib.Path, io.BytesIO, or io.StringIO."
             )
 
-        self._file: pathlib.Path | io.BytesIO | io.StringIO
         self._tmpfile: str | None = None
         self._delete_after = False  # delete file (e.g. tmp) afterwards
         self._mode = mode
@@ -216,7 +215,7 @@ def __init__(
         if isinstance(filelike, str):
             filelike = pathlib.Path(filelike)
 
-        self._file = filelike
+        self._file: pathlib.Path | io.BytesIO | io.StringIO = filelike
         self._memstream = isinstance(self._file, (io.BytesIO, io.StringIO))
 
         if obj and not self._memstream:

src/xtgeo/grid3d/_gridprop_import_xtgcpprop.py

@@ -1,18 +1,78 @@
 """GridProperty import function of xtgcpprop format."""
+from __future__ import annotations
 
 import json
+from collections.abc import Sequence
+from contextlib import contextmanager
+from io import BytesIO, StringIO
+from pathlib import Path
 from struct import unpack
+from typing import TYPE_CHECKING, Any, Generator
 
 import numpy as np
 
-import xtgeo
 import xtgeo.common.sys as xsys
 from xtgeo.common import null_logger
+from xtgeo.common.constants import UNDEF, UNDEF_INT
+from xtgeo.metadata.metadata import MetaDataCPProperty
 
 logger = null_logger(__name__)
 
+if TYPE_CHECKING:
+    from numpy.typing import DTypeLike
+
+    from xtgeo.common.sys import _XTGeoFile
+
+
+@contextmanager
+def _read_from_stream(
+    stream: BytesIO | StringIO,
+    size: int | None,
+    seek: int | None,
+) -> Generator[bytes, None, None]:
+    """Helper function to read from a stream with optional seeking."""
+    was_at = stream.tell()
+    if seek is not None:
+        stream.seek(seek)
+    try:
+        data = stream.read(size)
+        yield data if isinstance(data, bytes) else data.encode()
+    finally:
+        stream.seek(was_at)
+
+
+@contextmanager
+def _read_filelike(
+    filelike: Path | BytesIO | StringIO,
+    size: int | None = None,
+    seek: int | None = None,
+) -> Generator[bytes, None, None]:
+    """Context manager for reading a specified number of bytes from a file-like object.
+
+    Accepts either a Path, BytesIO, or StringIO object as input. The function reads
+    up to 'offset' bytes from the file-like object and yields these bytes.
+
+    For BytesIO and StringIO, the read operation preserves the original
+    file cursor position.
+    """
+
+    if isinstance(filelike, Path):
+        with filelike.open("rb") as f:
+            if seek is not None:
+                f.seek(seek)
+            yield f.read(size)
+    elif isinstance(filelike, (BytesIO, StringIO)):
+        with _read_from_stream(stream=filelike, size=size, seek=seek) as f:
+            yield f
+    else:
+        raise TypeError("Filelike must be one of: Path, BytesIO or StringIO.")
 
-def import_xtgcpprop(mfile, ijrange=None, zerobased=False):
+
+def import_xtgcpprop(
+    mfile: _XTGeoFile,
+    ijrange: Sequence[int] | None = None,
+    zerobased: bool = False,
+) -> dict[str, Any]:
     """Using pure python for experimental xtgcpprop import.
 
     Args:
@@ -23,21 +83,19 @@ def import_xtgcpprop(mfile, ijrange=None, zerobased=False):
 
     """
     offset = 36
-    with open(mfile.file, "rb") as fhandle:
-        buf = fhandle.read(offset)
 
-    # unpack header
-    swap, magic, nbyte, ncol, nrow, nlay = unpack("= i i i q q q", buf)
+    with _read_filelike(mfile.file, size=offset) as header:
+        # unpack header
+        swap, magic, nbyte, ncol, nrow, nlay = unpack("= i i i q q q", header)
 
     if swap != 1 or magic not in (1351, 1352):
         raise ValueError("Invalid file format (wrong swap id or magic number).")
 
     if magic == 1351:
-        dtype = np.float32 if nbyte == 4 else np.float64
+        dtype: DTypeLike = np.float32 if nbyte == 4 else np.float64
     else:
-        dtype = "int" + str(nbyte * 8)
+        dtype = f"int{nbyte * 8}"
 
-    vals = None
     narr = ncol * nrow * nlay
 
     ncolnew = nrownew = 0
@@ -51,35 +109,38 @@ def import_xtgcpprop(mfile, ijrange=None, zerobased=False):
         vals = xsys.npfromfile(mfile.file, dtype=dtype, count=narr, offset=offset)
 
     # read metadata which will be at position offet + nfloat*narr +13
-    pos = offset + nbyte * narr + 13
-
-    with open(mfile.file, "rb") as fhandle:
-        fhandle.seek(pos)
-        jmeta = fhandle.read().decode()
+    with _read_filelike(
+        mfile.file,
+        seek=offset + nbyte * narr + 13,
+    ) as _meta:
+        meta = json.loads(_meta, object_pairs_hook=dict)
 
-    meta = json.loads(jmeta, object_pairs_hook=dict)
     req = meta["_required_"]
 
-    reqattrs = xtgeo.MetaDataCPProperty.REQUIRED
-
-    result = dict()
-    for myattr in reqattrs:
-        result[myattr] = req[myattr]
+    result = {att: req[att] for att in MetaDataCPProperty.REQUIRED}
 
     if ijrange:
         result["ncol"] = ncolnew
         result["nrow"] = nrownew
 
     result["values"] = np.ma.masked_equal(
         vals.reshape((result["ncol"], result["nrow"], result["nlay"])),
-        xtgeo.UNDEF_INT if result["discrete"] else xtgeo.UNDEF,
+        UNDEF_INT if result["discrete"] else UNDEF,
     )
     return result
 
 
 def _import_xtgcpprop_partial(
-    mfile, nbyte, dtype, offset, ijrange, zerobased, ncol, nrow, nlay
-):
+    mfile: _XTGeoFile,
+    nbyte: int,
+    dtype: DTypeLike,
+    offset: int,
+    ijrange: Sequence[int],
+    zerobased: bool,
+    ncol: int,
+    nrow: int,
+    nlay: int,
+) -> tuple[np.ndarray, int, int]:
     """Partial import of a property."""
     i1, i2, j1, j2 = ijrange
     if not zerobased:

src/xtgeo/grid3d/_gridprop_lowlevel.py

@@ -68,7 +68,7 @@ def update_values_from_carray(
     values = np.asanyarray(values, order="C")
 
     # make it float64 or whatever(?) and mask it
-    self.values = values # type: ignore
+    self.values = values  # type: ignore
     self.mask_undef()
 
     # optionally delete the C array if needed