<commit_before>"""Celery tasks"""
import numpy as np

from celery import Celery
import _wax_compiled_helpers as cch
import logging
import snappy
from wax.Configuration import SAMPLE_TYPE, MAX_DRIFT, PADDING, THRESHOLD
from wax.Database.InputDBInterface import MongoDBInput
from wax.Database.OutputDBInterface import MongoDBOutput

#Specify mongodb host and datababse to connect to                                                                                              
BROKER_URL = 'mongodb://xedaqje.no-ip.biz:27017/jobs'

celery = Celery('EOD_TASKS',
                broker=BROKER_URL,
                backend=BROKER_URL)

def save_decision(events, last, t0, t1):
    if last != None:
        # Check for event that spans many search blocks.  This super event 'should'
        # never happen.
        if last['range'][1] > t1 and last['range'][0] < t1 - PADDING:
            raise RuntimeError("Event spans two search blocks (mega-event?)")

        if last['range'][0] > t0 + PADDING and last['range'][1] < t1:
            events.append(last)  # Save event

@celery.task
def process_time_range(t0, t1,
                        collection_name, hostname,
                        threshold=THRESHOLD):
    """Process a time chunk

    .. todo:: Must this know padding?  Maybe just hand cursor so can mock?

    collection_name
    hostname

    :param t0: Initial time to query.
    :type t0: int.
    :param t1: Final time.
    :type t1: int.
    :returns:  int -- number of bytes processed
    :raises: AssertionError
    """
    input = MongoDBInput(collection_name=collection_name,
                         hostname=hostname)

    if input.initialized:
        output = MongoDBOutput(collection_name=input.get_collection_name(),
                                    hostname=hostname)
    else:
        logging.debug("Cannot setup output DB.")
        raise RuntimeError()

    data_docs = input.get_data_docs(t0, t1)

    untriggered_size = 0
    reduction_factor = 100
    n = int(np.ceil((t1 - t0) / reduction_factor))
    cch.setup(n)

    # Every data doc has a start time and end time.  These ranges are used
    # later to compute overlaps with event ranges.
    doc_ranges = np.zeros((len(data_docs), 2), dtype=np.int32)

    for i, doc in enumerate(data_docs):
        time_correction = doc['time'] - t0

        samples = get_samples_from_doc(doc, input.is_compressed)
        doc['size'] = samples.size * 2
        # Record time range of these samples
        doc_ranges[i] = (time_correction / reduction_factor,
                         (time_correction + samples.size) / reduction_factor)

        cch.add_samples(samples, time_correction, reduction_factor)
        untriggered_size += doc['size']

    #sum_data = cch.get_sum()

    logging.debug("Size of data process in bytes: %d", untriggered_size)

    # If no data analyzed, return
    logging.debug("Size of data analyzed: %d", untriggered_size)
    if untriggered_size == 0:
        logging.debug('No data found in [%d, %d]' % (t0, t1))
        return 0

    ranges = cch.build_events(threshold,  # Threshold
                              int(MAX_DRIFT / reduction_factor))

    #  One event has many samples, thus to mapping represented as samples -> event
    mappings = cch.overlaps(doc_ranges.flatten())
    events = []
<commit_msg>Add assert to check time of input data is int<commit_after>"""Celery tasks"""
import numpy as np

from celery import Celery
import _wax_compiled_helpers as cch
import logging
import snappy
from wax.Configuration import SAMPLE_TYPE, MAX_DRIFT, PADDING, THRESHOLD
from wax.Database.InputDBInterface import MongoDBInput
from wax.Database.OutputDBInterface import MongoDBOutput

#Specify mongodb host and datababse to connect to                                                                                              
BROKER_URL = 'mongodb://xedaqje.no-ip.biz:27017/jobs'

celery = Celery('EOD_TASKS',
                broker=BROKER_URL,
                backend=BROKER_URL)

def save_decision(events, last, t0, t1):
    if last != None:
        # Check for event that spans many search blocks.  This super event 'should'
        # never happen.
        if last['range'][1] > t1 and last['range'][0] < t1 - PADDING:
            raise RuntimeError("Event spans two search blocks (mega-event?)")

        if last['range'][0] > t0 + PADDING and last['range'][1] < t1:
            events.append(last)  # Save event

@celery.task
def process_time_range(t0, t1,
                        collection_name, hostname,
                        threshold=THRESHOLD):
    """Process a time chunk

    .. todo:: Must this know padding?  Maybe just hand cursor so can mock?

    collection_name
    hostname

    :param t0: Initial time to query.
    :type t0: int.
    :param t1: Final time.
    :type t1: int.
    :returns:  int -- number of bytes processed
    :raises: AssertionError
    """
    input = MongoDBInput(collection_name=collection_name,
                         hostname=hostname)

    if input.initialized:
        output = MongoDBOutput(collection_name=input.get_collection_name(),
                                    hostname=hostname)
    else:
        logging.debug("Cannot setup output DB.")
        raise RuntimeError()

    data_docs = input.get_data_docs(t0, t1)

    untriggered_size = 0
    reduction_factor = 100
    n = int(np.ceil((t1 - t0) / reduction_factor))
    cch.setup(n)

    # Every data doc has a start time and end time.  These ranges are used
    # later to compute overlaps with event ranges.
    doc_ranges = np.zeros((len(data_docs), 2), dtype=np.int32)

    for i, doc in enumerate(data_docs):
        time_correction = doc['time'] - t0

        samples = get_samples_from_doc(doc, input.is_compressed)
        doc['size'] = samples.size * 2
        # Record time range of these samples
        doc_ranges[i] = (time_correction / reduction_factor,
                         (time_correction + samples.size) / reduction_factor)

        assert type(time_correction) == int
        cch.add_samples(samples, time_correction, reduction_factor)
        untriggered_size += doc['size']

    #sum_data = cch.get_sum()

    logging.debug("Size of data process in bytes: %d", untriggered_size)

    # If no data analyzed, return
    logging.debug("Size of data analyzed: %d", untriggered_size)
    if untriggered_size == 0:
        logging.debug('No data found in [%d, %d]' % (t0, t1))
        return 0

    ranges = cch.build_events(threshold,  # Threshold
                              int(MAX_DRIFT / reduction_factor))

    #  One event has many samples, thus to mapping represented as samples -> event
    mappings = cch.overlaps(doc_ranges.flatten())
    events = []

{'fraction_non_alphanumeric': 0.09242695289206918, 'fraction_numerical': 0.011329755515802028, 'mean_word_length': 2.371356783919598, 'pattern_counts': {'":': 0, '<': 7, '<?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': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4408689', 'n_tokens_mistral': 2062, 'n_tokens_neox': 1969, 'n_words': 570}