<commit_before># Copyright (c) <NAME>.
# See LICENSE for details.

'''
Usage: peaky.py [options]
Options:
  -f, --fast     Act quickly
      --version  Display Twisted version and exit.
      --help     Display this help and exit.
'''

from sys import argv
from os import environ

from twisted.python.usage import Options
from twisted.python.usage import UsageError
from twisted.internet.protocol import ProcessProtocol
from twisted.internet import reactor

class Options(Options):
    optFlags = [
        ["fast", "f", "Act quickly"]]

opts = Options()
try:
    opts.parseOptions(argv[1:])
except UsageError as ue:
    print '%s: %s' % (argv[0], ue)
    print '%s: Try --help for usage details.' % (argv[0])
    exit(1)

if opts['fast']:
    print "fast",

class Top(ProcessProtocol):
    def __init__(self):
        self.data = ""

    def connectionMade(self):
#         self.transport.write(top)
        self.transport.closeStdin() # tell them we're done

    def outReceived(self, data):
        print "outReceived! with %d bytes!" % len(data)
        self.data = self.data + data

    def errReceived(self, data):
        print "errReceived! with %d bytes!" % len(data)
        print data

    def inConnectionLost(self):
        print "inConnectionLost! stdin is closed! (we probably did it)"

    def outConnectionLost(self):
        print "outConnectionLost! The child closed their stdout!"
        print self.data

    def errConnectionLost(self):
        print "errConnectionLost! The child closed their stderr."

    def processExited(self, reason):
        print "processExited, status %d" % (reason.value.exitCode,)

    def processEnded(self, reason):
        print "processEnded, status %d" % (reason.value.exitCode,)
        print "quitting"
        reactor.stop()

reactor.spawnProcess(
    processProtocol=Top(),
    executable="top",
    args=["top"],
    env={'TERM': environ['TERM']},
    usePTY=1)
reactor.run()<commit_msg>Parse ansi output from top.<commit_after># Copyright (c) <NAME>.
# See LICENSE for details.

'''
Usage: peaky.py [options]
Options:
  -l, --lines=   Max number of lines output from top. [default: 1024]
      --version  Display Twisted version and exit.
      --help     Display this help and exit.
'''

from sys import argv
from os import environ

from twisted.internet import reactor
from twisted.internet.protocol import ProcessProtocol
from twisted.python.usage import Options
from twisted.python.usage import UsageError
from twisted.conch.ui.ansi import AnsiParser
from twisted.conch.ui.ansi import ColorText

class Options(Options):
#     optFlags = [
#         ["fast", "f", "Act quickly"]]

    optParameters = [
        ['lines', 'l', 1024, 'Max number of lines output from top.', int]]

opts = Options()
try:
    opts.parseOptions(argv[1:])
except UsageError as ue:
    print '%s: %s' % (argv[0], ue)
    print '%s: Try --help for usage details.' % (argv[0])
    exit(1)

class PeakyAnsiParser(AnsiParser, object):
    def __init__(
            self,
            defaultFG=ColorText.WHITE,
            defaultBG=ColorText.BLACK):
        super(PeakyAnsiParser, self).__init__(
            defaultFG,
            defaultBG)
        self.color_text_list = []

    def writeString(self, color_text):
        text = color_text.text = color_text.text.replace('\r\n', '')
        if not text or text == 'B':
            return

        self.color_text_list.append(color_text)

class Top(ProcessProtocol):
    def __init__(self):
        self.data = ''

    def connectionMade(self):
        pass

    def outReceived(self, data):
        if 'top -' in data:
            def color_text_list():
                pap = PeakyAnsiParser()
                pap.parseString(self.data)
                return pap.color_text_list

            self.data = data
        else:
            self.data += data

    def errReceived(self, data):
        print "errReceived! with %d bytes!" % len(data)
        print data

    def inConnectionLost(self):
        print "inConnectionLost! stdin is closed! (we probably did it)"

    def outConnectionLost(self):
        print "outConnectionLost! The child closed their stdout!"

    def errConnectionLost(self):
        print "errConnectionLost! The child closed their stderr."

    def processExited(self, reason):
        print "processExited, status %d" % (reason.value.exitCode,)

    def processEnded(self, reason):
        print "processEnded, status %d" % (reason.value.exitCode,)
        print "quitting"
        reactor.stop()

def env_term():
    try:
        return environ['TERM']
    except KeyError:
        print 'TERM environment variable not set.'
        return None
        

reactor.spawnProcess(
    processProtocol=Top(),
    executable='top',
    args=['top'],
    env={
        'TERM': environ['TERM'],
        'COLUMNS': '512',
        'LINES': str(opts['lines']),
    },
    usePTY=1)

reactor.run()
{'fraction_non_alphanumeric': 0.1180668841761827, 'fraction_numerical': 0.004282218597063622, 'mean_word_length': 2.6659192825112106, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4459193', 'n_tokens_mistral': 1511, 'n_tokens_neox': 1459, 'n_words': 431}