XFESystem

"""Headless native-resolution smoke test; does not claim GPU VSync or a measured presentation rate."""
import argparse
import hashlib
import json
import pathlib
import shutil
import socket
import struct
import subprocess
import time
from PIL import Image
from ReadImage import inspect_image

ROOT = pathlib.Path(__file__).resolve().parents[2]
parser = argparse.ArgumentParser()
parser.add_argument('image')
parser.add_argument('--width', type=int, default=3840)
parser.add_argument('--height', type=int, default=2160)
parser.add_argument('--refresh-hint', type=int, default=240)
parser.add_argument('--diagnose-handoff', action='store_true')
options = parser.parse_args()
source = pathlib.Path(options.image).resolve()
report, files = inspect_image(source)
assert b'resolution: 0x0x32' in files['Limine.conf'], 'The display-chain test requires automatic resolution.'
out = ROOT / 'build/CSharpBoot' / ('DisplayChain-' + str(time.time_ns()))
out.mkdir(parents=True)
disk = out / 'System.img'
shutil.copyfile(source, disk)
variables = out / 'FirmwareVars.fd'
shutil.copyfile(ROOT / '.tools/firmware/edk2-i386-vars.fd', variables)
with socket.socket() as probe:
    probe.bind(('127.0.0.1', 0))
    port = probe.getsockname()[1]
args = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35', '-accel', 'tcg', '-cpu', 'max', '-smp', '1', '-m', '512M',
        '-vga', 'std', '-global', 'VGA.vgamem_mb=64', '-global', 'VGA.edid=on',
        '-global', 'VGA.xres=' + str(options.width), '-global', 'VGA.yres=' + str(options.height),
        '-global', 'VGA.xmax=' + str(options.width), '-global', 'VGA.ymax=' + str(options.height), '-global', 'VGA.refresh_rate=' + str(options.refresh_hint * 1000),
        '-drive', 'if=pflash,format=raw,readonly=on,file=' + str(ROOT / '.tools/firmware/edk2-x86_64-code.fd'),
        '-drive', 'if=pflash,format=raw,file=' + str(variables), '-drive', 'file=' + str(disk) + ',format=raw,if=ide,snapshot=on',
        '-display', 'none', '-serial', 'file:' + str(out / 'Serial.log'), '-monitor', 'none', '-no-reboot',
        '-netdev', 'user,id=Network', '-device', 'e1000e,netdev=Network', '-qmp', 'tcp:127.0.0.1:' + str(port) + ',server=on,wait=off']
result = {'Kind': 'HeadlessNativeDisplayChain', 'SourceImage': str(source), 'ImageSha256': report['ImageSha256'],
          'ImageVerification': report, 'QemuArgs': args, 'Passed': False, 'RefreshHintHz': options.refresh_hint, 'QemuRefreshRateMilliHz': options.refresh_hint * 1000,
          'RefreshMeaning': 'EDID target only; this test does not measure GPU VSync or frames per second', 'AutomaticInput': False, 'PhysicalDiskAccess': False}
process = subprocess.Popen(args, cwd=ROOT, stdout=(out / 'QemuStdout.log').open('wb'), stderr=(out / 'QemuStderr.log').open('wb'))
client = stream = None
def command(name, arguments=None):
    value = {'execute': name}
    if arguments: value['arguments'] = arguments
    stream.write(json.dumps(value).encode() + b'\n')
    stream.flush()
    while True:
        response = json.loads(stream.readline())
        if 'return' in response: return response['return']
        if 'error' in response: raise RuntimeError(response['error'])
try:
    deadline = time.monotonic() + 120
    while time.monotonic() < deadline:
        if process.poll() is not None: raise RuntimeError('QEMU exited before desktop readiness')
        try:
            client = socket.create_connection(('127.0.0.1', port), timeout=1)
            client.settimeout(15)
            break
        except OSError: time.sleep(0.1)
    assert client is not None, 'QMP startup timeout'
    stream = client.makefile('rwb')
    greeting = json.loads(stream.readline())
    assert 'QMP' in greeting
    command('qmp_capabilities')
    while time.monotonic() < deadline:
        serial = (out / 'Serial.log').read_text(errors='replace') if (out / 'Serial.log').exists() else ''
        assert '[FAILFAST]' not in serial and '*** UNHANDLED EXCEPTION' not in serial, serial[-4000:]
        if 'XFE_CSHARP_DESKTOP_READY' in serial: break
        if process.poll() is not None: raise RuntimeError('QEMU exited before the desktop')
        time.sleep(0.1)
    else: raise TimeoutError('Desktop startup timed out')
    time.sleep(0.5)
    pci = command('query-pci')
    result['PciDevices'] = pci
    for bus in pci:
        for device in bus.get('devices', []):
            if device.get('id', {}).get('vendor') != 0x1234 or device.get('id', {}).get('device') != 0x1111: continue
            for region in device.get('regions', []):
                if region.get('bar') != 2 or region.get('address', 0) <= 0: continue
                edid_path = out / 'QemuEdid.bin'
                command('pmemsave', {'val': region['address'], 'size': 384, 'filename': str(edid_path)})
                edid = edid_path.read_bytes()
                result['QemuEdidHeader'] = edid[:8].hex()
                result['QemuEdidExtensionCount'] = edid[126]
                result['QemuEdidBlockChecksums'] = [sum(edid[index:index+128]) & 255 for index in range(0, 384, 128)]
                if edid[256] == 0x70 and edid[261] == 3:
                    timing = edid[264:284]
                    clock = int.from_bytes(timing[:3], 'little') * 10000
                    h, hb = struct.unpack_from('<HH', timing, 4)
                    v, vb = struct.unpack_from('<HH', timing, 12)
                    result['QemuDisplayIdTiming'] = {'Width': h + 1, 'Height': v + 1, 'RefreshHz': clock / ((h + hb + 2) * (v + vb + 2))}
    if options.diagnose_handoff:
        for start in range(448 * 1024 * 1024, -1, -64 * 1024 * 1024):
            temporary = out / 'MemoryScan.bin'
            command('pmemsave', {'val': start, 'size': 64 * 1024 * 1024, 'filename': str(temporary)})
            memory = temporary.read_bytes()
            temporary.unlink()
            at = 0
            found = False
            while True:
                at = memory.find(b'XFEDISP1', at)
                if at < 0: break
                if at + 60192 <= len(memory) and struct.unpack_from('<II', memory, at + 8) == (1, 60192) and struct.unpack_from('<I', memory, at + 16)[0] <= 8:
                    value = memory[at:at+60192]
                    (out / 'DisplayHandoff.bin').write_bytes(value)
                    display = value[32:32+7520]
                    result['BootShimHandoff'] = {'PhysicalAddress': start + at, 'Flags': struct.unpack_from('<I', value, 28)[0], 'DisplayCount': struct.unpack_from('<I', value, 16)[0],
                        'Width': struct.unpack_from('<I', display, 40)[0], 'Height': struct.unpack_from('<I', display, 44)[0],
                        'DisplayFlags': struct.unpack_from('<I', display, 48)[0], 'RequestedMode': struct.unpack_from('<I', display, 92)[0],
                        'SelectedMode': struct.unpack_from('<I', display, 32)[0], 'EdidBytes': struct.unpack_from('<I', display, 56)[0],
                        'EdidSource': struct.unpack_from('<I', display, 60)[0]}
                    found = True
                    break
                at += 8
            if found: break
    screenshot = out / 'Screen.ppm'
    command('screendump', {'filename': str(screenshot)})
    with Image.open(screenshot) as image:
        result['NativeScreenshotSize'] = list(image.size)
        image.save(out / 'Screen.png')
        assert image.size == (options.width, options.height), 'Automatic GOP selection did not retain the EDID native resolution'
    result['NativeScreenshotSha256'] = hashlib.sha256((out / 'Screen.png').read_bytes()).hexdigest()
    result['DesktopReady'] = True
    result['Passed'] = True
except Exception as error:
    result['Error'] = str(error)
finally:
    if stream and process.poll() is None:
        try: command('quit')
        except Exception: pass
    try: process.wait(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill(); process.wait()
    if stream: stream.close()
    if client: client.close()
    result['QemuExitCode'] = process.returncode
    result['ImageUnchanged'] = hashlib.sha256(disk.read_bytes()).hexdigest() == report['ImageSha256']
    result['SourceImageUnchanged'] = hashlib.sha256(source.read_bytes()).hexdigest() == report['ImageSha256']
    result['Passed'] = result['Passed'] and process.returncode == 0 and result['ImageUnchanged'] and result['SourceImageUnchanged']
    (out / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(str(out / 'Result.json'))
    print(json.dumps(result, indent=2))
raise SystemExit(0 if result['Passed'] else 1)