Scripts/stdinreader

• Download example

…

stdinreader.py

#!/usr/bin/env python3

from Foundation import (
    NSObject,
    NSFileHandle,
    NSNotificationCenter,
    NSFileHandleReadCompletionNotification,
    NSFileHandleNotificationDataItem,
    NSFileHandleError,
)
from PyObjCTools import AppHelper


class FileObserver(NSObject):
    def initWithFileDescriptor_readCallback_errorCallback_(
        self, fileDescriptor, readCallback, errorCallback
    ):
        self = self.init()
        self.readCallback = readCallback
        self.errorCallback = errorCallback
        self.fileHandle = NSFileHandle.alloc().initWithFileDescriptor_(fileDescriptor)
        self.nc = NSNotificationCenter.defaultCenter()
        self.nc.addObserver_selector_name_object_(
            self,
            "fileHandleReadCompleted:",
            NSFileHandleReadCompletionNotification,
            self.fileHandle,
        )
        self.fileHandle.readInBackgroundAndNotify()
        return self

    def fileHandleReadCompleted_(self, aNotification):
        ui = aNotification.userInfo()
        newData = ui.objectForKey_(NSFileHandleNotificationDataItem)
        if newData is None:
            if self.errorCallback is not None:
                self.errorCallback(self, ui.objectForKey_(NSFileHandleError))
            self.close()
        else:
            self.fileHandle.readInBackgroundAndNotify()
            if self.readCallback is not None:
                self.readCallback(self, str(newData))

    def close(self):
        self.nc.removeObserver_(self)
        if self.fileHandle is not None:
            self.fileHandle.closeFile()
            self.fileHandle = None
        # break cycles in case these functions are closed over
        # an instance of us
        self.readCallback = None
        self.errorCallback = None

    def __del__(self):
        # Without this, if a notification fires after we are GC'ed
        # then the app will crash because NSNotificationCenter
        # doesn't retain observers.  In this example, it doesn't
        # matter, but it's worth pointing out.
        self.close()


def prompt():
    sys.stdout.write("write something: ")
    sys.stdout.flush()


def gotLine(observer, line):
    if line:
        print("you wrote:", line.rstrip())
        prompt()
    else:
        print("")
        AppHelper.stopEventLoop()


def gotError(observer, err):
    print("error:", err)
    AppHelper.stopEventLoop()


if __name__ == "__main__":
    import sys

    observer = FileObserver.alloc().initWithFileDescriptor_readCallback_errorCallback_(
        sys.stdin.fileno(), gotLine, gotError
    )
    prompt()
    AppHelper.runConsoleEventLoop(installInterrupt=True)