nusimulator2.py

# This will respond to keyboard input after either a 12ms delay (capital A) or
# no delay (lowercase a).
# For some reason this causes up to 50ms of difference in Terminal.app.
# Press Ctrl-C to stop.

import os, time, re, tty, termios

try:
    old_settings = termios.tcgetattr(0)
    tty.setraw(0)
    while True:
        read_data = os.read(0, 1024)
        if b'\x03' in read_data:
            # Ctrl-C
            break
        should_sleep = b'A' in read_data

        pre_sleep = time.time()
        if b'A' in read_data:
            sleep_time = 0.012
            time.sleep(sleep_time)
        else:
            sleep_time = 0.0
        post_sleep = time.time()
        actual_sleep_time = post_sleep - pre_sleep
        deviation = actual_sleep_time - sleep_time
        os.write(1, f'got {read_data} should_sleep={should_sleep} [deviation={deviation*1000:.2}ms]\r\n'.encode('ascii'))
finally:
    termios.tcsetattr(0, termios.TCSADRAIN, old_settings)

terminal-latency-test.swift

// Silly tool to measure 'end-to-end' terminal latency on macOS.
// The measured time includes the shell's latency and the terminal itself, but
// not keyboard or display lag.
//
// To use:
// - You need to be running Sonoma and Xcode 15 (or later).
// - Open two terminal windows.
// - In one terminal window, run `python3 nusimulator2.py`.
// - In the other:
//       xcrun -sdk macosx14.0 swiftc terminal-latency-test.swift
//       ./terminal-latency-test
// - A window picker should appear; pick the other window.
// - This program will create a screen capture session for that window, and
//   also start recording keystrokes.  (This data is not saved anywhere.)
// - Switch to the other window and type the letter 'a' repeatedly.  This
//   program will compare the time it received the keystroke to the time a new
//   frame is recorded.
//   - Note that `nusimulator2.py` will behave differently depending on whether
//     it's a capital or lowercase 'a'.
// - Ctrl-C when done.
//
// For each letter 'a', you should see something like:
//   got frame after 20.80ms skipped 0
// If 'skipped' is not 0, the measurement may be bogus.
//
// My results are somewhat noisy but not too bad.
//
// Originally based on ScreenCaptureKit-Recording-example from
// https://github.com/nonstrict-hq/ScreenCaptureKit-Recording-example
// by Nonstrict B.V., Mathijs Kadijk, and Tom Lokhorst, released under BSD license.

import AVFoundation
import CoreGraphics
import ScreenCaptureKit

let alsoPrintFPS = false

// Permission checks
guard AXIsProcessTrusted() else {
    fatalError("No accessibility permission (grant in System Settings)")
}
guard CGPreflightScreenCaptureAccess() else {
    fatalError("No screen capture permission (grant in System Settings)")
}

actor Comparer {
    var keyTime: Date? = nil
    var skippedFrames: Int = 0

    var lastSecondDate: Date? = nil
    var framesInLastSecond: Int = 0

    func noteKey(at date: Date) {
        keyTime = date
    }
    func noteFrame(at date: Date) {
        if let keyTime {
            let diff = keyTime.distance(to: date)
            print(String(format: "got frame after %.2fms skipped %d", diff * 1000, skippedFrames))
            self.keyTime = nil
            skippedFrames = 0
        } else {
            skippedFrames += 1
        }

        if alsoPrintFPS {
            framesInLastSecond += 1
            if lastSecondDate == nil || lastSecondDate!.distance(to: date) >= 1 {
                print("\(framesInLastSecond) fps")
                lastSecondDate = date
                framesInLastSecond = 0
            }
        }
    }
}
let comparer = Comparer()

_ = NSEvent.addGlobalMonitorForEvents(matching: [.keyDown]) { (event) in
    //print("got event \(event.keyCode)")
    if event.keyCode == 0 { // keycode 0 is the letter 'a'
        Task { await comparer.noteKey(at: Date()) }
    }
}!

private class StreamHandler: NSObject, SCStreamOutput, SCContentSharingPickerObserver {
    var stream: SCStream?
    func start() {
        let picker = SCContentSharingPicker.shared
        picker.add(self)
        picker.isActive = true
        picker.present(using: .window)
        print("presented")
    }

    func contentSharingPicker(_ picker: SCContentSharingPicker, didCancelFor stream: SCStream?) {
        fatalError("picker did cancel")
    }

    func contentSharingPickerStartDidFailWithError(_ error: Error) {
        fatalError("picker failed with error: \(error)")
    }

    func contentSharingPicker(_ picker: SCContentSharingPicker, didUpdateWith filter: SCContentFilter, for _stream: SCStream?) {
        let configuration = SCStreamConfiguration()
        configuration.queueDepth = 6
        configuration.capturesAudio = false
        configuration.colorSpaceName = CGColorSpace.sRGB
        configuration.colorMatrix = CGDisplayStream.yCbCrMatrix_ITU_R_709_2

        // Create SCStream and add local StreamOutput object to receive samples
        let stream = SCStream(filter: filter, configuration: configuration, delegate: nil)
        try! stream.addStreamOutput(self, type: .screen, sampleHandlerQueue: DispatchQueue(label: "stream queue"))
        Task {
            try await stream.startCapture()
            print("started capture")
        }
        self.stream = stream
    }

    func stream(_ stream: SCStream, didOutputSampleBuffer sampleBuffer: CMSampleBuffer, of type: SCStreamOutputType) {
        if type != .screen {
            return
        }

        // Return early if the sample buffer is invalid
        guard sampleBuffer.isValid else {
            print("got bad frame with invalid sample buffer")
            return
        }

        // Retrieve the array of metadata attachments from the sample buffer
        guard let attachmentsArray = CMSampleBufferGetSampleAttachmentsArray(sampleBuffer, createIfNecessary: false) as? [[SCStreamFrameInfo: Any]],
              let attachments = attachmentsArray.first
        else {
            print("got bad frame with no attachments in sample buffer")
            return
        }

        // Validate the status of the frame. If it isn't `.complete`, return
        guard let statusRawValue = attachments[SCStreamFrameInfo.status] as? Int,
              let status = SCFrameStatus(rawValue: statusRawValue) else {
            print("got bad frame with no status; attachments: \(attachments)")
            return
        }

        switch status {
            case .complete:
                Task { await comparer.noteFrame(at: Date()) }
                //print("got frame with timestamp \(sampleBuffer.presentationTimeStamp)")
            case .idle:
                break // ignore
            default:
                print("got bad frame with status = \(status)")
        }
    }
}

StreamHandler().start()

NSApplication.shared.run() // must do this for keylogging to work