240 lines
12 KiB
Python
240 lines
12 KiB
Python
#!/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 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 isinstance(expected, (float, int)) and not isinstance(expected, bool):
|
|
ok = (isinstance(actual, (float, int)) and not isinstance(actual, bool)
|
|
and math.isfinite(actual) and abs(actual - expected) <= tolerance)
|
|
else:
|
|
ok = actual == expected
|
|
if not ok:
|
|
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())
|