#!/usr/bin/env python3 """Reproduce the saved steel-door data calculations; Python 3.9+, standard library. Run beside the distributed source-inputs.json, CSVs and combined JSON, or pass --data-dir. This is a calculation and file-consistency check, NOT a new check of external sources. No network access is made and no verification date is changed. Source values are published references, not physical measurements. Calculated weight values are excluded from the published-value spread. """ from __future__ import annotations import argparse import csv import json import sys from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path from typing import Any D = Decimal FILES = { "gauge_chart": "steel-door-gauge-chart.csv", "coating_reconciliation": "steel-door-coating-reconciliation.csv", "duty_level_endurance": "steel-door-duty-level-endurance.csv", "component_thickness": "steel-door-component-thickness.csv", } COUNTS = {"gauge_chart": 28, "coating_reconciliation": 55, "duty_level_endurance": 6, "component_thickness": 23} def decimal(value: Any) -> Decimal: """Construct exact base-10 values from saved strings or integers.""" return D(str(value)) def formatted(value: Decimal, places: int) -> str: return format(value.quantize(D(1).scaleb(-places), rounding=ROUND_HALF_UP), f".{places}f") def read_csv(path: Path) -> list[dict[str, str]]: with path.open(encoding="utf-8-sig", newline="") as stream: reader = csv.DictReader(stream) if reader.fieldnames is None: raise ValueError(f"CSV has no header: {path.name}") rows = list(reader) if any(None in row or any(v is None for v in row.values()) for row in rows): raise ValueError(f"CSV has an inconsistent number of fields: {path.name}") return rows def run(data_dir: Path) -> tuple[dict[str, Any], dict[str, Any]]: inputs = json.loads((data_dir / "source-inputs.json").read_text(encoding="utf-8")) combined = json.loads((data_dir / "steel-door-gauge-dataset-v1-2.json").read_text(encoding="utf-8")) tables = {key: read_csv(data_dir / filename) for key, filename in FILES.items()} sources = read_csv(data_dir / "steel-door-source-register.csv") errors: list[str] = [] checks = 0 def check(condition: bool, label: str) -> None: nonlocal checks checks += 1 if not condition: errors.append(label) def same_decimal(actual: Any, expected: Any, label: str) -> None: check(decimal(actual) == decimal(expected), label) def display(actual: str, expected: Decimal, places: int, label: str) -> None: check(actual == formatted(expected, places), label) date = inputs["verification_date"] check(combined["verified"] == date, "Combined JSON/source-input verification dates differ") check(combined["version"] == inputs["version"], "Dataset versions differ") check(len(sources) == 20, "Source register must contain 20 sources") source_ids = {r["source_id"] for r in sources} check(sources == inputs["sources"], "Source register differs from source-inputs.json") check(sources == combined["sources"], "Source register differs from combined JSON") for key, rows in tables.items(): check(len(rows) == COUNTS[key], f"Unexpected record count for {key}") other = combined["tables"][key] check(len(rows) == len(other), f"CSV/JSON record count differs for {key}") for index, row in enumerate(rows): check(row["date_checked"] == date, f"{key}[{index}]: date differs") check(set(row["source_ids"].split(";")) <= source_ids, f"{key}[{index}]: unknown source ID") if index < len(other): normalized = {k: "" if v is None else str(v) for k, v in other[index].items()} check(row == normalized, f"{key}[{index}]: CSV/JSON field mismatch") maps = {key: {int(g): decimal(v) for g, v in inputs[key].items()} for key in ["steel_source_galvanized", "steel_source_nominal", "steel_source_lower_ordering_limit", "dasma_bare_min", "sdi_min", "hmma_min"]} galv = maps["steel_source_galvanized"] nominal = maps["steel_source_nominal"] lower = maps["steel_source_lower_ordering_limit"] dasma = maps["dasma_bare_min"] sdi = maps["sdi_min"] hmma = maps["hmma_min"] weights = {int(g): decimal(v) for g, v in inputs["statutory_weight_column_oz_sqft"].items()} weight_constant = decimal(inputs["clopay_weight_conversion_constant_oz_sqft_per_in"]) check(weight_constant > 0, "Weight conversion constant must be positive") if weight_constant <= 0: raise ValueError("Nonpositive weight conversion constant") check([int(r["gauge_no"]) for r in tables["gauge_chart"]] == list(range(3, 31)), "Gauge rows must cover 3–30 once in ascending order") computed: dict[str, Any] = {"gauge_computations": [], "coating_computations": [], "endurance_computations": [], "component_gauge_mapping": []} source_fields = [ ("gsg_galvanized_nominal_in", galv), ("msg_nominal_published_in", nominal), ("supplier_lower_ordering_limit_in", lower), ("dasma_tds154_min_in", dasma), ("sdi_117_26_min_in", sdi), ("hmma_803_min_in", hmma), ] for row in tables["gauge_chart"]: g = int(row["gauge_no"]) for field, mapping in source_fields: if g in mapping: same_decimal(row[field], mapping[g], f"Gauge {g}: {field} source mismatch") else: check(row[field] == "", f"Gauge {g}: unprovided source entry in {field}") same_decimal(row["weight_basis_oz_per_sqft"], weights[g], f"Gauge {g}: weight input") weight_result = weights[g] / weight_constant display(row["weight_basis_computed_in"], weight_result, 4, f"Gauge {g}: weight calculation") hollow = sdi.get(g, hmma.get(g)) published = [m[g] for m in [galv, nominal, lower, dasma] if g in m] if hollow is not None: published.append(hollow) display(row["calculated_mm_of_hollow_metal_min"], hollow * D("25.4"), 4, f"Gauge {g}: metric calculation") else: check(row["calculated_mm_of_hollow_metal_min"] == "", f"Gauge {g}: missing metric input") for field, key in [("sdi_a250_8_table2_pub_mm", "sdi_table2_printed_mm"), ("hmma_803_pub_mm", "hmma_printed_mm")]: check(row[field] == inputs[key].get(str(g), ""), f"Gauge {g}: printed metric field") smallest, largest = min(published), max(published) spread = D(100) * (largest / smallest - 1) display(row["lowest_published_in"], smallest, 4, f"Gauge {g}: published minimum") display(row["highest_published_in"], largest, 4, f"Gauge {g}: published maximum") display(row["spread_pct"], spread, 1, f"Gauge {g}: spread") check(int(row["distinct_published_values"]) == len(set(published)), f"Gauge {g}: distinct values") expected_match = ("yes" if lower[g] == dasma[g] else "no") if g in lower and g in dasma else "" check(row["supplier_limit_equals_dasma_min"] == expected_match, f"Gauge {g}: supplier/DASMA match") for field, coating in [("dasma_g60_in", "G-60"), ("dasma_g90_in", "G-90")]: if g in dasma: increment = decimal(inputs["dasma_coating_reference"][coating]["published_increment_in"]) display(row[field], dasma[g] + increment, 4, f"Gauge {g}: {field}") else: check(row[field] == "", f"Gauge {g}: unsupported coated entry") computed["gauge_computations"].append({ "gauge_no": g, "weight_basis_computed_in": formatted(weight_result, 4), "lowest_published_in": formatted(smallest, 4), "highest_published_in": formatted(largest, 4), "spread_pct": formatted(spread, 1)}) shared = sorted(set(lower) & set(dasma)) matches = [g for g in shared if lower[g] == dasma[g]] check(len(shared) == 20 and len(matches) == 18, "Expected 18/20 supplier/DASMA matches") check(sorted(set(shared)-set(matches)) == [8, 19], "Supplier/DASMA exception gauges differ") common = sorted(set(sdi) & set(hmma)) check(len(common) == 7 and all(sdi[g] == hmma[g] for g in common), "Seven shared SDI/HMMA minimums") nominal_common = sorted(set(weights) & set(nominal)) nominal_matches = [g for g in nominal_common if decimal(formatted(weights[g]/weight_constant, 3)) == nominal[g]] check(len(nominal_common) == 26 and len(nominal_matches) == 25, "Expected 25/26 direct-rounded nominal matches") check(sorted(set(nominal_common)-set(nominal_matches)) == [9], "Nominal calculation exception must be 9 gauge") coating_keys = set() reference_factor = decimal(inputs["sdi_reference_coating_inches_per_oz_sqft_both_sides"]) for index, row in enumerate(tables["coating_reconciliation"]): g = int(row["gauge_no"]) name = row["coating_class"] check(g in inputs["coating_selection_gauges"], f"Coating[{index}]: unselected gauge") coating_keys.add((g, name)) raw = inputs["dasma_coating_reference"][name] weight, increment = decimal(raw["weight_oz_sqft_both_sides"]), decimal(raw["published_increment_in"]) calculated = weight * reference_factor same_decimal(row["coating_oz_per_sqft_both_sides"], weight, f"Coating[{index}]: weight") display(row["calculated_added_in_sdi117_reference_rule"], calculated, 5, f"Coating[{index}]: calculation") same_decimal(row["dasma_bare_in"], dasma[g], f"Coating[{index}]: bare source") display(row["dasma_coated_in"], dasma[g] + increment, 4, f"Coating[{index}]: coated source") same_decimal(row["published_added_in"], increment, f"Coating[{index}]: increment") agrees = decimal(formatted(calculated, 4)) == increment check(row["agrees_to_4dp"] == ("yes" if agrees else "no"), f"Coating[{index}]: agreement") check(agrees, f"Coating[{index}]: source reconciliation differs") computed["coating_computations"].append({"gauge_no": g, "coating_class": name, "calculated_increment_in": formatted(calculated, 5), "agrees_to_4dp": agrees}) expected_keys = {(g, name) for g in inputs["coating_selection_gauges"] for name in inputs["dasma_coating_reference"]} check(coating_keys == expected_keys and len(coating_keys) == 55, "Coating subset duplicate/missing key") for index, row in enumerate(tables["duty_level_endurance"]): for key, value in inputs["endurance_qualification_inputs"][index].items(): check(row[key] == value, f"Endurance[{index}]: source field {key}") days = decimal(row["cycles_required"]) / (D(15) * D(60) * D(24)) display(row["calculated_uninterrupted_days_at_15_cycles_per_min"], days, 1, f"Endurance[{index}]: calculated days") computed["endurance_computations"].append({"class_name": row["class_name"], "calculated_days_at_15_cycles_per_min": formatted(days, 1)}) for index, row in enumerate(tables["component_thickness"]): for key, value in inputs["component_clause_inputs"][index].items(): check(row[key] == value, f"Component[{index}]: saved clause field {key}") minimum = decimal(row["min_thickness_in"]) gauge = int(row["msg_reference"]) check(hmma[gauge] == minimum, f"Component[{index}]: HMMA gauge-reference mapping") check(bool(row["qualification"].strip()), f"Component[{index}]: missing qualification") computed["component_gauge_mapping"].append({"component": row["component"], "minimum_in": str(minimum), "reference_gauge": gauge}) component_values = [decimal(row["min_thickness_in"]) for row in tables["component_thickness"]] ratio = max(component_values) / min(component_values) check(formatted(ratio, 1) == "10.4", "Selected component ratio") spreads = {r["gauge_no"]: r["spread_pct"] for r in computed["gauge_computations"]} check(spreads[16] == "20.8", "Headline 16-gauge spread") selected_spreads = [decimal(spreads[g]) for g in [12, 14, 16, 18, 20]] check(min(selected_spreads) == D("16.1") and max(selected_spreads) == D("25.0"), "Selected five-gauge spread range") report = {"status": "PASS" if not errors else "FAIL", "checks": checks, "mismatches": errors, "dataset_version": inputs["version"], "source_verification_date_unchanged": date, "record_counts": {k: len(v) for k, v in tables.items()}, "headline_16_gauge_spread_pct": spreads[16], "supplier_dasma_matches": f"{len(matches)}/{len(shared)}", "direct_3dp_nominal_matches": f"{len(nominal_matches)}/{len(nominal_common)}", "coating_comparisons": len(coating_keys), "component_max_min_ratio": formatted(ratio, 1), "scope": "Checks saved source inputs and arithmetic only. No external sources reverified; no physical measurements."} return report, computed def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--data-dir", type=Path, default=Path(__file__).resolve().parent, help="Directory containing the distributed inputs, CSVs and JSON") parser.add_argument("--output", type=Path, help="Also write the check report as JSON") parser.add_argument("--write-recomputed", type=Path, help="Write the independently recomputed calculation fields as JSON") args = parser.parse_args() try: report, computed = run(args.data_dir) for path, obj in [(args.output, report), (args.write_recomputed, computed)]: if path is not None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(obj, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print(json.dumps(report, indent=2, ensure_ascii=False)) return 0 if report["status"] == "PASS" else 1 except (OSError, ValueError, KeyError, InvalidOperation, IndexError, TypeError) as exc: print(f"Unable to validate dataset: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())