"""Generate release-level manifest, schema, and timing metadata. Run from any location with Python 3.12+ and NumPy available: py -3.12 processing/generate_release_metadata.py The intended release excludes D_250W/Raw_data/Processed_data because it is a duplicate derived-data tree. The manifest represents the release directory, not the legacy ZIP archive until that archive is rebuilt from the same scope. """ from __future__ import annotations import csv import hashlib import json from pathlib import Path from typing import Any import numpy as np ROOT = Path(__file__).resolve().parents[1] VERSION = "1.0.0" EXCLUDED_PREFIX = ("D_250W", "Raw_data", "Processed_data") SELF_HASH = "SELF_REFERENTIAL_NOT_APPLICABLE" def release_files() -> list[Path]: return [ path for path in sorted(ROOT.rglob("*")) if path.is_file() and path.name != "manifest.csv" and path.relative_to(ROOT).parts[:3] != EXCLUDED_PREFIX and "__pycache__" not in path.relative_to(ROOT).parts ] def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def case_id(relative_path: Path) -> str: return { "A_150W": "Boiling-412", "B_180W": "Boiling-413", "C_230W": "Boiling-416", "D_250W": "Boiling-417", }.get(relative_path.parts[0], "release-level") def file_class(relative_path: Path) -> str: if "Raw_data" in relative_path.parts: return "raw_instrument_export" if "Processed_data" in relative_path.parts: return "derived_analysis_product" if relative_path.parts[0] in {"processing", "calibration"}: return "release_documentation_or_code" return "release_metadata" def infer_dtype(values: list[str]) -> str: nonempty = [value for value in values if value not in {"", "NA", "N/A", "null", "None"}] if not nonempty: return "string_or_missing" try: [int(value) for value in nonempty] return "int64" except ValueError: pass try: [float(value) for value in nonempty] return "float64" except ValueError: return "string" def unit_from_name(name: str) -> str: suffixes = { "_s": "s", "_ms": "ms", "_Hz": "Hz", "_kHz": "kHz", "_kPa": "kPa", "_Pa": "Pa", "_C": "degC", "_K": "K", "_W": "W", "_W_cm2": "W cm^-2", "_W_cm2K": "W cm^-2 K^-1", "_V2": "V^2", "_dB": "dB", } for suffix, unit in suffixes.items(): if name.endswith(suffix): return unit return "dimensionless_or_not_applicable" def field_description(name: str) -> str: if name == "time_reference_temperature_s": return "Time in seconds on the retained Temperature.lvm reference clock." if name == "event_name": return "Operational event label defined in event_definition.md." if name == "definition": return "Machine-readable operational definition assigned by processing." if name == "evidence_status": return "Evidence class; screening and not_confirmed are not validated physical-event labels." if name.startswith("temperature_"): return "Temperature-derived quantity; unit and coordinate/time qualifier are encoded in the field name." if "heat_flux" in name or name.startswith("q_"): return "Reconstructed one-dimensional temperature-gradient heat-flux proxy; not a direct heat-flux measurement." if "hydrophone" in name or "acoustic" in name: return "Hydrophone/AE-derived quantity; voltage-PSD products are not calibrated acoustic power." if name.endswith("_available"): return "Boolean availability flag for the named modality or product." if name.endswith("_path") or name.endswith("_file"): return "Relative or recorded source/output path associated with the case." return "Published field; semantic qualifiers and units are encoded in the field name." def csv_schema(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) rows = [] for _, row in zip(range(25), reader): rows.append(row) fields = reader.fieldnames or [] return { "format": "CSV with one header row and comma delimiter", "missing_value_rule": "Empty strings indicate missing values unless a field-specific status flag is supplied.", "fields": [ { "name": name, "dtype": infer_dtype([row.get(name, "") for row in rows]), "shape": "scalar per row", "units": unit_from_name(name), "description": field_description(name), } for name in fields ], } def lvm_schema(path: Path) -> dict[str, Any]: lines = path.read_text(encoding="utf-8-sig", errors="replace").splitlines() header_index = next((i for i, line in enumerate(lines) if line.startswith("X_Value\t")), None) units_index = next((i for i, line in enumerate(lines) if line.startswith("Y_Unit_Label\t")), None) fields = lines[header_index].split("\t") if header_index is not None else [] units = lines[units_index].split("\t") if units_index is not None else [] return { "format": "LabVIEW Measurement (.lvm), tab-delimited, instrument header followed by data table", "missing_value_rule": "Use the raw export values; no release-wide missing-value sentinel is imposed.", "fields": [ { "name": name, "dtype": "float64" if index < len(fields) - 1 else "string_or_empty", "shape": "scalar per sample row", "units": units[index] if index < len(units) and units[index] else "not recorded", "description": "Native LabVIEW data-table column; instrument label retained verbatim.", } for index, name in enumerate(fields) ], } def npz_schema(path: Path) -> dict[str, Any]: with np.load(path, allow_pickle=False) as archive: fields = [ { "name": key, "dtype": str(archive[key].dtype), "shape": list(archive[key].shape), "units": unit_from_name(key), "description": field_description(key), } for key in archive.files ] return { "format": "NumPy compressed archive (.npz)", "missing_value_rule": "Follow NaN/boolean/status conventions of the corresponding processed CSV and summary metadata.", "fields": fields, } def json_schema(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): return {"format": "JSON", "fields": []} fields = [] for key, item in value.items(): shape = "scalar" if isinstance(item, list): shape = "array" elif isinstance(item, dict): shape = "object" fields.append({ "name": key, "dtype": type(item).__name__, "shape": shape, "units": unit_from_name(key), "description": field_description(key), }) return {"format": "UTF-8 JSON object", "missing_value_rule": "JSON null indicates unavailable values.", "fields": fields} def generic_schema(path: Path) -> dict[str, Any]: extension = path.suffix.lower() formats = { ".mp4": "High-speed-video export for qualitative inspection; no independently verified common-clock registration.", ".wfs": "EasyAE/AEWin waveform binary; parse with validated compatible software.", ".dta": "EasyAE/AEWin configuration/export binary or text container; parse with validated compatible software.", ".txt": "EasyAE/AEWin event export; preserve native layout and parser provenance.", ".xlsx": "Spreadsheet analysis product; corresponding CSV/JSON products are the machine-readable release interface.", ".png": "Rendered diagnostic figure; presentation derivative, not a numerical source of record.", ".md": "Markdown documentation.", } return {"format": formats.get(extension, "Release artifact"), "fields": []} def build_dictionary(files: list[Path]) -> dict[str, Any]: entries: dict[str, Any] = {} for path in files: relative = path.relative_to(ROOT) if path.suffix.lower() == ".csv": schema = csv_schema(path) elif path.suffix.lower() == ".lvm": schema = lvm_schema(path) elif path.suffix.lower() == ".npz": schema = npz_schema(path) elif path.suffix.lower() == ".json": schema = json_schema(path) else: schema = generic_schema(path) entries[relative.as_posix()] = {"file_class": file_class(relative), "schema": schema} entries["data_dictionary.json"] = { "file_class": "release_metadata", "schema": { "format": "UTF-8 JSON object", "fields": [ {"name": "schema_version", "dtype": "string", "shape": "scalar", "units": "not_applicable", "description": "Version of the data-dictionary schema."}, {"name": "release_version", "dtype": "string", "shape": "scalar", "units": "not_applicable", "description": "Dataset release version covered by this dictionary."}, {"name": "scope", "dtype": "string", "shape": "scalar", "units": "not_applicable", "description": "Release inclusion/exclusion boundary."}, {"name": "time_base", "dtype": "string", "shape": "scalar", "units": "not_applicable", "description": "Release-wide processed-data time-base statement."}, {"name": "missing_value_policy", "dtype": "string", "shape": "scalar", "units": "not_applicable", "description": "Release-wide missing-value convention."}, {"name": "files", "dtype": "object", "shape": "path-keyed object", "units": "not_applicable", "description": "Per-file class and schema definitions."}, ], }, } entries["manifest.csv"] = { "file_class": "release_metadata", "schema": { "format": "CSV with one header row and comma delimiter", "missing_value_rule": "No ordinary missing values; manifest.csv uses SELF_REFERENTIAL_NOT_APPLICABLE for its own SHA-256 field.", "fields": [ {"name": "path", "dtype": "string", "shape": "scalar per row", "units": "not_applicable", "description": "Forward-slash relative path in the intended release."}, {"name": "case_identifier", "dtype": "string", "shape": "scalar per row", "units": "not_applicable", "description": "Boiling test identifier or release-level label."}, {"name": "file_class", "dtype": "string", "shape": "scalar per row", "units": "not_applicable", "description": "Raw export, derived product, release documentation/code, or release metadata."}, {"name": "byte_size", "dtype": "int64_or_string", "shape": "scalar per row", "units": "bytes", "description": "File size in bytes; self-referential for manifest.csv."}, {"name": "sha256", "dtype": "string", "shape": "scalar per row", "units": "not_applicable", "description": "SHA-256 hex digest; SELF_REFERENTIAL_NOT_APPLICABLE for manifest.csv."}, {"name": "release_version", "dtype": "string", "shape": "scalar per row", "units": "not_applicable", "description": "Dataset release version."}, ], }, } return { "schema_version": "1.0.0", "release_version": VERSION, "scope": "All files in the intended release except D_250W/Raw_data/Processed_data, which is a duplicate derived-data tree, and interpreter-cache artifacts under __pycache__.", "time_base": "Processed thermal quantities use the retained Temperature.lvm reference clock. See timing_reconciliation.csv for offsets and limitations.", "missing_value_policy": "Raw exports are preserved. Derived products use explicit availability/status fields where present; empty CSV values and JSON null indicate unavailable values.", "files": entries, } def relative_or_unavailable(value: str | None) -> str: if not value or value == "not_available": return "not_available" candidate = Path(value) try: return candidate.relative_to(ROOT).as_posix() except ValueError: return value def build_timing_rows() -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for case_dir in sorted(ROOT.glob("*_?*W")): if not case_dir.is_dir(): continue alignment_path = case_dir / "Processed_data" / "time_alignment.json" if not alignment_path.exists(): continue alignment = json.loads(alignment_path.read_text(encoding="utf-8")) case = case_id(Path(case_dir.name) / "x") for record in alignment["records"]: rows.append({ "case_id": case, "modality": record["modality"], "native_path": relative_or_unavailable(record.get("source_path")), "processed_path": relative_or_unavailable(str(alignment_path)), "native_time_definition": record.get("method", "not_recorded"), "processed_time_definition": alignment["reference_time_definition"], "source_start_clock": record.get("source_start_clock") or "not_available", "reference_start_clock": record.get("reference_start_clock") or "not_available", "applied_offset_s": record.get("applied_offset_s") if record.get("applied_offset_s") is not None else "not_available", "offset_tolerance_s": record.get("tolerance_s") if record.get("tolerance_s") is not None else "not_available", "status": record.get("status", "not_recorded"), "limitation": alignment["limits"], }) rows.extend([ { "case_id": case, "modality": "thermal_npz", "native_path": f"{case_dir.name}/Raw_data/Temperature.lvm", "processed_path": f"{case_dir.name}/Processed_data/thermal_timeseries.npz", "native_time_definition": "Temperature.lvm Time (sec)", "processed_time_definition": "Array time convention defined by NPZ field names; reference-clock alignment is inherited from thermal processing.", "source_start_clock": "see temperature row", "reference_start_clock": "see temperature row", "applied_offset_s": 0.0, "offset_tolerance_s": 0.001, "status": "derived_on_temperature_reference", "limitation": alignment["limits"], }, { "case_id": case, "modality": "high_speed_video", "native_path": "not_released_as_native_camera_file", "processed_path": f"{case_dir.name}/Processed_data/Boiling-{case.split('-')[-1]}_video.mp4", "native_time_definition": "not recorded", "processed_time_definition": "Transcoded MP4 clock not independently verified against acquisition clocks.", "source_start_clock": "not_available", "reference_start_clock": "not_available", "applied_offset_s": "not_available", "offset_tolerance_s": "not_available", "status": "qualitative_only", "limitation": "Do not infer absolute video-to-thermal, video-to-hydrophone, or video-to-AE phase/lag.", }, ]) return rows def write_timing() -> None: output = ROOT / "timing_reconciliation.csv" rows = build_timing_rows() with output.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def write_dictionary(files: list[Path]) -> None: output = ROOT / "data_dictionary.json" output.write_text(json.dumps(build_dictionary(files), indent=2) + "\n", encoding="utf-8") def write_manifest(files: list[Path]) -> None: output = ROOT / "manifest.csv" with output.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=["path", "case_identifier", "file_class", "byte_size", "sha256", "release_version"]) writer.writeheader() for path in files: relative = path.relative_to(ROOT) writer.writerow({ "path": relative.as_posix(), "case_identifier": case_id(relative), "file_class": file_class(relative), "byte_size": path.stat().st_size, "sha256": sha256(path), "release_version": VERSION, }) writer.writerow({ "path": "manifest.csv", "case_identifier": "release-level", "file_class": "release_metadata", "byte_size": "self-referential", "sha256": SELF_HASH, "release_version": VERSION, }) def main() -> None: initial_files = release_files() write_timing() files_for_dictionary = release_files() write_dictionary(files_for_dictionary) write_manifest(release_files()) print(f"Wrote metadata for {len(release_files()) + 1} release artifacts (including manifest.csv).") if __name__ == "__main__": main()