diff --git a/idelib/dataset.py b/idelib/dataset.py index 9de11a3d..8385df16 100644 --- a/idelib/dataset.py +++ b/idelib/dataset.py @@ -49,6 +49,8 @@ from bisect import bisect_right from collections.abc import Iterable, Sequence from datetime import datetime +from threading import Lock +import warnings from functools import partial import os.path @@ -61,7 +63,7 @@ import numpy as np from .transforms import Transform, CombinedPoly, PolyPoly -from .parsers import getParserTypes, getParserRanges +from .parsers import getParserTypes, getParserRanges, ChannelDataBlock SCHEMA_FILE = 'mide_ide.xml' @@ -254,6 +256,8 @@ def __init__(self, stream, name=None, exitCondition=None, quiet=True): self.loadCancelled = False self.loading = True self.filename = getattr(stream, "name", None) + + self._channelDataLock = Lock() # Subsets: used when importing multiple files into the same dataset. self.subsets = [] @@ -489,6 +493,11 @@ def updateTransforms(self): for ch in self.channels.values(): ch.updateTransforms() + def fillCaches(self): + for channel in self.channels.values(): + for ea in channel.sessions.values(): + ea.fillCache() + #=============================================================================== # @@ -1216,6 +1225,7 @@ def __init__(self, parentChannel, session=None, parentList=None): self.removeMean = False self.hasMinMeanMax = True + self._rollingMeanSpan = None self.rollingMeanSpan = self.DEFAULT_MEAN_SPAN self.transform = None @@ -1233,6 +1243,46 @@ def __init__(self, parentChannel, session=None, parentList=None): self._mean = None + _format = self.parent.parser.format + if isinstance(self.parent, SubChannel): + self._npType = self.parent.parent.getSession()._npType[self.subchannelId] + elif len(_format) == 0: + self._npType = np.uint8 + else: + if isinstance(_format, bytes): + _format = _format.decode() + + if _format[0] in ['<', '>', '=']: + endian = _format[0] + dtypes = [endian + ChannelDataBlock.TO_NP_TYPESTR[x] for x in _format[1:]] + else: + dtypes = [ChannelDataBlock.TO_NP_TYPESTR[x] for x in str(_format)] + + self._npType = np.dtype([(str(i), dtype) for i, dtype in enumerate(dtypes)]) + + self._channelDataLock = parentChannel.dataset._channelDataLock + self._cacheArray = None + self._cacheBytes = None + self._fullyCached = False + self._cacheStart = None + self._cacheEnd = None + self._cacheBlockStart = None + self._cacheBlockEnd = None + self._cacheLen = 0 + + + @property + def rollingMeanSpan(self): + return self._rollingMeanSpan + + + @rollingMeanSpan.setter + def rollingMeanSpan(self, value): + if value != -1: + warnings.warn('Rolling mean has been deprecated, this behavior has ' + 'been replaced with total mean removal.') + self._rollingMeanSpan = value + def updateTransforms(self, recurse=True): """ (Re-)Build and (re-)apply the transformation functions. @@ -1304,6 +1354,14 @@ def copy(self, newParent=None): newList.noBivariates = self.noBivariates newList._blockIndices = self._blockIndices newList._blockTimes = self._blockTimes + newList._channelDataLock = self._channelDataLock + newList._cacheArray = self._cacheArray + newList._cacheBytes = self._cacheBytes + newList._fullyCached = self._fullyCached + newList._cacheStart = self._cacheStart + newList._cacheEnd = self._cacheEnd + newList._cacheBlockStart = self._cacheBlockStart + newList._cacheBlockEnd = self._cacheBlockEnd return newList @@ -1315,76 +1373,80 @@ def append(self, block): :attention: Added elements must be in chronological order! """ - if block.numSamples is None: - block.numSamples = block.getNumSamples(self.parent.parser) + with self._channelDataLock: + if block.numSamples is None: + block.numSamples = block.getNumSamples(self.parent.parser) - # Set the session first/last times if they aren't already set. - # Possibly redundant if all sessions are 'closed.' - if self.session.firstTime is None: - self.session.firstTime = block.startTime - else: - self.session.firstTime = min(self.session.firstTime, block.startTime) + # Set the session first/last times if they aren't already set. + # Possibly redundant if all sessions are 'closed.' + if self.session.firstTime is None: + self.session.firstTime = block.startTime + else: + self.session.firstTime = min(self.session.firstTime, block.startTime) - if self.session.lastTime is None: - self.session.lastTime = block.endTime - else: - self.session.lastTime = max(self.session.lastTime, block.endTime) + if self.session.lastTime is None: + self.session.lastTime = block.endTime + else: + self.session.lastTime = max(self.session.lastTime, block.endTime) + + # Check that the block actually contains at least one sample. + if block.numSamples < 1: + # Ignore blocks with empty payload. Could occur in FW <17. + # TODO: Make sure this doesn't hide too many errors! + logger.warning("Ignoring block with bad payload size for %r" % self) + return + + block.cache = self.parent.cache + oldLength = self._length + + block.blockIndex = len(self._data) + block.indexRange = (oldLength, oldLength + block.numSamples) + + # _singleSample hint not explicitly set; set it based on this block. + # There will be problems if the first block has only one sample, but + # future ones don't. This shouldn't happen, though. + if self._singleSample is None: + self._singleSample = block.numSamples == 1 + if self._parentList is not None: + self._parentList._singleSample = self._singleSample + if self.parent.singleSample is None: + self.parent.singleSample = self._singleSample + if self.parent.parent is not None: + self.parent.parent.singleSample = self._singleSample + + # HACK (somewhat): Single-sample-per-block channels get min/mean/max + # which is just the same as the value of the sample. Set the values, + # but don't set hasMinMeanMax. + if self._singleSample is True:# and not self.hasMinMeanMax: + block.minMeanMax = np.tile(block.payload, 3) + block.parseMinMeanMax(self.parent.parser) + self.hasMinMeanMax = False + elif block.minMeanMax is not None: + block.parseMinMeanMax(self.parent.parser) + self.hasMinMeanMax = True #self.hasMinMeanMax and True + else: + # XXX: Attempt to calculate min/mean/max here instead of + # in _computeMinMeanMax(). Causes issues with pressure for some + # reason - it starts removing mean and won't plot. + vals = self.parseBlock(block) + block.min = vals.min(axis=-1) + block.mean = vals.mean(axis=-1) + block.max = vals.max(axis=-1) + self.hasMinMeanMax = True + # self.hasMinMeanMax = False + # self.allowMeanRemoval = False + + # Cache the index range for faster searching + self._blockIndices.append(oldLength) + self._blockTimes.append(block.startTime) + + self._hasSubsamples = self._hasSubsamples or block.numSamples > 1 + + self._data.append(block) + self._length += block.numSamples + + block._payload = np.frombuffer(block._payloadEl.dump(), dtype=self._npType) - # Check that the block actually contains at least one sample. - if block.numSamples < 1: - # Ignore blocks with empty payload. Could occur in FW <17. - # TODO: Make sure this doesn't hide too many errors! - logger.warning("Ignoring block with bad payload size for %r" % self) - return - - block.cache = self.parent.cache - oldLength = self._length - - block.blockIndex = len(self._data) - block.indexRange = (oldLength, oldLength + block.numSamples) - - # _singleSample hint not explicitly set; set it based on this block. - # There will be problems if the first block has only one sample, but - # future ones don't. This shouldn't happen, though. - if self._singleSample is None: - self._singleSample = block.numSamples == 1 - if self._parentList is not None: - self._parentList._singleSample = self._singleSample - if self.parent.singleSample is None: - self.parent.singleSample = self._singleSample - if self.parent.parent is not None: - self.parent.parent.singleSample = self._singleSample - - # HACK (somewhat): Single-sample-per-block channels get min/mean/max - # which is just the same as the value of the sample. Set the values, - # but don't set hasMinMeanMax. - if self._singleSample is True:# and not self.hasMinMeanMax: - block.minMeanMax = np.tile(block.payload, 3) - block.parseMinMeanMax(self.parent.parser) - self.hasMinMeanMax = False - elif block.minMeanMax is not None: - block.parseMinMeanMax(self.parent.parser) - self.hasMinMeanMax = True #self.hasMinMeanMax and True - else: - # XXX: Attempt to calculate min/mean/max here instead of - # in _computeMinMeanMax(). Causes issues with pressure for some - # reason - it starts removing mean and won't plot. - vals = self.parseBlock(block) - block.min = vals.min(axis=-1) - block.mean = vals.mean(axis=-1) - block.max = vals.max(axis=-1) - self.hasMinMeanMax = True -# self.hasMinMeanMax = False -# self.allowMeanRemoval = False - - # Cache the index range for faster searching - self._blockIndices.append(oldLength) - self._blockTimes.append(block.startTime) - - self._hasSubsamples = self._hasSubsamples or block.numSamples > 1 - - self._data.append(block) - self._length += block.numSamples @property def _firstTime(self): @@ -1682,7 +1744,7 @@ def __getitem__(self, idx, display=False): else: xform = self._fullXform - if isinstance(idx, int): + if isinstance(idx, (int, np.integer)): if idx >= len(self): raise IndexError("EventArray index out of range") @@ -1809,20 +1871,13 @@ def itervalues(self, start=None, end=None, step=1, subchannels=True, :return: an iterable of structured array value blocks in the specified index range. """ - # TODO: Optimize; times don't need to be computed since they aren't used - iterBlockValues = ( - np.stack(values) - for _, values in self._blockSlice(start, end, step, display) - ) - if self.hasSubchannels and subchannels is not True: - chIdx = np.asarray(subchannels) - return (vals - for blockVals in iterBlockValues - for vals in blockVals[chIdx].T) - else: - return (vals - for blockVals in iterBlockValues - for vals in blockVals.T) + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + + out = self.arrayValues(start=start, end=end, step=step) + + yield from out.T def arrayValues(self, start=None, end=None, step=1, subchannels=True, @@ -1841,12 +1896,38 @@ def arrayValues(self, start=None, end=None, step=1, subchannels=True, """ # TODO: Optimize; times don't need to be computed since they aren't used # -> take directly from _blockSlice - arrayEvents = self.arraySlice(start, end, step, display) - if self.hasSubchannels and subchannels is not True: - return arrayEvents[np.asarray(subchannels)+1] + if not isinstance(start, slice): + start = slice(start, end, step) + start, end, step = start.indices(len(self)) + + if self.useAllTransforms: + xform = self._fullXform + if display: + xform = self._displayXform or xform + else: + xform = self._comboXform + + rawData = self._accessCache(start, end, step) + + if isinstance(self.parent, SubChannel): + out = np.empty((1, len(rawData))) + else: + out = np.empty((len(rawData.dtype), len(rawData))) + + if isinstance(self.parent, SubChannel): + xform.polys[self.subchannelId].inplace(rawData, out=out) else: - return arrayEvents[1:] + for i, (k, _) in enumerate(rawData.dtype.descr): + xform.polys[i].inplace(rawData[k], out=out[i]) + + if self.removeMean: + out[1:] -= out[1:].mean(axis=1, keepdims=True) + + if subchannels is True: + return out + else: + return out[list(subchannels)] def _blockSlice(self, start=None, end=None, step=1, display=False): @@ -1910,10 +1991,13 @@ def iterSlice(self, start=None, end=None, step=1, display=False): 'display' transform) will be applied to the data. :return: an iterable of events in the specified index range. """ - for times, values in self._blockSlice(start, end, step, display): - blockEvents = np.append(times[np.newaxis], values, axis=0) - for event in blockEvents.T: - yield event + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + + out = self.arraySlice(start=start, end=end, step=step, display=display) + + yield from out.T def arraySlice(self, start=None, end=None, step=1, display=False): @@ -1927,15 +2011,37 @@ def arraySlice(self, start=None, end=None, step=1, display=False): 'display' transform) will be applied to the data. :return: a structured array of events in the specified index range. """ - raw_slice = [ - [times[np.newaxis].T, values.T] - for times, values in self._blockSlice(start, end, step, display) - ] - if not raw_slice: - no_of_chs = (len(self.parent.types) if self.hasSubchannels else 1) - return np.empty((no_of_chs+1, 0), dtype=np.float) - return np.block(raw_slice).T + if not isinstance(start, slice): + start = slice(start, end, step) + start, end, step = start.indices(len(self)) + + if self.useAllTransforms: + xform = self._fullXform + if display: + xform = self._displayXform or xform + else: + xform = self._comboXform + + rawData = self._accessCache(start, end, step) + + if isinstance(self.parent, SubChannel): + out = np.empty((2, len(rawData))) + else: + out = np.empty((len(rawData.dtype) + 1, len(rawData))) + + self._inplaceTime(start, end, step, out=out[0]) + + if isinstance(self.parent, SubChannel): + xform.polys[self.subchannelId].inplace(rawData, out=out[1], timestamp=out[0]) + else: + for i, (k, _) in enumerate(rawData.dtype.descr): + xform.polys[i].inplace(rawData[k], out=out[i + 1], timestamp=out[0]) + + if self.removeMean: + out[1:] -= out[1:].mean(axis=1, keepdims=True) + + return out def _blockJitterySlice(self, start=None, end=None, step=1, jitter=0.5, @@ -1961,6 +2067,12 @@ def _blockJitterySlice(self, start=None, end=None, step=1, jitter=0.5, jitter = 0.5 scaledJitter = jitter * abs(step) + indices = np.arange(start, end, step) + if scaledJitter > 0.5: + indices[1:-1] += np.rint( + scaledJitter*np.random.uniform(-1, 1, max(0, len(indices) - 2)) + ).astype(indices.dtype) + startBlockIdx = self._getBlockIndexWithIndex(start) if start > 0 else 0 endBlockIdx = self._getBlockIndexWithIndex(end-1, start=startBlockIdx) @@ -2013,13 +2125,16 @@ def iterJitterySlice(self, start=None, end=None, step=1, jitter=0.5, 'display' transform) will be applied to the data. :return: an iterable of events in the specified index range. """ + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + self._computeMinMeanMax() + + data = self.arrayJitterySlice(start=start, end=end, step=step, jitter=jitter, display=display) + + yield from data.T - for times, values in self._blockJitterySlice(start, end, step, jitter, - display): - blockEvents = np.append(times[np.newaxis], values, axis=0) - for event in blockEvents.T: - yield event def arrayJitterySlice(self, start=None, end=None, step=1, jitter=0.5, @@ -2036,19 +2151,63 @@ def arrayJitterySlice(self, start=None, end=None, step=1, jitter=0.5, 'display' transform) will be applied to the data. :return: a structured array of events in the specified index range. """ + + if not isinstance(start, slice): + start = slice(start, end, step) + start, end, step = start.indices(len(self)) + + # Check for non-jittered cases + if jitter is True: + jitter = 0.5 + scaledJitter = jitter * abs(step) + + if scaledJitter <= 0.5: + return self.arraySlice(start=start, end=end, step=step, display=display) + + # begin as normal self._computeMinMeanMax() - - raw_slice = [ - [times[np.newaxis].T, values.T] - for times, values in self._blockJitterySlice( - start, end, step, jitter, display - ) - ] - if not raw_slice: - no_of_chs = (len(self.parent.types) if self.hasSubchannels else 1) - return np.empty((no_of_chs+1, 0), dtype=np.float) - return np.block(raw_slice).T + if not isinstance(start, slice): + start = slice(start, end, step) + start, end, step = start.indices(len(self)) + + if self.useAllTransforms: + xform = self._fullXform + if display: + xform = self._displayXform or xform + else: + xform = self._comboXform + + # grab all raw data + rawData = self._accessCache(None, None, 1) + + # slightly janky way of enforcing output length + if isinstance(self.parent, SubChannel): + out = np.empty((2, len(self._accessCache(start, end, step)))) + else: + out = np.empty((len(rawData.dtype) + 1, len(self._accessCache(start, end, step)))) + + # save on space by being really clever and storing indices in timestamps + indices = out[0].view(np.int64) + indices[:] = np.arange(start, end, step, dtype=np.int64) + if len(indices) > 2: + indices[1:-1] += np.rint(np.random.uniform( + -scaledJitter, scaledJitter, (len(indices) - 2,) + )).astype(np.int64) + + # now index raw data + rawData = rawData[indices] + + # now times + self._inplaceTimeFromIndices(indices, out=out[0]) + + if isinstance(self.parent, SubChannel): + xform.polys[self.subchannelId].inplace(rawData, out=out[1], timestamp=out[0]) + else: + for i, (k, _) in enumerate(rawData.dtype.descr): + xform.polys[i].inplace(rawData[k], out=out[i + 1], timestamp=out[0]) + + return out def getEventIndexBefore(self, t): @@ -2059,14 +2218,17 @@ def getEventIndexBefore(self, t): :return: The index of the event preceding the given time, -1 if the time occurs before the first event. """ - if t <= self._data[0].startTime: + if t < self._data[0].startTime: return -1 + + if t >= self._data[-1].endTime: + return self._data[-1].indexRange[1] + blockIdx = self._getBlockIndexWithTime(t) try: block = self._data[blockIdx] except IndexError: - blockIdx = len(self._data)-1 - block = self._data[blockIdx] + block = self._data[-1] return int(block.indexRange[0] + \ ((t - block.startTime) / self._getBlockSampleTime(blockIdx))) @@ -2079,11 +2241,16 @@ def getEventIndexNear(self, t): """ if t <= self._data[0].startTime: return 0 + + if t >= self._data[-1].endTime: + return self._data[-1].indexRange[1] + idx = self.getEventIndexBefore(t) - events = self[idx:idx+2] - if events[0][0] == t or len(events) == 1: + events = self[idx:idx+2][0] + if len(events) == 1: return idx - return min((t - events[0][0], idx), (events[1][0] - t, idx+1))[1] + + return idx + abs(events - t).argmin() def getRangeIndices(self, startTime, endTime): @@ -2130,6 +2297,10 @@ def iterRange(self, startTime=None, endTime=None, step=1, display=False): :keyword endTime: The second time, or `None` to use the end of the session. """ + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + startIdx, endIdx = self.getRangeIndices(startTime, endTime) return self.iterSlice(startIdx,endIdx,step,display=display) @@ -2181,6 +2352,10 @@ def iterMinMeanMax(self, startTime=None, endTime=None, padding=0, :return: An iterator producing sets of three events (min, mean, and max, respectively). """ + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + if not self.hasMinMeanMax: self._computeMinMeanMax() @@ -2277,9 +2452,36 @@ def arrayMinMeanMax(self, startTime=None, endTime=None, padding=0, and max, respectively). """ - return np.moveaxis([i for i in iterator(self.iterMinMeanMax( - startTime, endTime, padding, times, display - ))], 0, -1) + startBlock, endBlock = self._getBlockRange(startTime, endTime) + shape = (3, max(1, len(self._npType)) + int(times), endBlock - startBlock) + scid = self.subchannelId + isSubchannel = isinstance(self.parent, SubChannel) + + out = np.empty(shape) + + for i, d in enumerate(self._data[startBlock:endBlock]): + if isSubchannel: + if times: + out[:, 0, i] = d.startTime + out[0, 1, i] = d.min[scid] + out[1, 1, i] = d.mean[scid] + out[2, 1, i] = d.max[scid] + else: + out[0, 0, i] = d.min[scid] + out[1, 0, i] = d.mean[scid] + out[2, 0, i] = d.max[scid] + else: + if times: + out[:, 0, i] = d.startTime + out[0, 1:, i] = d.min + out[1, 1:, i] = d.mean + out[2, 1:, i] = d.max + else: + out[0, :, i] = d.min + out[1, :, i] = d.mean + out[2, :, i] = d.max + + return out def getMinMeanMax(self, startTime=None, endTime=None, padding=0, @@ -2485,10 +2687,11 @@ def getSampleTime(self, idx=None): :return: The time between samples (us) """ sr = self.parent.sampleRate - if idx is None and sr is not None: - return 1.0 / sr - else: - idx = 0 + if idx is None: + if sr is not None: + return 1.0/sr + else: + idx = 0 return self._getBlockSampleTime(self._getBlockIndexWithIndex(idx)) @@ -2625,6 +2828,10 @@ def iterResampledRange(self, startTime, stopTime, maxPoints, padding=0, :todo: Optimize iterResampledRange(); not very efficient, particularly not with single-sample blocks. """ + + warnings.warn(DeprecationWarning('iter methods should be expected to be ' + 'removed in future versions of idelib')) + startIdx, stopIdx = self.getRangeIndices(startTime, stopTime) numPoints = (stopIdx - startIdx) startIdx = max(startIdx-padding, 0) @@ -2748,11 +2955,19 @@ def exportCsv(self, stream, start=None, stop=None, step=1, subchannels=True, if headers: stream.write('"Time"%s%s\n' % (delimiter, delimiter.join(['"%s"' % n for n in names]))) + + data = _self.arraySlice(start, stop, step) + if useUtcTime and _self.session.utcStartTime: + if useIsoFormat: + times = data[0] + data = data.astype([('time', ' nextUpdateTime or thisOffset > nextUpdatePos: # Update progress bar updater(count=eventsRead+samplesRead, - percent=(thisOffset-firstDataPos+0.0)/dataSize) + percent=(1/3) + (2/3)*(thisOffset-firstDataPos)/dataSize) nextUpdatePos = thisOffset + ticSize nextUpdateTime = thisTime + updateInterval @@ -504,6 +506,8 @@ def readData(doc, source=None, updater=nullUpdater, numUpdates=500, updateInterv # (typically the last) doc.fileDamaged = True + doc.fillCaches() + doc.loading = False updater(done=True) return eventsRead diff --git a/testing/test_dataset.py b/testing/test_dataset.py index a5e59036..73ef70f3 100644 --- a/testing/test_dataset.py +++ b/testing/test_dataset.py @@ -8,12 +8,16 @@ `calibration.AccelTransform`. These classes may be refactored out in the future. """ - +import struct from io import StringIO, BytesIO import sys import unittest import mock +import pytest + +import numpy as np # type: ignore + from idelib.dataset import (Cascading, Channel, Dataset, @@ -25,16 +29,53 @@ Transformable, WarningRange, ) - from idelib.transforms import Transform, CombinedPoly, PolyPoly from idelib.transforms import AccelTransform, Univariate from idelib import importer from idelib import parsers -import numpy as np # type: ignore +from testing.utils import nullcontext from .file_streams import makeStreamLike + +# ============================================================================== +# Fixtures +# ============================================================================== + +_fileStrings = {} + + +def _load_file(filePath): + if filePath not in _fileStrings: + with open(filePath, 'rb') as f: + _fileStrings[filePath] = f.read() + out = BytesIO(_fileStrings[filePath]) + out.name = filePath + return out + + +@pytest.fixture +def testIDE(): + doc = importer.openFile(_load_file('./test.ide')) + importer.readData(doc) + return doc + + +@pytest.fixture +def SSX70065IDE(): + doc = importer.openFile(_load_file('./testing/SSX70065.IDE')) + importer.readData(doc) + return doc + + +@pytest.fixture +def SSX_DataIDE(): + doc = importer.openFile(_load_file('./testing/SSX_Data.IDE')) + importer.readData(doc) + return doc + + #=============================================================================== # #=============================================================================== @@ -53,20 +94,20 @@ def __init__(self): self.startTime = 0 self.sampleTime = 0 self.numSamples = 1 - - + + def __getitem__(self, index): return self.data[index] - - + + def __len__(self): return len(self.data) - - + + def updateTransforms(self): self.isUpdated = True - - + + def parseWith(self, x, start, end, step, subchannel): return (x, start, end, step, subchannel) @@ -79,95 +120,95 @@ def parseByIndexWith(self, parser, indices, subchannel): # #=============================================================================== -class CascadingTestCase(unittest.TestCase): +class TestCascading(unittest.TestCase): """ Test case for methods in the Cascading class. """ - + def setUp(self): self.casc1 = Cascading() self.casc1.name = 'parent' self.casc2 = Cascading() self.casc2.name = 'child' self.casc2.parent = self.casc1 - - + + def tearDown(self): self.casc1 = None self.casc2 = None - - + + def testHierarchy(self): """ Test for hierarchy method. """ self.assertEqual(self.casc2.hierarchy(), [self.casc1, self.casc2]) - - + + def testPath(self): """ Test for path method. """ self.assertEqual(self.casc1.path(), 'parent') self.assertEqual(self.casc2.path(), 'parent:child') self.casc1.path = lambda : None self.assertEqual(self.casc2.path(), 'child') - - + + def testRepr(self): """ Test that casting to a string creates the correct string. """ self.assertIn("= 0) - self.assertTrue(stopIdx <= len(eventArray)) - self.assertTrue(len(range(startIdx, stopIdx, step)) <= maxPoints) + dat = eventArray.arraySlice() - self.assertEqual( - EventArray.iterResampledRange(eventArray, startTime, stopTime, - maxPoints, jitter=0.1,), - mock.sentinel.b - ) - startIdx, stopIdx, step, jitter = ( - eventArray.iterJitterySlice.call_args[0] - ) - self.assertTrue(startIdx >= 0) - self.assertTrue(stopIdx <= len(eventArray)) - self.assertTrue(len(range(startIdx, stopIdx, step)) <= maxPoints) + # Run tests + np.testing.assert_array_almost_equal( + np.stack(list(eventArray.iterResampledRange(0, 1e6, 9))).T, + dat[:, [0, 112, 224, 336, 448, 560, 672, 784, 896]], + ) - def testArrayResampledRange(self): + def testArrayResampledRange(self, testIDE): """ Test for arrayResampledRange method. """ - - # Stub data/methods - eventArray = mock.Mock(spec=EventArray) - eventArray.configure_mock( - __len__=lambda self: 100, - ) - eventArray.getRangeIndices.return_value = 0, 105 - eventArray.arraySlice.return_value = mock.sentinel.a - eventArray.arrayJitterySlice.return_value = mock.sentinel.b - startTime = mock.sentinel.startTime - stopTime = mock.sentinel.stopTime - maxPoints = 43 + eventArray = testIDE.channels[8].getSession() - # Run tests - self.assertEqual( - EventArray.arrayResampledRange(eventArray, startTime, stopTime, - maxPoints), - mock.sentinel.a - ) - startIdx, stopIdx, step = ( - eventArray.arraySlice.call_args[0] - ) - self.assertTrue(startIdx >= 0) - self.assertTrue(stopIdx <= len(eventArray)) - self.assertTrue(len(range(startIdx, stopIdx, step)) <= maxPoints) + dat = eventArray.arraySlice() - self.assertEqual( - EventArray.arrayResampledRange(eventArray, startTime, stopTime, - maxPoints, jitter=0.1,), - mock.sentinel.b - ) - startIdx, stopIdx, step, jitter = ( - eventArray.arrayJitterySlice.call_args[0] - ) - self.assertTrue(startIdx >= 0) - self.assertTrue(stopIdx <= len(eventArray)) - self.assertTrue(len(range(startIdx, stopIdx, step)) <= maxPoints) + # Run tests + np.testing.assert_array_almost_equal( + eventArray.arrayResampledRange(0, 1e6, 9), + dat[:, [0, 112, 224, 336, 448, 560, 672, 784, 896]], + ) - def testExportCSV(self): + @pytest.mark.skip("this doesn't actually do anything") + def testExportCSV(self, eventArray1): """ Test for exportCsv method.""" self.mockData() - self.eventArray1._data[0].minMeanMax = 1 - self.eventArray1._data[0].blockIndex = 2 - self.eventArray1._data[0].min = [3] - self.eventArray1._data[0].mean = [4] - self.eventArray1._data[0].max = [5] + eventArray1._data[0].minMeanMax = 1 + eventArray1._data[0].blockIndex = 2 + eventArray1._data[0].min = [3] + eventArray1._data[0].mean = [4] + eventArray1._data[0].max = [5] + + def testMeanRemovalSingleBlock(self, testIDE): + """ Testing mean removal for spans less than one block """ + + eventArray = testIDE.channels[8].getSession() + eventArray.removeMean = False + + unremovedData = eventArray[:] + eventArray.rollingMeanSpan = 1 + eventArray.removeMean = True + + # for d in eventArray._data: + # unremovedData[1:, slice(*d.indexRange)] -= d.mean[:, np.newaxis] + unremovedData[1:] -= unremovedData[1:].mean(axis=1, keepdims=True) + + removedData = eventArray[:] + + np.testing.assert_array_equal(removedData, unremovedData) + + def testMeanRemovalFullFile(self, testIDE): + """ Testing mean removal spanning the full file """ + + eventArray = testIDE.channels[8].getSession() + eventArray.removeMean = False + + unremovedData = eventArray[:] + unremovedData[1:] -= unremovedData[1:].mean(axis=1)[:, np.newaxis] + + eventArray.rollingMeanSpan = -1 + eventArray.removeMean = True + + removedData = eventArray[:] + + np.testing.assert_array_equal(removedData, unremovedData) #=============================================================================== # #=============================================================================== -class PlotTestCase(unittest.TestCase): +class TestPlot: """ Unit test for the Plot class. """ - - def setUp(self): - self.dataset = importer.importFile('./testing/SSX70065.IDE') - self.dataset.addSession(0, 1, 2) - self.dataset.addSensor(0) - - self.fakeParser = GenericObject() - self.fakeParser.types = [0] - self.fakeParser.format = [] - - self.channel1 = Channel( - self.dataset, channelId=0, name="channel1", parser=self.fakeParser, - displayRange=[0]) - self.eventList1 = EventArray(self.channel1, session=self.dataset.sessions[0]) - - self.channel1.addSubChannel(subchannelId=0) - - self.subChannel1 = SubChannel(self.channel1, 0) - - self.plot1 = Plot(self.eventList1, 0, name="Plot1") - - - def tearDown(self): - self.dataset.close() - self.dataset = None - - self.fakeParser = None - self.channel1 = None - self.eventList1 = None - self.subChannel1 = None - self.plot1 = None - - - def mockData(self): - """ mock up a bit of fake data so I don't have to worry that external - classes are working during testing. - """ - fakeData = GenericObject() - fakeData.startTime = 0 - fakeData.indexRange = [0, 3] - fakeData.sampleTime = 1 - fakeData.numSamples = 1 - self.eventList1._data = [fakeData] - - - def testConstructor(self): + + @pytest.fixture + def channel32(self, SSX70065IDE): + return SSX70065IDE.channels[32] + + @pytest.fixture + def eventArray(self, channel32): + return channel32.getSession() + + @pytest.fixture + def plot1(self, eventArray): + return Plot(eventArray, 0, name='Plot1') + + def testConstructor(self, plot1, eventArray): """ Test for the constructor. """ - self.assertEqual(self.plot1.source, self.eventList1) - self.assertEqual(self.plot1.id, 0) - self.assertEqual(self.plot1.session, self.eventList1.session) - self.assertEqual(self.plot1.dataset, self.eventList1.dataset) - self.assertEqual(self.plot1.name, "Plot1") - self.assertEqual(self.plot1.units, self.eventList1.units) - self.assertEqual(self.plot1.attributes, None) - - - def testGetEventIndexBefore(self): + + plotParams = ( + plot1.source, + plot1.id, + plot1.session, + plot1.dataset, + plot1.name, + plot1.units, + plot1.attributes, + ) + + targetParams = ( + eventArray, + 0, + eventArray.session, + eventArray.dataset, + 'Plot1', + eventArray.units, + None, + ) + + assert plotParams == targetParams + + @pytest.mark.parametrize('t', [0, 1, 10, 100, 1000, 10000, 100000]) + def testGetEventIndexBefore(self, eventArray, plot1, t): """ Test for getEventIndexBefore method. """ - self.mockData() - - self.assertEqual( - self.plot1.getEventIndexBefore(0), - self.eventList1.getEventIndexBefore(0)) - - + + assert plot1.getEventIndexBefore(t) == eventArray.getEventIndexBefore(t) + + @pytest.mark.skip('not implemented') def testGetRange(self): """ Test for getRange method. """ - print('gotta get to this')#self.plot1.getRange(0, 1)) - # TODO + pass #=============================================================================== -#--- Data test cases +#--- Data test cases #=============================================================================== -class DataTestCase(unittest.TestCase): +class TestData: """ Basic tests of data fidelity against older, "known good" CSV exports. Exports were generated using the library as of the release of 1.8.0. @@ -2026,140 +1754,153 @@ class DataTestCase(unittest.TestCase): errors. """ - def setUp(self): - self.dataset = importer.importFile('./testing/SSX_Data.IDE') - self.delta = 0.0015 + @pytest.fixture + def dataset(self, SSX_DataIDE): + return SSX_DataIDE + @pytest.fixture + def channel8(self, dataset): + return dataset.channels[8] - def testCalibratedExport(self): - """ Test regular export, with bivariate polynomials applied. - """ - out = StringIO() - accel = self.dataset.channels[8].getSession() + @pytest.fixture + def accelArray(self, channel8): + return channel8.getSession() - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + @pytest.fixture + def out(self): + return StringIO() - np.testing.assert_allclose(new.T, old, rtol=1e-4) + @staticmethod + def generateCsvArray(filestream, eventArray, **kwargs): + eventArray.exportCsv(filestream, **kwargs) + filestream.seek(0) + return np.genfromtxt(filestream, delimiter=', ').T - - def testUncalibratedExport(self): - """ Test export with no per-channel polynomials.""" + def testCalibratedExport(self, accelArray, out): + """ Test regular export, with bivariate polynomials applied. + """ - out = StringIO() - accel = self.dataset.channels[8].getSession() + new = self.generateCsvArray(out, accelArray) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + np.testing.assert_equal(new[1:], old[1:]) - np.testing.assert_allclose(new.T, old, rtol=1e-4) + def testUncalibratedExport(self, accelArray, out): + """ Test export with no per-channel polynomials.""" + new = self.generateCsvArray(out, accelArray) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 + np.testing.assert_equal(new[1:], old[1:]) - def testNoBivariates(self): + def testNoBivariates(self, accelArray, out): """ Test export with bivariate polynomial references disabled (values only offset, not temperature-corrected). """ - out = StringIO() - accel = self.dataset.channels[8].getSession() - accel.noBivariates = True - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + accelArray.noBivariates = True - np.testing.assert_allclose(new.T, old, rtol=1e-4) + new = self.generateCsvArray(out, accelArray) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 + np.testing.assert_equal(new[1:], old[1:]) - def testRollingMeanRemoval(self): + def testRollingMeanRemoval(self, accelArray, out): """ Test regular export, with the rolling mean removed from the data. """ - - out = StringIO() - accel = self.dataset.channels[8].getSession() - accel.removeMean = True - accel.rollingMeanSpan = 5000000 - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + removeMean = True + meanSpan = 5000000 + + accelArray.removeMean = removeMean + accelArray.rollingMeanSpan = meanSpan + + new = self.generateCsvArray(out, accelArray, removeMean=removeMean, meanSpan=meanSpan) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 - np.testing.assert_allclose(new.T, old, rtol=1e-4) + np.testing.assert_equal(new[1:], old[1:]) - - def testTotalMeanRemoval(self): + def testTotalMeanRemoval(self, accelArray, out): """ Test regular export, calibrated, with the total mean removed from the data. """ - out = StringIO() - accel = self.dataset.channels[8].getSession() - accel.removeMean = True - accel.rollingMeanSpan = -1 + removeMean = True + meanSpan = -1 - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + accelArray.removeMean = removeMean + accelArray.rollingMeanSpan = meanSpan - np.testing.assert_allclose(new.T, old, rtol=1e-4) - + new = self.generateCsvArray(out, accelArray, removeMean=removeMean, meanSpan=meanSpan) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 - def testCalibratedRollingMeanRemoval(self): + np.testing.assert_equal(new[1:], old[1:]) + + def testCalibratedRollingMeanRemoval(self, accelArray, out): """ Test regular export, calibrated, with the rolling mean removed from the data. """ - out = StringIO() - accel = self.dataset.channels[8].getSession() - accel.removeMean = True - accel.rollingMeanSpan = 5000000 - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + removeMean = True + meanSpan = 5000000 + + accelArray.removeMean = removeMean + accelArray.rollingMeanSpan = meanSpan + + new = self.generateCsvArray(out, accelArray, removeMean=removeMean, meanSpan=meanSpan) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 - np.testing.assert_allclose(new.T, old, rtol=1e-4) + np.testing.assert_equal(new[1:], old[1:]) - - def testCalibratedTotalMeanRemoval(self): + def testCalibratedTotalMeanRemoval(self, accelArray, out): """ Test regular export, with the total mean removed from the data. """ - out = StringIO() - accel = self.dataset.channels[8].getSession() - accel.removeMean = True - accel.rollingMeanSpan = -1 - accel.exportCsv(out) - out.seek(0) - new = np.genfromtxt(out, delimiter=', ') - old = accel.__getitem__(slice(None), display=True) + removeMean = True + meanSpan = -1 - np.testing.assert_allclose(new.T, old, rtol=1e-4) - + accelArray.removeMean = removeMean + accelArray.rollingMeanSpan = meanSpan -#=============================================================================== + new = self.generateCsvArray(out, accelArray, removeMean=removeMean, meanSpan=meanSpan) + old = accelArray.__getitem__(slice(None), display=True) + old = np.round(1e6*old)/1e6 + + np.testing.assert_equal(new[1:], old[1:]) + + def testTimestamps(self, accelArray, out): + """ Tests the timestamps, which are the same on all exports + """ + + new = self.generateCsvArray(out, accelArray) + old = accelArray[:] + old = np.round(1e6*old)/1e6 + + np.testing.assert_allclose(new[0], old[0], rtol=1e-10) + + +# =============================================================================== # -#=============================================================================== +# ============================================================================== DEFAULTS = { "sensors": { 0x00: {"name": "832M1 Accelerometer"}, 0x01: {"name": "MPL3115 Temperature/Pressure"} }, - + "channels": { 0x00: {"name": "Accelerometer XYZ", - # "parser": struct.Struct("