"""Build release metadata for the 30-case subatmospheric-boiling archive. Run from any working directory with Python, NumPy, and the standard library: python processing/build_release_metadata.py The script writes only release metadata in the dataset root. Raw and processed measurement files are read but never modified. """ from __future__ import annotations import csv import hashlib import json import re import subprocess from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parents[1] DOI = "10.5281/zenodo.22234711" def case_id(path: Path) -> str: match = re.search(r"Boiling-(\d+)", str(path)) return f"Boiling-{match.group(1)}" if match else "" def file_class(path: Path) -> str: parts = {part.lower() for part in path.parts} if "plot images" in parts: return "derived_plot" if "processed_data" in parts: return "processed_or_derived" if "raw_data" in parts: return "native_or_source" if path.name in {"README.md", "LICENSE", "CITATION.cff"} or "processing" in parts: return "release_documentation_or_code" return "release_documentation" 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 unit_for(field: str) -> str: lower = field.lower() if "time" in lower or lower.endswith("_s"): return "s" if "pressure" in lower and "kpa" in lower: return "kPa" if "temperature" in lower or "temp" in lower or "superheat" in lower: return "degC" if "heat_flux" in lower or "q_" in lower: return "W/cm^2" if "htc" in lower: return "W/(cm^2 K)" if "fps" in lower or "frequency" in lower or "hz" in lower: return "Hz" if "pixel" in lower or "width" in lower or "height" in lower: return "pixel" if "fraction" in lower or lower.startswith("vf"): return "dimensionless" if "count" in lower or "frames" in lower or "samples" in lower: return "count" return "see source schema" def description_for(field: str) -> str: known = { "Surface_Temperature_degC": "Reconstructed apparent surface temperature from embedded copper-block thermocouples.", "Heat_Flux_W_cm_2": "Fourier-law heat-flux estimate from the fitted through-thickness temperature gradient.", "Wall_Superheat_degC": "Reconstructed surface temperature minus pressure-derived saturation temperature.", "CHF_value": "Selected heating-branch CHF event value in the processed record.", "CHF_Time_value": "Selected CHF event time in the processed record.", "NBR_Heat_Flux": "Selected cooling-branch heat flux at return to the nucleate-boiling branch, where present.", "NBR_index": "Stored index of the selected NBR event in the processed array, where present.", "T_Surf_Max_Surface_Temperature": "Maximum reconstructed surface temperature during the post-CHF excursion.", "Bubble_count_array": "Number of segmented bubbles in each analyzed side-view frame; image-derived.", "Vapor_fraction_array": "Projected side-view vapor-pixel fraction in each analyzed frame; image-derived.", } if field in known: return known[field] if field.startswith("ONB"): return "Stored onset-of-nucleate-boiling event field or image-derived ONB quantity." if field.startswith("CHF"): return "Stored critical-heat-flux event field or image-derived CHF quantity." if field.startswith("NBR"): return "Stored nucleate-boiling-return event field or image-derived NBR quantity." if "HTC" in field: return "Processed heat-transfer coefficient field; inspect the source schema for the exact filtering convention." if "Vapor" in field or "Bubble" in field: return "Image-derived side-view bubble or vapor metric; not a volumetric void fraction or wall dry-area measurement." return "Field retained from the source CSV/NPZ schema; consult the source file and event-definition document." def build_manifest() -> None: output = ROOT / "manifest.csv" rows = [] for path in sorted(p for p in ROOT.rglob("*") if p.is_file() and p.name != output.name): rows.append( { "path": path.relative_to(ROOT).as_posix(), "case_id": case_id(path), "file_class": file_class(path), "byte_size": path.stat().st_size, "sha256": sha256(path), "record_doi": DOI, } ) with output.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def build_dictionary() -> None: fields = {} csv_files = sorted(ROOT.rglob("*.csv")) for path in csv_files: with path.open("r", newline="", encoding="utf-8-sig") as handle: reader = csv.reader(handle) header = next(reader, []) for field in header: fields.setdefault(field, path.relative_to(ROOT).as_posix()) npz_files = sorted(ROOT.rglob("*_full_data.npz")) if npz_files: with np.load(npz_files[0], allow_pickle=False) as archive: for field in archive.files: fields.setdefault(field, npz_files[0].relative_to(ROOT).as_posix()) rows = [ { "field": field, "data_class": "processed_or_derived", "unit": unit_for(field), "description": description_for(field), "source_example": source, "missing_value_rule": "blank or source-specific; inspect source file", } for field, source in sorted(fields.items()) ] with (ROOT / "data_dictionary.csv").open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def mp4_fps(path: Path) -> str: try: result = subprocess.run( [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=r_frame_rate", "-of", "default=noprint_wrappers=1:nokey=1", str(path), ], capture_output=True, text=True, check=True, ) return result.stdout.strip() except (FileNotFoundError, subprocess.CalledProcessError): return "not checked" def build_timing_table() -> None: rows = [] for npz in sorted(ROOT.rglob("*_full_data.npz")): with np.load(npz, allow_pickle=False) as archive: thermo = np.asarray(archive["Time_Thermocouple_seconds"]) pressure = np.asarray(archive["Time_Pressure_seconds"]) fps = np.asarray(archive["fps"]).reshape(-1)[0] frames = np.asarray(archive["Number_of_Frames"]).reshape(-1)[0] video = next(npz.parent.glob("*_video.mp4"), None) rows.append( { "case_id": case_id(npz), "npz_path": npz.relative_to(ROOT).as_posix(), "thermocouple_dt_s": float(np.median(np.diff(thermo))), "thermocouple_sampling_hz": float(1.0 / np.median(np.diff(thermo))), "pressure_dt_s": float(np.median(np.diff(pressure))), "pressure_sampling_hz": float(1.0 / np.median(np.diff(pressure))), "npz_frame_count": int(frames), "npz_video_fps": float(fps), "mp4_container_fps": mp4_fps(video) if video else "not present", "status": "recorded array rates; synchronization and drift require separate verification", } ) with (ROOT / "timing_reconciliation.csv").open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) if __name__ == "__main__": build_dictionary() build_timing_table() build_manifest() print(f"Wrote release metadata under {ROOT}")