"""Acquisition provenance and reproducibility for IoT field sessions.
For publication and reviewer credibility, every IoT field session should
be able to emit a reproducible audit trail: what was recorded, where, by
whom, with which instrument, which windows were rejected, and the exact
bytes of the raw files. This module builds that trail as plain JSON-able
structures.
The routines are dependency-free (standard library only) and do not
import :mod:`pycsamt.iot.session`, so they can be reused independently.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import os
import platform
import zipfile
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import (
Any,
)
from ..api.property import PyCSAMTObject
from . import _common as _c
__all__ = [
"ProvenanceRecord",
"AcquisitionManifest",
"hash_raw_file",
"hash_bytes",
"hash_mapping",
"sign_mapping",
"verify_signature",
"verify_manifest",
"hash_chain",
"verify_hash_chain",
"log_qc_decision",
"build_acquisition_manifest",
"export_acquisition_manifest",
"export_station_audit",
"export_reproducibility_bundle",
]
#: JSON separators giving a canonical, whitespace-free encoding.
_CANONICAL = (",", ":")
def _tool_version() -> str:
try:
from importlib.metadata import version
return version("pycsamt")
except Exception: # noqa: BLE001 - version metadata may be absent
try:
import pycsamt
return getattr(pycsamt, "__version__", "unknown")
except Exception: # noqa: BLE001
return "unknown"
def _utc_iso(timestamp: float | None = None) -> str:
if timestamp is None:
dt = datetime.now(timezone.utc)
else:
dt = datetime.fromtimestamp(float(timestamp), tz=timezone.utc)
return dt.isoformat()
[docs]
def hash_bytes(data: bytes, *, algo: str = "sha256") -> str:
"""Return the hex digest of *data* using *algo*."""
hasher = hashlib.new(algo)
hasher.update(data)
return hasher.hexdigest()
[docs]
def hash_mapping(mapping: Mapping[str, Any], *, algo: str = "sha256") -> str:
"""Return a stable hash of *mapping* (key-sorted JSON, UTF-8)."""
encoded = json.dumps(
dict(mapping), sort_keys=True, default=str, separators=(",", ":")
).encode("utf-8")
return hash_bytes(encoded, algo=algo)
def _key_bytes(key: str | bytes) -> bytes:
if isinstance(key, str):
return key.encode("utf-8")
return bytes(key)
[docs]
def sign_mapping(
mapping: Mapping[str, Any],
key: str | bytes,
*,
algo: str = "sha256",
) -> str:
"""Return an HMAC signature over the canonical JSON of *mapping*.
Unlike :func:`hash_mapping`, which anyone can recompute, an HMAC
signature also proves the manifest was produced by a party holding the
shared *key* -- useful for asserting a field audit trail has not been
tampered with in transit.
"""
encoded = json.dumps(
dict(mapping), sort_keys=True, default=str, separators=_CANONICAL
).encode("utf-8")
return hmac.new(_key_bytes(key), encoded, algo).hexdigest()
[docs]
def verify_signature(
mapping: Mapping[str, Any],
signature: str,
key: str | bytes,
*,
algo: str = "sha256",
) -> bool:
"""Return whether *signature* is a valid HMAC of *mapping* under *key*.
Uses a constant-time comparison so verification does not leak timing
information.
"""
expected = sign_mapping(mapping, key, algo=algo)
return hmac.compare_digest(expected, str(signature))
[docs]
def verify_manifest(signed: Mapping[str, Any], key: str | bytes) -> bool:
"""Verify a signed manifest produced by :meth:`AcquisitionManifest.sign`.
Checks the HMAC *signature* over the wrapped ``manifest`` payload and,
when present, that the payload's own ``content_hash`` still matches.
"""
data = dict(signed)
payload = data.get("manifest")
signature = data.get("signature")
if not isinstance(payload, Mapping) or signature is None:
return False
algo = str(data.get("signature_algo", "hmac-sha256")).split("-", 1)[-1]
if not verify_signature(payload, signature, key, algo=algo):
return False
content_hash = dict(payload).get("content_hash")
if content_hash is not None:
recomputed = hash_mapping(
{k: v for k, v in dict(payload).items() if k != "content_hash"}
)
if not hmac.compare_digest(str(content_hash), recomputed):
return False
return True
[docs]
def hash_chain(
entries: Iterable[Mapping[str, Any]],
*,
algo: str = "sha256",
genesis: str = "",
) -> list[dict[str, Any]]:
"""Return a tamper-evident hash chain over *entries*.
Each returned entry is a copy of the input with ``seq``, ``prev_hash``,
and ``entry_hash`` added, where ``entry_hash`` folds in the previous
entry's hash. Altering, inserting, or reordering any entry breaks the
chain from that point on (detected by :func:`verify_hash_chain`). This
suits a running log such as per-window QC decisions.
"""
chained: list[dict[str, Any]] = []
prev = genesis
for i, entry in enumerate(entries):
item = dict(entry)
item["seq"] = i
item["prev_hash"] = prev
digest = hash_mapping(
{k: v for k, v in item.items() if k != "entry_hash"}, algo=algo
)
item["entry_hash"] = digest
prev = digest
chained.append(item)
return chained
[docs]
def verify_hash_chain(
chained: Iterable[Mapping[str, Any]],
*,
algo: str = "sha256",
genesis: str = "",
) -> bool:
"""Return whether a :func:`hash_chain` sequence is intact and in order."""
prev = genesis
for i, entry in enumerate(chained):
item = dict(entry)
if item.get("seq") != i or item.get("prev_hash") != prev:
return False
stored = item.get("entry_hash")
recomputed = hash_mapping(
{k: v for k, v in item.items() if k != "entry_hash"}, algo=algo
)
if stored is None or not hmac.compare_digest(str(stored), recomputed):
return False
prev = str(stored)
return True
[docs]
def hash_raw_file(
path: str,
*,
algo: str = "sha256",
chunk_size: int = 1 << 20,
) -> dict[str, Any]:
"""Hash a raw acquisition file and return its integrity record.
Returns a mapping with ``path``, ``bytes``, ``algo``, ``digest``, and
``modified_utc``. Raises :class:`FileNotFoundError` when the file is
missing.
"""
if not os.path.isfile(path):
raise FileNotFoundError(f"raw file not found: {path}")
hasher = hashlib.new(algo)
size = 0
with open(path, "rb") as handle:
while True:
block = handle.read(chunk_size)
if not block:
break
hasher.update(block)
size += len(block)
return dict(
path=os.path.abspath(path),
name=os.path.basename(path),
bytes=size,
algo=algo,
digest=hasher.hexdigest(),
modified_utc=_utc_iso(os.path.getmtime(path)),
)
[docs]
def log_qc_decision(
station: str,
decision: str,
*,
channel: str | None = None,
reasons: Iterable[str] | None = None,
operator: str | None = None,
timestamp: float | None = None,
window: tuple[float, float] | None = None,
) -> dict[str, Any]:
"""Return a normalised QC-decision record for the audit trail."""
return dict(
station=_c.as_nonempty_str(station, "station"),
channel=(str(channel).lower() if channel is not None else None),
decision=_c.as_nonempty_str(decision, "decision").lower(),
reasons=list(reasons or []),
operator=_c.as_optional_str(operator, "operator"),
logged_utc=_utc_iso(timestamp),
window=(list(window) if window is not None else None),
)
[docs]
@dataclass
class ProvenanceRecord(PyCSAMTObject):
"""Per-station occupation provenance for one field session."""
station_id: str
instrument_serial: str | None = None
firmware: str | None = None
operator: str | None = None
lat: float | None = None
lon: float | None = None
elevation: float | None = None
ex_azimuth_deg: float | None = None
ey_azimuth_deg: float | None = None
occupation_start: float | None = None
occupation_end: float | None = None
sample_rate_hz: float | None = None
gps_quality: str | None = None
battery_status: str | None = None
accepted_band_hz: tuple[float, float] | None = None
field_notes: str = ""
qc_decisions: list[dict[str, Any]] = field(default_factory=list)
rejected_windows: list[Any] = field(default_factory=list)
processing_steps: list[str] = field(default_factory=list)
raw_files: list[dict[str, Any]] = field(default_factory=list)
def __post_init__(self) -> None:
self.validate()
[docs]
def validate(self) -> None:
self.station_id = _c.as_nonempty_str(self.station_id, "station_id")
self.lat = _c.as_latitude(self.lat)
self.lon = _c.as_longitude(self.lon)
self.elevation = _c.as_elevation(self.elevation)
if self.sample_rate_hz is not None:
self.sample_rate_hz = _c.as_positive(
self.sample_rate_hz, "sample_rate_hz"
)
if self.occupation_start is not None:
self.occupation_start = _c.as_timestamp(
self.occupation_start, "occupation_start"
)
if self.occupation_end is not None:
self.occupation_end = _c.as_timestamp(
self.occupation_end, "occupation_end"
)
[docs]
@property
def occupation_seconds(self) -> float | None:
"""Occupation duration in seconds, if both bounds are known."""
if self.occupation_start is None or self.occupation_end is None:
return None
return float(self.occupation_end - self.occupation_start)
[docs]
def add_raw_file(
self, path: str, *, algo: str = "sha256"
) -> dict[str, Any]:
"""Hash *path* and attach the integrity record."""
record = hash_raw_file(path, algo=algo)
self.raw_files.append(record)
return record
[docs]
def as_dict(self) -> dict[str, Any]:
return dict(
station_id=self.station_id,
instrument_serial=self.instrument_serial,
firmware=self.firmware,
operator=self.operator,
lat=self.lat,
lon=self.lon,
elevation=self.elevation,
ex_azimuth_deg=self.ex_azimuth_deg,
ey_azimuth_deg=self.ey_azimuth_deg,
occupation_start=self.occupation_start,
occupation_end=self.occupation_end,
occupation_seconds=self.occupation_seconds,
sample_rate_hz=self.sample_rate_hz,
gps_quality=self.gps_quality,
battery_status=self.battery_status,
accepted_band_hz=(
list(self.accepted_band_hz)
if self.accepted_band_hz is not None
else None
),
field_notes=self.field_notes,
qc_decisions=list(self.qc_decisions),
rejected_windows=list(self.rejected_windows),
processing_steps=list(self.processing_steps),
raw_files=list(self.raw_files),
)
[docs]
@dataclass
class AcquisitionManifest(PyCSAMTObject):
"""Reproducible manifest describing one IoT acquisition session."""
survey_id: str
created_utc: str = field(default_factory=_utc_iso)
tool: str = "pycsamt.iot"
tool_version: str = field(default_factory=_tool_version)
method: str | None = None
operator: str | None = None
records: list[ProvenanceRecord] = field(default_factory=list)
devices: list[dict[str, Any]] = field(default_factory=list)
files: list[dict[str, Any]] = field(default_factory=list)
qc_decisions: list[dict[str, Any]] = field(default_factory=list)
processing_steps: list[str] = field(default_factory=list)
environment: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
self.validate()
[docs]
def validate(self) -> None:
self.survey_id = _c.as_nonempty_str(self.survey_id, "survey_id")
self.records = [
r
if isinstance(r, ProvenanceRecord)
else ProvenanceRecord(**dict(r))
for r in list(self.records or [])
]
if not self.environment:
self.environment = _default_environment()
[docs]
def add_record(
self, record: ProvenanceRecord | Mapping[str, Any]
) -> None:
self.records.append(
record
if isinstance(record, ProvenanceRecord)
else ProvenanceRecord(**dict(record))
)
[docs]
def add_file(self, path: str, *, algo: str = "sha256") -> dict[str, Any]:
record = hash_raw_file(path, algo=algo)
self.files.append(record)
return record
[docs]
def add_qc_decision(self, **kwargs: Any) -> dict[str, Any]:
decision = log_qc_decision(**kwargs)
self.qc_decisions.append(decision)
return decision
[docs]
def add_processing_step(self, step: str) -> None:
self.processing_steps.append(_c.as_nonempty_str(step, "step"))
[docs]
def as_dict(self) -> dict[str, Any]:
payload = dict(
survey_id=self.survey_id,
created_utc=self.created_utc,
tool=self.tool,
tool_version=self.tool_version,
method=self.method,
operator=self.operator,
environment=dict(self.environment),
devices=list(self.devices),
records=[r.as_dict() for r in self.records],
files=list(self.files),
qc_decisions=list(self.qc_decisions),
processing_steps=list(self.processing_steps),
metadata=dict(self.metadata),
)
payload["content_hash"] = hash_mapping(payload)
return payload
[docs]
def to_json(self, *, indent: int = 2) -> str:
return json.dumps(self.as_dict(), indent=indent, default=str)
[docs]
def write(self, path: str, *, indent: int = 2) -> str:
parent = os.path.dirname(os.path.abspath(path))
if parent:
os.makedirs(parent, exist_ok=True)
with open(path, "w", encoding="utf-8") as handle:
handle.write(self.to_json(indent=indent))
return os.path.abspath(path)
[docs]
def chained_qc_decisions(self) -> list[dict[str, Any]]:
"""Return the QC decisions as a tamper-evident hash chain."""
return hash_chain(self.qc_decisions)
[docs]
def sign(
self, key: str | bytes, *, algo: str = "sha256"
) -> dict[str, Any]:
"""Return an HMAC-signed envelope around the manifest.
The result wraps the manifest payload under ``manifest`` alongside
a ``signature`` and ``signature_algo``, so it can be written out and
later checked with :func:`verify_manifest` by a holder of *key*.
"""
payload = self.as_dict()
return dict(
manifest=payload,
signature=sign_mapping(payload, key, algo=algo),
signature_algo=f"hmac-{algo}",
)
[docs]
def write_signed(
self,
path: str,
key: str | bytes,
*,
algo: str = "sha256",
indent: int = 2,
) -> str:
"""Write an HMAC-signed manifest envelope to *path* (JSON)."""
parent = os.path.dirname(os.path.abspath(path))
if parent:
os.makedirs(parent, exist_ok=True)
with open(path, "w", encoding="utf-8") as handle:
json.dump(
self.sign(key, algo=algo), handle, indent=indent, default=str
)
return os.path.abspath(path)
def _default_environment() -> dict[str, Any]:
return dict(
python=platform.python_version(),
platform=platform.platform(),
machine=platform.machine(),
)
[docs]
def build_acquisition_manifest(
survey_id: str,
*,
records: Iterable[ProvenanceRecord | Mapping[str, Any]] | None = None,
devices: Iterable[Mapping[str, Any]] | None = None,
qc_decisions: Iterable[Mapping[str, Any]] | None = None,
processing_steps: Iterable[str] | None = None,
method: str | None = None,
operator: str | None = None,
metadata: Mapping[str, Any] | None = None,
) -> AcquisitionManifest:
"""Assemble an :class:`AcquisitionManifest` from session components."""
manifest = AcquisitionManifest(
survey_id=survey_id,
method=method,
operator=operator,
records=[
r
if isinstance(r, ProvenanceRecord)
else ProvenanceRecord(**dict(r))
for r in list(records or [])
],
devices=[dict(d) for d in list(devices or [])],
qc_decisions=[dict(q) for q in list(qc_decisions or [])],
processing_steps=list(processing_steps or []),
metadata=dict(metadata or {}),
)
return manifest
[docs]
def export_acquisition_manifest(
survey_id: str,
path: str,
*,
indent: int = 2,
**kwargs: Any,
) -> str:
"""Build an acquisition manifest and write it to *path*.
Thin convenience over :func:`build_acquisition_manifest` +
:meth:`AcquisitionManifest.write`; ``kwargs`` are forwarded to the
builder (``records``, ``devices``, ``qc_decisions``, ``method``, ...).
Returns the written path.
"""
manifest = build_acquisition_manifest(survey_id, **kwargs)
return manifest.write(path, indent=indent)
[docs]
def export_station_audit(
record: ProvenanceRecord | Mapping[str, Any],
path: str,
*,
indent: int = 2,
) -> str:
"""Write a single-station provenance audit to *path* (JSON)."""
prov = (
record
if isinstance(record, ProvenanceRecord)
else ProvenanceRecord(**dict(record))
)
parent = os.path.dirname(os.path.abspath(path))
if parent:
os.makedirs(parent, exist_ok=True)
audit = prov.as_dict()
audit["audit_hash"] = hash_mapping(audit)
with open(path, "w", encoding="utf-8") as handle:
json.dump(audit, handle, indent=indent, default=str)
return os.path.abspath(path)
[docs]
def export_reproducibility_bundle(
manifest: AcquisitionManifest,
out_dir: str,
*,
include_raw: bool = False,
zip_bundle: bool = False,
) -> dict[str, Any]:
"""Write a manifest and per-station audits into *out_dir*.
Parameters
----------
manifest : AcquisitionManifest
The session manifest to export.
out_dir : str
Destination directory (created if needed).
include_raw : bool
When ``True``, copy referenced raw files into ``out_dir/raw`` if
they exist on disk.
zip_bundle : bool
When ``True``, also produce ``<out_dir>.zip`` of the bundle.
Returns
-------
dict
Paths written: ``manifest``, ``audits``, optional ``raw`` and
``zip``.
"""
os.makedirs(out_dir, exist_ok=True)
manifest_path = os.path.join(out_dir, "acquisition_manifest.json")
manifest.write(manifest_path)
audits_dir = os.path.join(out_dir, "audits")
os.makedirs(audits_dir, exist_ok=True)
audit_paths: list[str] = []
for record in manifest.records:
audit_path = os.path.join(audits_dir, f"{record.station_id}.json")
export_station_audit(record, audit_path)
audit_paths.append(audit_path)
result: dict[str, Any] = dict(
manifest=os.path.abspath(manifest_path),
audits=audit_paths,
)
if include_raw:
import shutil
raw_dir = os.path.join(out_dir, "raw")
os.makedirs(raw_dir, exist_ok=True)
copied: list[str] = []
seen: list[dict[str, Any]] = list(manifest.files)
for record in manifest.records:
seen.extend(record.raw_files)
for entry in seen:
src = entry.get("path")
if src and os.path.isfile(src):
dst = os.path.join(raw_dir, os.path.basename(src))
shutil.copy2(src, dst)
copied.append(dst)
result["raw"] = copied
if zip_bundle:
zip_path = os.path.abspath(out_dir.rstrip(os.sep)) + ".zip"
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for root, _dirs, files in os.walk(out_dir):
for name in files:
full = os.path.join(root, name)
zf.write(full, os.path.relpath(full, out_dir))
result["zip"] = zip_path
return result
