abse/smartwatch-flex-top-only

scripts/check-oled-clock-shorts.py

"""Scan the changed bottom copper; reuse other layers only after exact equality."""
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
import hashlib, json, os, re, subprocess

root=Path(__file__).resolve().parent.parent
baseline=root/'baseline-v0217'
out=root/'reports/oled-clock-style/shorts'
out.mkdir(parents=True,exist_ok=True)
sha=lambda p:hashlib.sha256(p.read_bytes()).hexdigest()
read=lambda p:json.loads(p.read_text())
source_by_scope={'full':root/'dist/index/circuit.json','external':root/'dist/index/external-shorts.circuit.json'}
prior=read(baseline/'shorts-manifest.json')['runs']

def unchanged_layers(source):
    # The exact full board apart from bottom-only wires, bottom pours and
    # diagnostic warnings. All nets, pads, through vias and mixed-layer traces
    # remain in this comparison; no logical connectivity aliases are used.
    return [e for e in source if not (
        'warning' in e['type'] or
        (e['type']=='pcb_trace' and all(p['route_type']=='wire' and p['layer']=='bottom' for p in e['route'])) or
        (e['type']=='pcb_copper_pour' and e['layer']=='bottom'))]

records=[]
for r in prior:
    if r['layer']=='bottom':continue
    old=baseline/('index.circuit.json' if r['scope']=='full' else 'external-shorts.circuit.json');new=source_by_scope[r['scope']]
    baseline_log=baseline/'logs'/Path(r['log']).name
    assert r['exitCode']==0 and r['reportedShorts']==0
    assert sha(old)==r['sourceSha256'] and sha(baseline_log)==r['logSha256']
    old_projection=unchanged_layers(read(old));new_projection=unchanged_layers(read(new))
    assert old_projection==new_projection,(r['scope'],r['layer'])
    projection_sha=hashlib.sha256(json.dumps(new_projection,separators=(',',':')).encode()).hexdigest()
    records.append({**r,'sourceFile':str(new.relative_to(root)),'sourceSha256':sha(new),
                    'log':str(baseline_log.relative_to(root)),
                    'baselineSourceSha256':r['sourceSha256'],
                    'verificationMethod':'Reused passing scan after exact equality of every non-bottom wire, mixed-layer trace, pour, via, pad, net and all other non-diagnostic records.',
                    'unchangedLayerProjectionSha256':projection_sha,'freshScan':False})

env=dict(os.environ,NODE_USE_ENV_PROXY='1',XDG_CONFIG_HOME='/workspace/.cloud-tools/config',
         XDG_CACHE_HOME='/workspace/.cache',XDG_STATE_HOME='/workspace/.cloud-tools/state')
bun='/workspace/.cloud-tools/node_modules/.bin/bun'
def run(job):
    scope,mode=job;source=source_by_scope[scope];name=f'{scope}-{mode}-bottom'
    cwd=out/name;cwd.mkdir(exist_ok=True)
    p=subprocess.run([bun,str(root/'scripts/check-shorts-complete.mjs'),str(source),
                      '--mode',mode,'--layer','bottom','--pixels-per-mm','25'],
                      cwd=cwd,env=env,stdin=subprocess.DEVNULL,stdout=subprocess.PIPE,
                      stderr=subprocess.STDOUT,text=True)
    log=out/f'{name}.log';log.write_text(p.stdout)
    match=re.search(r'Detected (\d+) shorts? in ',p.stdout)
    count=0 if 'No shorts detected in ' in p.stdout else int(match[1]) if match else None
    r=dict(scope=scope,mode=mode,layer='bottom',pixelsPerMM=25,sourceFile=str(source.relative_to(root)),
           sourceSha256=sha(source),log=str(log.relative_to(root)),logSha256=sha(log),
           artifactsDirectory=str(cwd.relative_to(root)),exitCode=p.returncode,
           reportedShorts=count,freshScan=True,verificationMethod='Fresh native scan of changed bottom copper.')
    print(r,flush=True)
    return r
jobs=[(scope,mode) for scope in ['full','external'] for mode in ['pcb','gerber']]
manifest=out/'manifest.json'
completed_records=read(manifest)['runs'] if manifest.exists() else []
retry=out/'external-retry-records.json'
if retry.exists():completed_records+=read(retry)
if completed_records:
    for r in completed_records:
        if (r['scope'],r['mode']) not in jobs:continue
        if r.get('freshScan') and r['exitCode']==0 and r['reportedShorts']==0 and sha(source_by_scope[r['scope']])==r['sourceSha256'] and sha(root/r['log'])==r['logSha256']:
            records.append(r)
            jobs.remove((r['scope'],r['mode']))
with ThreadPoolExecutor(max_workers=2) as pool:
    for r in pool.map(run,jobs):
        records.append(r)
        (out/'manifest.json').write_text(json.dumps({'pixelsPerMM':25,'runs':records},indent=2)+'\n')
(out/'manifest.json').write_text(json.dumps({'pixelsPerMM':25,'runs':records},indent=2)+'\n')
assert len(records)==16 and all(r['exitCode']==0 and r['reportedShorts']==0 for r in records)
assert sum(r['freshScan'] for r in records)==4
print('PASS: four fresh bottom scans and twelve exactly unchanged layer scans.',flush=True)