Provenance and Reproducibility#

Provenance records explain what happened during acquisition and make a field session reproducible after the instruments have left the site. In an IoT workflow, that audit trail should include station occupation metadata, devices, QC decisions, raw-file integrity hashes, processing steps, and the environment used to create the manifest.

This page uses one real file from the repository demo data, data/AMT/WILLY_DATA/L18PLT/18-001A.edi, for the raw-file hash. The IoT QC packets are synthetic because provenance is documenting acquisition events around the data, not reprocessing the EDI file itself.

Hash A Raw File#

Use pycsamt.iot.hash_raw_file() to create an integrity record. The full digest is stored in manifests; the example prints only a prefix to keep the documentation readable.

 1from pathlib import Path
 2
 3from pycsamt.iot import hash_raw_file
 4
 5raw_file = Path("data/AMT/WILLY_DATA/L18PLT/18-001A.edi")
 6raw_hash = hash_raw_file(str(raw_file))
 7print(
 8    {
 9        "name": raw_hash["name"],
10        "bytes": raw_hash["bytes"],
11        "algo": raw_hash["algo"],
12        "digest_prefix": raw_hash["digest"][:16],
13    }
14)

Output:

{
  "name": "18-001A.edi",
  "bytes": 22758,
  "algo": "sha256",
  "digest_prefix": "a7bc0c7b290c1847"
}

Log QC Decisions#

Use pycsamt.iot.log_qc_decision() for normalized audit records. It standardizes station IDs, channel names, decisions, reasons, timestamps, and optional acquisition windows.

 1from pycsamt.iot import log_qc_decision
 2
 3qc_reject = log_qc_decision(
 4    "001A",
 5    "reject",
 6    channel="ey",
 7    reasons=["finite_coverage_below_threshold"],
 8    operator="pyCSAMT documentation",
 9    timestamp=1_700_000_120.0,
10    window=(1_700_000_060.0, 1_700_000_120.0),
11)
12print(qc_reject)

Output:

{
  "station": "001A",
  "channel": "ey",
  "decision": "reject",
  "reasons": [
    "finite_coverage_below_threshold"
  ],
  "operator": "pyCSAMT documentation",
  "logged_utc": "2023-11-14T22:15:20+00:00",
  "window": [
    1700000060.0,
    1700000120.0
  ]
}

Build A Session Manifest#

The easiest route is to build a pycsamt.iot.FieldSession, add devices, stations, and packets, then call pycsamt.iot.FieldSession.to_manifest(). The manifest can still be enriched afterward with raw-file hashes, extra QC decisions, and processing steps.

  1from pycsamt.iot import (
  2    DeviceConfig,
  3    FieldSession,
  4    MonitoringConfig,
  5    StationConfig,
  6    TelemetryPacket,
  7)
  8
  9device = DeviceConfig(
 10    "l18-node-01",
 11    station="001A",
 12    protocol="file",
 13    sample_rate_hz=512.0,
 14    channels=["ex", "ey", "hx", "hy"],
 15    role="recorder",
 16    metadata={
 17        "instrument_serial": "DOC-REC-001",
 18        "firmware": "demo-2.0",
 19        "source_edi": raw_file.as_posix(),
 20    },
 21)
 22station = StationConfig(
 23    "001A",
 24    profile="L18",
 25    position_m=0.0,
 26    channels=["ex", "ey", "hx", "hy"],
 27    dipole_length_m=50.0,
 28    ex_azimuth_deg=90.0,
 29    ey_azimuth_deg=0.0,
 30    device_ids=[device.device_id],
 31    operator="pyCSAMT documentation",
 32    notes="Station id and raw EDI file are from the repository demo data.",
 33)
 34session = FieldSession(
 35    "WILLY-L18-PROVENANCE-DEMO",
 36    devices=[device],
 37    stations=[station],
 38    method="amt",
 39    operator="pyCSAMT documentation",
 40    monitoring_config=MonitoringConfig(
 41        method="amt",
 42        required_channels=["ex", "ey", "hx", "hy"],
 43        frequency_band_hz=(1.0, 1000.0),
 44    ),
 45    metadata={"real_data_path": "data/AMT/WILLY_DATA/L18PLT"},
 46)
 47
 48for index, (accepted, decision, reasons) in enumerate(
 49    [
 50        (True, "accept", []),
 51        (False, "reject", ["finite_coverage_below_threshold"]),
 52    ]
 53):
 54    session.add_packet(
 55        TelemetryPacket.from_device(
 56            device,
 57            timestamp=1_700_000_000.0 + 60.0 * index,
 58            survey_id=session.survey_id,
 59            kind="qc",
 60            payload={
 61                "method": "amt",
 62                "station": "001A",
 63                "channels": ["ex", "ey", "hx", "hy"],
 64                "frequency_band_hz": [1.0, 1000.0],
 65                "accepted": accepted,
 66                "decision": decision,
 67                "reasons": reasons,
 68                "battery_v": 12.4 - 0.1 * index,
 69            },
 70        )
 71    )
 72
 73manifest = session.to_manifest()
 74manifest.created_utc = "2023-11-14T22:13:20+00:00"
 75manifest.environment = {
 76    "python": "3.10",
 77    "platform": "documentation",
 78    "machine": "example",
 79}
 80for record in manifest.records:
 81    for item in record.qc_decisions:
 82        item["logged_utc"] = "2023-11-14T22:13:20+00:00"
 83manifest.add_file(str(raw_file))
 84manifest.add_qc_decision(
 85    station="001A",
 86    decision="reject",
 87    channel="ey",
 88    reasons=["finite_coverage_below_threshold"],
 89    operator="pyCSAMT documentation",
 90    timestamp=1_700_000_120.0,
 91    window=(1_700_000_060.0, 1_700_000_120.0),
 92)
 93manifest.add_processing_step("loaded station metadata from L18 demo line")
 94manifest.add_processing_step("recorded synthetic IoT QC packet decisions")
 95
 96manifest_dict = manifest.as_dict()
 97print(
 98    {
 99        "survey_id": manifest_dict["survey_id"],
100        "method": manifest_dict["method"],
101        "operator": manifest_dict["operator"],
102        "n_records": len(manifest_dict["records"]),
103        "n_devices": len(manifest_dict["devices"]),
104        "n_files": len(manifest_dict["files"]),
105        "n_qc_decisions": len(manifest_dict["qc_decisions"]),
106        "content_hash_prefix": manifest_dict["content_hash"][:16],
107    }
108)

Output:

{
  "survey_id": "WILLY-L18-PROVENANCE-DEMO",
  "method": "amt",
  "operator": "pyCSAMT documentation",
  "n_records": 1,
  "n_devices": 1,
  "n_files": 1,
  "n_qc_decisions": 3,
  "content_hash_prefix": "f609c0a3c301389a"
}

Inspect A Station Record#

Each station record collects occupation metadata and QC decisions for that station. Session-derived records are convenient when packets already carry station, method, channel, and battery metadata.

 1record = manifest.as_dict()["records"][0]
 2print(
 3    {
 4        "station_id": record["station_id"],
 5        "operator": record["operator"],
 6        "sample_rate_hz": record["sample_rate_hz"],
 7        "battery_status": record["battery_status"],
 8        "accepted_band_hz": record["accepted_band_hz"],
 9        "qc_decisions": record["qc_decisions"],
10    }
11)

Output:

{
  "station_id": "001A",
  "operator": "pyCSAMT documentation",
  "sample_rate_hz": null,
  "battery_status": null,
  "accepted_band_hz": [
    1.0,
    1000.0
  ],
  "qc_decisions": [
    {
      "station": "001A",
      "channel": null,
      "decision": "accept",
      "reasons": [],
      "operator": "pyCSAMT documentation",
      "logged_utc": "2023-11-14T22:13:20+00:00",
      "window": null
    },
    {
      "station": "001A",
      "channel": null,
      "decision": "reject",
      "reasons": [
        "finite_coverage_below_threshold"
      ],
      "operator": "pyCSAMT documentation",
      "logged_utc": "2023-11-14T22:13:20+00:00",
      "window": null
    }
  ]
}

Build A Manual Audit#

Use pycsamt.iot.ProvenanceRecord and pycsamt.iot.build_acquisition_manifest() when you are assembling a manifest from a station sheet, field notes, and file hashes instead of a live FieldSession.

 1from pycsamt.iot import ProvenanceRecord, build_acquisition_manifest
 2
 3qc_accept = log_qc_decision(
 4    "001A",
 5    "accept",
 6    channel="ex",
 7    reasons=["finite_coverage_ok", "spike_fraction_ok"],
 8    operator="pyCSAMT documentation",
 9    timestamp=1_700_000_060.0,
10    window=(1_700_000_000.0, 1_700_000_060.0),
11)
12manual_record = ProvenanceRecord(
13    station_id="001A",
14    instrument_serial="DOC-REC-001",
15    firmware="demo-2.0",
16    operator="pyCSAMT documentation",
17    ex_azimuth_deg=90.0,
18    ey_azimuth_deg=0.0,
19    occupation_start=1_700_000_000.0,
20    occupation_end=1_700_000_120.0,
21    sample_rate_hz=512.0,
22    gps_quality="locked",
23    battery_status="12.3 V",
24    accepted_band_hz=(1.0, 1000.0),
25    field_notes="Manual audit record built from station sheet and QC logs.",
26    qc_decisions=[qc_accept, qc_reject],
27    processing_steps=["edge finite-coverage screening", "operator review"],
28)
29manual_record.add_raw_file(str(raw_file))
30
31manual_manifest = build_acquisition_manifest(
32    "WILLY-L18-MANUAL-AUDIT",
33    records=[manual_record],
34    devices=[device.as_dict()],
35    qc_decisions=[qc_accept, qc_reject],
36    processing_steps=["manual audit assembled for documentation"],
37    method="amt",
38    operator="pyCSAMT documentation",
39    metadata={"source": "documentation example"},
40)
41manual_manifest.created_utc = "2023-11-14T22:13:20+00:00"
42manual_manifest.environment = {
43    "python": "3.10",
44    "platform": "documentation",
45    "machine": "example",
46}
47print(manual_manifest.as_dict()["content_hash"][:16])

Output:

12d8f383d727517a

Export A Reproducibility Bundle#

Use pycsamt.iot.export_reproducibility_bundle() to write the manifest and one JSON audit per station. Set zip_bundle=True when you also want a zip archive for hand-off. Set include_raw=True only when you intentionally want raw files copied into the bundle.

 1from pycsamt.iot import (
 2    export_reproducibility_bundle,
 3    export_station_audit,
 4)
 5
 6out_dir = "docs/source/user_guide/iot/_provenance_demo_bundle"
 7bundle = export_reproducibility_bundle(
 8    manual_manifest,
 9    out_dir,
10    zip_bundle=True,
11)
12audit_path = export_station_audit(
13    manual_record,
14    f"{out_dir}/single_station_audit.json",
15)
16print(
17    {
18        "manifest_name": Path(bundle["manifest"]).name,
19        "audit_count": len(bundle["audits"]),
20        "zip_name": Path(bundle["zip"]).name,
21        "single_station_audit": Path(audit_path).name,
22    }
23)

Output:

{
  "manifest_name": "acquisition_manifest.json",
  "audit_count": 1,
  "zip_name": "_provenance_demo_bundle.zip",
  "single_station_audit": "single_station_audit.json"
}

Field Interpretation#

The manifest gives reviewers enough information to understand and verify the acquisition trail: which station was occupied, which device was used, which QC decisions were made, which raw file was referenced, and whether the manifest content changed after export. The content_hash is a stable digest of the manifest payload. Raw-file hashes protect the integrity of data files, while station audits preserve the field context around those files.

For production surveys, write the manifest beside the processed products and keep the raw-file hash records even when raw files are too large to bundle. That makes later reporting, reprocessing, and external review much less fragile.