<commit_before>import re
from socket import AF_INET, SOCK_DGRAM, socket
import threading
import time
import types

try:
    from setproctitle import setproctitle
except ImportError:
    setproctitle = None

from daemon import Daemon


__all__ = ['Server']

def _clean_key(k):
    return re.sub(
        '[^a-zA-Z_\-0-9\.]',
        '',
        k.replace('/','-').replace(' ','_')
    )

TIMER_MSG = '''stats.timers.%(key)s.lower %(min)s %(ts)s
stats.timers.%(key)s.count %(count)s %(ts)s
stats.timers.%(key)s.mean %(mean)s %(ts)s
stats.timers.%(key)s.upper %(max)s %(ts)s
stats.timers.%(key)s.upper_%(pct_threshold)s %(max_threshold)s %(ts)s
'''

class Server(object):
    
    def __init__(self, pct_threshold=90, debug=False, graphite_host='localhost', graphite_port=2003):
        self.buf = 1024
        self.flush_interval = 10000
        self.pct_threshold = pct_threshold
        self.graphite_host = graphite_host
        self.graphite_port = graphite_port
        self.debug = debug

        self.counters = {}
        self.timers = {}
        self.flusher = 0

    def process(self, data):
        key, val = data.split(':')
        key = _clean_key(key)

        sample_rate = 1;
        fields = val.split('|')

        if (fields[1] == 'ms'):
            if key not in self.timers:
                self.timers[key] = []
            self.timers[key].append(float(fields[0] or 0))
        else:
            if len(fields) == 3:
                sample_rate = float(re.match('^@([\d\.]+)', fields[2]).groups()[0])
            if key not in self.counters:
                self.counters[key] = 0;
            self.counters[key] += float(fields[0] or 1) * (1 / sample_rate)

    def flush(self):
        ts = int(time.time())
        stats = 0
        stat_string = ''
        for k, v in self.counters.items():
            v = float(v) / (self.flush_interval / 1000)
            msg = 'stats.%s %s %s\n' % (k, v, ts)
            stat_string += msg

            self.counters[k] = 0
            stats += 1

        for k, v in self.timers.items():
            if len(v) > 0:
                v.sort()
                count = len(v)
                min = v[0]
                max = v[-1]

                mean = min
                max_threshold = max

                if count > 1:
                    thresh_index = int((self.pct_threshold / 100.0) * count)
                    max_threshold = v[thresh_index - 1]
                    total = sum(v[:thresh_index-1])
                    mean = total / thresh_index

                self.timers[k] = []

                stat_string += TIMER_MSG % {
                    'key':k,
                    'mean':mean,
                    'max': max,
                    'min': min,
                    'count': count,
                    'max_threshold': max_threshold,
                    'pct_threshold': self.pct_threshold,
                    'ts': ts,
                }
                stats += 1

        stat_string += 'statsd.numStats %s %d' % (stats, ts)

        graphite = socket()
        graphite.connect((self.graphite_host, self.graphite_port))
        graphite.sendall(stat_string)
        graphite.close()
        self._set_timer()

        if self.debug:
            print stat_string


    def _set_timer(self):
        self._timer = threading.Timer(self.flush_interval/1000, self.flush)
        self._timer.start()

    def serve(self, hostname='', port=8125, graphite_host='localhost', graphite_port=2003):
        assert type(port) is types.IntType, 'port is not an integer: %s' % (port)<commit_msg>Closes #20 where the server did not send a newline with the numStats string.<commit_after>import re
from socket import AF_INET, SOCK_DGRAM, socket
import threading
import time
import types

try:
    from setproctitle import setproctitle
except ImportError:
    setproctitle = None

from daemon import Daemon


__all__ = ['Server']

def _clean_key(k):
    return re.sub(
        '[^a-zA-Z_\-0-9\.]',
        '',
        k.replace('/','-').replace(' ','_')
    )

TIMER_MSG = '''stats.timers.%(key)s.lower %(min)s %(ts)s
stats.timers.%(key)s.count %(count)s %(ts)s
stats.timers.%(key)s.mean %(mean)s %(ts)s
stats.timers.%(key)s.upper %(max)s %(ts)s
stats.timers.%(key)s.upper_%(pct_threshold)s %(max_threshold)s %(ts)s
'''

class Server(object):
    
    def __init__(self, pct_threshold=90, debug=False, graphite_host='localhost', graphite_port=2003):
        self.buf = 1024
        self.flush_interval = 10000
        self.pct_threshold = pct_threshold
        self.graphite_host = graphite_host
        self.graphite_port = graphite_port
        self.debug = debug

        self.counters = {}
        self.timers = {}
        self.flusher = 0

    def process(self, data):
        key, val = data.split(':')
        key = _clean_key(key)

        sample_rate = 1;
        fields = val.split('|')

        if (fields[1] == 'ms'):
            if key not in self.timers:
                self.timers[key] = []
            self.timers[key].append(float(fields[0] or 0))
        else:
            if len(fields) == 3:
                sample_rate = float(re.match('^@([\d\.]+)', fields[2]).groups()[0])
            if key not in self.counters:
                self.counters[key] = 0;
            self.counters[key] += float(fields[0] or 1) * (1 / sample_rate)

    def flush(self):
        ts = int(time.time())
        stats = 0
        stat_string = ''
        for k, v in self.counters.items():
            v = float(v) / (self.flush_interval / 1000)
            msg = 'stats.%s %s %s\n' % (k, v, ts)
            stat_string += msg

            self.counters[k] = 0
            stats += 1

        for k, v in self.timers.items():
            if len(v) > 0:
                v.sort()
                count = len(v)
                min = v[0]
                max = v[-1]

                mean = min
                max_threshold = max

                if count > 1:
                    thresh_index = int((self.pct_threshold / 100.0) * count)
                    max_threshold = v[thresh_index - 1]
                    total = sum(v[:thresh_index-1])
                    mean = total / thresh_index

                self.timers[k] = []

                stat_string += TIMER_MSG % {
                    'key':k,
                    'mean':mean,
                    'max': max,
                    'min': min,
                    'count': count,
                    'max_threshold': max_threshold,
                    'pct_threshold': self.pct_threshold,
                    'ts': ts,
                }
                stats += 1

        stat_string += "statsd.numStats %s %d\n" % (stats, ts)

        graphite = socket()
        graphite.connect((self.graphite_host, self.graphite_port))
        graphite.sendall(stat_string)
        graphite.close()
        self._set_timer()

        if self.debug:
            print stat_string


    def _set_timer(self):
        self._timer = threading.Timer(self.flush_interval/1000, self.flush)
        self._timer.start()

    def serve(self, hostname='', port=8125, graphite_host='localhost', graphite_port=2003):
        assert type(port) is types.IntType, 'port is not an integer: %s' % (port)
{'fraction_non_alphanumeric': 0.1342869045965838, 'fraction_numerical': 0.016664352173309263, 'mean_word_length': 1.6813104988830976, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '5151883', 'n_tokens_mistral': 2455, 'n_tokens_neox': 2254, 'n_words': 494}