File size: 3,299 Bytes
80b01cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """Read numeric NPZ input without allowing embedded Python pickles."""
from __future__ import annotations
from pathlib import Path
from typing import Any
def read_npz(path: str | Path, *, require_y: bool = False) -> dict[str, Any]:
import numpy as np
path = Path(path)
with np.load(path, allow_pickle=False) as archive:
if "X" not in archive:
raise ValueError("NPZ must contain X with shape (n_windows,n_timepoints,n_channels)")
if require_y and "y" not in archive:
raise ValueError("NPZ must contain encoded integer y for this command")
try:
out = {key: np.asarray(archive[key]) for key in
("X", "y", "subject", "time_index", "label_names", "sampling_rate_hz")
if key in archive}
except ValueError as exc:
raise ValueError("NPZ object arrays are not supported; export numeric arrays and "
"Unicode strings, not pickled metadata") from exc
X = out["X"]
if X.ndim != 3 or not all(X.shape):
raise ValueError("X must have nonempty shape (n_windows,n_timepoints,n_channels)")
if "y" in out:
y = out["y"]
if y.ndim != 1 or len(y) != len(X) or not np.issubdtype(y.dtype, np.integer):
raise ValueError("y must be a one-dimensional integer array aligned with X")
if y.size and y.min() < 0:
raise ValueError("y must use nonnegative global class IDs")
for key in ("subject", "time_index"):
if key in out and (out[key].ndim != 1 or len(out[key]) != len(X)):
raise ValueError(f"{key} must be one-dimensional and aligned with X")
if "label_names" in out:
if out["label_names"].ndim != 1:
raise ValueError("label_names must be a one-dimensional global class vocabulary")
out["label_names"] = [str(value) for value in out["label_names"].tolist()]
if "y" in out and out["y"].size and out["y"].max() >= len(out["label_names"]):
raise ValueError("y is incompatible with label_names")
return out
def as_har_data(data: dict[str, Any], *, n_classes: int | None = None, source: str = "NPZ") -> Any:
import numpy as np
from wisp.core.data import HARData
if "y" not in data:
raise ValueError("y is required to construct a training/evaluation dataset")
names = data.get("label_names")
if names is None:
if n_classes is None or n_classes < 1:
raise ValueError("Supply the global label_names in NPZ or explicit --n-classes; "
"a fold's observed labels must not redefine the class axis")
names = [str(index) for index in range(n_classes)]
elif n_classes is not None and n_classes != len(names):
raise ValueError("n_classes differs from the supplied label_names vocabulary")
if data["y"].size and data["y"].max() >= len(names):
raise ValueError("y contains a class ID outside the global vocabulary")
rate = data.get("sampling_rate_hz")
return HARData(X=data["X"], y=data["y"], subject=data.get("subject"),
time_index=data.get("time_index"), label_names=names, source=source,
sampling_rate_hz=None if rate is None else float(np.asarray(rate).reshape(-1)[0]))
|