abse/smartwatch-flex-top-only

scripts/run-short-audits.py

from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import hashlib, json, os, re, subprocess, sys

root = Path(__file__).resolve().parent.parent
full, external = (root / p for p in sys.argv[1:3])
out = root / 'reports/shorts'
out.mkdir(parents=True, exist_ok=True)
env = dict(os.environ, XDG_CONFIG_HOME='/workspace/.cloud-tools/config',
           XDG_CACHE_HOME='/workspace/.cache', XDG_STATE_HOME='/workspace/.cloud-tools/state',
           NODE_USE_ENV_PROXY='1')
bun = '/workspace/.cloud-tools/node_modules/.bin/bun'
sha = lambda p: hashlib.sha256(p.read_bytes()).hexdigest()

def run(job):
    scope, mode, layer = job
    source = full if scope == 'full' else external
    name = f'{scope}-{mode}-{layer}'
    cwd = out / 'runs' / name
    cwd.mkdir(parents=True, exist_ok=True)
    log = out / f'{name}.log'
    result = subprocess.run([bun, str(root/'scripts/check-shorts-complete.mjs'), str(source),
                             '--mode', mode, '--layer', layer, '--pixels-per-mm', '25'],
                            cwd=cwd, env=env, stdin=subprocess.DEVNULL,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    log.write_text(result.stdout)
    match = re.search(r'Detected (\d+) shorts? in ', result.stdout)
    count = 0 if 'No shorts detected in ' in result.stdout else int(match[1]) if match else None
    record = dict(scope=scope, mode=mode, layer=layer, 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=result.returncode, reportedShorts=count)
    print(f'{name}: exit {result.returncode}, reported shorts {count}', flush=True)
    return record

jobs = [(scope, mode, layer) for scope in ['full', 'external']
        for mode in ['pcb', 'gerber'] for layer in ['top', 'inner1', 'inner2', 'bottom']]
records = []
if (out/'manifest.json').exists():
    for prior in json.loads((out/'manifest.json').read_text())['runs']:
        source=full if prior['scope']=='full' else external
        if prior['exitCode']==0 and prior['reportedShorts']==0 and prior['sourceSha256']==sha(source):
            records.append(prior)
completed={(r['scope'],r['mode'],r['layer']) for r in records}
jobs=[job for job in jobs if job not in completed]
with ThreadPoolExecutor(max_workers=2) as executor:
    for future in as_completed([executor.submit(run, job) for job in jobs]):
        records.append(future.result())
        (out/'manifest.json').write_text(json.dumps(dict(pixelsPerMM=25, runs=records), indent=2)+'\n')
assert len(records) == 16 and all(r['reportedShorts'] is not None for r in records)
assert all(r['exitCode'] == 0 and r['reportedShorts'] == 0 for r in records)