<commit_before>import os
from xml.etree import cElementTree as ET

from malcolm.compat import et_to_string
from malcolm.core import method_takes, REQUIRED, Info
from malcolm.core.vmetas import StringMeta, PointGeneratorMeta
from malcolm.parts.builtin.childpart import ChildPart
from malcolm.controllers.runnablecontroller import RunnableController
from malcolm.parts.ADCore.datasettablepart import DatasetProducedInfo, \
    dataset_types


SUFFIXES = "NXY3456789"

attribute_dataset_types = ["detector", "monitor", "position"]


# Produced by plugins in part_info
class NDAttributeDatasetInfo(Info):
    def __init__(self, name, type, attr, rank):
        self.name = name
        assert type in attribute_dataset_types, \
            "Dataset type %s not in %s" % (type, dataset_types)
        self.type = type
        self.attr = attr
        self.rank = rank


class CalculatedNDAttributeDatasetInfo(Info):
    def __init__(self, name, attr):
        self.name = name
        self.attr = attr


class NDArrayDatasetInfo(Info):
    def __init__(self, name, rank):
        self.name = name
        self.rank = rank


class HDFWriterPart(ChildPart):
    # Attributes
    datasets = None

    # Future for the start action
    start_future = None
    array_future = None
    done_when_reaches = 0

    # The HDF5 layout file we write to say where the datasets go
    layout_filename = None

    def _create_dataset_infos(self, part_info, generator, filename):
        # Update the dataset table
        uniqueid = "/entry/NDAttributes/NDArrayUniqueId"
        generator_rank = len(generator.dimensions)

        # Get the detector name from the primary source
        ndarray_infos = NDArrayDatasetInfo.filter_values(part_info)
        assert len(ndarray_infos) in (0, 1), \
            "More than one NDArrayDatasetInfo defined %s" % ndarray_infos

        # Add the primary datasource
        if ndarray_infos:
            ndarray_info = ndarray_infos[0]
            yield DatasetProducedInfo(
                name="%s.data" % ndarray_info.name, filename=filename,
                type="primary", rank=ndarray_info.rank + generator_rank,
                path="/entry/detector/detector",
                uniqueid=uniqueid)

            # Add any secondary datasources
            for calculated_info in \
                    CalculatedNDAttributeDatasetInfo.filter_values(part_info):
                yield DatasetProducedInfo(
                    name="%s.%s" % (ndarray_info.name, calculated_info.name),
                    filename=filename, type="secondary",
                    rank=ndarray_info.rank + generator_rank,
                    path="/entry/%s/%s" % (
                        calculated_info.name, calculated_info.name),
                    uniqueid=uniqueid)

        # Add all the other datasources
        for dataset_info in NDAttributeDatasetInfo.filter_values(part_info):
            if dataset_info.type == "detector":
                # Something like I0
                name = "%s.data" % dataset_info.name
                type = "primary"
            elif dataset_info.type == "monitor":
                # Something like Iref
                name = "%s.data" % dataset_info.name
                type = "monitor"
            elif dataset_info.type == "position":
                # Something like x
                name = "%s.value" % dataset_info.name
                type = "position_value"
            else:
                raise AttributeError("Bad dataset type %r, should be in %s" % (
                    dataset_info.type, attribute_dataset_types))
            yield DatasetProducedInfo(
                name=name, filename=filename, type=type,
                rank=dataset_info.rank + generator_rank,
                path="/entry/%s/%s" % (dataset_info.name, dataset_info.name),
                uniqueid=uniqueid)

        # Add any setpoint dimensions
        for dim in generator.axes:
            yield DatasetProducedInfo(
                name="%s.value_set" % dim, filename=filename,
                type="position_set", rank=1,
                path="/entry/detector/%s_set" % dim, uniqueid="")

    @RunnableController.Reset
    def reset(self, task):
        super(HDFWriterPart, self).reset(task)
        self.abort(task)

    @RunnableController.Configure
    @method_takes(
        "generator", PointGeneratorMeta("Generator instance"), REQUIRED,
        "filePath", StringMeta("Full file path to write data to"), REQUIRED,
        "fileTemplate", StringMeta(
            """Printf style template to generate full path from. Arguments are:
            1) %s: directory name including trailing slash from filePath
            2) %s: file name from filePath
            """), "%s%s")
    def configure(self, task, completed_steps, steps_to_do, part_info, params):
        self.done_when_reaches = completed_steps + steps_to_do
        # For first run then open the file
        # Enable position mode before setting any position related things
        task.put(self.child["positionMode"], True)
        # Setup our required settings
        # TODO: this should be different for windows detectors
        file_path = params.filePath.rstrip(os.sep)
        file_dir, filename = file_path.rsplit(os.sep, 1)
        assert "." in filename, \
            "File extension for %r should be supplied" % filename
        futures = task.put_many_async(self.child, dict(
            enableCallbacks=True,
            fileWriteMode="Stream",
            swmrMode=True,
            positionMode=True,
            dimAttDatasets=True,
            lazyOpen=True,
            arrayCounter=0,
            filePath=file_dir + os.sep,
            fileName=filename,
            fileTemplate=params.fileTemplate))
        futures += self._set_dimensions(task, params.generator)
        xml = self._make_layout_xml(params.generator, part_info)
        self.layout_filename = os.path.join(
            file_dir, "%s-layout.xml" % self.params.mri)
        open(self.layout_filename, "w").write(xml)
        futures += task.put_async(self.child["xml"], self.layout_filename)
        # Wait for the previous puts to finish
        task.wait_all(futures)
        # Reset numCapture back to 0
        task.put(self.child["numCapture"], 0)
        # We want the HDF writer to flush this often:
        flush_time = 1  # seconds
        # (In particular this means that HDF files can be read cleanly by
        # SciSoft at the start of a scan.)
        assert params.generator.duration > 0, \
            "Duration %s for generator must be >0 to signify constant exposure"\
            % params.generator.duration
        n_frames_between_flushes = max(2, round(
            flush_time/params.generator.duration))
        task.put(self.child["flushDataPerNFrames"], n_frames_between_flushes)
        task.put(self.child["flushAttrPerNFrames"], n_frames_between_flushes)
        # Start the plugin
        self.start_future = task.post_async(self.child["start"])
        # Start a future waiting for the first array
        self.array_future = task.when_matches_async(
            self.child["arrayCounter"], 1)
        # Return the dataset information
        dataset_infos = list(self._create_dataset_infos(
            part_info, params.generator, filename))
        return dataset_infos

    @RunnableController.PostRunReady
    @RunnableController.Seek
    def seek(self, task, completed_steps, steps_to_do, part_info):
        self.done_when_reaches = completed_steps + steps_to_do
        # Just reset the array counter
        task.put(self.child["arrayCounter"], 0)
        # Start a future waiting for the first array
        self.array_future = task.when_matches_async(
            self.child["arrayCounter"], 1)

    @RunnableController.Run
    @RunnableController.Resume
    def run(self, task, update_completed_steps):
        task.wait_all(self.array_future)
        task.unsubscribe_all()
        task.subscribe(self.child["uniqueId"], update_completed_steps, self)
        # TODO: what happens if we miss the last frame?
        task.when_matches(self.child["uniqueId"], self.done_when_reaches)
<commit_msg>Update HDFWriterPart frames per flush calculation<commit_after>import os
import math as m
from xml.etree import cElementTree as ET

from malcolm.compat import et_to_string
from malcolm.core import method_takes, REQUIRED, Info
from malcolm.core.vmetas import StringMeta, PointGeneratorMeta
from malcolm.parts.builtin.childpart import ChildPart
from malcolm.controllers.runnablecontroller import RunnableController
from malcolm.parts.ADCore.datasettablepart import DatasetProducedInfo, \
    dataset_types


SUFFIXES = "NXY3456789"

attribute_dataset_types = ["detector", "monitor", "position"]


# Produced by plugins in part_info
class NDAttributeDatasetInfo(Info):
    def __init__(self, name, type, attr, rank):
        self.name = name
        assert type in attribute_dataset_types, \
            "Dataset type %s not in %s" % (type, dataset_types)
        self.type = type
        self.attr = attr
        self.rank = rank


class CalculatedNDAttributeDatasetInfo(Info):
    def __init__(self, name, attr):
        self.name = name
        self.attr = attr


class NDArrayDatasetInfo(Info):
    def __init__(self, name, rank):
        self.name = name
        self.rank = rank


class HDFWriterPart(ChildPart):
    # Attributes
    datasets = None

    # Future for the start action
    start_future = None
    array_future = None
    done_when_reaches = 0

    # The HDF5 layout file we write to say where the datasets go
    layout_filename = None

    def _create_dataset_infos(self, part_info, generator, filename):
        # Update the dataset table
        uniqueid = "/entry/NDAttributes/NDArrayUniqueId"
        generator_rank = len(generator.dimensions)

        # Get the detector name from the primary source
        ndarray_infos = NDArrayDatasetInfo.filter_values(part_info)
        assert len(ndarray_infos) in (0, 1), \
            "More than one NDArrayDatasetInfo defined %s" % ndarray_infos

        # Add the primary datasource
        if ndarray_infos:
            ndarray_info = ndarray_infos[0]
            yield DatasetProducedInfo(
                name="%s.data" % ndarray_info.name, filename=filename,
                type="primary", rank=ndarray_info.rank + generator_rank,
                path="/entry/detector/detector",
                uniqueid=uniqueid)

            # Add any secondary datasources
            for calculated_info in \
                    CalculatedNDAttributeDatasetInfo.filter_values(part_info):
                yield DatasetProducedInfo(
                    name="%s.%s" % (ndarray_info.name, calculated_info.name),
                    filename=filename, type="secondary",
                    rank=ndarray_info.rank + generator_rank,
                    path="/entry/%s/%s" % (
                        calculated_info.name, calculated_info.name),
                    uniqueid=uniqueid)

        # Add all the other datasources
        for dataset_info in NDAttributeDatasetInfo.filter_values(part_info):
            if dataset_info.type == "detector":
                # Something like I0
                name = "%s.data" % dataset_info.name
                type = "primary"
            elif dataset_info.type == "monitor":
                # Something like Iref
                name = "%s.data" % dataset_info.name
                type = "monitor"
            elif dataset_info.type == "position":
                # Something like x
                name = "%s.value" % dataset_info.name
                type = "position_value"
            else:
                raise AttributeError("Bad dataset type %r, should be in %s" % (
                    dataset_info.type, attribute_dataset_types))
            yield DatasetProducedInfo(
                name=name, filename=filename, type=type,
                rank=dataset_info.rank + generator_rank,
                path="/entry/%s/%s" % (dataset_info.name, dataset_info.name),
                uniqueid=uniqueid)

        # Add any setpoint dimensions
        for dim in generator.axes:
            yield DatasetProducedInfo(
                name="%s.value_set" % dim, filename=filename,
                type="position_set", rank=1,
                path="/entry/detector/%s_set" % dim, uniqueid="")

    @RunnableController.Reset
    def reset(self, task):
        super(HDFWriterPart, self).reset(task)
        self.abort(task)

    @RunnableController.Configure
    @method_takes(
        "generator", PointGeneratorMeta("Generator instance"), REQUIRED,
        "filePath", StringMeta("Full file path to write data to"), REQUIRED,
        "fileTemplate", StringMeta(
            """Printf style template to generate full path from. Arguments are:
            1) %s: directory name including trailing slash from filePath
            2) %s: file name from filePath
            """), "%s%s")
    def configure(self, task, completed_steps, steps_to_do, part_info, params):
        self.done_when_reaches = completed_steps + steps_to_do
        # For first run then open the file
        # Enable position mode before setting any position related things
        task.put(self.child["positionMode"], True)
        # Setup our required settings
        # TODO: this should be different for windows detectors
        file_path = params.filePath.rstrip(os.sep)
        file_dir, filename = file_path.rsplit(os.sep, 1)
        assert "." in filename, \
            "File extension for %r should be supplied" % filename
        futures = task.put_many_async(self.child, dict(
            enableCallbacks=True,
            fileWriteMode="Stream",
            swmrMode=True,
            positionMode=True,
            dimAttDatasets=True,
            lazyOpen=True,
            arrayCounter=0,
            filePath=file_dir + os.sep,
            fileName=filename,
            fileTemplate=params.fileTemplate))
        futures += self._set_dimensions(task, params.generator)
        xml = self._make_layout_xml(params.generator, part_info)
        self.layout_filename = os.path.join(
            file_dir, "%s-layout.xml" % self.params.mri)
        open(self.layout_filename, "w").write(xml)
        futures += task.put_async(self.child["xml"], self.layout_filename)
        # Wait for the previous puts to finish
        task.wait_all(futures)
        # Reset numCapture back to 0
        task.put(self.child["numCapture"], 0)
        assert params.generator.duration > 0, \
            "Duration %s for generator must be >0 to signify constant " \
            "exposure" % params.generator.duration
        # We want the HDF writer to flush this often:
        flush_time = 1  # seconds
        # (In particular this means that HDF files can be read cleanly by
        # SciSoft at the start of a scan.)
        if params.generator.duration > flush_time:
            # We are going slower than 1/flush_time Hz, so flush every frame
            n_frames_between_flushes = 1
        else:
            # Limit update rate to be every flush_time seconds
            n_frames_between_flushes = int(m.ceil(
                flush_time / params.generator.duration))
            # But make sure we flush in this round of frames
            n_frames_between_flushes = min(
                steps_to_do, n_frames_between_flushes)
        task.put(self.child["flushDataPerNFrames"], n_frames_between_flushes)
        task.put(self.child["flushAttrPerNFrames"], n_frames_between_flushes)
        # Start the plugin
        self.start_future = task.post_async(self.child["start"])
        # Start a future waiting for the first array
        self.array_future = task.when_matches_async(
            self.child["arrayCounter"], 1)
        # Return the dataset information
        dataset_infos = list(self._create_dataset_infos(
            part_info, params.generator, filename))
        return dataset_infos

    @RunnableController.PostRunReady
    @RunnableController.Seek
    def seek(self, task, completed_steps, steps_to_do, part_info):
        self.done_when_reaches = completed_steps + steps_to_do
        # Just reset the array counter
        task.put(self.child["arrayCounter"], 0)
        # Start a future waiting for the first array
        self.array_future = task.when_matches_async(
            self.child["arrayCounter"], 1)

    @RunnableController.Run
    @RunnableController.Resume
    def run(self, task, update_completed_steps):
        task.wait_all(self.array_future)
        task.unsubscribe_all()
        task.subscribe(self.child["uniqueId"], update_completed_steps, self)
        # TODO: what happens if we miss the last frame?
        task.when_matches(self.child["uniqueId"], self.done_when_reaches)

{'fraction_non_alphanumeric': 0.08489157058366689, 'fraction_numerical': 0.003285739888882251, 'mean_word_length': 2.3426517571884986, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, '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': '3613931', 'n_tokens_mistral': 4687, 'n_tokens_neox': 4552, 'n_words': 1184}