class TdmsFile(object):
    """ Reads and stores data from a TDMS file.

    There are two main ways to create a new TdmsFile object.
    TdmsFile.read will read all data into memory::

        tdms_file = TdmsFile.read(tdms_file_path)

    or you can use TdmsFile.open to read file metadata but not immediately read all data,
    for cases where a file is too large to easily fit in memory or you don't need to
    read data for all channels::

        with TdmsFile.open(tdms_file_path) as tdms_file:
            # Use tdms_file
            ...

    This class acts like a dictionary, where the keys are names of groups in the TDMS
    files and the values are TdmsGroup objects.
    A TdmsFile can be indexed by group name to access a group within the TDMS file, for example::

        tdms_file = TdmsFile.read(tdms_file_path)
        group = tdms_file[group_name]

    Iterating over a TdmsFile produces the names of groups in this file,
    or you can use the groups method to directly access all groups::

        for group in tdms_file.groups():
            # Use group
            ...
    """

    @staticmethod
    def read(file, raw_timestamps=False, memmap_dir=None):
        """ Creates a new TdmsFile object and reads all data in the file

        :param file: Either the path to the tdms file to read
            as a string or pathlib.Path, or an already opened file.
        :param raw_timestamps: By default TDMS timestamps are read as numpy datetime64
            but this loses some precision.
            Setting this to true will read timestamps as a custom TdmsTimestamp type.
        :param memmap_dir: The directory to store memory mapped data files in,
            or None to read data into memory. The data files are created
            as temporary files and are deleted when the channel data is no
            longer used. tempfile.gettempdir() can be used to get the default
            temporary file directory.
        """
        return TdmsFile(file, raw_timestamps=raw_timestamps, memmap_dir=memmap_dir)

    @staticmethod
    def open(file, raw_timestamps=False, memmap_dir=None):
        """ Creates a new TdmsFile object and reads metadata, leaving the file open
            to allow reading channel data

        :param file: Either the path to the tdms file to read
            as a string or pathlib.Path, or an already opened file.
        :param raw_timestamps: By default TDMS timestamps are read as numpy datetime64
            but this loses some precision.
            Setting this to true will read timestamps as a custom TdmsTimestamp type.
        :param memmap_dir: The directory to store memory mapped data files in,
            or None to read data into memory. The data files are created
            as temporary files and are deleted when the channel data is no
            longer used. tempfile.gettempdir() can be used to get the default
            temporary file directory.
        """
        return TdmsFile(
            file, raw_timestamps=raw_timestamps, memmap_dir=memmap_dir, read_metadata_only=True, keep_open=True)

    @staticmethod
    def read_metadata(file, raw_timestamps=False):
        """ Creates a new TdmsFile object and only reads the metadata

        :param file: Either the path to the tdms file to read
            as a string or pathlib.Path, or an already opened file.
        :param raw_timestamps: By default TDMS timestamps are read as numpy datetime64
            but this loses some precision.
            Setting this to true will read timestamps as a custom TdmsTimestamp type.
        """
        return TdmsFile(file, raw_timestamps=raw_timestamps, read_metadata_only=True)

    def __init__(self, file, raw_timestamps=False, memmap_dir=None, read_metadata_only=False, keep_open=False):
        """Initialise a new TdmsFile object

        :param file: Either the path to the tdms file to read
            as a string or pathlib.Path, or an already opened file.
        :param raw_timestamps: By default TDMS timestamps are read as numpy datetime64
            but this loses some precision.
            Setting this to true will read timestamps as a custom TdmsTimestamp type.
        :param memmap_dir: The directory to store memory mapped data files in,
            or None to read data into memory. The data files are created
            as temporary files and are deleted when the channel data is no
            longer used. tempfile.gettempdir() can be used to get the default
            temporary file directory.
        :param read_metadata_only: If this parameter is enabled then only the
            metadata of the TDMS file will read.
        :param keep_open: Keeps the file open so data can be read if only metadata
            is read initially.
        """

        self._memmap_dir = memmap_dir
        self._raw_timestamps = raw_timestamps
        self._groups = OrderedDict()
        self._properties = OrderedDict()
        self._channel_data = {}
        self._tdms_version = 0
        self.data_read = False

        self._reader = TdmsReader(file)
        try:
            self._read_file(self._reader, read_metadata_only, keep_open)
        finally:
            if not keep_open:
                self._reader.close()

    def groups(self):
        """Returns a list of the groups in this file

        :rtype: List of TdmsGroup.
        """

        return list(self._groups.values())

    @property
    def tdms_version(self):
        """ The TDMS format version of this file
        """
        return self._tdms_version

    @property
    def properties(self):
        """ Return the properties of this file as a dictionary

        These are the properties associated with the root TDMS object.
        """

        return self._properties

    def as_dataframe(self, time_index=False, absolute_time=False, scaled_data=True):
        """
        Converts the TDMS file to a DataFrame. DataFrame columns are named using the TDMS object paths.

        :param time_index: Whether to include a time index for the dataframe.
        :param absolute_time: If time_index is true, whether the time index
            values are absolute times or relative to the start time.
        :param scaled_data: By default the scaled data will be used.
            Set to False to use raw unscaled data.
            For DAQmx data, there will be one column per DAQmx raw scaler and column names will include the scale id.
        :return: The full TDMS file data.
        :rtype: pandas.DataFrame
        """

        return pandas_export.from_tdms_file(self, time_index, absolute_time, scaled_data)

    def as_hdf(self, filepath, mode='w', group='/'):
        """
        Converts the TDMS file into an HDF5 file

        :param filepath: The path of the HDF5 file you want to write to.
        :param mode: The write mode of the HDF5 file. This can be 'w' or 'a'
        :param group: A group in the HDF5 file that will contain the TDMS data.
        """
        return hdf_export.from_tdms_file(self, filepath, mode, group)

    def data_chunks(self):
        """ A generator that streams chunks of data from disk.
        This method may only be used when the TDMS file was opened without reading all data immediately.

        :rtype: Generator that yields :class:`DataChunk` objects
        """
        channel_offsets = defaultdict(int)
        for chunk in self._reader.read_raw_data():
            _convert_data_chunk(chunk, self._raw_timestamps)
            yield DataChunk(self, chunk, channel_offsets)
            for path, data in chunk.channel_data.items():
                channel_offsets[path] += len(data)

    def close(self):
        """ Close the underlying file if it was opened by this TdmsFile

            If this TdmsFile was initialised with an already open file
            then the reference to it is released but the file is not closed.
        """
        if self._reader is not None:
            self._reader.close()
            self._reader = None

    def __len__(self):
        """ Returns the number of groups in this file
        """
        return len(self._groups)

    def __iter__(self):
        """ Returns an iterator over the names of groups in this file
        """
        return iter(self._groups)

    def __getitem__(self, group_name):
        """ Retrieve a TDMS group from the file by name
        """
        try:
            return self._groups[group_name]
        except KeyError:
            raise KeyError("There is no group named '%s' in the TDMS file" % group_name)

    def _ipython_key_completions_(self):
        """ Return possible group names for tab-completion when indexing
        """
        return list(self._groups.keys())

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def _read_file(self, tdms_reader, read_metadata_only, keep_open):
        tdms_reader.read_metadata(require_segment_indexes=keep_open)
        self._tdms_version = tdms_reader.tdms_version

        # Use object metadata to build group and channel objects
        group_properties = OrderedDict()
        group_channels = OrderedDict()
        object_properties = {
            path_string: self._convert_properties(obj.properties)
            for path_string, obj in tdms_reader.object_metadata.items()}
        try:
            self._properties = object_properties['/']
        except KeyError:
            pass

        for (path_string, obj) in tdms_reader.object_metadata.items():
            properties = object_properties[path_string]
            path = ObjectPath.from_string(path_string)
            if path.is_root:
                pass
            elif path.is_group:
                group_properties[path.group] = properties
            else:
                # Object is a channel
                try:
                    channel_group_properties = object_properties[path.group_path()]
                except KeyError:
                    channel_group_properties = OrderedDict()
                channel = TdmsChannel(
                    path, obj.data_type, obj.scaler_data_types, obj.num_values,
                    properties, channel_group_properties, self._properties,
                    tdms_reader, self._raw_timestamps, self._memmap_dir)
                if path.group in group_channels:
                    group_channels[path.group].append(channel)
                else:
                    group_channels[path.group] = [channel]

        # Create group objects containing channels and properties
        for group_name, properties in group_properties.items():
            try:
                channels = group_channels[group_name]
            except KeyError:
                channels = []
            group_path = ObjectPath(group_name)
            self._groups[group_name] = TdmsGroup(group_path, properties, channels)
        for group_name, channels in group_channels.items():
            if group_name not in self._groups:
                # Group with channels but without any corresponding object metadata in the file:
                group_path = ObjectPath(group_name)
                self._groups[group_name] = TdmsGroup(group_path, {}, channels)

        if not read_metadata_only:
            self._read_data(tdms_reader)

    def _read_data(self, tdms_reader):
        with Timer(log, "Allocate space"):
            # Allocate space for data
            for group in self.groups():
                for channel in group.channels():
                    self._channel_data[channel.path] = get_data_receiver(
                        channel, len(channel), self._raw_timestamps, self._memmap_dir)

        with Timer(log, "Read data"):
            # Now actually read all the data
            for chunk in tdms_reader.read_raw_data():
                for (path, data) in chunk.channel_data.items():
                    channel_data = self._channel_data[path]
                    if data.data is not None:
                        channel_data.append_data(data.data)
                    elif data.scaler_data is not None:
                        for scaler_id, scaler_data in data.scaler_data.items():
                            channel_data.append_scaler_data(scaler_id, scaler_data)

            for group in self.groups():
                for channel in group.channels():
                    channel_data = self._channel_data[channel.path]
                    if channel_data is not None:
                        channel._set_raw_data(channel_data)

        self.data_read = True

    def _convert_properties(self, properties):
        def convert_prop(val):
            if isinstance(val, TdmsTimestamp) and not self._raw_timestamps:
                # Convert timestamps to numpy datetime64 if raw timestamps are not requested
                return val.as_datetime64()
            return val
        return OrderedDict((k, convert_prop(v)) for (k, v) in properties.items()) class LogManager(object):
    def __init__(self):
    def get_logger(self, module_name):
    def set_level(self, level): class BytesIoTestFile(GeneratedFile):
    def load(self, *args, **kwargs):
        file = self.get_bytes_io_file()
        return tdms.TdmsFile(file, *args, **kwargs) class GeneratedFile(object):
    """Generate a TDMS file for testing"""

    def __init__(self):
        self._content = []

    def add_segment(self, toc, metadata, data, incomplete=False, binary_data=False, version=4713):
        metadata_bytes = _hex_to_bytes(metadata)
        data_bytes = data if binary_data else _hex_to_bytes(data)
        if toc is not None:
            lead_in = b'TDSm'
            toc_mask = long(0)
            for toc_item in toc:
                if toc_item == "kTocMetaData":
                    toc_mask = toc_mask | long(1) << 1
                elif toc_item == "kTocRawData":
                    toc_mask = toc_mask | long(1) << 3
                elif toc_item == "kTocDAQmxRawData":
                    toc_mask = toc_mask | long(1) << 7
                elif toc_item == "kTocInterleavedData":
                    toc_mask = toc_mask | long(1) << 5
                elif toc_item == "kTocBigEndian":
                    toc_mask = toc_mask | long(1) << 6
                elif toc_item == "kTocNewObjList":
                    toc_mask = toc_mask | long(1) << 2
                else:
                    raise ValueError("Unrecognised TOC value: %s" % toc_item)
            lead_in += struct.pack('<i', toc_mask)
            lead_in += struct.pack('<l', version)
            next_segment_offset = len(metadata_bytes) + len(data_bytes)
            raw_data_offset = len(metadata_bytes)
            if incomplete:
                lead_in += _hex_to_bytes('FF' * 8)
            else:
                lead_in += struct.pack('<Q', next_segment_offset)
            lead_in += struct.pack('<Q', raw_data_offset)
        else:
            lead_in = b''
        self._content.append((lead_in, metadata_bytes, data_bytes))

    def get_tempfile(self, **kwargs):
        named_file = tempfile.NamedTemporaryFile(suffix=".tdms", **kwargs)
        file = named_file.file
        file.write(self._get_contents())
        file.seek(0)
        return named_file

    @contextmanager
    def get_tempfile_with_index(self):
        directory = tempfile.mkdtemp()
        tdms_path = os.path.join(directory, 'test_file.tdms')
        tdms_index_path = os.path.join(directory, 'test_file.tdms_index')
        with open(tdms_path, 'wb') as file:
            file.write(self._get_contents())
        with open(tdms_index_path, 'wb') as file:
            file.write(self._get_index_contents())
        try:
            yield tdms_path
        finally:
            os.unlink(tdms_path)
            os.unlink(tdms_index_path)
            os.rmdir(directory)

    def load(self, *args, **kwargs):
        with tempfile.NamedTemporaryFile(suffix=".tdms") as named_file:
            file = named_file.file
            file.write(self._get_contents())
            file.seek(0)
            return tdms.TdmsFile(file, *args, **kwargs)

    def get_bytes_io_file(self):
        file = BytesIO()
        file.write(self._get_contents())
        file.seek(0)
        return file

    def _get_contents(self):
        contents = b''
        for segment in self._content:
            contents += segment[0]
            contents += segment[1]
            contents += segment[2]
        return contents

    def _get_index_contents(self):
        contents = b''
        for segment in self._content:
            lead_in = segment[0]
            if len(lead_in) >= 4:
                lead_in = b'TDSh' + lead_in[4:]
            contents += lead_in
            contents += segment[1]
        return contents def basic_segment():
    """Basic TDMS segment with one group and two channels"""

    toc = ("kTocMetaData", "kTocRawData", "kTocNewObjList")
    metadata = (
        # Number of objects
        "04 00 00 00"
        # Length of the first object path
        "08 00 00 00"
        # Object path (/'Group')
        "2F 27 47 72"
        "6F 75 70 27"
        # Raw data index
        "FF FF FF FF"
        # Num properties
        "02 00 00 00"
        # Name length
        "04 00 00 00"
        # Property name (prop)
        "70 72 6F 70"
        # Property data type (string)
        "20 00 00 00"
        # Length of string value
        "05 00 00 00"
        # Value
        "76 61 6C 75 65"
        # Length of second property name
        "03 00 00 00"
        # Property name (num)
        "6E 75 6D"
        # Data type of property
        "03 00 00 00"
        # Value
        "0A 00 00 00"
        # Length of the second object path
        "13 00 00 00"
        # Second object path (/'Group'/'Channel1')
        "2F 27 47 72"
        "6F 75 70 27"
        "2F 27 43 68"
        "61 6E 6E 65"
        "6C 31 27"
        # Length of index information
        "14 00 00 00"
        # Raw data data type
        "03 00 00 00"
        # Dimension
        "01 00 00 00"
        # Number of raw data values
        "02 00 00 00"
        "00 00 00 00"
        # Number of properties (0)
        "00 00 00 00"
        # Length of the third object path
        "13 00 00 00"
        # Third object path (/'Group'/'Channel2')
        "2F 27 47 72"
        "6F 75 70 27"
        "2F 27 43 68"
        "61 6E 6E 65"
        "6C 32 27"
        # Length of index information
        "14 00 00 00"
        # Raw data data type
        "03 00 00 00"
        # Dimension
        "01 00 00 00"
        # Number of data values
        "02 00 00 00"
        "00 00 00 00"
        # Set time properties for the second channel
        "02 00 00 00"
        "0F 00 00 00" +
        string_hexlify('wf_start_offset') +
        "0A 00 00 00" +
        hexlify_value("<d", 0.0) +
        "0C 00 00 00" +
        string_hexlify('wf_increment') +
        "0A 00 00 00" +
        hexlify_value("<d", 0.1) +
        # Length of the object path
        "01 00 00 00"
        # Object path (/)
        "2F"
        # Raw data index
        "FF FF FF FF"
        # Num properties
        "01 00 00 00"
        # Length of property name
        "03 00 00 00"
        # Property name (num)
        "6E 75 6D"
        # Data type of property
        "03 00 00 00"
        # Value
        "0F 00 00 00"
    )
    data = (
        # Data for segment
        "01 00 00 00"
        "02 00 00 00"
        "03 00 00 00"
        "04 00 00 00"
    )
    return toc, metadata, data def channel_metadata(channel_name, data_type, num_values, properties=None):
    return (
        # Length of the object path
        hexlify_value('<I', len(channel_name)) +
        # Object path
        string_hexlify(channel_name) +
        # Length of index information
        "14 00 00 00" +
        # Raw data data type
        hexlify_value('<I', data_type) +
        # Dimension
        "01 00 00 00" +
        # Number of raw data values
        hexlify_value('<Q', num_values) +
        hex_properties(properties)
    ) def compare_arrays(actual_data, expected_data):
    try:
        np.testing.assert_almost_equal(actual_data, expected_data)
    except TypeError:
        # Cannot compare given types
        assert len(actual_data) == len(expected_data)
        for (actual, expected) in zip(actual_data, expected_data):
            assert actual == expected def hexlify_value(struct_type, value):
    """Return hex string representation of a value"""
    return binascii.hexlify(struct.pack(struct_type, value)).decode('utf-8') def segment_objects_metadata(*args):
    """ Metadata for multiple objects in a segment
    """
    num_objects_hex = hexlify_value("<I", len(args))
    return num_objects_hex + "".join(args) def string_hexlify(input_string):
    """Return hex string representation of string"""
    return binascii.hexlify(input_string.encode('utf-8')).decode('utf-8') TDS_TYPE_INT8 = 1
TDS_TYPE_INT16 = 2
TDS_TYPE_INT32 = 3
TDS_TYPE_BOOL = 0x21
TDS_TYPE_COMPLEX64 = 0x08000c
TDS_TYPE_COMPLEX128 = 0x10000d
TDS_TYPE_FLOAT32 = 9
TDS_TYPE_FLOAT64 = 10
TDS_TYPE_FLOAT32_WITH_UNIT = 0x19
TDS_TYPE_FLOAT64_WITH_UNIT = 0x1A
def scenario(func):
    def as_param():
def get_scenarios():
def single_segment_with_one_channel():
def single_segment_with_two_channels():
def single_segment_with_two_channels_interleaved():
def single_segment_with_interleaved_data_of_different_width():
def no_metadata_segment():
def object_with_no_metadata_in_segment():
def add_new_channel():
def repeated_objects_without_data_in_segment_and_added_object():
def increase_channel_size():
def remove_a_channel():
def alternating_data_objects_with_new_obj_list():
def alternating_data_objects_reusing_obj_list():
def chunked_segment():
def chunked_interleaved_segment():
def less_data_than_expected():
def less_data_than_expected_interleaved():
def segment_with_zero_data_length():
def incomplete_last_segment():
def incomplete_last_row_of_interleaved_data():
def bool_data():
def float_data():
def float_data_with_unit():
def complex_data():
def scaled_data():
def timestamp_data():
def interleaved_timestamp_data():
def interleaved_timestamp_and_numpy_data():
def segment_without_data():
def channel_without_data():
def channel_without_data_or_data_type():
def extra_padding_after_metadata():
def timestamp_data_chunk(times):
def timestamp_hexlify(time):
from collections import defaultdict
from shutil import copyfile
from hypothesis import (assume, given, example, settings, strategies, HealthCheck)
from nptdms import TdmsFile
from nptdms.log import log_manager
from nptdms.test.util import (
    BytesIoTestFile,
    GeneratedFile,
    basic_segment,
    channel_metadata,
    compare_arrays,
    hexlify_value,
    segment_objects_metadata,
    string_hexlify,
)
from nptdms.test import scenarios
import logging
import os
import sys
import tempfile
import weakref
import numpy as np
import pytest
    import pathlib
def test_indexing_scaled_channel_with_integer():
    """ Test indexing into a channel with an integer index when the channel is scaled
    """
    test_file, expected_data = scenarios.scaled_data().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            for ((group, channel), expected_channel_data) in expected_data.items():
                channel_object = tdms_file[group][channel]
                values = []
                for i in range(len(channel_object)):
                    values.append(channel_object[i])
                compare_arrays(values, expected_channel_data)


def test_indexing_channel_with_ellipsis():
    """ Test indexing into a channel with ellipsis returns all data
    """
    test_file, expected_data = scenarios.chunked_segment().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            for ((group, channel), expected_channel_data) in expected_data.items():
                channel_object = tdms_file[group][channel]
                compare_arrays(channel_object[...], expected_channel_data)


@pytest.fixture(scope="module")
def opened_tdms_file():
    """ Allow re-use of an opened TDMS file
    """
    test_file, expected_data = scenarios.chunked_segment().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            yield tdms_file, expected_data


@given(
    start=strategies.integers(-10, 10) | strategies.none(),
    stop=strategies.integers(-10, 10) | strategies.none(),
    step=strategies.integers(-5, 5).filter(lambda i: i != 0) | strategies.none(),
)
@settings(max_examples=1000)
def test_indexing_channel_with_slice(opened_tdms_file, start, stop, step):
    """ Test indexing into a channel with a slice
    """
    tdms_file, expected_data = opened_tdms_file
    for ((group, channel), expected_channel_data) in expected_data.items():
        channel_object = tdms_file[group][channel]
        compare_arrays(channel_object[start:stop:step], expected_channel_data[start:stop:step])


@pytest.mark.parametrize('index', [-9, 8])
def test_indexing_channel_with_invalid_integer_raises_error(index):
    """ Test indexing into a channel with an invalid integer index
    """
    test_file, expected_data = scenarios.chunked_segment().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            for ((group, channel), expected_channel_data) in expected_data.items():
                channel_object = tdms_file[group][channel]
                with pytest.raises(IndexError):
                    _ = channel_object[index]


def test_indexing_channel_with_zero_step_raises_error():
    """ Test indexing into a channel with a slice with zero step size raises an error
    """
    test_file, expected_data = scenarios.chunked_segment().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            for ((group, channel), expected_channel_data) in expected_data.items():
                channel_object = tdms_file[group][channel]
                with pytest.raises(ValueError) as exc_info:
                    _ = channel_object[::0]
                assert str(exc_info.value) == "Step size cannot be zero"


@pytest.mark.parametrize('index', ["test", None])
def test_indexing_channel_with_invalid_type_raises_error(index):
    """ Test indexing into a channel with an invalid index type
    """
    test_file, expected_data = scenarios.chunked_segment().values
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            for ((group, channel), expected_channel_data) in expected_data.items():
                channel_object = tdms_file[group][channel]
                with pytest.raises(TypeError) as exc_info:
                    _ = channel_object[index]
                assert "Invalid index type" in str(exc_info.value)


def test_invalid_offset_in_read_data_throws():
    """ Exception is thrown when reading a subset of data with an invalid offset
    """
    test_file, expected_data = scenarios.single_segment_with_one_channel().values
    group, channel = list(expected_data.keys())[0]
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            with pytest.raises(ValueError) as exc_info:
                tdms_file[group][channel].read_data(-1, 5)
            assert "offset must be non-negative" in str(exc_info.value)


def test_invalid_length_in_read_data_throws():
    """ Exception is thrown when reading a subset of data with an invalid length
    """
    test_file, expected_data = scenarios.single_segment_with_one_channel().values
    group, channel = list(expected_data.keys())[0]
    with test_file.get_tempfile() as temp_file:
        with TdmsFile.open(temp_file.file) as tdms_file:
            with pytest.raises(ValueError) as exc_info:
                tdms_file[group][channel].read_data(0, -5)
            assert "length must be non-negative" in str(exc_info.value)


def test_read_data_after_close_throws():
    """ Trying to read after opening and closing without reading data should throw
    """
    test_file, expected_data = scenarios.single_segment_with_one_channel().values
    group, channel = list(expected_data.keys())[0]
    with test_file.get_tempfile() as temp_file:

        with TdmsFile.open(temp_file.file) as tdms_file: