<commit_before>"""Implementation of odin_data Meta Writer

This module is passed meta data messages for a single acquisition which it writes to disk.
Will need to be subclassed by detector specific implementation.

<NAME>, Diamond Light Source
"""
import os
from time import time
import logging

import h5py

from odin_data import _version as versioneer
from odin_data.util import construct_version_dict
from .hdf5dataset import HDF5Dataset, Int64HDF5Dataset

# Data message parameters
FRAME = "frame"
OFFSET = "offset"
RANK = "rank"
CREATE_DURATION = "create_duration"
WRITE_DURATION = "write_duration"
FLUSH_DURATION = "flush_duration"
CLOSE_DURATION = "close_duration"
MESSAGE_TYPE_ID = "parameter"

# Configuration items
DIRECTORY = "directory"
FILE_PREFIX = "file_prefix"
FLUSH_FRAME_FREQUENCY = "flush_frame_frequency"
FLUSH_TIMEOUT = "flush_timeout"


class MetaWriterConfig(object):

    def __init__(self, sensor_shape):
        """
        Args:
            sensor_shape(tuple): Detector sensor size (y, x)

        """
        self.sensor_shape = sensor_shape


def require_open_hdf5_file(func):
    """A decorator to verify the HDF5 file is open before calling the wrapped method

    If the HDF5 file is currently open for writing, call the method, else log the reason
    the file is not open

    NOTE: This should only be used on MetaWriter methods (that take self as the first
    argument)

    """

    def wrapper(*args, **kwargs):
        writer = args[0]  # Extract class instance (self) from args

        if writer.file_open:
            # It is safe to call the wrapped method
            return func(*args, **kwargs)

        # It is not safe to call the wrapped method - log the reason why
        if writer.finished:
            reason = "Already finished writing"
        else:
            reason = "Have not received startacquisition yet"
        writer._logger.error(
            "%s | Cannot call %s - File not open - %s",
            writer._name,
            func.__name__,
            reason,
        )

    return wrapper


class MetaWriter(object):
    """This class handles meta data messages and writes parameters to disk"""

    FILE_SUFFIX = "_meta.h5"
    CONFIGURE_PARAMETERS = [
        DIRECTORY,
        FILE_PREFIX,
        FLUSH_FRAME_FREQUENCY,
        FLUSH_TIMEOUT,
    ]
    # Detector-specific parameters received on per-frame meta message
    DETECTOR_WRITE_FRAME_PARAMETERS = []

    def __init__(self, name, directory, endpoints, config):
        """
        Args:
            name(str): Unique name to construct file path and to include in
                       log messages
            directory(str): Directory to create the meta file in
            endpoints(list): Endpoints parent MetaListener will receive data on
            config(MetaWriterConfig): Configuration options

        """
        self._logger = logging.getLogger(self.__class__.__name__)

        # Config
        self.directory = directory
        self.file_prefix = None
        self.flush_frame_frequency = 100
        self.flush_timeout = 1

        # Status
        self.full_file_path = None
        self.write_count = 0
        self.finished = False
        self.write_timeout_count = 0

        # Internal parameters
        self._name = name
        self._processes_running = [False] * len(endpoints)
        self._config = config
        self._last_flushed = time()  # Seconds since epoch
        self._frames_since_flush = 0
        self._hdf5_file = None
        self._datasets = dict(
            (dataset.name, dataset)
            for dataset in self._define_datasets() + self._define_detector_datasets()
        )
        # Child class parameters
        self._frame_data_map = dict()  # Map of frame number to detector data
        self._frame_offset_map = dict()  # Map of frame number to offset in dataset
        self._writers_finished = False
        self._detector_finished = True  # See stop_when_detector_finished

    @staticmethod
    def _define_datasets():
        return [
            Int64HDF5Dataset(FRAME),
            Int64HDF5Dataset(OFFSET),
            Int64HDF5Dataset(CREATE_DURATION, cache=False),
            Int64HDF5Dataset(WRITE_DURATION),
            Int64HDF5Dataset(FLUSH_DURATION),
            Int64HDF5Dataset(CLOSE_DURATION, cache=False),
        ]

    @staticmethod
    def _define_detector_datasets():
        return []

    @property
    def file_open(self):
        return self._hdf5_file is not None

    @property
    def active_process_count(self):
        return self._processes_running.count(True)

    def _generate_full_file_path(self):
        prefix = self.file_prefix if self.file_prefix is not None else self._name
        self.full_file_path = os.path.join(
            self.directory, "{}{}".format(prefix, self.FILE_SUFFIX)
        )
        return self.full_file_path

    def _create_file(self, file_path, dataset_size):
        self._logger.debug(
            "%s | Opening file %s - Expecting %d frames",<commit_msg>Make _define_detector_datasets an instance method<commit_after>"""Implementation of odin_data Meta Writer

This module is passed meta data messages for a single acquisition which it writes to disk.
Will need to be subclassed by detector specific implementation.

<NAME>, Diamond Light Source
"""
import os
from time import time
import logging

import h5py

from odin_data import _version as versioneer
from odin_data.util import construct_version_dict
from .hdf5dataset import HDF5Dataset, Int64HDF5Dataset

# Data message parameters
FRAME = "frame"
OFFSET = "offset"
RANK = "rank"
CREATE_DURATION = "create_duration"
WRITE_DURATION = "write_duration"
FLUSH_DURATION = "flush_duration"
CLOSE_DURATION = "close_duration"
MESSAGE_TYPE_ID = "parameter"

# Configuration items
DIRECTORY = "directory"
FILE_PREFIX = "file_prefix"
FLUSH_FRAME_FREQUENCY = "flush_frame_frequency"
FLUSH_TIMEOUT = "flush_timeout"


class MetaWriterConfig(object):

    def __init__(self, sensor_shape):
        """
        Args:
            sensor_shape(tuple): Detector sensor size (y, x)

        """
        self.sensor_shape = sensor_shape


def require_open_hdf5_file(func):
    """A decorator to verify the HDF5 file is open before calling the wrapped method

    If the HDF5 file is currently open for writing, call the method, else log the reason
    the file is not open

    NOTE: This should only be used on MetaWriter methods (that take self as the first
    argument)

    """

    def wrapper(*args, **kwargs):
        writer = args[0]  # Extract class instance (self) from args

        if writer.file_open:
            # It is safe to call the wrapped method
            return func(*args, **kwargs)

        # It is not safe to call the wrapped method - log the reason why
        if writer.finished:
            reason = "Already finished writing"
        else:
            reason = "Have not received startacquisition yet"
        writer._logger.error(
            "%s | Cannot call %s - File not open - %s",
            writer._name,
            func.__name__,
            reason,
        )

    return wrapper


class MetaWriter(object):
    """This class handles meta data messages and writes parameters to disk"""

    FILE_SUFFIX = "_meta.h5"
    CONFIGURE_PARAMETERS = [
        DIRECTORY,
        FILE_PREFIX,
        FLUSH_FRAME_FREQUENCY,
        FLUSH_TIMEOUT,
    ]
    # Detector-specific parameters received on per-frame meta message
    DETECTOR_WRITE_FRAME_PARAMETERS = []

    def __init__(self, name, directory, endpoints, config):
        """
        Args:
            name(str): Unique name to construct file path and to include in
                       log messages
            directory(str): Directory to create the meta file in
            endpoints(list): Endpoints parent MetaListener will receive data on
            config(MetaWriterConfig): Configuration options

        """
        self._logger = logging.getLogger(self.__class__.__name__)

        # Config
        self.directory = directory
        self.file_prefix = None
        self.flush_frame_frequency = 100
        self.flush_timeout = 1

        # Status
        self.full_file_path = None
        self.write_count = 0
        self.finished = False
        self.write_timeout_count = 0

        # Internal parameters
        self._name = name
        self._processes_running = [False] * len(endpoints)
        self._config = config
        self._last_flushed = time()  # Seconds since epoch
        self._frames_since_flush = 0
        self._hdf5_file = None
        self._datasets = dict(
            (dataset.name, dataset)
            for dataset in self._define_datasets() + self._define_detector_datasets()
        )
        # Child class parameters
        self._frame_data_map = dict()  # Map of frame number to detector data
        self._frame_offset_map = dict()  # Map of frame number to offset in dataset
        self._writers_finished = False
        self._detector_finished = True  # See stop_when_detector_finished

    @staticmethod
    def _define_datasets():
        return [
            Int64HDF5Dataset(FRAME),
            Int64HDF5Dataset(OFFSET),
            Int64HDF5Dataset(CREATE_DURATION, cache=False),
            Int64HDF5Dataset(WRITE_DURATION),
            Int64HDF5Dataset(FLUSH_DURATION),
            Int64HDF5Dataset(CLOSE_DURATION, cache=False),
        ]

    def _define_detector_datasets(self):
        return []

    @property
    def file_open(self):
        return self._hdf5_file is not None

    @property
    def active_process_count(self):
        return self._processes_running.count(True)

    def _generate_full_file_path(self):
        prefix = self.file_prefix if self.file_prefix is not None else self._name
        self.full_file_path = os.path.join(
            self.directory, "{}{}".format(prefix, self.FILE_SUFFIX)
        )
        return self.full_file_path

    def _create_file(self, file_path, dataset_size):
        self._logger.debug(
            "%s | Opening file %s - Expecting %d frames",
{'fraction_non_alphanumeric': 0.07169848846459825, 'fraction_numerical': 0.007557677008750994, 'mean_word_length': 2.538705137227305, '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': '5520305', 'n_tokens_mistral': 2865, 'n_tokens_neox': 2788, 'n_words': 844}