<commit_before>import time as ttime
from itertools import count
from collections import namedtuple, deque
import uuid
import signal
import numpy as np


class Msg(namedtuple('Msg_base', ['message', 'obj', 'args', 'kwargs'])):
    __slots__ = ()

    def __repr__(self):
        return '{}: ({}), {}, {}'.format(
            self.message, self.obj, self.args, self.kwargs)


class Base:
    def __init__(self, name, fields):
        self._name = name
        self._fields = fields

    def describe(self):
        return {k: {'source': self._name, 'dtype': 'number'}
                for k in self._fields}

    def __repr__(self):
        return '{}: {}'.format(self._klass, self._name)


class Reader(Base):
    _klass = 'reader'

    def __init__(self, *args, **kwargs):
        super(Reader, self).__init__(*args, **kwargs)
        self._cnt = 0

    def read(self):
        data = dict()
        for k in self._fields:
            data[k] = self._cnt
            self._cnt += 1

        return data

    def trigger(self, *, blocking=True):
        if blocking:
            pass


class Mover(Base):
    _klass = 'mover'

    def __init__(self, *args, **kwargs):
        super(Mover, self).__init__(*args, **kwargs)
        self._data = {k: 0 for k in self._fields}
        self._staging = None

    def read(self):
        return dict(self._data)

    def set(self, new_values):
        if set(new_values) - set(self._data):
            raise ValueError('setting non-existent field')
        self._staging = new_values

    def trigger(self, *, blocking=True):
        if blocking:
            pass

        if self._staging:
            for k, v in self._staging.items():
                self._data[k] = v

        self._staging = None

    def settle(self):
        pass


class SynGaus(Mover):

    def __init__(self, name, motor_name, det_name, center, Imax, sigma=1):
        super(SynGaus, self).__init__(name, (motor_name, det_name))
        self._motor = motor_name
        self._det = det_name
        self.center = center
        self.Imax = Imax
        self.sigma = sigma

    def set(self, new_values):
        if self._det in new_values:
            raise ValueError("you can't set the detector value")
        super(SynGaus, self).set(new_values)
        m = self._staging[self._motor]
        v = self.Imax * np.exp(-(m - self.center)**2 / (2 * self.sigma**2))
        self._staging[self._det] = v


class FlyMagic(Base):
    def kickoff(self):
        pass

    def collect(self):
        pass


def MoveRead_gen(motor, detector):
    try:
        for j in range(10):
            yield Msg('set', motor, ({'x': j}, ), {})
            yield Msg('trigger', motor, (), {})
            yield Msg('trigger', detector, (), {})
            yield Msg('read', detector, (), {})
            yield Msg('read', motor, (), {})
    finally:
        print('Generator finished')


def SynGaus_gen(syngaus):
    try:
        for x in np.linspace(-5, 5, 100):
            yield Msg('set', syngaus, ({'x': x}, ), {})
            yield Msg('trigger', syngaus, (), {})
            yield Msg('wait', None, (.1, ), {})
            yield Msg('read', syngaus, (), {})

    finally:
        print('generator finished')


class RunEngine:
    def __init__(self):
        self.panic = False
        self._sigint_handler = None
        self._read_cache = deque()
        self._proc_registry = {
            'create': self._create,
            'save': self._save,
            'read': self._read,
            'null': self._null,
            'set': self._set,
            'trigger': self._trigger,
            'wait': self._wait
            }
        self._cache = None

    def run(self, g, additional_callbacks=None):
        with SignalHandler(signal.SIGINT) as self._sigint_handler:
            # When run_engine gets pushed to a thread, this SIGINT
            # handling will still work.
            self.run_engine(self, g, additional_callbacks=None)

    def run_engine(self, g, additional_callbacks=None):
        run_start_uid = uid()
        doc = dict(uid=run_start_uid,
                time=ttime.time(), beamline_id=beamline_id, owner=owner,
                scan_id=scan_id, **custom)
        self.emit_start(doc)
        self._cache = deque()
        r = None
        while True:
            try:
                if self.panic:
                    exit_status = 'fail'
                    break
                if self._sigint_handler.interrupted:
                    exit_status = 'abort'
                    break
                msg = g.send(r)
                r = self._proc_registry[msg.message](msg)
<commit_msg>FIX: One cache.<commit_after>import time as ttime
from itertools import count
from collections import namedtuple, deque
import uuid
import signal
import numpy as np


class Msg(namedtuple('Msg_base', ['message', 'obj', 'args', 'kwargs'])):
    __slots__ = ()

    def __repr__(self):
        return '{}: ({}), {}, {}'.format(
            self.message, self.obj, self.args, self.kwargs)


class Base:
    def __init__(self, name, fields):
        self._name = name
        self._fields = fields

    def describe(self):
        return {k: {'source': self._name, 'dtype': 'number'}
                for k in self._fields}

    def __repr__(self):
        return '{}: {}'.format(self._klass, self._name)


class Reader(Base):
    _klass = 'reader'

    def __init__(self, *args, **kwargs):
        super(Reader, self).__init__(*args, **kwargs)
        self._cnt = 0

    def read(self):
        data = dict()
        for k in self._fields:
            data[k] = self._cnt
            self._cnt += 1

        return data

    def trigger(self, *, blocking=True):
        if blocking:
            pass


class Mover(Base):
    _klass = 'mover'

    def __init__(self, *args, **kwargs):
        super(Mover, self).__init__(*args, **kwargs)
        self._data = {k: 0 for k in self._fields}
        self._staging = None

    def read(self):
        return dict(self._data)

    def set(self, new_values):
        if set(new_values) - set(self._data):
            raise ValueError('setting non-existent field')
        self._staging = new_values

    def trigger(self, *, blocking=True):
        if blocking:
            pass

        if self._staging:
            for k, v in self._staging.items():
                self._data[k] = v

        self._staging = None

    def settle(self):
        pass


class SynGaus(Mover):

    def __init__(self, name, motor_name, det_name, center, Imax, sigma=1):
        super(SynGaus, self).__init__(name, (motor_name, det_name))
        self._motor = motor_name
        self._det = det_name
        self.center = center
        self.Imax = Imax
        self.sigma = sigma

    def set(self, new_values):
        if self._det in new_values:
            raise ValueError("you can't set the detector value")
        super(SynGaus, self).set(new_values)
        m = self._staging[self._motor]
        v = self.Imax * np.exp(-(m - self.center)**2 / (2 * self.sigma**2))
        self._staging[self._det] = v


class FlyMagic(Base):
    def kickoff(self):
        pass

    def collect(self):
        pass


def MoveRead_gen(motor, detector):
    try:
        for j in range(10):
            yield Msg('set', motor, ({'x': j}, ), {})
            yield Msg('trigger', motor, (), {})
            yield Msg('trigger', detector, (), {})
            yield Msg('read', detector, (), {})
            yield Msg('read', motor, (), {})
    finally:
        print('Generator finished')


def SynGaus_gen(syngaus):
    try:
        for x in np.linspace(-5, 5, 100):
            yield Msg('set', syngaus, ({'x': x}, ), {})
            yield Msg('trigger', syngaus, (), {})
            yield Msg('wait', None, (.1, ), {})
            yield Msg('read', syngaus, (), {})

    finally:
        print('generator finished')


class RunEngine:
    def __init__(self):
        self.panic = False
        self._sigint_handler = None
        self._read_cache = deque()
        self._proc_registry = {
            'create': self._create,
            'save': self._save,
            'read': self._read,
            'null': self._null,
            'set': self._set,
            'trigger': self._trigger,
            'wait': self._wait
            }

    def run(self, g, additional_callbacks=None):
        with SignalHandler(signal.SIGINT) as self._sigint_handler:
            # When run_engine gets pushed to a thread, this SIGINT
            # handling will still work.
            self.run_engine(self, g, additional_callbacks=None)

    def run_engine(self, g, additional_callbacks=None):
        run_start_uid = uid()
        doc = dict(uid=run_start_uid,
                time=ttime.time(), beamline_id=beamline_id, owner=owner,
                scan_id=scan_id, **custom)
        self.emit_start(doc)
        self._read_cache.clear()
        r = None
        while True:
            try:
                if self.panic:
                    exit_status = 'fail'
                    break
                if self._sigint_handler.interrupted:
                    exit_status = 'abort'
                    break
                msg = g.send(r)
                r = self._proc_registry[msg.message](msg)

{'fraction_non_alphanumeric': 0.12882238126219908, 'fraction_numerical': 0.0032530904359141183, 'mean_word_length': 1.8975808985234055, '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': '7596378', 'n_tokens_mistral': 2862, 'n_tokens_neox': 2765, 'n_words': 701}