<commit_before>import time as ttime
from ophyd.areadetector import (PerkinElmerDetector, ImagePlugin,
                                TIFFPlugin, StatsPlugin, HDF5Plugin,
                                ProcessPlugin, ROIPlugin)
from ophyd.device import BlueskyInterface
from ophyd.areadetector.trigger_mixins import SingleTrigger, MultiTrigger
from ophyd.areadetector.filestore_mixins import (FileStoreIterativeWrite,
                                                 FileStoreHDF5IterativeWrite,
                                                 FileStoreTIFFSquashing,
                                                 FileStoreTIFF)
from ophyd import Signal, EpicsSignal, EpicsSignalRO # Tim test
from ophyd import Component as C
from ophyd import StatusBase

# from shutter import sh1

#shctl1 = EpicsSignal('XF:28IDC-ES:1{Det:PE1}cam1:ShutterMode', name='shctl1')
shctl1 = EpicsMotor('XF:28IDC-ES:1{Sh2:Exp-Ax:5}Mtr', name='shctl1')


class XPDShutter(Device):
    cmd = C(EpicsSignal, 'Cmd-Cmd')
    close_sts = C(EpicsSignalRO, 'Sw:Cls1-Sts')
    open_sts = C(EpicsSignalRO, 'Sw:Opn1-Sts')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._st = None
        self._target = None
        self.close_sts.subscribe(self._watcher_close,
                                 self.close_sts.SUB_VALUE)

        self.open_sts.subscribe(self._watcher_open,
                                 self.open_sts.SUB_VALUE)

    def set(self, value, *, wait=False, **kwargs):
        if value not in ('Open', 'Close'):
            raise ValueError(
                "must be 'Open' or 'Close', not {!r}".format(value))
        if wait:
            raise RuntimeError()
        if self._st is not None:
            raise RuntimeError()
        self._target = value
        self._st = st = DeviceStatus(self, timeout=None)
        self.cmd.put(value)

        return st

    def _watcher_open(self, *, old_value=None, value=None, **kwargs):
        print("in open watcher", old_value, value)
        if self._target != 'Open':
            return
        if self._st is None:
            return

        if new_value:
            self._st._finished()
            self._target = None
            self._st = None
        print("in open watcher")

    def _watcher_close(self, *, old_value=None, value=None, **kwargs):
        print("in close watcher", old_value, value)
        if self._target != 'Close':
            return

        if self._st is None:
            return

        if new_value:
            self._st._finished()
            self._target = None
            self._st = None

        pass


class SavedImageSignal(Signal):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stashed_datakey = {}

    def describe(self):
        ret = super().describe()
        ret[self.name].update(self.stashed_datakey)
        return ret


def take_dark(cam, light_field, dark_field_name):
    # close shutter

    # take the dark frame
    cam.stage()
    st = cam.trigger()
    while not st.done:
        ttime.sleep(.1)
    ret = cam.read()
    desc = cam.describe()
    cam.unstage()

    # save the df uid
    df = ret[light_field]
    df_sig = getattr(cam, dark_field_name)
    df_sig.put(**df)
    # save the darkfrom description
    df_sig.stashed_datakey = desc[light_field]




class XPDTIFFPlugin(TIFFPlugin, FileStoreTIFFSquashing,
                    FileStoreIterativeWrite):
    pass


class XPDHDF5Plugin(HDF5Plugin, FileStoreHDF5IterativeWrite):
    pass


class XPDPerkinElmer(PerkinElmerDetector):
    image = C(ImagePlugin, 'image1:')

    tiff = C(XPDTIFFPlugin, 'TIFF1:',
             write_path_template='H:/pe1_data/%Y/%m/%d/',
             read_path_template='/direct/XF28ID1/pe1_data/%Y/%m/%d/',
             cam_name='cam',  # used to configure "tiff squashing"
             proc_name='proc',  # ditto
             read_attrs=[],
             root='/direct/XF28ID1/',
             fs=db.fs)

    # hdf5 = C(XPDHDF5Plugin, 'HDF1:',
    #          write_path_template='G:/pe1_data/%Y/%m/%d/',
    #          read_path_template='/direct/XF28ID1/pe1_data/%Y/%m/%d/',
    #          root='/direct/XF28ID1/', fs=db.fs)

    proc = C(ProcessPlugin, 'Proc1:')

    # These attributes together replace `num_images`. They control
    # summing images before they are stored by the detector (a.k.a. "tiff
    # squashing").
    images_per_set = C(Signal, value=1, add_prefix=())
    number_of_sets = C(Signal, value=1, add_prefix=())

    stats1 = C(StatsPlugin, 'Stats1:')
    stats2 = C(StatsPlugin, 'Stats2:')
    stats3 = C(StatsPlugin, 'Stats3:')
    stats4 = C(StatsPlugin, 'Stats4:')<commit_msg>MNT: fs -> reg translation<commit_after>import time as ttime
from ophyd.areadetector import (PerkinElmerDetector, ImagePlugin,
                                TIFFPlugin, StatsPlugin, HDF5Plugin,
                                ProcessPlugin, ROIPlugin)
from ophyd.device import BlueskyInterface
from ophyd.areadetector.trigger_mixins import SingleTrigger, MultiTrigger
from ophyd.areadetector.filestore_mixins import (FileStoreIterativeWrite,
                                                 FileStoreHDF5IterativeWrite,
                                                 FileStoreTIFFSquashing,
                                                 FileStoreTIFF)
from ophyd import Signal, EpicsSignal, EpicsSignalRO # Tim test
from ophyd import Component as C
from ophyd import StatusBase

# from shutter import sh1

#shctl1 = EpicsSignal('XF:28IDC-ES:1{Det:PE1}cam1:ShutterMode', name='shctl1')
shctl1 = EpicsMotor('XF:28IDC-ES:1{Sh2:Exp-Ax:5}Mtr', name='shctl1')


class XPDShutter(Device):
    cmd = C(EpicsSignal, 'Cmd-Cmd')
    close_sts = C(EpicsSignalRO, 'Sw:Cls1-Sts')
    open_sts = C(EpicsSignalRO, 'Sw:Opn1-Sts')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._st = None
        self._target = None
        self.close_sts.subscribe(self._watcher_close,
                                 self.close_sts.SUB_VALUE)

        self.open_sts.subscribe(self._watcher_open,
                                 self.open_sts.SUB_VALUE)

    def set(self, value, *, wait=False, **kwargs):
        if value not in ('Open', 'Close'):
            raise ValueError(
                "must be 'Open' or 'Close', not {!r}".format(value))
        if wait:
            raise RuntimeError()
        if self._st is not None:
            raise RuntimeError()
        self._target = value
        self._st = st = DeviceStatus(self, timeout=None)
        self.cmd.put(value)

        return st

    def _watcher_open(self, *, old_value=None, value=None, **kwargs):
        print("in open watcher", old_value, value)
        if self._target != 'Open':
            return
        if self._st is None:
            return

        if new_value:
            self._st._finished()
            self._target = None
            self._st = None
        print("in open watcher")

    def _watcher_close(self, *, old_value=None, value=None, **kwargs):
        print("in close watcher", old_value, value)
        if self._target != 'Close':
            return

        if self._st is None:
            return

        if new_value:
            self._st._finished()
            self._target = None
            self._st = None

        pass


class SavedImageSignal(Signal):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.stashed_datakey = {}

    def describe(self):
        ret = super().describe()
        ret[self.name].update(self.stashed_datakey)
        return ret


def take_dark(cam, light_field, dark_field_name):
    # close shutter

    # take the dark frame
    cam.stage()
    st = cam.trigger()
    while not st.done:
        ttime.sleep(.1)
    ret = cam.read()
    desc = cam.describe()
    cam.unstage()

    # save the df uid
    df = ret[light_field]
    df_sig = getattr(cam, dark_field_name)
    df_sig.put(**df)
    # save the darkfrom description
    df_sig.stashed_datakey = desc[light_field]




class XPDTIFFPlugin(TIFFPlugin, FileStoreTIFFSquashing,
                    FileStoreIterativeWrite):
    pass


class XPDHDF5Plugin(HDF5Plugin, FileStoreHDF5IterativeWrite):
    pass


class XPDPerkinElmer(PerkinElmerDetector):
    image = C(ImagePlugin, 'image1:')

    tiff = C(XPDTIFFPlugin, 'TIFF1:',
             write_path_template='H:/pe1_data/%Y/%m/%d/',
             read_path_template='/direct/XF28ID1/pe1_data/%Y/%m/%d/',
             cam_name='cam',  # used to configure "tiff squashing"
             proc_name='proc',  # ditto
             read_attrs=[],
             root='/direct/XF28ID1/',
             reg=db.reg)

    # hdf5 = C(XPDHDF5Plugin, 'HDF1:',
    #          write_path_template='G:/pe1_data/%Y/%m/%d/',
    #          read_path_template='/direct/XF28ID1/pe1_data/%Y/%m/%d/',
    #          root='/direct/XF28ID1/', reg=db.reg)

    proc = C(ProcessPlugin, 'Proc1:')

    # These attributes together replace `num_images`. They control
    # summing images before they are stored by the detector (a.k.a. "tiff
    # squashing").
    images_per_set = C(Signal, value=1, add_prefix=())
    number_of_sets = C(Signal, value=1, add_prefix=())

    stats1 = C(StatsPlugin, 'Stats1:')
    stats2 = C(StatsPlugin, 'Stats2:')
    stats3 = C(StatsPlugin, 'Stats3:')
    stats4 = C(StatsPlugin, 'Stats4:')
{'fraction_non_alphanumeric': 0.11420968600319319, 'fraction_numerical': 0.011708355508249068, 'mean_word_length': 2.1872455902306647, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1360743', 'n_tokens_mistral': 3063, 'n_tokens_neox': 2962, 'n_words': 626}