<commit_before># -*- coding: utf-8 -*-
"""
"""
from __future__ import absolute_import

from collections import deque
from functools import wraps
import logging
import mimetypes
import time

try:
    import gevent
    import gevent.event
except ImportError:
    gevent = None

if gevent:
    from gevent.monkey import get_original
    Thread, Lock = get_original('threading', ['Thread', 'Lock'])
    sleep = gevent.sleep
else:
    from threading import Thread, Lock
    from time import sleep

log = logging.getLogger(__name__)

class StreamTimeout(Exception):
    pass

class VideoFrame(object):
    def __init__(self, image_data, content_type):
        self.image_data = image_data
        self.content_type = content_type

    @classmethod
    def from_file(cls, filename):
        content_type, encoding = mimetypes.guess_type(filename)
        if not content_type:
            raise ValueError("Can not guess content type")
        with open(filename, 'rb') as fp:
            return cls(fp.read(), content_type)

class StaticVideoStream(object):
    """ A video stream from static images.  For testing.
    """
    def __init__(self, image_filenames, loop=True, frame_rate=4.0):
        if not image_filenames:
            raise ValueError("No images given")
        self.frames = map(VideoFrame.from_file, image_filenames)
        self.loop = loop
        self.start = time.time()
        self.frame_rate = frame_rate
        self.closed = False

    def close(self):
        self.closed = True

    def __iter__(self):
        frame = self.get_frame()
        while frame:
            yield frame
            frame = self.get_frame(frame)

    def get_frame(self, current_frame=None, timeout=None):
        index = self.frame_rate * (time.time() - self.start)
        frame = self._get_frame(index)
        if current_frame is None or frame is not current_frame:
            return frame

        wait = (int(index) + 1 - index) / self.frame_rate
        if timeout and timeout < wait:
            sleep(timeout)
            raise StreamTimeout()
        sleep(wait)
        return self._get_frame(index + 1)

    def _get_frame(self, index):
        index = int(index)
        if self.closed:
            return None
        if self.loop:
            index = index % len(self.frames)
        elif index >= len(self.frames):
            return None                 # stream done
        return self.frames[index]

def synchronized(method):
    @wraps(method)
    def wrapper(self, *args):
        with self.mutex:
            return method(self, *args)
    return wrapper

class VideoStreamer(Thread):
    """ Stream video in a separate thread.

    This has good support of the timeout argument to get_frame.

    """
    def __init__(self, stream, buffer_size=4):
        super(VideoStreamer, self).__init__()

        self.stream = stream
        self.framebuf = deque(maxlen=buffer_size)
        self.mutex = Lock()

        self.new_frame_event = gevent.event.Event()

        # gevent magic:
        # Hook to to set the new_frame event in the main thread
        async = gevent.get_hub().loop.async()
        async.start(self._signal_new_frame)
        self.signal_new_frame = async.send

        self.daemon = True
        self.closed = False
        self.start()

    def close(self):
        self.closed = True
        self.stream.close()

    def run(self):
        log.info("Capture thread starting: %r", self.stream)
        frame = None
        while not self.closed:
            try:
                frame = self.stream.get_frame(frame, timeout=1.0)
            except StreamTimeout:
                pass
            else:
                if frame is None:
                    self.closed = True
                else:
                    self._buffer_frame(frame)
        log.info("Capture thread terminating: %r", self.stream)

    @synchronized
    def _buffer_frame(self, frame):
        self.framebuf.appendleft(frame)
        self.signal_new_frame()

    @synchronized
    def _signal_new_frame(self):
        new_frame_event = self.new_frame_event
        self.new_frame_event = gevent.event.Event()
        new_frame_event.set()

    def __iter__(self):
        frame = self.get_frame()
        while frame:
            yield frame
            frame = self.get_frame(frame)

    def get_frame(self, current_frame=None, timeout=None):
        if self.closed:
            return None
        frame = self._get_frame(current_frame)
        while isinstance(frame, gevent.event.Event):
            if not frame.wait(timeout):
                raise StreamTimeout()
            frame = self._get_frame(current_frame)
        return frame<commit_msg>Close stream from capture thread<commit_after># -*- coding: utf-8 -*-
"""
"""
from __future__ import absolute_import

from collections import deque
from functools import wraps
import logging
import mimetypes
import time

try:
    import gevent
    import gevent.event
except ImportError:
    gevent = None

if gevent:
    from gevent.monkey import get_original
    Thread, Lock = get_original('threading', ['Thread', 'Lock'])
    sleep = gevent.sleep
else:
    from threading import Thread, Lock
    from time import sleep

log = logging.getLogger(__name__)

class StreamTimeout(Exception):
    pass

class VideoFrame(object):
    def __init__(self, image_data, content_type):
        self.image_data = image_data
        self.content_type = content_type

    @classmethod
    def from_file(cls, filename):
        content_type, encoding = mimetypes.guess_type(filename)
        if not content_type:
            raise ValueError("Can not guess content type")
        with open(filename, 'rb') as fp:
            return cls(fp.read(), content_type)

class StaticVideoStream(object):
    """ A video stream from static images.  For testing.
    """
    def __init__(self, image_filenames, loop=True, frame_rate=4.0):
        if not image_filenames:
            raise ValueError("No images given")
        self.frames = map(VideoFrame.from_file, image_filenames)
        self.loop = loop
        self.start = time.time()
        self.frame_rate = frame_rate
        self.closed = False

    def close(self):
        self.closed = True

    def __iter__(self):
        frame = self.get_frame()
        while frame:
            yield frame
            frame = self.get_frame(frame)

    def get_frame(self, current_frame=None, timeout=None):
        index = self.frame_rate * (time.time() - self.start)
        frame = self._get_frame(index)
        if current_frame is None or frame is not current_frame:
            return frame

        wait = (int(index) + 1 - index) / self.frame_rate
        if timeout and timeout < wait:
            sleep(timeout)
            raise StreamTimeout()
        sleep(wait)
        return self._get_frame(index + 1)

    def _get_frame(self, index):
        index = int(index)
        if self.closed:
            return None
        if self.loop:
            index = index % len(self.frames)
        elif index >= len(self.frames):
            return None                 # stream done
        return self.frames[index]

def synchronized(method):
    @wraps(method)
    def wrapper(self, *args):
        with self.mutex:
            return method(self, *args)
    return wrapper

class VideoStreamer(Thread):
    """ Stream video in a separate thread.

    This has good support of the timeout argument to get_frame.

    """
    def __init__(self, stream, buffer_size=4):
        super(VideoStreamer, self).__init__()

        self.stream = stream
        self.framebuf = deque(maxlen=buffer_size)
        self.mutex = Lock()

        self.new_frame_event = gevent.event.Event()

        # gevent magic:
        # Hook to to set the new_frame event in the main thread
        async = gevent.get_hub().loop.async()
        async.start(self._signal_new_frame)
        self.signal_new_frame = async.send

        self.daemon = True
        self.closed = False
        self.start()

    def close(self):
        self.closed = True

    def run(self):
        log.info("Capture thread starting: %r", self.stream)
        frame = None
        while not self.closed:
            try:
                frame = self.stream.get_frame(frame, timeout=1.0)
            except StreamTimeout:
                pass
            else:
                if frame is None:
                    self.closed = True
                else:
                    self._buffer_frame(frame)
        log.info("Capture thread terminating: %r", self.stream)
        self.stream.close()

    @synchronized
    def _buffer_frame(self, frame):
        self.framebuf.appendleft(frame)
        self.signal_new_frame()

    @synchronized
    def _signal_new_frame(self):
        new_frame_event = self.new_frame_event
        self.new_frame_event = gevent.event.Event()
        new_frame_event.set()

    def __iter__(self):
        frame = self.get_frame()
        while frame:
            yield frame
            frame = self.get_frame(frame)

    def get_frame(self, current_frame=None, timeout=None):
        if self.closed:
            return None
        frame = self._get_frame(current_frame)
        while isinstance(frame, gevent.event.Event):
            if not frame.wait(timeout):
                raise StreamTimeout()
            frame = self._get_frame(current_frame)
        return frame
{'fraction_non_alphanumeric': 0.08907401473886575, 'fraction_numerical': 0.0017088539997863932, 'mean_word_length': 2.128633478115603, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '4781416', 'n_tokens_mistral': 2667, 'n_tokens_neox': 2614, 'n_words': 707}