#!/usr/bin/env python3 """Reproduce this edition's document counts and optional receipt arithmetic. This script does not scrape sources or reproduce APQC/WERC's unpublished surveys. The default receipt input is explicitly fictional. Python 3.10+; standard library. Usage: python reproduce_dock_to_stock.py python reproduce_dock_to_stock.py --receipts my_log.csv \ --cutoff 2026-10-01T00:00:00-04:00 --output my_results.json Before using an operational file, select and document its arrival-based cohort. All timestamps must include a UTC offset. Completed results include only records that have reached the relevant endpoint by the supplied cutoff. Exception counts can overlap completion categories. Missing completion timestamps do not establish whether work is unfinished or its event was simply not recorded. """ from __future__ import annotations import argparse import csv import json import math import statistics from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable HERE = Path(__file__).resolve().parent TIME_FIELDS = ( 'carrier_arrival_timestamp', 'unload_start_timestamp', 'unload_complete_timestamp', 'receipt_posted_timestamp', 'putaway_complete_timestamp', 'available_in_wms_timestamp', ) REQUIRED_FIELDS = ( 'receipt_id', 'po_or_asn_number', 'carrier_arrival_timestamp', 'dock_door', 'unload_start_timestamp', 'unload_complete_timestamp', 'receipt_posted_timestamp', 'putaway_complete_timestamp', 'available_in_wms_timestamp', 'flow_through_flag', 'exception_flag', 'dock_to_stock_hours', 'notes', ) def read_csv(path: Path) -> list[dict[str, str]]: with path.open(encoding='utf-8-sig', newline='') as handle: return list(csv.DictReader(handle)) def timestamp(value: str) -> datetime | None: value = (value or '').strip() if not value: return None result = datetime.fromisoformat(value.replace('Z', '+00:00')) if result.tzinfo is None or result.utcoffset() is None: raise ValueError('Every timestamp must include a UTC offset.') return result.astimezone(timezone.utc) def hours(start: datetime, end: datetime) -> float: return (end - start).total_seconds() / 3600 def fail_unless(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def document_checks(directory: Path = HERE) -> dict[str, Any]: """Compute counts, then verify that CSV and JSON records agree exactly.""" stems = ( 'dock-to-stock-benchmark-ledger', 'dock-to-stock-definition-census', 'dock-to-stock-ohio-cincinnati-context', 'dock-to-stock-system-and-regulatory-semantics', 'dock-to-stock-sources', ) datasets: dict[str, list[dict[str, str]]] = {} for stem in stems: csv_rows = read_csv(directory / (stem + '.csv')) json_rows = json.loads((directory / (stem + '.json')).read_text()) # CSV is text-only; JSON retains numeric types. Compare normalized cells. normalized = [ {key: str(value) if value is not None else '' for key, value in row.items()} for row in json_rows ] fail_unless(csv_rows == normalized, f'CSV/JSON mismatch: {stem}') datasets[stem] = csv_rows ledger = datasets['dock-to-stock-benchmark-ledger'] definitions = datasets['dock-to-stock-definition-census'] numerical = [r for r in ledger if r['states_duration'] == 'Y'] direct = [r for r in numerical if r['survey_summary_record'] == 'Y'] system = Counter(r['explicit_system_event'] for r in definitions) waiting = sum(r['wait_time_explicitly_included'] == 'Yes' for r in definitions) formulas = sum(r['formula_stated'] == 'Yes' for r in definitions) expected = { 'evidence_records': len(ledger), 'numerical_evidence_records': len(numerical), 'direct_benchmark_or_producer_summary_records': len(direct), 'other_numerical_records': len(numerical) - len(direct), 'definition_passages': len(definitions), 'explicit_system_event': dict(system), 'explicit_waiting': waiting, 'formula_present': formulas, } fail_unless(len(ledger) == 24 and len(numerical) == 23 and len(direct) == 9, 'Ledger count differs from this published edition.') fail_unless(len(definitions) == 20 and system == {'Yes': 8, 'Implied': 5, 'No': 7}, 'Definition count differs from this published edition.') fail_unless(waiting == 2 and formulas == 8, 'Waiting/formula count differs from this published edition.') by_year = {r['report_year']: r for r in ledger if r['report_year']} first = float(by_year['2019']['value_hours_high']) last = float(by_year['2025']['value_hours_high']) change = (last - first) / first * 100 fail_unless(math.isclose(change, 75), 'Cutoff change does not reproduce.') expected['werc_cutoff_change_percent_2019_to_2025'] = change expected['cutoff_interpretation'] = ( 'Change in published cutoff values; not a fixed-panel change in warehouse times.' ) expected['csv_json_pairs_match'] = True return expected def receipt_analysis(rows: Iterable[dict[str, str]], cutoff: datetime) -> dict[str, Any]: """Compute two endpoints, with validation and explicit record accounting. The stored dock_to_stock_hours column is not trusted as an input. It is recomputed from events. Flow-through rows are reported as exclusions under the published template's conservative receipt-level rule. Duplicate IDs invalidate all rows sharing that ID rather than selectively keeping one. """ if cutoff.tzinfo is None or cutoff.utcoffset() is None: raise ValueError('Cutoff must include a UTC offset.') cutoff = cutoff.astimezone(timezone.utc) data = list(rows) ids = Counter((r.get('receipt_id') or '').strip() for r in data) details: list[dict[str, Any]] = [] durations = {'recorded_and_put_away': [], 'available_stock': []} missing_ages = {'recorded_and_put_away': [], 'available_stock': []} exception_count = 0 for row in data: identifier = (row.get('receipt_id') or '').strip() detail: dict[str, Any] = {'receipt_id': identifier} details.append(detail) try: fail_unless(set(REQUIRED_FIELDS).issubset(row), 'Missing required CSV column.') fail_unless(bool(identifier) and not identifier.startswith('LEGEND'), 'Missing receipt ID or a legend row mixed into data.') fail_unless(ids[identifier] == 1, 'Duplicate receipt ID; all duplicates rejected.') fail_unless(row['flow_through_flag'] in ('Y', 'N'), 'flow_through_flag must be Y or N.') fail_unless(row['exception_flag'] in ('Y', 'N'), 'exception_flag must be Y or N.') times = {field: timestamp(row[field]) for field in TIME_FIELDS} start = times['carrier_arrival_timestamp'] fail_unless(start is not None, 'Missing dock-arrival timestamp.') # The declaration is an actual-receipt event contract. A planned or # pre-advice entry must not masquerade as a completed receipt event. fail_unless(all(t is None or t >= start for t in times.values()), 'An event precedes the declared dock arrival; review event provenance.') us, ue = times['unload_start_timestamp'], times['unload_complete_timestamp'] fail_unless(us is None or ue is None or ue >= us, 'Unload completion precedes unload start.') if start > cutoff: detail['status'] = 'excluded_after_cutoff' continue if row['exception_flag'] == 'Y': exception_count += 1 if row['flow_through_flag'] == 'Y': detail['status'] = 'excluded_flow_through' continue detail['status'] = 'eligible_arrival' detail['exception_flag'] = row['exception_flag'] posted = times['receipt_posted_timestamp'] putaway = times['putaway_complete_timestamp'] available = times['available_in_wms_timestamp'] endpoints = { 'recorded_and_put_away': (posted, putaway), 'available_stock': (posted, putaway, available), } for name, required in endpoints.items(): if any(t is None or t > cutoff for t in required): detail[name] = { 'status': 'missing_completion_at_cutoff', 'hours': None, 'age_hours': hours(start, cutoff), 'note': 'Could mean unfinished work or missing event data; inspect receipt status.', } missing_ages[name].append(hours(start, cutoff)) else: end = max(required) value = hours(start, end) detail[name] = {'status': 'complete', 'hours': value, 'completion_utc': end.isoformat()} durations[name].append(value) except (ValueError, TypeError, KeyError) as exc: detail['status'] = 'rejected_for_review' detail['reason'] = str(exc) summary: dict[str, Any] = { 'input_rows': len(data), 'cutoff_utc': cutoff.isoformat(), 'row_status_counts': dict(Counter(d['status'] for d in details)), 'exception_flagged_arrivals_at_or_before_cutoff': exception_count, 'exception_count_is_overlapping_not_additive': True, } for name, values in durations.items(): ages = missing_ages[name] summary[name] = { 'completed_n': len(values), 'missing_completion_n': len(ages), 'mean_hours': statistics.mean(values) if values else None, 'median_receipt_hours': statistics.median(values) if values else None, 'oldest_missing_completion_age_hours': max(ages) if ages else None, } return {'summary': summary, 'records': details} def self_test() -> dict[str, str]: example = read_csv(HERE / 'dock-to-stock-worked-example.csv')[0] cutoff = timestamp('2026-09-08T17:00:00-04:00') result = receipt_analysis([example], cutoff) fail_unless(result['summary']['recorded_and_put_away']['mean_hours'] == 3.0, 'Fictional recorded-and-put-away result failed.') fail_unless(result['summary']['available_stock']['mean_hours'] == 5.0, 'Fictional available-stock result failed.') missing = dict(example, receipt_id='TEST-MISSING', putaway_complete_timestamp='', available_in_wms_timestamp='') flow = dict(example, receipt_id='TEST-FLOW', flow_through_flag='Y') bad = dict(example, receipt_id='TEST-INVALID', putaway_complete_timestamp='2026-09-08T07:59:00-04:00') early_available = dict(example, receipt_id='TEST-EARLY-AVAILABLE', available_in_wms_timestamp='2026-09-08T10:00:00-04:00') late_post = dict(example, receipt_id='TEST-LATE-POST', receipt_posted_timestamp='2026-09-08T12:00:00-04:00') mixed = receipt_analysis([example, missing, flow, bad, early_available, late_post], cutoff) status = mixed['summary']['row_status_counts'] fail_unless(status == {'eligible_arrival': 4, 'excluded_flow_through': 1, 'rejected_for_review': 1}, 'Status accounting failed.') rec = {r['receipt_id']: r for r in mixed['records']} fail_unless(rec['TEST-MISSING']['recorded_and_put_away']['hours'] is None, 'Missing completion must not become zero.') fail_unless(rec['TEST-MISSING']['recorded_and_put_away']['age_hours'] == 9.0, 'Missing completion age failed.') fail_unless(rec['TEST-EARLY-AVAILABLE']['available_stock']['hours'] == 3.0, 'Earlier availability must not override later putaway.') fail_unless(rec['TEST-LATE-POST']['recorded_and_put_away']['hours'] == 4.0, 'Later posting must be included.') dup = receipt_analysis([example, example], cutoff) fail_unless(dup['summary']['row_status_counts'] == {'rejected_for_review': 2}, 'Duplicate rejection failed.') # UTC arithmetic across the repeated fall-back hour. start = timestamp('2026-11-01T01:30:00-04:00') end = timestamp('2026-11-01T01:30:00-05:00') fail_unless(hours(start, end) == 1.0, 'UTC-offset arithmetic failed.') no_offset = dict(example, receipt_id='TEST-NO-OFFSET', carrier_arrival_timestamp='2026-09-08T08:00:00') rejected = receipt_analysis([no_offset], cutoff) fail_unless(rejected['records'][0]['status'] == 'rejected_for_review', 'Timezone-free timestamp should be rejected.') return {'result': 'passed', 'fixtures': 'All test events are explicitly fictional.'} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--receipts', type=Path, default=None) parser.add_argument('--cutoff', default=None, help='Required with custom receipts; ISO 8601 timestamp with UTC offset.') parser.add_argument('--output', type=Path, default=None) args = parser.parse_args() if args.receipts and not args.cutoff: parser.error('--cutoff is required with --receipts') try: checks = document_checks() tests = self_test() receipt_file = args.receipts or HERE / 'dock-to-stock-worked-example.csv' cutoff = timestamp(args.cutoff or '2026-09-08T17:00:00-04:00') fail_unless(cutoff is not None, 'Cutoff is required.') result = { 'dataset_version': '2026-09-12', 'document_checks': checks, 'self_tests': tests, 'receipt_input': str(receipt_file), 'input_is_fictional_default': args.receipts is None, 'receipt_analysis': receipt_analysis(read_csv(receipt_file), cutoff), } text = json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + '\n' if args.output: args.output.write_text(text, encoding='utf-8') else: print(text, end='') except (OSError, ValueError, csv.Error, json.JSONDecodeError) as exc: parser.exit(1, f'Validation failed: {exc}\n') if __name__ == '__main__': main()