test: add tools/validate_network.py for issue #1
This commit is contained in:
dekim2 committed
1 parent
5e81e9cfe9
commit
fbff81d189
1 file changed
+239
@@ -0,0 +1,239 @@
|
||||
#!/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())
|
||||
Reference in new issue
Block a user