#!/usr/bin/env python3 """Read-only INP/network.json comparison. Exit 0=match, 1=mismatch, 2=input error.""" import argparse from collections import Counter, defaultdict import hashlib import json import math from pathlib import Path import sys SECTIONS = {'JUNCTIONS':'junction', 'OUTFALLS':'outfall', 'STORAGE':'storage', 'CONDUITS':'conduit', 'PUMPS':'pump', 'WEIRS':'weir'} def values_match(actual, expected, tolerance): """Apply the same numeric contract at every nesting level.""" if isinstance(expected, bool): return isinstance(actual, bool) and actual == expected if isinstance(expected, (float, int)): return (isinstance(actual, (float, int)) and not isinstance(actual, bool) and math.isfinite(actual) and math.isfinite(expected) and abs(actual - expected) <= tolerance) if isinstance(expected, list): return (isinstance(actual, list) and len(actual) == len(expected) and all(values_match(a, e, tolerance) for a, e in zip(actual, expected))) if isinstance(expected, dict): return (isinstance(actual, dict) and actual.keys() == expected.keys() and all(values_match(actual[k], v, tolerance) for k, v in expected.items())) return type(actual) is type(expected) and actual == expected def parse_inp(path): raw = Path(path).read_bytes() for encoding in ('utf-8-sig', 'cp949'): try: text = raw.decode(encoding) break except UnicodeDecodeError: continue else: raise ValueError('INP encoding must be UTF-8 or CP949') sections = defaultdict(list) section = None for line, value in enumerate(text.splitlines(), 1): value = value.split(';', 1)[0].strip() if not value: continue if value.startswith('[') and value.endswith(']'): section = value[1:-1].upper() elif section: sections[section].append((line, value.split())) return raw, sections def validate(inp, network, tolerance=1e-6): raw, sections = parse_inp(inp) data = json.loads(Path(network).read_text(encoding='utf-8-sig')) errors, exclusions = [], [] checks = 0 def compare(entity, field, actual, expected): nonlocal checks checks += 1 if not values_match(actual, expected, tolerance): errors.append({'id': entity, 'field': field, 'actual': actual, 'expected': expected}) def number(tokens, index, default=None): if index >= len(tokens): if default is not None: return default raise ValueError(f'missing numeric field: {tokens}') value = float(tokens[index]) if not math.isfinite(value): raise ValueError(f'non-finite input: {tokens}') return value def indexed(section): result = {} for line, tokens in sections[section]: if tokens[0] in result: errors.append({'id': tokens[0], 'field': section, 'error': 'duplicate INP ID'}) result[tokens[0]] = (line, tokens) return result options = {t[0]: t[1] for _, t in sections['OPTIONS']} units = options.get('FLOW_UNITS', 'CFS').upper() if units not in ('CFS', 'GPM', 'MGD', 'CMS', 'LPS', 'MLD'): raise ValueError('unsupported FLOW_UNITS: ' + units) factor = .3048 if units in ('CFS', 'GPM', 'MGD') else 1.0 offset = options.get('LINK_OFFSETS', 'DEPTH').upper() if offset not in ('DEPTH', 'ELEVATION'): raise ValueError('unsupported LINK_OFFSETS: ' + offset) compare('meta', 'sha256', data['meta'].get('sha256'), hashlib.sha256(raw).hexdigest()) compare('meta', 'flowUnits', data['meta'].get('flowUnits'), units) compare('meta', 'offsetMode', data['meta'].get('offsetMode'), offset) compare('meta', 'physicalConversion', data['meta'].get('physicalConversion'), factor) coordinates = {id: [number(t, 1), number(t, 2)] for id, (_, t) in indexed('COORDINATES').items()} xs = indexed('XSECTIONS') vertices = defaultdict(list) for _, t in sections['VERTICES']: vertices[t[0]].append([number(t, 1), number(t, 2)]) source_nodes, source_links = {}, {} expected_counts = {} for section, kind in SECTIONS.items(): source = indexed(section) expected_counts[kind] = len(source) destination = source_nodes if kind in ('junction','outfall','storage') else source_links for id, record in source.items(): if id in destination: errors.append({'id': id, 'field': 'id', 'error': 'duplicate across facility types'}) destination[id] = (kind, *record) if not source_nodes or not source_links: raise ValueError('INP has no supported network nodes or links') nodes = {n['id']: n for n in data['nodes']} links = {l['id']: l for l in data['links']} compare('nodes', 'unique IDs', len(nodes), len(data['nodes'])) compare('links', 'unique IDs', len(links), len(data['links'])) compare('nodes', 'IDs', sorted(nodes), sorted(source_nodes)) compare('links', 'IDs', sorted(links), sorted(source_links)) expected_connections = defaultdict(list) for id, (_, _, t) in source_links.items(): expected_connections[t[1]].append(id) expected_connections[t[2]].append(id) for id, (kind, line, t) in source_nodes.items(): if id not in nodes: continue n = nodes[id] depth = number(t, 2) * factor if kind != 'outfall' else None inv = number(t, 1) * factor for field, expected in {'type':kind, 'line':line, 'raw':t, 'invert':inv, 'maxDepth':depth, 'top':None if depth is None else inv+depth, 'xy':coordinates.get(id)}.items(): compare(id, field, n.get(field), expected) if kind == 'outfall': has_stage=t[2].upper() not in ('FREE','NORMAL') gate_index=4 if has_stage else 3 for field, expected in {'boundary':t[2], 'stage':t[3] if has_stage else None, 'flapGate':t[gate_index] if len(t)>gate_index else 'NO'}.items(): compare(id,field,n.get(field),expected) elif kind == 'storage': compare(id,'storageShape',n.get('storageShape'),t[4]) compare(id,'storageCurve',n.get('storageCurve'),t[5] if t[4]=='TABULAR' else None) compare(id, 'connections', sorted(n.get('connections', [])), sorted(expected_connections[id])) if id not in coordinates: exclusions.append({'id':id, 'type':kind, 'reason':'missing_coordinate'}) rendered_conduits=0 shape_counts=Counter() for id, (kind, line, t) in source_links.items(): if id not in links: continue link = links[id] for field, expected in {'type':kind, 'line':line, 'raw':t, 'fromNode':t[1], 'toNode':t[2], 'vertices':vertices[id]}.items(): compare(id, field, link.get(field), expected) missing = t[1] not in coordinates or t[2] not in coordinates path = None if missing else [coordinates[t[1]], *vertices[id], coordinates[t[2]]] compare(id, 'pathXY', link.get('pathXY'), path) distance = None if missing else sum(math.dist(a,b) for a,b in zip(path,path[1:])) compare(id, 'planDistance', link.get('planDistance'), distance) reason = 'missing_coordinate' if missing else None if kind == 'pump': for field, expected in {'pumpCurve':t[3],'initialStatus':t[4], 'startupDepth':number(t,5,0)*factor, 'shutoffDepth':number(t,6,0)*factor}.items(): compare(id,field,link.get(field),expected) elif kind == 'weir': crest=number(t,4)*factor elevation=crest if offset=='ELEVATION' else number(source_nodes[t[1]][2],1)*factor+crest for field, expected in {'weirType':t[3],'crestOffset':crest,'crestElevation':elevation, 'dischargeCoefficient':number(t,5)}.items(): compare(id,field,link.get(field),expected) if kind == 'conduit': if id not in xs: raise ValueError('missing XSECTIONS: ' + id) _, x = xs[id] shape = x[1].upper() shape_counts[shape]+=1 if shape not in ('CIRCULAR', 'RECT_CLOSED', 'RECT_OPEN', 'TRAPEZOIDAL'): raise ValueError('unsupported conduit shape: ' + shape) height = number(x,2) * factor width = height if shape == 'CIRCULAR' else number(x,3) * factor section = link.get('section') or {} for field, expected in {'shape':shape,'height':height,'width':width, 'diameter':height if shape=='CIRCULAR' else None, 'barrels':number(x,6,1),'raw':x}.items(): compare(id, 'section.'+field, section.get(field), expected) if shape == 'TRAPEZOIDAL': compare(id, 'section.sideSlopes', section.get('sideSlopes'), [number(x,4),number(x,5)]) start = number(t,5,0)*factor end = number(t,6,0)*factor inv_start = start if offset=='ELEVATION' else number(source_nodes[t[1]][2],1)*factor+start inv_end = end if offset=='ELEVATION' else number(source_nodes[t[2]][2],1)*factor+end for field, expected in {'length':number(t,3)*factor,'roughness':number(t,4), 'inletOffset':start,'outletOffset':end,'invertStart':inv_start, 'invertEnd':inv_end,'centerStart':inv_start+height/2, 'centerEnd':inv_end+height/2}.items(): compare(id, field, link.get(field), expected) length=number(t,3)*factor for field, expected in {'lengthDelta':None if distance is None else distance-length, 'lengthDifferenceRatio':None if distance is None or length<=0 else abs(distance-length)/length, 'slopeInput':None if length<=0 else (inv_start-inv_end)/length, 'slopePlan':None if distance is None or distance<=1e-9 else (inv_start-inv_end)/distance}.items(): compare(id,field,link.get(field),expected) if distance is not None and distance <= 1e-9: reason = 'coincident_endpoints' if reason is None: rendered_conduits+=1 compare(id, 'renderable', link.get('renderable'), reason is None) if reason: exclusions.append({'id':id,'type':kind,'reason':reason}) compare('summary','counts',data['summary'].get('counts'),expected_counts) compare('summary','shapes',data['summary'].get('shapes'),dict(shape_counts)) compare('summary','renderedConduits',data['summary'].get('renderedConduits'),rendered_conduits) compare('summary','vertexCount',data['summary'].get('vertexCount'),sum(len(v) for v in vertices.values())) compare('summary','coordinateCount',data['summary'].get('coordinateCount'), sum(id in coordinates for id in source_nodes)) for excluded in exclusions: ids = excluded['id'] allowed = ('missing_coordinate',) if ids in nodes else ('unplaced_link','coincident_endpoints') compare(ids,'exclusion documented',any(ids in i.get('ids',[]) and i.get('code') in allowed for i in data['issues']),True) return {'status':'passed' if not errors else 'failed','checks':checks, 'counts':expected_counts,'mismatches':errors,'excluded_facilities':exclusions, 'tolerance_absolute':tolerance, 'scope':'Input/JSON consistency only; no hydraulic or surveyed geometry validation.'} def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--inp',required=True,type=Path) parser.add_argument('--network',type=Path,default=Path(__file__).resolve().parents[1]/'web/data/network.json') parser.add_argument('--output',type=Path) args=parser.parse_args() try: report=validate(args.inp,args.network) code=0 if report['status']=='passed' else 1 except (OSError,ValueError,KeyError,IndexError,TypeError) as exc: report={'status':'input_error','error':str(exc)} code=2 text=json.dumps(report,ensure_ascii=False,indent=2) if args.output: args.output.parent.mkdir(parents=True,exist_ok=True) args.output.write_text(text+'\n',encoding='utf-8') print(text) return code if __name__=='__main__': sys.exit(main())