<commit_before>from sortedcontainers import SortedDict

class Topics:
    def __init__(self):
        self.topics = dict()
        self.transfers = dict()

    def process(self,topic, payload, options=None):
        if not topic in self.topics:
            self.topics[topic] = dict()
            self.topics[topic]['raw'] = [] # For raw data
            self.topics[topic]['xy'] = SortedDict() # For indexed data
            if options:
                self.topics[topic]['type'] = 'indexed'
            else:
                self.topics[topic]['type'] = 'linear'

        t = self.topics[topic]
        t['raw'].append(payload)

        # If topic is of type indexed data, also store in ordered dict
        if options:
            t['xy'][options['index']] = payload

        # If there is an active transfer, transfer received data to the queue
        if topic in self.transfers:
            # For indexed data, provide the index for x and payload for y
            if t['type'] == 'indexed' and options is not None:
                self.transfers[topic]['queue'].put([options['index'], payload])
            # For linear data, provide sample id for x and payload for y
            elif t['type'] == 'linear':
                self.transfers[topic]['queue'].put([self.transfers[topic]['lastindex'], payload])
                self.transfers[topic]['lastindex'] += 1

    def ls(self):
        return self.topics.keys()

    def samples(self,topic,amount=1):
        if not topic in self.topics:
            return None
        if amount == 0 or amount is None:
            return self.topics[topic]['raw']
        # TODO : Print indexed data ?
        return self.topics[topic]['raw'][-amount:]

    def count(self,topic):
        if not topic in self.topics:
            return 0
        # TODO : what to do on indexed data ?
        return len(self.topics[topic]['raw'])

    def exists(self,topic):
        return topic in self.topics

    def transfer(self,topic,queue):
        # If the topic data is not already transfered to some queue
        if not topic in self.transfers:
            self.transfers[topic] = dict()
            self.transfers[topic]['queue'] = queue
            self.transfers[topic]['lastindex'] = 0
            # If there is already existing data under the topic
            if topic in self.topics:
                if self.topics[topic]['type'] == 'indexed':
                    for key, value in self.topics[topic]['xy'].iteritems():
                        queue.put([key, value])
                elif self.topics[topic]['type'] == 'linear':
                    for item in self.topics[topic]['raw']:
                        queue.put([self.transfers[topic]['lastindex'], item])
                        self.transfers[topic]['lastindex'] += 1

    def xytype(self,topic):
        return self.topics[topic]['type']<commit_msg>Implemented untransfer method<commit_after>from sortedcontainers import SortedDict

class Topics:
    def __init__(self):
        self.topics = dict()
        self.transfers = dict()

    def process(self,topic, payload, options=None):
        if not topic in self.topics:
            self.topics[topic] = dict()
            self.topics[topic]['raw'] = [] # For raw data
            self.topics[topic]['xy'] = SortedDict() # For indexed data
            if options:
                self.topics[topic]['type'] = 'indexed'
            else:
                self.topics[topic]['type'] = 'linear'

        t = self.topics[topic]
        t['raw'].append(payload)

        # If topic is of type indexed data, also store in ordered dict
        if options:
            t['xy'][options['index']] = payload

        # If there is an active transfer, transfer received data to the queue
        if topic in self.transfers:
            # For indexed data, provide the index for x and payload for y
            if t['type'] == 'indexed' and options is not None:
                self.transfers[topic]['queue'].put([options['index'], payload])
            # For linear data, provide sample id for x and payload for y
            elif t['type'] == 'linear':
                self.transfers[topic]['queue'].put([self.transfers[topic]['lastindex'], payload])
                self.transfers[topic]['lastindex'] += 1

    def ls(self):
        return self.topics.keys()

    def samples(self,topic,amount=1):
        if not topic in self.topics:
            return None
        if amount == 0 or amount is None:
            return self.topics[topic]['raw']
        # TODO : Print indexed data ?
        return self.topics[topic]['raw'][-amount:]

    def count(self,topic):
        if not topic in self.topics:
            return 0
        # TODO : what to do on indexed data ?
        return len(self.topics[topic]['raw'])

    def exists(self,topic):
        return topic in self.topics

    def transfer(self,topic,queue):
        # If the topic data is not already transfered to some queue
        if not topic in self.transfers:
            self.transfers[topic] = dict()
            self.transfers[topic]['queue'] = queue
            self.transfers[topic]['lastindex'] = 0
            # If there is already existing data under the topic
            if topic in self.topics:
                if self.topics[topic]['type'] == 'indexed':
                    for key, value in self.topics[topic]['xy'].iteritems():
                        queue.put([key, value])
                elif self.topics[topic]['type'] == 'linear':
                    for item in self.topics[topic]['raw']:
                        queue.put([self.transfers[topic]['lastindex'], item])
                        self.transfers[topic]['lastindex'] += 1

    def untransfer(self,topic):
        # If the topic data is already transfered to some queue
        if topic in self.transfers:
            # Remove it from the transfer list
            del self.transfers[topic]

    def xytype(self,topic):
        return self.topics[topic]['type']
{'fraction_non_alphanumeric': 0.1190796103459859, 'fraction_numerical': 0.002015451797111186, 'mean_word_length': 2.0015120967741935, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5291113', 'n_tokens_mistral': 1633, 'n_tokens_neox': 1565, 'n_words': 483}