#!/usr/bin/env python3 """Reproduce descriptive dock-equipment comparisons from the accompanying CSVs. Python 3.10+, standard library only. This script does not access the network, perform engineering calculations, or authorize equipment use. It summarizes published records and manufacturer-guide entries, not physical test results. Usage: python reproduce-dock-comparison.py python reproduce-dock-comparison.py --check python reproduce-dock-comparison.py --output recalculated.json """ from __future__ import annotations import argparse import csv import json import math import statistics import sys from collections import defaultdict from pathlib import Path from typing import Any VERSION = "2026-09-14" def read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) if not reader.fieldnames: raise ValueError(f"Missing CSV header: {path.name}") records = list(reader) if not records: raise ValueError(f"No records: {path.name}") return records def positive_number(record: dict[str, str], field: str) -> float: try: value = float(record[field]) except (KeyError, TypeError, ValueError) as exc: raise ValueError(f"Invalid {field!r} in record {record.get('model', record)}") from exc if not math.isfinite(value) or value <= 0: raise ValueError(f"Expected positive finite {field!r}: {value!r}") return value def family_of(record: dict[str, str]) -> str | None: if record["sample_role"] != "core": return None model = record["model"] if model.startswith("E-"): return "Vestil E" if model.startswith("EH-"): return "Vestil EH" if record["manufacturer"] == "Bluff Manufacturing" and model.startswith("A"): return "Bluff A" return None def summarize(directory: Path) -> tuple[dict[str, Any], dict[str, Any]]: records = read_csv(directory / f"dock-plate-vs-dock-board-{VERSION}.csv") capacity_guide = read_csv(directory / f"bluff-selection-capacity-inputs-{VERSION}.csv") length_guide = read_csv(directory / f"bluff-selection-length-inputs-{VERSION}.csv") by_model = {row["model"]: row for row in records} if len(by_model) != len(records): raise ValueError("Duplicate model identifiers in model CSV") ids = [row["record_id"] for row in records] if len(set(ids)) != len(ids): raise ValueError("Duplicate record_id values") if any(row["sample_role"] not in {"core", "supplementary"} for row in records): raise ValueError("Unexpected sample_role") core = [row for row in records if row["sample_role"] == "core"] groups: dict[str, list[dict[str, str]]] = defaultdict(list) for row in records: for field in ("published_width_in", "published_length_in", "published_capacity_lb"): positive_number(row, field) family = family_of(row) if family: groups[family].append(row) if set(groups) != {"Vestil E", "Vestil EH", "Bluff A"}: raise ValueError("Unexpected core plate series") size = lambda row: (positive_number(row, "published_width_in"), positive_number(row, "published_length_in")) e = {size(row): row for row in groups["Vestil E"]} eh = {size(row): row for row in groups["Vestil EH"]} paired = sorted(e.keys() & eh.keys()) ratios = [positive_number(eh[key], "published_capacity_lb") / positive_number(e[key], "published_capacity_lb") for key in paired] e_thickness = {positive_number(row, "plate_thickness_in") for row in e.values()} eh_thickness = {positive_number(row, "plate_thickness_in") for row in eh.values()} if len(e_thickness) != 1 or len(eh_thickness) != 1: raise ValueError("Thickness comparison requires a single documented thickness per series") thin, thick = next(iter(e_thickness)), next(iter(eh_thickness)) endpoints: list[dict[str, Any]] = [] indices: dict[str, Any] = {} for family in ("Vestil E", "Vestil EH", "Bluff A"): rows = groups[family] for width in sorted({positive_number(row, "published_width_in") for row in rows}): width_rows = sorted((row for row in rows if positive_number(row, "published_width_in") == width), key=lambda row: positive_number(row, "published_length_in")) first, last = width_rows[0], width_rows[-1] short_length, long_length = positive_number(first, "published_length_in"), positive_number(last, "published_length_in") short_cap, long_cap = positive_number(first, "published_capacity_lb"), positive_number(last, "published_capacity_lb") endpoints.append({"family": family, "width_in": int(width), "shortest_key": f"{int(width)}{int(short_length)}", "longest_key": f"{int(width)}{int(long_length)}", "shortest_capacity_lb": int(short_cap), "longest_capacity_lb": int(long_cap), "length_multiple": long_length / short_length, "capacity_remaining_percent": long_cap / short_cap * 100}) values = [positive_number(row, "published_capacity_lb") * positive_number(row, "published_length_in") / positive_number(row, "published_width_in") for row in rows] indices[family] = {"n": len(values), "mean_index_lb": statistics.mean(values), "population_cv_percent": statistics.pstdev(values) / statistics.mean(values) * 100} selected = [by_model[model] for model in ("LP60-48", "E-6048", "A6048", "EH-6048")] if any(size(row) != (60, 48) for row in selected): raise ValueError("A selected 60 x 48 catalog-size record changed") capacities = [positive_number(row, "published_capacity_lb") for row in selected] low, high = min(capacities), max(capacities) ahtd = [row for row in core if row["model"].startswith("AHTD-")] ahtd_caps = {positive_number(row, "published_capacity_lb") for row in ahtd} if len(ahtd_caps) != 1: raise ValueError("The AHTD constant-rating claim no longer holds") ahtd_cap = next(iter(ahtd_caps)) plates = [row for rows in groups.values() for row in rows] lengths = [positive_number(row, "published_length_in") for row in ahtd] seven = next(row for row in length_guide if float(row["height_low_in"]) == float(row["height_high_in"]) == 7) plate_seven = positive_number(seven, "plate_length_in") board_seven = positive_number(seven, "propane_board_length_in") result = { "total_records": len(records), "core_records": len(core), "supplementary_records": len(records) - len(core), "core_manufacturers": len({row["manufacturer"] for row in core}), "total_manufacturers": len({row["manufacturer"] for row in records}), "paired_vestil_sizes": len(paired), "median_eh_to_e_ratio": statistics.median(ratios), "thickness_increase_percent": (thick / thin - 1) * 100, "capacity_retention_min_percent": min(row["capacity_remaining_percent"] for row in endpoints), "capacity_retention_max_percent": max(row["capacity_remaining_percent"] for row in endpoints), "same_size_min_lb": int(low), "same_size_max_lb": int(high), "same_size_high_over_low_percent": (high - low) / low * 100, "highest_selected_plate_to_ahtd_ratio": max(positive_number(row, "published_capacity_lb") for row in plates) / ahtd_cap, "ahtd_family_length_ratio": max(lengths) / min(lengths), "plate_vs_propane_length_at_7_in_percent": (plate_seven - board_seven) / board_seven * 100, "size_index_statistics": indices, "length_endpoint_comparisons": endpoints, } guide_multiples = [{"truck_type": row["truck_type"], "forklift_bracket_lower_lb": int(row["forklift_bracket_lower_lb"]), "forklift_bracket_upper_lb": int(row["forklift_bracket_upper_lb"]), "board_capacity_lb": int(row["board_capacity_lb"]), "computed_upper_endpoint_multiple": positive_number(row, "board_capacity_lb") / positive_number(row, "forklift_bracket_upper_lb")} for row in capacity_guide] hl = [] for row in length_guide: converted: dict[str, Any] = {"height_low_in": float(row["height_low_in"]), "height_high_in": float(row["height_high_in"])} for field in ("plate_length_in", "propane_board_length_in", "electric_board_length_in"): length = float(row[field]) if row[field] else None converted[field] = length converted[field.replace("length_in", "H_over_L_percent")] = None if length is None else [converted["height_low_in"] / length * 100, converted["height_high_in"] / length * 100] hl.append(converted) auxiliary = {"scope": "Manufacturer guide entries; arithmetic is descriptive, not equipment-selection advice or ramp-grade measurement.", "forklift_guide_endpoint_multiples": guide_multiples, "height_to_catalog_length_ratios": hl} return result, auxiliary def compare(expected: Any, actual: Any, path: str = "result") -> None: if isinstance(expected, dict): if not isinstance(actual, dict) or expected.keys() != actual.keys(): raise ValueError(f"Key mismatch at {path}") for key in expected: compare(expected[key], actual[key], f"{path}.{key}") elif isinstance(expected, list): if not isinstance(actual, list) or len(expected) != len(actual): raise ValueError(f"List mismatch at {path}") for index, (left, right) in enumerate(zip(expected, actual)): compare(left, right, f"{path}[{index}]") elif isinstance(expected, (int, float)) and not isinstance(expected, bool): if not isinstance(actual, (int, float)) or not math.isclose(expected, actual, rel_tol=1e-12, abs_tol=1e-12): raise ValueError(f"Numeric mismatch at {path}: {expected!r} versus {actual!r}") elif expected != actual: raise ValueError(f"Value mismatch at {path}: {expected!r} versus {actual!r}") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--directory", type=Path, default=Path(__file__).resolve().parent, help="Directory containing the supplied CSV and JSON files") parser.add_argument("--output", type=Path, help="Write recomputed model summary JSON to this path") parser.add_argument("--auxiliary-output", type=Path, help="Write recomputed manufacturer-guide arithmetic to this path") parser.add_argument("--check", action="store_true", help="Compare recomputed results with the supplied results JSON") args = parser.parse_args() try: result, auxiliary = summarize(args.directory) if args.check: stored_path = args.directory / f"dock-plate-vs-dock-board-calculations-{VERSION}.json" stored = json.loads(stored_path.read_text(encoding="utf-8")) compare(stored, result) print("PASS: all published model-summary calculations reproduce from the CSV inputs.", file=sys.stderr) text = json.dumps(result, ensure_ascii=False, indent=2) + "\n" if args.output: args.output.write_text(text, encoding="utf-8") elif not args.check: print(text, end="") if args.auxiliary_output: args.auxiliary_output.write_text(json.dumps(auxiliary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") except (OSError, ValueError, KeyError, StopIteration, csv.Error) as exc: print(f"Verification failed: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())