t_progress.py 文件源码

def __t_pty_tracker(self, trackerclass, **kwargs):
                def __drain(masterf):
                        while True:
                                chunksz = 1024
                                termdata = masterf.read(chunksz)
                                if len(termdata) < chunksz:
                                        # assume we hit EOF
                                        break

                #
                # - Allocate a pty
                # - Create a thread to drain off the master side; without
                #   this, the slave side will block when trying to write.
                # - Connect the prog tracker to the slave side
                # - Set it running
                #
                (master, slave) = pty.openpty()
                slavef = os.fdopen(slave, "w")
                masterf = os.fdopen(master, "rb")

                t = threading.Thread(target=__drain, args=(masterf,))
                t.start()

                p = trackerclass(output_file=slavef, **kwargs)
                progress.test_progress_tracker(p, gofast=True)
                slavef.close()

                t.join()
                masterf.close()