"""Bounded, identity-pinned ingestion of the Stat2Data Volts trace. This optional research adapter does not execute R or admit a physical model. The single acquisition is preserved; splitting it does not create independent experimental runs. No download or analysis happens at import time. """ from __future__ import annotations import hashlib import importlib from pathlib import Path import warnings import zlib from tnfr.validation.nodal_prediction import NodalMeasurementRun SOURCE_SHA256 = "e50e505783dbf9b6d477165c92328e805e5d8bb0d27e3458862e2d5ba7f00a52" SOURCE_BYTES = 767 MAX_COMPRESSED_BYTES = 1024 MAX_EXPANDED_BYTES = 65536 RDATA_VERSION = "1.1.0" RUN_ID = "Stat2Data-Volts-single-trace" def _rdata_module(): try: module = importlib.import_module("rdata") except ImportError as exc: raise ImportError( "Volts ingestion requires the optional research dependency rdata==1.1.0" ) from exc if module.__version__ != RDATA_VERSION: raise ImportError( "Volts ingestion requires rdata==1.1.0 for reproducible decoding" ) return module def _bounded_expand(raw: bytes) -> bytes: """Expand exactly one complete gzip/zlib stream, never nested compression.""" if not raw or len(raw) > MAX_COMPRESSED_BYTES: raise ValueError("compressed input exceeds the Volts byte bound") decoder = zlib.decompressobj(wbits=47) try: expanded = decoder.decompress(raw, MAX_EXPANDED_BYTES + 1) except zlib.error as exc: raise ValueError("Volts input must be a valid gzip/zlib stream") from exc if len(expanded) > MAX_EXPANDED_BYTES or decoder.unconsumed_tail: raise ValueError("expanded input exceeds the Volts byte bound") if not decoder.eof or decoder.unused_data: raise ValueError( "Volts compression must be complete with no trailing members or bytes" ) if not expanded.startswith((b"RDX2\nX\n", b"RDX3\nX\n")): raise ValueError( "expected uncompressed XDR RData; nested compression is forbidden" ) return expanded def _validate_plain_tree(parsed, module) -> None: """Reject executable/custom R types before invoking any class conversion.""" types = module.parser.RObjectType allowed = { types.NILVALUE, types.SYM, types.LIST, types.CHAR, types.INT, types.REAL, types.STR, types.VEC, types.REF, } pending = [(parsed.object, 0)] seen = set() while pending: obj, depth = pending.pop() if depth > 32 or len(seen) >= 512: raise ValueError("R object structure exceeds the Volts bound") if id(obj) in seen: continue seen.add(id(obj)) kind = obj.info.type if kind not in allowed or (obj.info.object and kind != types.VEC): raise ValueError("only plain data.frame and numeric R objects are accepted") if kind == types.REF: target = obj.referenced_object if target is None or target.info.type != types.SYM: raise ValueError( "only previously defined R symbol references are accepted" ) pending.append((target, depth + 1)) for child in (obj.attributes, obj.tag): if child is not None: pending.append((child, depth + 1)) if kind in {types.LIST, types.STR, types.VEC}: if len(obj.value) > 64: raise ValueError("R vector length exceeds the Volts bound") pending.extend((child, depth + 1) for child in obj.value) elif kind == types.SYM: pending.append((obj.value, depth + 1)) elif kind in {types.INT, types.REAL}: if obj.value.ndim != 1 or len(obj.value) > 50: raise ValueError("R numeric vector exceeds the Volts row bound") if obj.attributes is not None: raise ValueError("Volts numeric columns must have no custom attributes") elif kind == types.CHAR: if obj.value is None or len(obj.value) > 128: raise ValueError("invalid or oversized R metadata string") try: obj.value.decode("ascii") except UnicodeDecodeError as exc: raise ValueError( "Volts metadata must contain only ASCII characters" ) from exc def _validated_run(converted) -> NodalMeasurementRun: import numpy as np import pandas as pd if type(converted) is not dict or tuple(converted) != ("Volts",): raise ValueError("RData must contain exactly one dataset named Volts") frame = converted["Volts"] if type(frame) is not pd.DataFrame or frame.shape != (50, 2): raise ValueError( "Volts must be a plain data.frame with exactly 50 rows and two columns" ) if tuple(frame.columns) != ("Voltage", "Time"): raise ValueError("Volts columns must be exactly Voltage, Time in that order") for column in frame.columns: if frame[column].dtype.kind not in "fiu": raise ValueError( "Volts columns must be real numeric, not boolean or textual" ) values = frame.to_numpy(dtype=float, na_value=np.nan) if not np.isfinite(values).all(): raise ValueError("Volts measurements and timestamps must all be finite") return NodalMeasurementRun( run_id=RUN_ID, acquisition_id=RUN_ID, channel_ids=("Voltage",), timestamps=tuple(values[:, 1]), samples=(tuple(values[:, 0]),), value_unit="V", time_unit="s", ) def _decode_expanded(expanded: bytes) -> tuple[NodalMeasurementRun, dict]: """Internal synthetic-test seam; public ingestion first authenticates bytes.""" if len(expanded) > MAX_EXPANDED_BYTES: raise ValueError("expanded input exceeds the Volts byte bound") if not expanded.startswith((b"RDX2\nX\n", b"RDX3\nX\n")): raise ValueError( "expected uncompressed XDR RData; nested compression is forbidden" ) module = _rdata_module() parsed = module.parser.parse_data( expanded, expand_altrep=False, altrep_constructor_dict={}, extension=".rda", ) _validate_plain_tree(parsed, module) def plain_frame(obj, attrs): if set(attrs) != {"names", "class", "row.names"} or tuple(attrs["class"]) != ( "data.frame", ): raise ValueError( "only the plain data.frame class and metadata are accepted" ) return module.conversion.dataframe_constructor(obj, attrs) with warnings.catch_warnings(): warnings.simplefilter("error") converted = module.conversion.convert( parsed, constructor_dict={"data.frame": plain_frame}, default_encoding="ascii", ) run = _validated_run(converted) return run, { "expanded_bytes": len(expanded), "expanded_sha256": hashlib.sha256(expanded).hexdigest(), "rdata_version": module.__version__, "serialization_format": parsed.versions.format, } def load_volts(path: str | Path) -> tuple[NodalMeasurementRun, dict]: """Load only the preregistered archive; preserve every original observation.""" with Path(path).open("rb") as source: raw = source.read(MAX_COMPRESSED_BYTES + 1) if len(raw) > MAX_COMPRESSED_BYTES: raise ValueError("compressed input exceeds the Volts byte bound") digest = hashlib.sha256(raw).hexdigest() if len(raw) != SOURCE_BYTES or digest != SOURCE_SHA256: raise ValueError( "Volts source size or SHA-256 differs from the preregistered archive" ) expanded = _bounded_expand(raw) run, metadata = _decode_expanded(expanded) metadata.update( { "source_bytes": len(raw), "source_sha256": digest, "dataset": "Volts", "rows": 50, "columns": ("Voltage", "Time"), "acquisition_id": RUN_ID, "split_scope": "within_single_acquisition", "physical_status": "not_admitted", } ) return run, metadata