XFESystem

"""Compare cold/warm full Monaco initialization in two verified native images.

Actual PS/2 navigation, model readback, presented software-rendered glyphs, and
keyboard editing. No pre-warming, injected JS, host browser or substituted frame.
"""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import socket
import statistics
import subprocess
import time

from PIL import Image
from TestBoot import Qmp
from IndependentBingEvidence import PRESENTED, window_geometry
from TestBrowserPipeline import _monaco_frame_stats, _monaco_fourth_line_delta

ROOT = Path(__file__).resolve().parents[2]
ADDRESS = 'http://xfesystem.local/Browser/Monaco/Index.html'
VALUE = '// XFESystem | XFEstudio\nconst answer = 6 * 7;\nconsole.log(answer);'
MODELS = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)


def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def run(receipt_path, output, iterations):
    receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig'))
    assert receipt['Passed'] and receipt['ContainsCosmos'] is False
    image = Path(receipt['ImagePath'])
    assert digest(image) == receipt['ImageSha256']
    output.mkdir(parents=True)
    result = dict(Kind='XfeNativeFullMonacoStartup', Passed=False,
                  ImageSha256=receipt['ImageSha256'], VerifiedReceipt=str(receipt_path),
                  VerifiedReceiptSha256=digest(receipt_path), HarnessSha256=digest(__file__),
                  VirtualCpus=8, ExecutingCpus=1, HostBrowserUsed=False, Prewarmed=False,
                  HostScreenshotSubstitution=False, Cases=[])
    serial_path = output / 'Serial.log'
    qemu = qmp = None

    def text():
        return serial_path.read_text(encoding='utf-8', errors='replace') if serial_path.exists() else ''

    def wait(predicate, seconds=120):
        end = time.monotonic() + seconds
        while time.monotonic() < end:
            assert qemu.poll() is None, 'Guest exited'
            log = text()
            assert not any(m in log for m in ('XFE_NATIVE_FAIL', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST',
                'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_EDITOR_BACKGROUND|SessionError|', 'XFE_BROWSER_PAINT_FAILED'))
            if predicate(log):
                return log
            time.sleep(.02)
        raise AssertionError('Full editor initialization timeout: ' + text()[-4096:])

    def key(chord):
        qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
        time.sleep(.12)

    def type_text(value):
        codes = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', ' ': 'spc'}
        for char in value:
            key(codes.get(char, 'shift-' + char.lower() if char.isupper() else char))

    def frame(name):
        qmp.call('screendump', {'filename': str(output / (name + '.ppm'))})
        with Image.open(output / (name + '.ppm')) as image:
            rgb = image.convert('RGB')
        rgb.save(output / (name + '.png'))
        return rgb

    def browser_visible(log):
        try:
            window_geometry(log, 'Browser')
            return True
        except AssertionError:
            return False

    try:
        shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', output / 'FirmwareVars.fd')
        with socket.socket() as reservation:
            reservation.bind(('127.0.0.1', 0))
            port = reservation.getsockname()[1]
        command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35', '-accel', 'tcg',
            '-cpu', 'max', '-smp', '8', '-m', '1024M', '-vga', 'std', '-global', 'VGA.vgamem_mb=64',
            '-display', 'none', '-net', 'none', '-rtc', 'base=utc',
            '-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
            '-drive', f'if=pflash,format=raw,file={output / "FirmwareVars.fd"}',
            '-drive', f'file={image},format=raw,if=ide,snapshot=on', '-serial', f'file:{serial_path}',
            '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
        result['QemuArguments'] = command
        with (output / 'Qemu.log').open('w') as log:
            qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                                    creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
            end = time.monotonic() + 30
            while qmp is None:
                try:
                    qmp = Qmp(port)
                except OSError:
                    assert time.monotonic() < end
                    time.sleep(.1)
            wait(lambda s: 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in s)
            wait(lambda s: 'XFE_DESKTOP_STATE_END' in s)
            if 'XFE_DESKTOP_ACTIVE|Terminal' in text():
                key('alt-f4')
            key('ctrl-esc'); key('down'); key('down'); key('ret')
            wait(browser_visible)
            browser_id, _ = window_geometry(text(), 'Browser')
            wait(lambda s: f'XFE_EDITOR_BACKGROUND|Ready|Window|{browser_id}|' in s)
            initial_frame = None
            viewport = None
            for iteration in range(1, iterations + 1):
                key('ctrl-l'); type_text('about:Monaco')
                start = len(text())
                begun = time.monotonic_ns()
                qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})

                def ready(log):
                    models = MODELS.findall(log, start)
                    return models and models[-1] == VALUE and any(m[3] == ADDRESS for m in PRESENTED.finditer(log, start))

                log = wait(ready)
                _, (x, y, width, height) = window_geometry(log, 'Browser')
                viewport = (x + 2, y + 67, width - 4, height - 96)
                deadline = time.monotonic() + 60
                previous_pixels = None
                while True:
                    rgb = frame('Ready' + str(iteration))
                    stats = _monaco_frame_stats(rgb, viewport, 1)
                    web = rgb.crop((viewport[0], viewport[1], viewport[0] + viewport[2], viewport[1] + viewport[3]))
                    pixels = hashlib.sha256(web.tobytes()).hexdigest()
                    # A scanout can be sampled while it is being copied. Require
                    # the actual glyph frame to agree on two successive captures.
                    if stats['ReadyPainted'] and previous_pixels == pixels:
                        break
                    previous_pixels = pixels if stats['ReadyPainted'] else None
                    assert time.monotonic() < deadline, 'The full editor source and syntax glyphs were not painted'
                    wait(lambda s: True)
                    time.sleep(.1)
                elapsed = (time.monotonic_ns() - begun) / 1_000_000
                log = text()
                web.save(output / ('Viewport' + str(iteration) + '.png'))
                timings = []
                for line in log[start:].splitlines():
                    if not line.startswith('XFE_BROWSER_SCRIPT_TIMING|'):
                        continue
                    fields = line.split('|')
                    values = dict(zip(fields[2::2], fields[3::2]))
                    frequency = int(values['Frequency'])
                    timings.append(dict(Resource=fields[1], CacheHit=values.get('CacheHit'),
                        **{('Total' if k == 'Ticks' else k.removesuffix('Ticks')): int(v) * 1000 / frequency
                           for k, v in values.items() if k.endswith('Ticks')}))
                result['Cases'].append(dict(Iteration=iteration, Cold=iteration == 1,
                    EnterToFullPresentedMilliseconds=elapsed, ExactModel=VALUE, NativeGlyphs=stats,
                    ScriptTimingsMilliseconds=timings, WebPixelsSha256=hashlib.sha256(web.tobytes()).hexdigest()))
                initial_frame = rgb
                time.sleep(.15)
            # Native key packets must change the actual model AND add new glyphs.
            # Click the real text surface using relative PS/2 movement.
            pointer = [640, 400]
            target = [viewport[0] + 160, viewport[1] + 70]
            while pointer != target:
                delta = [max(-80, min(80, target[i] - pointer[i])) for i in range(2)]
                qmp.call('input-send-event', {'events': [
                    {'type': 'rel', 'data': {'axis': axis, 'value': delta[i]}} for i, axis in enumerate(('x', 'y'))]})
                pointer = [pointer[i] + delta[i] for i in range(2)]
                time.sleep(.025)
            for down in (True, False):
                qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
                time.sleep(.1)
            start = len(text()); key('ctrl-end'); key('ret'); type_text('xfenativeedit')
            expected = VALUE + '\nxfenativeedit'
            wait(lambda s: MODELS.findall(s, start) and MODELS.findall(s, start)[-1] == expected)
            edited = frame('Edited')
            stats = _monaco_frame_stats(edited, viewport, 1)
            delta = _monaco_fourth_line_delta(initial_frame, edited, stats['Lines'][3]['NativeBounds'])
            assert stats['Lines'][3]['HasTextGlyphDistribution'] and delta['NewInkPixels'] >= 35 and delta['NewInkColumns'] >= 24
            result.update(Passed=True, KeyboardEditedExactModel=True, EditedModel=expected, EditedGlyphDelta=delta)
    except Exception as error:
        result['Error'] = str(error)
    finally:
        if qemu is not None:
            if qemu.poll() is None and qmp is not None:
                try:
                    qmp.call('quit')
                except (OSError, RuntimeError):
                    pass
            try:
                qemu.wait(timeout=10)
            except subprocess.TimeoutExpired:
                qemu.terminate(); qemu.wait(timeout=10); result['ForcedCleanup'] = True
            result['QemuExitCode'] = qemu.returncode
        if qmp is not None:
            qmp.close()
        result['ImageUnchanged'] = digest(image) == receipt['ImageSha256']
        result['Passed'] &= result.get('QemuExitCode') == 0 and result['ImageUnchanged']
        (output / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps({k: v for k, v in result.items() if k not in ('Cases', 'QemuArguments')}, ensure_ascii=True))
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--baseline-proof', type=Path, required=True)
    parser.add_argument('--optimized-proof', type=Path, required=True)
    parser.add_argument('--iterations', type=int, default=5, choices=range(3, 10))
    args = parser.parse_args()
    output = ROOT / 'build/MonacoInit' / f'NativeComparison-{time.time_ns()}'
    before = run(args.baseline_proof.resolve(), output / 'Before', args.iterations)
    after = run(args.optimized_proof.resolve(), output / 'After', args.iterations)
    passed = before['Passed'] and after['Passed']
    comparison = dict(Kind='XfeNativeFullMonacoStartupComparison', Passed=passed, FullReleaseQualified=False,
        Method='QEMU TCG max, 8 vCPUs / 1 executing BSP; sequential cold/warm navigation to the unchanged official Monaco editor',
        Before=str(output / 'Before/Result.json'), After=str(output / 'After/Result.json'))
    if passed:
        old = [r['EnterToFullPresentedMilliseconds'] for r in before['Cases']]
        new = [r['EnterToFullPresentedMilliseconds'] for r in after['Cases']]
        models_equal = [r['ExactModel'] for r in before['Cases']] == [r['ExactModel'] for r in after['Cases']]
        pixels_equal = [r['WebPixelsSha256'] for r in before['Cases']] == [r['WebPixelsSha256'] for r in after['Cases']]
        passed = models_equal and pixels_equal
        comparison['Passed'] = passed
        comparison.update(BeforeColdMilliseconds=old[0], AfterColdMilliseconds=new[0], ColdSpeedup=old[0] / new[0],
            BeforeWarmMedianMilliseconds=statistics.median(old[1:]), AfterWarmMedianMilliseconds=statistics.median(new[1:]),
            AllNativeViewportPixelsEqual=pixels_equal, ExactModelsEqual=models_equal, BothImagesPassedActualKeyboardEditing=True)
        if not passed:
            comparison['Error'] = 'Native model or viewport pixel equality failed'
    (output / 'Comparison.json').write_text(json.dumps(comparison, indent=2), encoding='utf-8')
    print(json.dumps(comparison))
    raise SystemExit(0 if passed else 1)