<commit_before>import random
import threading
import numpy as np

import datetime
from rules import SendTriggers
from targets import Target
import config


class Stage:
    """"""

    def __init__(self, processor, target_space, send_triggers, source=config.site_name):
        self.processor = processor
        self.target_space = target_space
        self.send_triggers = send_triggers
        self.data_queues = {}
        for stream in config.data_streams:
            self.data_queues[stream] = []
        # NIMS grouped by target_id, so change to dict {target_id: [indices]}
        self.data_queues['nims'] = {}
        self.recent_targets = []
        # Adds ADCP (necessary for testing when not connected to ADCP)
        unixtime = (datetime.datetime.utcnow() - datetime.datetime(1970,1,1))
        self.addDataToStage('adcp', [unixtime.days*24*60*60 + unixtime.seconds, 1.2, 4.5])
        #self.startStageProcessing()

    def processDataBeforeStage(self, stream, data):
        """Performs whatever preprocessing necessitated for data from a
        particular stream, adds data to appropriate target list, then returns
        index for added data in TargetSpace.

        Assumes 'nims' passes a list inside a dict with different tracks.
        """
        if stream == 'adcp':
            data = [datetime.datetime.fromtimestamp(data[0]), data[1], data[2]]
            self.target_space.tables[stream].append(data)
            print(self.target_space.tables[stream])
            # return next empy index
            return len(self.target_space.tables[stream]) - 1
        elif stream == 'pamguard':
            # comm format matches desired, no need to change
            self.target_space.tables[stream].append(data)
            return len(self.target_space.tables[stream]) - 1
        elif stream == 'nims':
            indices = {}
            timestamp = data[0]
            for track in data[1]:
                new_data = [timestamp, track['id'], track['pings_visible'],
                        track['first_ping'], track['target_strength'], track['width'],
                        track['height'], track['size_sq_m'], track['speed_mps'],
                        track['min_angle_m'], track['min_range_m'], track['max_angle_m'],
                        track['max_range_m'], track['last_pos_angle'],
                        track['last_pos_range'], None]
                self.target_space.tables[stream].append(new_data)
                indices[track['id']] = len(self.target_space.tables[stream]) - 1
        elif stream in config.data_streams:
            raise ValueError("No stage processing functionality exists for" \
                             " data stream {0}.".format(stream))
        else:
            raise ValueError("Error processing data for stage. Stream {0} not" \
                              " defined in config file.".format(stream))

        return indices

    def addDataToStage(self, stream, data):
        """Calls processing function for data based on stream then adds data to
        stream-specific queue.
        """
        if stream not in config.data_streams:
            raise ValueError("Error adding data to stage. Data stream {0} not" \
                             " defined in config file.".format(stream))
        stage_indices = self.processDataBeforeStage(stream, data)
        if stream == 'nims':  # indexed
            for track_id in stage_indices:
                if track_id not in self.data_queues[stream]:<commit_msg>changed indexing to add values to last empty index in stage<commit_after>import random
import threading
import numpy as np

import datetime
from rules import SendTriggers
from targets import Target
import config


class Stage:
    """"""

    def __init__(self, processor, target_space, send_triggers, source=config.site_name):
        self.processor = processor
        self.target_space = target_space
        self.send_triggers = send_triggers
        self.data_queues = {}
        for stream in config.data_streams:
            self.data_queues[stream] = []
        # NIMS grouped by target_id, so change to dict {target_id: [indices]}
        self.data_queues['nims'] = {}
        self.recent_targets = []
        # Adds ADCP (necessary for testing when not connected to ADCP)
        unixtime = (datetime.datetime.utcnow() - datetime.datetime(1970,1,1))
        self.addDataToStage('adcp', [unixtime.days*24*60*60 + unixtime.seconds, 1.2, 4.5])
        #self.startStageProcessing()

    def processDataBeforeStage(self, stream, data):
        """Performs whatever preprocessing necessitated for data from a
        particular stream, adds data to appropriate target list, then returns
        index for added data in TargetSpace.

        Assumes 'nims' passes a list inside a dict with different tracks.
        """
        if stream == 'adcp':
            data = [datetime.datetime.fromtimestamp(data[0]), data[1], data[2]]
            self.target_space.tables[stream].append(data)
            print(self.target_space.tables[stream])
            if [] in self.target_space.tables[stream]:
                return self.target_space.tables[stream].index([])
            else:
                return len(self.target_space.tables[stream]) - 1
        elif stream == 'pamguard':
            # comm format matches desired, no need to change
            self.target_space.tables[stream].append(data)
            if [] in self.target_space.tables[stream]:
                return self.target_space.tables[stream].index([])
            else:
                return len(self.target_space.tables[stream]) - 1
        elif stream == 'nims':
            indices = {}
            timestamp = data[0]
            for track in data[1]:
                new_data = [timestamp, track['id'], track['pings_visible'],
                        track['first_ping'], track['target_strength'], track['width'],
                        track['height'], track['size_sq_m'], track['speed_mps'],
                        track['min_angle_m'], track['min_range_m'], track['max_angle_m'],
                        track['max_range_m'], track['last_pos_angle'],
                        track['last_pos_range'], None]
                self.target_space.tables[stream].append(new_data)
                if [] in self.target_space.tables[stream]:
                    indices[track['id']] = self.target_space.tables[stream].index([])
                else:
                    indices[track['id']] = len(self.target_space.tables[stream]) - 1
        elif stream in config.data_streams:
            raise ValueError("No stage processing functionality exists for" \
                             " data stream {0}.".format(stream))
        else:
            raise ValueError("Error processing data for stage. Stream {0} not" \
                              " defined in config file.".format(stream))

        return indices

    def addDataToStage(self, stream, data):
        """Calls processing function for data based on stream then adds data to
        stream-specific queue.
        """
        if stream not in config.data_streams:
            raise ValueError("Error adding data to stage. Data stream {0} not" \
                             " defined in config file.".format(stream))
        stage_indices = self.processDataBeforeStage(stream, data)
        if stream == 'nims':  # indexed
            for track_id in stage_indices:
                if track_id not in self.data_queues[stream]:
{'fraction_non_alphanumeric': 0.1034621578099839, 'fraction_numerical': 0.007246376811594203, 'mean_word_length': 2.2000858737655644, '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': '4847323', 'n_tokens_mistral': 2001, 'n_tokens_neox': 1936, 'n_words': 533}