MacOS Accessibility API Notifications

accessibility_notifications.swift

//
//  MacOS Accessibility Notifications
//  Notice: this version doesn't take into account new apps after this code is executed
//  Created by Phi Hiep Nguyen and AI on 2024-10-08.
//

import Cocoa
import Foundation

// AppObserver: Monitors and reports application activation events on macOS
class AppObserver {
    // Singleton instance for global access
    static let shared = AppObserver()
    
    // Dictionary to store observers for each application, keyed by process ID
    var observers: [pid_t: AXObserver] = [:]
    // Dictionary to track the last activation time for each app, used for debouncing
    var lastActivationTime: [String: Date] = [:]
    // Time interval to prevent duplicate notifications (500 milliseconds)
    let debounceInterval: TimeInterval = 0.5
    
    // Private initializer to enforce singleton pattern
    private init() {}
    
    // Starts the observation process for all running applications
    func start() {
        // Check for accessibility permissions
        guard self.checkAccessibilityPermissions() else {
            print("Accessibility permissions are required. Please grant them in System Preferences > Security & Privacy > Privacy > Accessibility.")
            print("After granting permissions, please restart the application.")
            exit(1)
        }
        
        let workspace = NSWorkspace.shared
        let runningApps = workspace.runningApplications
        
        // Set up observers for each running application
        for app in runningApps {
            addObserver(for: app)
        }
        
        // Start the run loop to process events
        RunLoop.current.run()
    }
    
    // Adds an accessibility observer for a given application
    func addObserver(for app: NSRunningApplication) {
        let pid = app.processIdentifier
        
        var observer: AXObserver?
        // Callback function triggered when an application is activated
        let callback: AXObserverCallback = { (observer, element, notification, refcon) in
            guard let appRefcon = refcon else { return }
            let appPid = Unmanaged<NSNumber>.fromOpaque(appRefcon).takeUnretainedValue().int32Value
            if let activatedApp = NSRunningApplication(processIdentifier: appPid) {
                let appName = activatedApp.localizedName ?? activatedApp.bundleIdentifier ?? "Unknown"
                // Use the shared instance to debounce notifications
                AppObserver.shared.handleNotification(appName: appName, notification: notification, element: element)
            }
        }
        
        // Create the accessibility observer
        let createError = AXObserverCreate(pid, callback, &observer)
        guard createError == .success, let observer = observer else {
            print("Failed to create observer for \(app.localizedName ?? "Unknown")")
            return
        }
        
        // Create an accessibility element for the application
        let axApp = AXUIElementCreateApplication(pid)
        // Store the process ID as the refcon (reference constant) for later use
        let appPidRef = Unmanaged.passUnretained(NSNumber(value: pid)).toOpaque()
        
        // Add notification for application activation
        var addError = AXObserverAddNotification(observer, axApp, kAXApplicationActivatedNotification as CFString, appPidRef)
        if addError != .success {
            print("Failed to add activation notification for \(app.localizedName ?? "Unknown")")
        }
        
        // Add notification for frontmost window changed
        addError = AXObserverAddNotification(observer, axApp, kAXFocusedWindowChangedNotification as CFString, appPidRef)
        if addError != .success {
            print("Failed to add window focus notification for \(app.localizedName ?? "Unknown")")
        }
        
        // Add the observer to the current run loop
        CFRunLoopAddSource(CFRunLoopGetCurrent(), AXObserverGetRunLoopSource(observer), .defaultMode)
        
        // Store the observer in the dictionary
        observers[pid] = observer
    }
    
    // Debounces notifications to prevent duplicate reports for rapid app switches
    func handleNotification(appName: String, notification: CFString, element: AXUIElement) {
        let currentTime = Date()
        if let lastTime = lastActivationTime[appName],
           currentTime.timeIntervalSince(lastTime) < debounceInterval {
            return // Ignore this notification as it's too soon after the last one
        }
        
        // Update the last activation time for this app
        lastActivationTime[appName] = currentTime
        
        if notification as String == kAXApplicationActivatedNotification as String {
            print("Notification: Application Activated for app: \(appName)")
        } else if notification as String == kAXFocusedWindowChangedNotification as String {
            var windowTitle: CFTypeRef?
            AXUIElementCopyAttributeValue(element, kAXTitleAttribute as CFString, &windowTitle)
            if let title = windowTitle as? String {
                print("Notification: Frontmost Window Changed for app: \(appName)")
                print("Current Window Title: \(title)")
            }
        }
    }
    
    // Clean up observers when the AppObserver is deallocated
    deinit {
        for (_, observer) in observers {
            CFRunLoopRemoveSource(CFRunLoopGetCurrent(), AXObserverGetRunLoopSource(observer), .defaultMode)
        }
    }
    
    // Check for accessibility permissions
    func checkAccessibilityPermissions() -> Bool {
        let checkOptPrompt = kAXTrustedCheckOptionPrompt.takeUnretainedValue() as NSString
        let options = [checkOptPrompt: true]
        return AXIsProcessTrustedWithOptions(options as CFDictionary)
    }
}

// Create a shared instance of AppObserver and start monitoring
let appObserver = AppObserver.shared
appObserver.start()