<commit_before>import collections
import ipc
import numpy
from numpy import abs
import h5py
import time, datetime
from datetime import datetime as DT
import pytz

class Stack:
    def __init__(self,name="stack",maxLen=100,reducePeriod=None,outPeriod=None):
        self._maxLen = maxLen
        self._outPeriod = outPeriod
        self._reducePeriod = reducePeriod
        self._name = name
        self.clear()
        self._outputs = ["std","mean","sum","median","min","max"]
        
    def clear(self):
        self._buffer = None
        self._currentIndex = 0
        n = ipc.mpi.nr_workers()
        if n > 1 and self._outPeriod is not None:
            self._outIndex = float(ipc.mpi.rank) / float(n-1) * (self._outPeriod-1)
        else:
            self._outIndex = 0
        if n > 1 and self._reducePeriod is not None:
            self._reduceIndex = float(ipc.mpi.rank) / float(n-1) * (self._reducePeriod-1)
        else:
            self._reduceIndex = 0
        self.last_std    = None
        self.last_mean   = None
        self.last_sum    = None
        self.last_median = None
        self.last_min    = None
        self.last_max    = None
        self._reduced     = False

    def filled(self):
        return self._currentIndex >= self._maxLen
    
    def add(self,data):
        if self._buffer is None:
            s = tuple([self._maxLen] + list(data.shape))
            self._buffer = numpy.zeros(shape=s, dtype=data.dtype)
        self._buffer[self._currentIndex % self._maxLen,:] = data[:]
        self._currentIndex += 1
        
    def _getData(self):
        if self.filled():
            return self._buffer
        else:
            return self._buffer[:self._currentIndex]
        
    def std(self):
        self.last_std = self._getData().std(axis=0)
        return self.last_std
    
    def mean(self):
        self.last_mean = self._getData().mean(axis=0)
        return self.last_mean
    
    def sum(self):
        self.last_sum = self._getData().sum(axis=0)
        return self.last_sum
    
    def median(self):
        self.last_median = numpy.median(self._getData(),axis=0)
        return self.last_median
    
    def min(self):
        self.last_min = self._getData().min(axis=0)
        return self.last_min

    def max(self):
        self.last_max = self._getData().max(axis=0)
        return self.last_max

    def reduce(self):
        if self._reducePeriod is not None:
            if (self._currentIndex % self._reducePeriod) != self._outIndex:
                return
        if not self.filled():
            return
        for o in self._outputs:
            exec "self.%s()" % o       
        self._reduced = True

    def write(self,evt, directory=".", png=False, verbose=True):
        # Postpone writing?
        if self._outPeriod is not None:
            if (self._currentIndex % self._outPeriod) != self._outIndex:
                return
        if not self._reduced:
            if verbose:
                print "Postponing writing stack to file because stack is not reduced yet. Fill status %i/%i." % (self._currentIndex+1, self._maxLength)
            return
        # Timestamp for filename
        try:
            dt64 = evt["eventID"]["Timestamp"].datetime64
        except:
	    dt64 = numpy.datetime64(datetime.datetime.utcnow())
        # Convert to US/Pacific local time
        t_loc = datetime.datetime.now()
        t_utc = datetime.datetime.utcnow()
        delta = t_utc-t_loc
        dt64_loc = dt64 + delta.seconds * 1000000
        try:
            fid = evt["eventID"]["Timestamp"].fiducials
        except:
            fid = 0
        fn = "%s/%s-%s-fid%s-rk%i.h5" % (directory,self._name, str(dt64_loc)[:-5], fid, ipc.mpi.rank)
        # Write to H5<commit_msg>Fixed bug with background stack output with mpi.<commit_after>import collections
import ipc
import numpy
from numpy import abs
import h5py
import time, datetime
from datetime import datetime as DT
import pytz

class Stack:
    def __init__(self,name="stack",maxLen=100,reducePeriod=None,outPeriod=None):
        self._maxLen = maxLen
        self._outPeriod = outPeriod
        self._reducePeriod = reducePeriod
        self._name = name
        self.clear()
        self._outputs = ["std","mean","sum","median","min","max"]
        
    def clear(self):
        self._buffer = None
        self._currentIndex = 0
        n = ipc.mpi.nr_workers()
        if n > 1 and self._outPeriod is not None:
            self._outIndex = int(float(ipc.mpi.rank) / float(n-1) * (self._outPeriod-1))
        else:
            self._outIndex = 0
        if n > 1 and self._reducePeriod is not None:
            self._reduceIndex = int(float(ipc.mpi.rank) / float(n-1) * (self._reducePeriod-1))
        else:
            self._reduceIndex = 0
        self.last_std    = None
        self.last_mean   = None
        self.last_sum    = None
        self.last_median = None
        self.last_min    = None
        self.last_max    = None
        self._reduced     = False

    def filled(self):
        return self._currentIndex >= self._maxLen
    
    def add(self,data):
        if self._buffer is None:
            s = tuple([self._maxLen] + list(data.shape))
            self._buffer = numpy.zeros(shape=s, dtype=data.dtype)
        self._buffer[self._currentIndex % self._maxLen,:] = data[:]
        self._currentIndex += 1
        
    def _getData(self):
        if self.filled():
            return self._buffer
        else:
            return self._buffer[:self._currentIndex]
        
    def std(self):
        self.last_std = self._getData().std(axis=0)
        return self.last_std
    
    def mean(self):
        self.last_mean = self._getData().mean(axis=0)
        return self.last_mean
    
    def sum(self):
        self.last_sum = self._getData().sum(axis=0)
        return self.last_sum
    
    def median(self):
        self.last_median = numpy.median(self._getData(),axis=0)
        return self.last_median
    
    def min(self):
        self.last_min = self._getData().min(axis=0)
        return self.last_min

    def max(self):
        self.last_max = self._getData().max(axis=0)
        return self.last_max

    def reduce(self):
        if self._reducePeriod is not None:
            if (self._currentIndex % self._reducePeriod) != self._outIndex:
                return
        if not self.filled():
            return
        for o in self._outputs:
            exec "self.%s()" % o       
        self._reduced = True

    def write(self,evt, directory=".", png=False, verbose=True):
        # Postpone writing?
        if self._outPeriod is not None:
            if (self._currentIndex % self._outPeriod) != self._outIndex:
                return
        if not self._reduced:
            if verbose:
                print "Postponing writing stack to file because stack is not reduced yet. Fill status %i/%i." % (self._currentIndex+1, self._maxLen)
            return
        # Timestamp for filename
        try:
            dt64 = evt["eventID"]["Timestamp"].datetime64
        except:
	    dt64 = numpy.datetime64(datetime.datetime.utcnow())
        # Convert to US/Pacific local time
        t_loc = datetime.datetime.now()
        t_utc = datetime.datetime.utcnow()
        delta = t_utc-t_loc
        dt64_loc = dt64 + delta.seconds * 1000000
        try:
            fid = evt["eventID"]["Timestamp"].fiducials
        except:
            fid = 0
        fn = "%s/%s-%s-fid%s-rk%i.h5" % (directory,self._name, str(dt64_loc)[:-5], fid, ipc.mpi.rank)
        # Write to H5
{'fraction_non_alphanumeric': 0.10977224576271187, 'fraction_numerical': 0.012182203389830509, 'mean_word_length': 1.9332038834951457, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '315646', 'n_tokens_mistral': 2329, 'n_tokens_neox': 2254, 'n_words': 528}