KiCad - the KiCad Bridge

"""Apply the routing development overlay to an existing PINNED bridge cache.

Run with the bridge stopped: python install_routing_dev.py CACHE OVERLAY.zip.
Only the routing files and registration entries change; unrelated newer bridge
fixes stay installed. A timestamped backup beside the cache permits restoration.
"""
import ast
from datetime import datetime, timezone
import json
from pathlib import Path
import shutil
import sys
import zipfile

FILES = ['handlers/live_routing.py', 'handlers/route.py', 'parsers/pcb.py', 'requirements-routing.txt']
NAMES = ['routing_state', 'route_net', 'remove_route', 'routing_validate', 'board_pads', 'add_track', 'add_via', 'route']


def after_assignment(source, name, addition):
    node = next(n for n in ast.walk(ast.parse(source)) if isinstance(n, ast.Assign)
                and any(isinstance(t, ast.Name) and t.id == name for t in n.targets))
    lines = source.splitlines(keepends=True)
    indentation = ' ' * node.col_offset
    lines.insert(node.end_lineno, '\n' + '\n'.join(indentation + line for line in addition.splitlines()) + '\n')
    return ''.join(lines)


def install(cache, overlay):
    cache = Path(cache)
    if not (cache / '.dev-pin').is_file():
        raise ValueError('Refusing: use ab bridge_dev_mode {name:kicad,on:true} first')
    # Read and compile the entire overlay before touching the running cache.
    with zipfile.ZipFile(overlay) as z:
        contents = {name: z.read(name) for name in FILES}
        catalog = json.loads(z.read('routing-catalog.json'))
    if set(catalog) != set(NAMES):
        raise ValueError('Incomplete routing verb catalog')
    for name, data in contents.items():
        if name.endswith('.py'):
            compile(data, name, 'exec')
    source = (cache / 'server.py').read_text(encoding='utf-8')
    # Remove only our delimited registrations so reinstallation updates help too.
    import re
    source = re.sub(r'\n[ \t]*# ADOM_LIVE_ROUTING_DEV_BEGIN.*?# ADOM_LIVE_ROUTING_DEV_END\n', '\n', source, flags=re.S)
    if '# ADOM_LIVE_ROUTING_DEV' in source:
        raise ValueError('Old development overlay found; restore server.py from its backup before upgrading')
    def marked(text):
        return '# ADOM_LIVE_ROUTING_DEV_BEGIN\n' + text + '\n# ADOM_LIVE_ROUTING_DEV_END'
    source = after_assignment(source, '_VERB_CATALOG', marked('_VERB_CATALOG.update(' + repr(catalog) + ')'))
    addition = ('from handlers.live_routing import handle_routing_state, handle_route_net, handle_remove_route, handle_routing_validate\n'
                'from handlers.route import handle_board_pads, handle_add_track, handle_add_via, handle_route\n'
                'COMMAND_HANDLERS.update({' + ', '.join(repr(n) + ': handle_' + n for n in NAMES) + '})')
    source = after_assignment(source, 'COMMAND_HANDLERS', marked(addition))
    source = after_assignment(source, '_NO_GUI', marked('_NO_GUI.update(' + repr(set(NAMES)) + ')'))
    compile(source, str(cache / 'server.py'), 'exec')
    contents['server.py'] = source.encode('utf-8')
    manifest = json.loads((cache / 'bridge.json').read_text())
    for name in NAMES:
        if 'kicad_' + name not in manifest['verbs']:
            manifest['verbs'].append('kicad_' + name)
    manifest['version'] = '0.10.0-dev.per-net.1'
    contents['bridge.json'] = (json.dumps(manifest, indent=2) + '\n').encode()
    contents['BRIDGE_VERSION'] = manifest['version'].encode()
    contents['.bridge-version'] = manifest['version'].encode()
    backup = cache.parent / ('kicad-routing-backup-' + datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%f'))
    backup.mkdir()
    originals = {}
    for name in contents:
        original = cache / name
        originals[name] = original.read_bytes() if original.is_file() else None
        if originals[name] is not None:
            target = backup / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(originals[name])
    try:
        for name, data in contents.items():
            target = cache / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
    except Exception:
        for name, data in originals.items():
            if data is None:
                (cache / name).unlink(missing_ok=True)
            else:
                (cache / name).write_bytes(data)
        raise
    result = {'installed': str(cache), 'backup': str(backup), 'version': manifest['version']}
    print(json.dumps(result))
    return result

if __name__ == '__main__':
    install(*sys.argv[1:])