Skip to main content
`Ctrl`+`K`

# Source code for silx.io.fioh5

    # /*##########################################################################
    # Copyright (C) 2021 Timo Fuchs
    #
    # Permission is hereby granted, free of charge, to any person obtaining a copy
    # of this software and associated documentation files (the "Software"), to deal
    # in the Software without restriction, including without limitation the rights
    # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    # copies of the Software, and to permit persons to whom the Software is
    # furnished to do so, subject to the following conditions:
    #
    # The above copyright notice and this permission notice shall be included in
    # all copies or substantial portions of the Software.
    #
    # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
    # THE SOFTWARE.
    #
    # ############################################################################*/
    """This module provides a h5py-like API to access FioFile data.

    API description
    +++++++++++++++

    Fiofile data structure exposed by this API:

    ::

      /
          n.1/
              title = "…"
              start_time = "…"
              instrument/
                  fiofile/
                      comments = "…"
                      parameter = "…"
                  comment = "…"
                  parameter/
                      parameter_name = value

              measurement/
                  colname0 = …
                  colname1 = …
                  …

    The top level scan number ``n.1`` is determined from the filename as in
    ``prefix_n.fio``. (e.g. ``eh1_sixc_00045.fio`` would give ``45.1``)
    If no number is available, will use the filename instead.

    ``comments`` and ``parameter`` in group ``fiofile`` are the raw headers as they
    appear in the original file, as a string of lines separated by newline
    (``\\n``) characters. ``comment`` are the remaining comments,
    which were not parsed.

    The title is the content of the first comment header line
    (e.g ``"ascan  ss1vo -4.55687 -0.556875  40 0.2"``).
    The start_time is parsed from the second comment line.

    Datasets are stored in the data format specified in the fio file header.

    Scan data  (e.g. ``/1.1/measurement/colname0``) is accessed by column,
    the dataset name ``colname0`` being the column label as defined in the
    ``Col …`` header line.

    If a ``/`` character is present in a column label or in a motor name in the
    original FIO file, it will be substituted with a ``%`` character in the
    corresponding dataset name.

    MCA data is not yet supported.

    This reader requires a fio file as defined in
    src/sardana/macroserver/recorders/storage.py of the Sardana project
    (https://github.com/sardana-org/sardana).

    Accessing data
    ++++++++++++++

    Data and groups are accessed in :mod:`h5py` fashion::

        from silx.io.fioh5 import FioH5

        # Open a FioFile
        fiofh5 = FioH5("test_00056.fio")

        # using FioH5 as a regular group to access scans
        scan1group = fiofh5["56.1"]
        instrument_group = scan1group["instrument"]

        # alternative: full path access
        measurement_group = fiofh5["/56.1/measurement"]

        # accessing a scan data column by name as a 1D numpy array
        data_array = measurement_group["Pslit HGap"]

    :class:`FioH5` files and groups provide a :meth:`keys` method::

        >>> fiofh5.keys()
        ['96.1', '97.1', '98.1']
        >>> fiofh5['96.1'].keys()
        ['title', 'start_time', 'instrument', 'measurement']

    They can also be treated as iterators:

    .. code-block:: python

        from silx.io import is_dataset

        for scan_group in FioH5("test_00056.fio"):
            dataset_names = [item.name in scan_group["measurement"] if
                             is_dataset(item)]
            print("Found data columns in scan " + scan_group.name)
            print(", ".join(dataset_names))

    You can test for existence of data or groups::

        >>> "/1.1/measurement/Pslit HGap" in fiofh5
        True
        >>> "positioners" in fiofh5["/2.1/instrument"]
        True
        >>> "spam" in fiofh5["1.1"]
        False

    """

    __authors__ = ["T. Fuchs"]
    __license__ = "MIT"
    __date__ = "09/04/2021"

    import os

    import datetime
    import logging
    import io

    import h5py
    import numpy

    from silx import version as silx_version
    from . import commonh5

    from .spech5 import to_h5py_utf8

    logger1 = logging.getLogger(__name__)

    if h5py.version.version_tuple[0] < 3:
        text_dtype = h5py.special_dtype(vlen=str)  # old API
    else:
        text_dtype = "O"  # variable-length string (supported as of h5py > 3.0)

    ABORTLINENO = 5

    dtypeConverter = {
        "STRING": text_dtype,
        "DOUBLE": "f8",
        "FLOAT": "f4",
        "INTEGER": "i8",
        "BOOLEAN": "?",
    }

    def _bytestobool (val):
        """Convert bytes of a truth value to bool.

        Raises ValueError if 'val' is not supported.
        """
        if isinstance(val, bytes):
            val = val.decode()
        val.lower()
        if val in ('y', 'yes', 't', 'true', 'on', '1'):
            return True
        elif val in ('n', 'no', 'f', 'false', 'off', '0'):
            return False
        else:
            raise ValueError("Invalid truth value %r" % val)

    def is_fiofile(filename):
        """Test if a file is a FIO file, by checking if three consecutive lines
        start with *!*. Tests up to ABORTLINENO lines at the start of the file.

        :param str filename: File path
        :return: *True* if file is a FIO file, *False* if it is not a FIO file
        :rtype: bool
        """
        if not os.path.isfile(filename):
            return False
        # test for presence of three ! in first lines
        with open(filename, "rb") as f:
            chunk = f.read(2500)
        count = 0
        for i, line in enumerate(chunk.split(b"\n")):
            if line.startswith(b"!"):
                count += 1
                if count >= 3:
                    return True
            else:
                count = 0
            if i >= ABORTLINENO:
                break
        return False

    class FioFile(object):
        """This class opens a FIO file and reads the data."""

        def __init__(self, filepath):
            # parse filename
            filename = os.path.basename(filepath)
            fnowithsuffix = filename.split("_")[-1]
            try:
                self.scanno = int(fnowithsuffix.split(".")[0])
            except Exception:
                self.scanno = None
                logger1.warning("Cannot parse scan number of file %s", filename)

            with open(filepath, "r") as fiof:
                prev = 0
                line_counter = 0

                while True:
                    line = fiof.readline()
                    if line.startswith("!"):  # skip comments
                        prev = fiof.tell()
                        line_counter = 0
                        continue
                    if line.startswith("%c"):  # comment section
                        line_counter = 0
                        self.commentsection = ""
                        line = fiof.readline()
                        while not line.startswith("%") and not line.startswith("!"):
                            self.commentsection += line
                            prev = fiof.tell()
                            line = fiof.readline()
                    if line.startswith("%p"):  # parameter section
                        line_counter = 0
                        self.parameterssection = ""
                        line = fiof.readline()
                        while not line.startswith("%") and not line.startswith("!"):
                            self.parameterssection += line
                            prev = fiof.tell()
                            line = fiof.readline()
                    if line.startswith("%d"):  # data type definitions
                        line_counter = 0
                        self.datacols = []
                        self.names = []
                        self.dtypes = []
                        line = fiof.readline()
                        while line.startswith(" Col"):
                            splitline = line.split()
                            name = splitline[-2]
                            self.names.append(name)
                            dtype = dtypeConverter[splitline[-1]]
                            self.dtypes.append(dtype)
                            self.datacols.append((name, dtype))
                            prev = fiof.tell()
                            line = fiof.readline()
                        fiof.seek(prev)
                        break

                    line_counter += 1
                    if line_counter > ABORTLINENO:
                        raise IOError(
                            "Invalid fio file: Found no data "
                            "after %s lines" % ABORTLINENO
                        )
                np_datatype = \
                    numpy.dtype([(n, t) for (n,t) in zip(self.names, self.dtypes)])

                converter = {}
                for i, t in enumerate(self.dtypes):
                    if t == dtypeConverter['BOOLEAN']:
                        converter[i] = _bytestobool

                self.data = numpy.genfromtxt(
                    fiof,
                    dtype=np_datatype,
                    comments="!",
                    invalid_raise=True,
                    names=None,
                    deletechars='',
                    converters=converter
                )

                # ToDo: read only last line of file,
                # which sometimes contains the end of acquisition timestamp.

            self.parameter = {}

            # parse parameter section:
            try:
                for line in self.parameterssection.splitlines():
                    param, value = line.split(" = ")
                    self.parameter[param] = value
            except Exception:
                logger1.warning("Cannot parse parameter section")

            # parse default sardana comments: username and start time
            try:
                acquiMarker = "acquisition started at"  # indicates timestamp
                commentlines = self.commentsection.splitlines()
                if len(commentlines) >= 2:
                    self.title = commentlines[0]
                    l2 = commentlines[1]
                    acqpos = l2.lower().find(acquiMarker)
                    if acqpos < 0:
                        raise Exception("acquisition str not found")

                    self.user = l2[:acqpos][4:].strip()
                    self.start_time = l2[acqpos + len(acquiMarker) :].strip()
                    commentlines = commentlines[2:]
                self.comments = "\n".join(commentlines[2:])

            except Exception:
                logger1.warning("Cannot parse default comment section")
                self.comments = self.commentsection
                self.user = ""
                self.start_time = ""
                self.title = ""

    class FioH5NodeDataset(commonh5.Dataset):
        """This class inherits :class:`commonh5.Dataset`, to which it adds
        little extra functionality. The main additional functionality is the
        proxy behavior that allows to mimic the numpy array stored in this
        class.
        """

        def __init__(self, name, data, parent=None, attrs=None):
            # get proper value types, to inherit from numpy
            # attributes (dtype, shape, size)
            if isinstance(data, str):
                # use unicode (utf-8 when saved to HDF5 output)
                value = to_h5py_utf8(data)
            elif isinstance(data, float):
                # use 32 bits for float scalars
                value = numpy.float32(data)
            elif isinstance(data, int):
                value = numpy.int_(data)
            else:
                # Enforce numpy array
                array = numpy.array(data)
                data_kind = array.dtype.kind

                if data_kind in ["S", "U"]:
                    value = numpy.asarray(array, dtype=text_dtype)
                else:
                    value = array  # numerical data is already the correct datatype
            commonh5.Dataset.__init__(self, name, value, parent, attrs)

        def __getattr__(self, item):
            """Proxy to underlying numpy array methods."""
            if hasattr(self[()], item):
                return getattr(self[()], item)

            raise AttributeError("FioH5NodeDataset has no attribute %s" % item)

    class FioH5(commonh5.File):
        """This class reads a FIO file and exposes it as a *h5py.File*.

        It inherits :class:`silx.io.commonh5.Group` (via :class:`commonh5.File`),
        which implements most of its API.
        """

        def __init__(self, filename, order=1):
            """
            :param filename: Path to FioFile in filesystem
            :type filename: str
            """
            if isinstance(filename, io.IOBase):
                # see https://github.com/silx-kit/silx/issues/858
                filename = filename.name

            if not is_fiofile(filename):
                raise IOError("File %s is not a FIO file." % filename)

            try:
                fiof = FioFile(filename)  # reads complete file
            except Exception as e:
                raise IOError("FIO file %s cannot be read." % filename) from e

            attrs = {
                "NX_class": to_h5py_utf8("NXroot"),
                "file_time": to_h5py_utf8(datetime.datetime.now().isoformat()),
                "file_name": to_h5py_utf8(filename),
                "creator": to_h5py_utf8("silx fioh5 %s" % silx_version),
            }
            commonh5.File.__init__(self, filename, attrs=attrs)

            if fiof.scanno is not None:
                scan_key = "%s.%s" % (fiof.scanno, int(order))
            else:
                scan_key = os.path.splitext(os.path.basename(filename))[0]

            scan_group = FioScanGroup(scan_key, parent=self, scan=fiof)
            self.add_node(scan_group)

    class FioScanGroup(commonh5.Group):
        def __init__(self, scan_key, parent, scan):
            """

            :param parent: parent Group
            :param str scan_key: Scan key (e.g. "1.1")
            :param scan: FioFile object
            """
            if hasattr(scan, "user"):
                userattr = to_h5py_utf8(scan.user)
            else:
                userattr = to_h5py_utf8("")
            commonh5.Group.__init__(
                self,
                scan_key,
                parent=parent,
                attrs={"NX_class": to_h5py_utf8("NXentry"), "user": userattr},
            )

            # 'title', 'start_time' and 'user' are defaults
            # in Sardana created files:
            if hasattr(scan, "title"):
                title = scan.title
            else:
                title = scan_key  # use scan number as default title
            self.add_node(
                FioH5NodeDataset(name="title", data=to_h5py_utf8(title), parent=self)
            )

            if hasattr(scan, "start_time"):
                start_time = scan.start_time
                self.add_node(
                    FioH5NodeDataset(
                        name="start_time", data=to_h5py_utf8(start_time), parent=self
                    )
                )

            self.add_node(
                FioH5NodeDataset(
                    name="comments", data=to_h5py_utf8(scan.comments), parent=self
                )
            )

            self.add_node(FioInstrumentGroup(parent=self, scan=scan))
            self.add_node(FioMeasurementGroup(parent=self, scan=scan))

    class FioMeasurementGroup(commonh5.Group):
        def __init__(self, parent, scan):
            """

            :param parent: parent Group
            :param scan: FioFile object
            """
            commonh5.Group.__init__(
                self,
                name="measurement",
                parent=parent,
                attrs={"NX_class": to_h5py_utf8("NXcollection")},
            )

            for label in scan.names:
                safe_label = label.replace("/", "%")
                self.add_node(
                    FioH5NodeDataset(name=safe_label, data=scan.data[label], parent=self)
                )

    class FioInstrumentGroup(commonh5.Group):
        def __init__(self, parent, scan):
            """

            :param parent: parent Group
            :param scan: FioFile object
            """
            commonh5.Group.__init__(
                self,
                name="instrument",
                parent=parent,
                attrs={"NX_class": to_h5py_utf8("NXinstrument")},
            )

            self.add_node(FioParameterGroup(parent=self, scan=scan))
            self.add_node(FioFileGroup(parent=self, scan=scan))
            self.add_node(
                FioH5NodeDataset(
                    name="comment", data=to_h5py_utf8(scan.comments), parent=self
                )
            )

    class FioFileGroup(commonh5.Group):
        def __init__(self, parent, scan):
            """

            :param parent: parent Group
            :param scan: FioFile object
            """
            commonh5.Group.__init__(
                self,
                name="fiofile",
                parent=parent,
                attrs={"NX_class": to_h5py_utf8("NXcollection")},
            )

            self.add_node(
                FioH5NodeDataset(
                    name="comments", data=to_h5py_utf8(scan.commentsection), parent=self
                )
            )

            self.add_node(
                FioH5NodeDataset(
                    name="parameter", data=to_h5py_utf8(scan.parameterssection), parent=self
                )
            )

    class FioParameterGroup(commonh5.Group):
        def __init__(self, parent, scan):
            """

            :param parent: parent Group
            :param scan: FioFile object
            """
            commonh5.Group.__init__(
                self,
                name="parameter",
                parent=parent,
                attrs={"NX_class": to_h5py_utf8("NXcollection")},
            )

            for label in scan.parameter:
                safe_label = label.replace("/", "%")
                self.add_node(
                    FioH5NodeDataset(
                        name=safe_label,
                        data=to_h5py_utf8(scan.parameter[label]),
                        parent=self,
                    )
                )