<commit_before>__author__ = 'nathan.muir'

from camera import Camera, CameraFactory
import boto.ec2.cloudwatch
import sys, traceback, os
import json

class CloudWatchCameraFactory(CameraFactory):

    def camera(self, state):
        kwargs = {
            'freq': self.frequency,
            'task_names': None,
            'cloud_watch_namespace': 'celery',
            'aws_connection': boto.ec2.cloudwatch.CloudWatchConnection()
        }
        if 'TASK_MONITOR_CW_TASK_NAMES' in os.environ:
            kwargs['task_names'] = [x.strip() for x in os.environ['TASK_MONITOR_CW_TASK_NAMES'].split(',')]

        if 'TASK_MONITOR_CW_NAMESPACE' in os.environ:
            kwargs['cloud_watch_namespace'] = os.environ['TASK_MONITOR_CW_NAMESPACE'].strip()

        if 'TASK_MONITOR_CW_DIMS' in os.environ:
            pass
        return self.c(state, **kwargs)


class CloudWatchCamera(Camera):

    clear_after = True
    _event_names = ('waiting', 'running', 'completed', 'failed')

    def __init__(self, state, aws_connection, freq=1.0, task_names=None, cloud_watch_namespace='celery', base_dimensions=None):
        super(CloudWatchCamera, self).__init__(state, freq=freq)
        self.aws_connection = aws_connection
        self.cloud_watch_namespace = cloud_watch_namespace
        self.base_dimensions = base_dimensions
        if task_names is not None:
            self.task_names = task_names
        else:
            self.task_names = []
        self.metrics = None

    def on_shutter(self, state):
        try:
            self.metrics = self._build_metrics(state)
        except RuntimeError as r:
            print r


    def after_shutter(self):
        try:
            self.metrics.send()
        except:
            print "Exception in user code:"
            print '-'*60
            traceback.print_exc(file=sys.stdout)
        finally:
            self.metrics = None

    def _metric_list(self):
        return MetricList(self.cloud_watch_namespace, self.aws_connection, self.base_dimensions)

    def _build_metrics(self, state):
        metrics = self._metric_list()
        self._add_queue_times(metrics, state.time_to_start)
        self._add_run_times(metrics, state.time_to_process)
        self._add_task_events(metrics, state.totals)
        self._add_waiting_tasks(metrics, len(state.waiting_tasks))
        self._add_running_tasks(metrics, len(state.running_tasks))
        return metrics

    def _add_task_events(self, metrics, event_totals):
        task_events = self._fill_task_events(event_totals)
        for task_name, total_by_event in task_events.items():
            for event_name, total in total_by_event.items():
                metrics.add('CeleryTaskEvent%s' % event_name.capitalize(), unit='Count', value=total, dimensions={'task': task_name})

    def _fill_task_events(self, event_totals):
        task_events = {}
        for task_name in self.task_names:
            task_events[task_name] = dict((en, 0) for en in self._event_names)

        for task_name, totals in event_totals.items():
            if task_name not in task_events:
                task_events[task_name] = dict((en, 0) for en in self._event_names)
            for event_name, total in totals.items():
                task_events[task_name][event_name] = total
        return task_events

    @staticmethod
    def _add_queue_times(metrics, time_to_start):
        for task_name, stats in time_to_start.items():
            metrics.add('CeleryQueuedTime', unit='Seconds', dimensions={'task': task_name}, stats=stats.__dict__.copy())

    @staticmethod
    def _add_run_times(metrics, time_to_process):
        for task_name, stats in time_to_process.items():
            metrics.add('CeleryProcessingTime', unit='Seconds', dimensions={'task': task_name}, stats=stats.__dict__.copy())
    @staticmethod
    def _add_waiting_tasks(metrics, waiting_tasks):
        metrics.add('CeleryQueueSize', unit='Count', value=waiting_tasks, dimensions={'queue': 'celery'})

    @staticmethod
    def _add_running_tasks(metrics, running_tasks):
        metrics.add('CeleryRunningTasks', unit='Count', value=running_tasks, dimensions={'queue': 'celery'})




def xchunk(arr, size):
    for x in xrange(0, len(arr), size):<commit_msg>Renamed Task based metrics CeleryProcessingTime and CeleryQueuedTime to CeleryTaskProcessingTime and CeleryTaskQueuedTime<commit_after>__author__ = 'nathan.muir'

from camera import Camera, CameraFactory
import boto.ec2.cloudwatch
import sys, traceback, os
import json

class CloudWatchCameraFactory(CameraFactory):

    def camera(self, state):
        kwargs = {
            'freq': self.frequency,
            'task_names': None,
            'cloud_watch_namespace': 'celery',
            'aws_connection': boto.ec2.cloudwatch.CloudWatchConnection()
        }
        if 'TASK_MONITOR_CW_TASK_NAMES' in os.environ:
            kwargs['task_names'] = [x.strip() for x in os.environ['TASK_MONITOR_CW_TASK_NAMES'].split(',')]

        if 'TASK_MONITOR_CW_NAMESPACE' in os.environ:
            kwargs['cloud_watch_namespace'] = os.environ['TASK_MONITOR_CW_NAMESPACE'].strip()

        if 'TASK_MONITOR_CW_DIMS' in os.environ:
            pass
        return self.c(state, **kwargs)


class CloudWatchCamera(Camera):

    clear_after = True
    _event_names = ('waiting', 'running', 'completed', 'failed')

    def __init__(self, state, aws_connection, freq=1.0, task_names=None, cloud_watch_namespace='celery', base_dimensions=None):
        super(CloudWatchCamera, self).__init__(state, freq=freq)
        self.aws_connection = aws_connection
        self.cloud_watch_namespace = cloud_watch_namespace
        self.base_dimensions = base_dimensions
        if task_names is not None:
            self.task_names = task_names
        else:
            self.task_names = []
        self.metrics = None

    def on_shutter(self, state):
        try:
            self.metrics = self._build_metrics(state)
        except RuntimeError as r:
            print r


    def after_shutter(self):
        try:
            self.metrics.send()
        except:
            print "Exception in user code:"
            print '-'*60
            traceback.print_exc(file=sys.stdout)
        finally:
            self.metrics = None

    def _metric_list(self):
        return MetricList(self.cloud_watch_namespace, self.aws_connection, self.base_dimensions)

    def _build_metrics(self, state):
        metrics = self._metric_list()
        self._add_queue_times(metrics, state.time_to_start)
        self._add_run_times(metrics, state.time_to_process)
        self._add_task_events(metrics, state.totals)
        self._add_waiting_tasks(metrics, len(state.waiting_tasks))
        self._add_running_tasks(metrics, len(state.running_tasks))
        return metrics

    def _add_task_events(self, metrics, event_totals):
        task_events = self._fill_task_events(event_totals)
        for task_name, total_by_event in task_events.items():
            for event_name, total in total_by_event.items():
                metrics.add('CeleryTaskEvent%s' % event_name.capitalize(), unit='Count', value=total, dimensions={'task': task_name})

    def _fill_task_events(self, event_totals):
        task_events = {}
        for task_name in self.task_names:
            task_events[task_name] = dict((en, 0) for en in self._event_names)

        for task_name, totals in event_totals.items():
            if task_name not in task_events:
                task_events[task_name] = dict((en, 0) for en in self._event_names)
            for event_name, total in totals.items():
                task_events[task_name][event_name] = total
        return task_events

    @staticmethod
    def _add_queue_times(metrics, time_to_start):
        for task_name, stats in time_to_start.items():
            metrics.add('CeleryTaskQueuedTime', unit='Seconds', dimensions={'task': task_name}, stats=stats.__dict__.copy())

    @staticmethod
    def _add_run_times(metrics, time_to_process):
        for task_name, stats in time_to_process.items():
            metrics.add('CeleryTaskProcessingTime', unit='Seconds', dimensions={'task': task_name}, stats=stats.__dict__.copy())
    @staticmethod
    def _add_waiting_tasks(metrics, waiting_tasks):
        metrics.add('CeleryQueueSize', unit='Count', value=waiting_tasks, dimensions={'queue': 'celery'})

    @staticmethod
    def _add_running_tasks(metrics, running_tasks):
        metrics.add('CeleryRunningTasks', unit='Count', value=running_tasks, dimensions={'queue': 'celery'})




def xchunk(arr, size):
    for x in xrange(0, len(arr), size):
{'fraction_non_alphanumeric': 0.10388101324657216, 'fraction_numerical': 0.0020915640250987683, 'mean_word_length': 3.081081081081081, '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': '4538775', 'n_tokens_mistral': 2643, 'n_tokens_neox': 2609, 'n_words': 527}