XFESystem

"""Strict parsing/analysis of opt-in native software presentation timing.

No EDID value, QMP acknowledgement, or screenshot is a physical refresh-rate
measurement. The native record is a bounded post-capture ring dump, not a
per-frame serial stream that would change the measured hot path.
"""
import math
import re


BEGIN = 'XFE_POINTER_TIMING_BEGIN|1\n'
END = 'XFE_POINTER_TIMING_END\n'
META_FIELDS = {'Frequency', 'Capacity', 'StartTicks', 'EndTicks', 'FramesBefore', 'FramesAfter',
               'InputsBefore', 'InputsAfter', 'FirstInputTicks', 'LastInputTicks',
               'ObservedBefore', 'ObservedAfter', 'ConsumedBefore', 'ConsumedAfter',
               'LostBefore', 'LostAfter', 'DroppedFrames', 'CapturedFrames', 'Overflow', 'FramesPresentedDelta', 'InvalidInputReports'}
FRAME_FIELDS = {'Seq', 'BeginTicks', 'EndTicks', 'InputSeq', 'InputTicks', 'DispatchTicks', 'X', 'Y'}


def fields(line, prefix, required):
    parts = line.split('|')
    if parts[0] != prefix or len(parts) % 2 != 1:
        raise ValueError('Malformed timing field pairs: ' + line[:256])
    values = {}
    for name, raw in zip(parts[1::2], parts[2::2]):
        if not re.fullmatch(r'[A-Za-z][A-Za-z0-9]*', name) or name in values:
            raise ValueError('Invalid or duplicate timing field: ' + name)
        if not re.fullmatch(r'[0-9]+', raw) or len(raw) > 19 or int(raw) > (1 << 63) - 1:
            raise ValueError('Timing value is not a nonnegative signed-64 integer: ' + name)
        values[name] = int(raw)
    if not required.issubset(values):
        raise ValueError('Missing timing fields: ' + ','.join(sorted(required.difference(values))))
    return values


def parse_timing_dump(log):
    normalized = log.replace('\r\n', '\n')
    start = normalized.rfind(BEGIN)
    if start < 0:
        return None
    body = normalized[start + len(BEGIN):]
    stop = body.find(END)
    if stop < 0:
        return None
    # Kernel.WriteDiagnostic deliberately prefixes each bounded serial chunk
    # with a newline. Chunk boundaries may therefore add empty/space-only
    # records, but never justify discarding a nonblank FRAME or bad field.
    lines = [line for line in body[:stop].splitlines() if line.strip(' \t')]
    if not lines or not lines[0].startswith('META|'):
        raise ValueError('The native timing dump has no first META record')
    meta = fields(lines[0], 'META', META_FIELDS)
    if not 1 <= meta['Capacity'] <= 16_384:
        raise ValueError('The native timing ring capacity is out of bounds')
    if len(lines) - 1 > meta['Capacity']:
        raise ValueError('More native frames than the fixed timing ring capacity')
    frames = [fields(line, 'FRAME', FRAME_FIELDS) for line in lines[1:]]
    if len(frames) < 2:
        raise ValueError('At least two real presentations are required')
    return {'Meta': meta, 'Frames': frames}


def percentile(values, fraction):
    """Nearest-rank percentile; every measured gap, including stalls, remains."""
    ordered = sorted(values)
    return ordered[max(0, math.ceil(len(ordered) * fraction) - 1)]


def summarize_timing(dump, target_hz=240):
    if not 20 <= target_hz <= 1000:
        raise ValueError('Target refresh goal is outside 20..1000 Hz')
    meta, frames = dump['Meta'], dump['Frames']
    frequency = meta['Frequency']
    if frequency <= 0 or meta['EndTicks'] <= meta['StartTicks']:
        raise ValueError('Native monotonic timing frequency/window is invalid')
    if meta['FramesAfter'] - meta['FramesBefore'] != len(frames) or meta['FramesPresentedDelta'] != len(frames) or meta['CapturedFrames'] != len(frames) or meta['DroppedFrames'] != 0 or meta['Overflow'] != 0:
        raise ValueError('Native presentation ring is incomplete; no frame may be silently omitted')
    if meta['InvalidInputReports'] or meta['InputsBefore'] != meta['ObservedBefore'] or meta['InputsAfter'] != meta['ObservedAfter']:
        raise ValueError('Native input counters or reports are inconsistent')
    if not 0 <= meta['ConsumedBefore'] <= meta['ConsumedAfter'] <= meta['ObservedAfter'] or meta['ConsumedBefore'] > meta['ObservedBefore']:
        raise ValueError('Native consumed report counters are inconsistent')
    if meta['InputsAfter'] <= meta['InputsBefore'] or meta['LostAfter'] != meta['LostBefore']:
        raise ValueError('Input reports were absent or lost during the native sample')
    consumed_delta = meta['ConsumedAfter'] - meta['ConsumedBefore']
    delivered_sequences = {frame['InputSeq'] for frame in frames if frame['InputSeq'] > meta['InputsBefore']}
    if len(delivered_sequences) > consumed_delta:
        raise ValueError('Presented distinct new input sequences exceed actual consumed device reports')
    if not meta['StartTicks'] <= meta['FirstInputTicks'] < meta['LastInputTicks'] <= meta['EndTicks']:
        raise ValueError('Native continuous-input timestamps are outside the sample')
    previous_end, previous_input_seq, previous_input_ticks, previous_dispatch_ticks = meta['StartTicks'], 0, 0, 0
    for index, frame in enumerate(frames):
        if frame['Seq'] != meta['FramesBefore'] + index + 1:
            raise ValueError('Native presentation sequences are not consecutive')
        if not previous_end <= frame['BeginTicks'] <= frame['EndTicks'] <= meta['EndTicks']:
            raise ValueError('Native presentation begin/end timestamps overlap or run backwards')
        if frame['InputSeq'] < previous_input_seq or frame['InputSeq'] > meta['InputsAfter']:
            raise ValueError('Native consumed-input sequence is invalid')
        if frame['InputSeq'] > meta['InputsBefore']:
            if not meta['FirstInputTicks'] <= frame['InputTicks'] <= frame['DispatchTicks'] <= frame['BeginTicks']:
                raise ValueError('A frame uses future or undispatched input')
            if frame['InputTicks'] < previous_input_ticks or frame['DispatchTicks'] < previous_dispatch_ticks:
                raise ValueError('Consumed input timestamps run backwards')
            previous_input_ticks, previous_dispatch_ticks = frame['InputTicks'], frame['DispatchTicks']
        previous_end, previous_input_seq = frame['EndTicks'], frame['InputSeq']
    gaps = [(last['EndTicks'] - first['EndTicks']) / frequency for first, last in zip(frames, frames[1:])]
    if not gaps or min(gaps) <= 0:
        raise ValueError('Presentation-completion timestamps must strictly advance')
    active = [frame for frame in frames if meta['FirstInputTicks'] <= frame['EndTicks'] <= meta['LastInputTicks']]
    if len(active) < 2:
        raise ValueError('No complete native active-motion presentation interval')
    active_gaps = [(last['EndTicks'] - first['EndTicks']) / frequency for first, last in zip(active, active[1:])]
    boundary_gaps = [(active[0]['EndTicks'] - meta['FirstInputTicks']) / frequency,
                     (meta['LastInputTicks'] - active[-1]['EndTicks']) / frequency]
    fresh = [frame for frame in frames if frame['InputSeq'] > meta['InputsBefore']]
    latencies = [(frame['EndTicks'] - frame['InputTicks']) / frequency for frame in fresh]
    distinct_positions = len({(frame['X'], frame['Y']) for frame in fresh})
    window_seconds = (meta['EndTicks'] - meta['StartTicks']) / frequency
    active_seconds = (meta['LastInputTicks'] - meta['FirstInputTicks']) / frequency
    input_reports = meta['InputsAfter'] - meta['InputsBefore']
    active_hz = len(active) / active_seconds
    period = 1 / target_hz
    input_rate = (input_reports - 1) / active_seconds
    # A completed framebuffer copy can repaint animation or the same cursor.
    # Only a newer consumed device report AND a different presented position
    # qualify as a pointer update. Fix the earliest actual presentation carrying
    # a new consumed report as the baseline, never the probe's zero-initialized
    # position. The baseline is not itself a movement or an optimizable start.
    baseline = next((frame for frame in active if frame['InputSeq'] > meta['InputsBefore']), None)
    pointer_frames = []
    if baseline:
        previous_position, previous_sequence = (baseline['X'], baseline['Y']), baseline['InputSeq']
        for frame in active:
            if frame['Seq'] <= baseline['Seq']:
                continue
            position = (frame['X'], frame['Y'])
            if position != previous_position and frame['InputSeq'] > previous_sequence:
                pointer_frames.append(frame)
            previous_position, previous_sequence = position, frame['InputSeq']
    pointer_endpoints = ([baseline] + pointer_frames) if baseline else []
    pointer_gaps = [(last['EndTicks'] - first['EndTicks']) / frequency for first, last in zip(pointer_endpoints, pointer_endpoints[1:])]
    pointer_seconds = ((pointer_frames[-1]['EndTicks'] - baseline['EndTicks']) / frequency
                       if baseline and pointer_frames else 0)
    first_response = (baseline['EndTicks'] - meta['FirstInputTicks']) / frequency if baseline else None
    tail_gap = ((meta['LastInputTicks'] - (pointer_frames[-1] if pointer_frames else baseline)['EndTicks']) / frequency
                if baseline else None)
    # First-response latency is separate from the first actual movement period.
    # Neither it nor a tail stall may disappear by shortening the steady window.
    pointer_boundaries = [first_response, tail_gap] if baseline else [active_seconds]
    pointer_hz = len(pointer_frames) / pointer_seconds if pointer_seconds else 0
    pointer_sufficient = baseline is not None and len(pointer_frames) >= 2
    present_goal = distinct_positions >= 2 and input_rate >= target_hz and active_hz >= target_hz and all(gap <= period for gap in active_gaps + boundary_gaps)
    pointer_goal = pointer_sufficient and input_rate >= target_hz and pointer_hz >= target_hz and all(gap <= period for gap in pointer_gaps + pointer_boundaries)
    def distribution(samples):
        return {'MinimumMs': min(samples) * 1000, 'MedianMs': percentile(samples, .5) * 1000,
                'P95Ms': percentile(samples, .95) * 1000, 'P99Ms': percentile(samples, .99) * 1000,
                'MaximumMs': max(samples) * 1000, 'Samples': len(samples)}
    return {
        'EvidenceValid': True,
        'TargetHz': target_hz,
        'TargetPeriodMs': period * 1000,
        'SoftwareMeaning': 'Actual completed Canvas.Present operations / native monotonic time; not physical monitor scanout.',
        'WindowSeconds': window_seconds,
        'ActualPresentations': len(frames),
        'SoftwarePresentHzWholeCapture': len(frames) / window_seconds,
        'AllFrameIntervals': distribution(gaps),
        'ContinuousMotionSeconds': active_seconds,
        'PresentationsDuringContinuousMotion': len(active),
        'SoftwarePresentHzDuringContinuousMotion': active_hz,
        'ActiveFrameIntervals': distribution(active_gaps),
        'ActiveBoundaryGapsMs': [gap * 1000 for gap in boundary_gaps],
        'ActiveIntervalsOverTargetBudget': sum(gap > period for gap in active_gaps),
        'ActualInputReports': input_reports,
        'ActualConsumedReports': meta['ConsumedAfter'] - meta['ConsumedBefore'],
        'UnconsumedReportsAtStop': meta['ObservedAfter'] - meta['ConsumedAfter'],
        'NativeInputReportsPerSecond': input_rate,
        'LatestConsumedInputToPresent': distribution(latencies) if latencies else None,
        'DistinctConsumedInputSequencesPresented': len({frame['InputSeq'] for frame in fresh}),
        'DistinctActualPointerPositionsPresented': distinct_positions,
        'FreshPointerPresentationMeaning': 'An actual completed Present with a newer consumed device-report sequence and a changed position from the preceding actual Present. The earliest in-window Present with a new consumed report is a fixed baseline, not a movement; rate is actual transitions / baseline-to-last-transition duration.',
        'FreshPointerBaselineFrameSequence': baseline['Seq'] if baseline else None,
        'FreshPointerBaselineEndTicks': baseline['EndTicks'] if baseline else None,
        'FreshPointerCadenceWindowSeconds': pointer_seconds,
        'FreshPointerCadenceEvidenceSufficient': pointer_sufficient,
        'FreshPointerFirstResponseMs': first_response * 1000 if first_response is not None else None,
        'FreshPointerTailGapMs': tail_gap * 1000 if tail_gap is not None else None,
        'FreshPointerPresentations': len(pointer_frames),
        'FreshPointerPresentHz': pointer_hz,
        'FreshPointerFrameIntervals': distribution(pointer_gaps) if pointer_gaps else None,
        'FreshPointerBoundaryMeaning': 'First actual input to fixed baseline Present, and last real pointer transition to last actual input; both checked independently of steady transition rate.',
        'FreshPointerBoundaryGapsMs': [gap * 1000 for gap in pointer_boundaries],
        'FreshPointerIntervalsOverTargetBudget': sum(gap > period for gap in pointer_gaps),
        'RepeatedPointerPresentationRatio': 1 - len(pointer_frames) / len(active),
        'InputPressureSufficientForTarget': input_rate >= target_hz,
        # No tolerance silently substitutes a lower rate for the user's goal.
        'SoftwarePresentCadenceGoalMet': present_goal,
        'SoftwarePointerCadenceGoalMet': pointer_goal,
        'SoftwareCadenceGoalMet': present_goal and pointer_goal,
        'NativeScanoutHzMeasured': None,
        'VSyncVerified': False,
        'PhysicalMonitorRefreshMatched': False,
        'PresentationLatencyMeaning': 'Latest actual guest report consumed by a frame to completed framebuffer copy; excludes host clock and panel scanout.',
    }


def cross_check_host_clock(summary, host_motion_seconds, *, relative_tolerance=.02, boundary_allowance_seconds=.05):
    """Reject gross guest-clock calibration errors and never boost FPS by correction.

    The bounded input queue can shift the two measured endpoints slightly. It
    cannot justify seconds of disagreement or make a short guest second count
    as a real second. Even within the explicit tolerance, the larger elapsed
    span is used to conservatively recheck the software rate and gap budget.
    """
    if not 0 <= relative_tolerance <= .05 or not 0 <= boundary_allowance_seconds <= .1:
        raise ValueError('Clock tolerance exceeds the bounded validation policy')
    native_seconds = summary.get('ContinuousMotionSeconds')
    finite = isinstance(native_seconds, (int, float)) and isinstance(host_motion_seconds, (int, float)) and \
        math.isfinite(native_seconds) and math.isfinite(host_motion_seconds) and native_seconds > 0 and host_motion_seconds > 0
    result = dict(summary)
    comparison = {'HostSpanMeaning': 'First QMP input queued to last input acknowledgement on the host monotonic clock.',
                  'NativeSpanMeaning': 'First to last newly consumed device-report timestamp in guest ticks / declared guest frequency.',
                  'RelativeTolerance': relative_tolerance, 'BoundaryAllowanceSeconds': boundary_allowance_seconds,
                  'HostMotionSeconds': host_motion_seconds if finite else None,
                  'NativeMotionSeconds': native_seconds if finite else None, 'Passed': False}
    result['ClockConsistency'] = comparison
    if not finite:
        comparison['Failure'] = 'Missing, non-finite or nonpositive cross-clock elapsed interval'
        result['EvidenceValid'] = result['SoftwareCadenceGoalMet'] = False
        return result
    allowed = boundary_allowance_seconds + host_motion_seconds * relative_tolerance
    difference = abs(native_seconds - host_motion_seconds)
    comparison.update({'NativeSecondsPerHostSecond': native_seconds / host_motion_seconds,
                       'AbsoluteDifferenceSeconds': difference, 'AllowedDifferenceSeconds': allowed,
                       'Passed': difference <= allowed})
    conservative_scale = max(1, host_motion_seconds / native_seconds)
    native_hz = summary['SoftwarePresentHzDuringContinuousMotion']
    conservative_hz = native_hz / conservative_scale
    max_gap_ms = summary['ActiveFrameIntervals']['MaximumMs'] * conservative_scale
    boundary_gaps_ms = [gap * conservative_scale for gap in summary['ActiveBoundaryGapsMs']]
    pointer_hz = summary['FreshPointerPresentHz'] / conservative_scale
    pointer_intervals = summary['FreshPointerFrameIntervals']
    pointer_max_gap_ms = pointer_intervals['MaximumMs'] * conservative_scale if pointer_intervals else None
    pointer_boundaries_ms = [gap * conservative_scale for gap in summary['FreshPointerBoundaryGapsMs']]
    result.update({'ConservativeHostAnchoredSoftwarePresentHz': conservative_hz,
                   'ConservativeHostAnchoredMaximumFrameIntervalMs': max_gap_ms,
                   'ConservativeHostAnchoredBoundaryGapsMs': boundary_gaps_ms,
                   'ConservativeHostAnchoredFreshPointerPresentHz': pointer_hz,
                   'ConservativeHostAnchoredFreshPointerMaximumIntervalMs': pointer_max_gap_ms,
                   'ConservativeHostAnchoredFreshPointerBoundaryGapsMs': pointer_boundaries_ms,
                   'ClockCorrectionMeaning': 'Use the full input-window native/host comparison to derive a conservative time scale, also applied to the fixed-baseline pointer subwindow; correction can only lower rates and increase all gaps. Raw native measurements remain unchanged.'})
    result['EvidenceValid'] = bool(summary['EvidenceValid'] and comparison['Passed'])
    result['SoftwarePresentCadenceGoalMet'] = bool(summary['SoftwarePresentCadenceGoalMet'] and result['EvidenceValid'] and
        conservative_hz >= summary['TargetHz'] and max_gap_ms <= summary['TargetPeriodMs'] and
        all(gap <= summary['TargetPeriodMs'] for gap in boundary_gaps_ms))
    result['SoftwarePointerCadenceGoalMet'] = bool(summary['SoftwarePointerCadenceGoalMet'] and result['EvidenceValid'] and
        pointer_hz >= summary['TargetHz'] and pointer_max_gap_ms is not None and pointer_max_gap_ms <= summary['TargetPeriodMs'] and
        all(gap <= summary['TargetPeriodMs'] for gap in pointer_boundaries_ms))
    result['SoftwareCadenceGoalMet'] = result['SoftwarePresentCadenceGoalMet'] and result['SoftwarePointerCadenceGoalMet']
    if not comparison['Passed']:
        comparison['Failure'] = 'Native clock span disagrees with actual host motion beyond bounded input queue/clock tolerance; no native FPS qualification is valid.'
    return result