pycsamt.iot.provenance

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 pycsamt.iot.session, so they can be reused independently.

Functions

build_acquisition_manifest(survey_id, *[, ...])	Assemble an AcquisitionManifest from session components.
export_acquisition_manifest(survey_id, path, *)	Build an acquisition manifest and write it to path.
export_reproducibility_bundle(manifest, ...)	Write a manifest and per-station audits into out_dir.
export_station_audit(record, path, *[, indent])	Write a single-station provenance audit to path (JSON).
hash_bytes(data, *[, algo])	Return the hex digest of data using algo.
hash_chain(entries, *[, algo, genesis])	Return a tamper-evident hash chain over entries.
hash_mapping(mapping, *[, algo])	Return a stable hash of mapping (key-sorted JSON, UTF-8).
hash_raw_file(path, *[, algo, chunk_size])	Hash a raw acquisition file and return its integrity record.
log_qc_decision(station, decision, *[, ...])	Return a normalised QC-decision record for the audit trail.
sign_mapping(mapping, key, *[, algo])	Return an HMAC signature over the canonical JSON of mapping.
verify_hash_chain(chained, *[, algo, genesis])	Return whether a hash_chain() sequence is intact and in order.
verify_manifest(signed, key)	Verify a signed manifest produced by AcquisitionManifest.sign().
verify_signature(mapping, signature, key, *)	Return whether signature is a valid HMAC of mapping under key.

Classes

AcquisitionManifest(survey_id[, ...])	Reproducible manifest describing one IoT acquisition session.
ProvenanceRecord(station_id[, ...])	Per-station occupation provenance for one field session.

class pycsamt.iot.provenance.ProvenanceRecord(station_id, instrument_serial=None, firmware=None, operator=None, lat=None, lon=None, elevation=None, ex_azimuth_deg=None, ey_azimuth_deg=None, occupation_start=None, occupation_end=None, sample_rate_hz=None, gps_quality=None, battery_status=None, accepted_band_hz=None, field_notes='', qc_decisions=<factory>, rejected_windows=<factory>, processing_steps=<factory>, raw_files=<factory>)

Bases: PyCSAMTObject

Per-station occupation provenance for one field session.

Parameters:

• station_id (str)

• instrument_serial (str | None)

• firmware (str | None)

• operator (str | None)

• lat (float | None)

• lon (float | None)

• elevation (float | None)

• ex_azimuth_deg (float | None)

• ey_azimuth_deg (float | None)

• occupation_start (float | None)

• occupation_end (float | None)

• sample_rate_hz (float | None)

• gps_quality (str | None)

• battery_status (str | None)

• accepted_band_hz (tuple[float, float] | None)

• field_notes (str)

• qc_decisions (list[dict[str, Any]])

• rejected_windows (list[Any])

• processing_steps (list[str])

• raw_files (list[dict[str, Any]])

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]]

rejected_windows: list[Any]

processing_steps: list[str]

raw_files: list[dict[str, Any]]

validate()

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

property occupation_seconds: float | None

Occupation duration in seconds, if both bounds are known.

add_raw_file(path, *, algo='sha256')

Hash path and attach the integrity record.

Parameters:

• path (str)

• algo (str)

Return type:

dict[str, Any]

as_dict()

Return type:

dict[str, Any]

class pycsamt.iot.provenance.AcquisitionManifest(survey_id, created_utc=<factory>, tool='pycsamt.iot', tool_version=<factory>, method=None, operator=None, records=<factory>, devices=<factory>, files=<factory>, qc_decisions=<factory>, processing_steps=<factory>, environment=<factory>, metadata=<factory>)

Bases: PyCSAMTObject

Reproducible manifest describing one IoT acquisition session.

Parameters:

• survey_id (str)

• created_utc (str)

• tool (str)

• tool_version (str)

• method (str | None)

• operator (str | None)

• records (list[ProvenanceRecord])

• devices (list[dict[str, Any]])

• files (list[dict[str, Any]])

• qc_decisions (list[dict[str, Any]])

• processing_steps (list[str])

• environment (dict[str, Any])

• metadata (dict[str, Any])

survey_id: str

created_utc: str

tool: str = 'pycsamt.iot'

tool_version: str

method: str | None = None

operator: str | None = None

records: list[ProvenanceRecord]

devices: list[dict[str, Any]]

files: list[dict[str, Any]]

qc_decisions: list[dict[str, Any]]

processing_steps: list[str]

environment: dict[str, Any]

metadata: dict[str, Any]

validate()

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

add_record(record)

Parameters:

record (ProvenanceRecord | Mapping[str, Any])

Return type:

None

add_file(path, *, algo='sha256')

Parameters:

• path (str)

• algo (str)

Return type:

dict[str, Any]

add_qc_decision(**kwargs)

Parameters:

kwargs (Any)

Return type:

dict[str, Any]

add_processing_step(step)

Parameters:

step (str)

Return type:

None

as_dict()

Return type:

dict[str, Any]

to_json(*, indent=2)

Parameters:

indent (int)

Return type:

str

write(path, *, indent=2)

Parameters:

• path (str)

• indent (int)

Return type:

str

chained_qc_decisions()

Return the QC decisions as a tamper-evident hash chain.

Return type:

list[dict[str, Any]]

sign(key, *, algo='sha256')

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 verify_manifest() by a holder of key.

Parameters:

• key (str | bytes)

• algo (str)

Return type:

dict[str, Any]

write_signed(path, key, *, algo='sha256', indent=2)

Write an HMAC-signed manifest envelope to path (JSON).

Parameters:

• path (str)

• key (str | bytes)

• algo (str)

• indent (int)

Return type:

str

pycsamt.iot.provenance.hash_raw_file(path, *, algo='sha256', chunk_size=1048576)

Hash a raw acquisition file and return its integrity record.

Returns a mapping with path, bytes, algo, digest, and modified_utc. Raises FileNotFoundError when the file is missing.

Parameters:

• path (str)

• algo (str)

• chunk_size (int)

Return type:

dict[str, Any]

pycsamt.iot.provenance.hash_bytes(data, *, algo='sha256')

Return the hex digest of data using algo.

Parameters:

• data (bytes)

• algo (str)

Return type:

str

pycsamt.iot.provenance.hash_mapping(mapping, *, algo='sha256')

Return a stable hash of mapping (key-sorted JSON, UTF-8).

Parameters:

• mapping (Mapping[str, Any])

• algo (str)

Return type:

str

pycsamt.iot.provenance.sign_mapping(mapping, key, *, algo='sha256')

Return an HMAC signature over the canonical JSON of mapping.

Unlike 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.

Parameters:

• mapping (Mapping[str, Any])

• key (str | bytes)

• algo (str)

Return type:

str

pycsamt.iot.provenance.verify_signature(mapping, signature, key, *, algo='sha256')

Return whether signature is a valid HMAC of mapping under key.

Uses a constant-time comparison so verification does not leak timing information.

Parameters:

• mapping (Mapping[str, Any])

• signature (str)

• key (str | bytes)

• algo (str)

Return type:

bool

pycsamt.iot.provenance.verify_manifest(signed, key)

Verify a signed manifest produced by AcquisitionManifest.sign().

Checks the HMAC signature over the wrapped manifest payload and, when present, that the payload’s own content_hash still matches.

Parameters:

• signed (Mapping[str, Any])

• key (str | bytes)

Return type:

bool

pycsamt.iot.provenance.hash_chain(entries, *, algo='sha256', genesis='')

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 verify_hash_chain()). This suits a running log such as per-window QC decisions.

Parameters:

• entries (Iterable[Mapping[str, Any]])

• algo (str)

• genesis (str)

Return type:

list[dict[str, Any]]

pycsamt.iot.provenance.verify_hash_chain(chained, *, algo='sha256', genesis='')

Return whether a hash_chain() sequence is intact and in order.

Parameters:

• chained (Iterable[Mapping[str, Any]])

• algo (str)

• genesis (str)

Return type:

bool

pycsamt.iot.provenance.log_qc_decision(station, decision, *, channel=None, reasons=None, operator=None, timestamp=None, window=None)

Return a normalised QC-decision record for the audit trail.

Parameters:

• station (str)

• decision (str)

• channel (str | None)

• reasons (Iterable[str] | None)

• operator (str | None)

• timestamp (float | None)

• window (tuple[float, float] | None)

Return type:

dict[str, Any]

pycsamt.iot.provenance.build_acquisition_manifest(survey_id, *, records=None, devices=None, qc_decisions=None, processing_steps=None, method=None, operator=None, metadata=None)

Assemble an AcquisitionManifest from session components.

Parameters:

• survey_id (str)

• records (Iterable[ProvenanceRecord | Mapping[str, Any]] | None)

• devices (Iterable[Mapping[str, Any]] | None)

• qc_decisions (Iterable[Mapping[str, Any]] | None)

• processing_steps (Iterable[str] | None)

• method (str | None)

• operator (str | None)

• metadata (Mapping[str, Any] | None)

Return type:

AcquisitionManifest

pycsamt.iot.provenance.export_acquisition_manifest(survey_id, path, *, indent=2, **kwargs)

Build an acquisition manifest and write it to path.

Thin convenience over build_acquisition_manifest() + AcquisitionManifest.write(); kwargs are forwarded to the builder (records, devices, qc_decisions, method, …). Returns the written path.

Parameters:

• survey_id (str)

• path (str)

• indent (int)

• kwargs (Any)

Return type:

str

pycsamt.iot.provenance.export_station_audit(record, path, *, indent=2)

Write a single-station provenance audit to path (JSON).

Parameters:

• record (ProvenanceRecord | Mapping[str, Any])

• path (str)

• indent (int)

Return type:

str

pycsamt.iot.provenance.export_reproducibility_bundle(manifest, out_dir, *, include_raw=False, zip_bundle=False)

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:

Paths written: manifest, audits, optional raw and zip.

Return type:

dict