Commit f1d4926a authored by Marian Dovgialo's avatar Marian Dovgialo

#45153 Merge branch 'development' into 'master'

Releasing 1.1.3

See merge request !20
parents c009e152 262aef09
Pipeline #17044 passed with stages
in 1 minute and 16 seconds
......@@ -43,7 +43,7 @@ def get_config():
cfg.style = "pep440"
cfg.tag_prefix = ""
cfg.parentdir_prefix = "''"
cfg.versionfile_source = "obci/_version.py"
cfg.versionfile_source = "obci_readmanager/_version.py"
cfg.verbose = False
return cfg
......
......@@ -5,7 +5,7 @@
"""Mixin for readmanager to create MNE.Raw objects."""
import json
from typing import Union
from copy import copy
from copy import copy, deepcopy
import numpy
from ..signal.signal_constants import sample_type_from_numpy
......@@ -30,11 +30,19 @@ def add_stim_chnl(raw):
stim_data = numpy.zeros((1, len(raw.times)))
info = mne.create_info(['OBCI_STIM'], raw.info['sfreq'], ['stim'])
stim_raw = mne.io.RawArray(stim_data, info)
raw.add_channels([stim_raw], force_update_info=True)
raw.add_channels([stim_raw], force_update_info=False)
def _description_from_tag(tag):
tag_desc = copy(tag)
tag_custom_describtion = tag_desc['desc']
for key in tag_custom_describtion:
try:
value = tag_custom_describtion[key]
if ';' in value:
tag_custom_describtion[key] = value.replace(';', ':')
except TypeError:
pass
tag_desc.pop('start_timestamp')
tag_desc.pop('end_timestamp')
return json.dumps(tag_desc)
......@@ -47,9 +55,6 @@ def tags_from_mne_annotations(ans):
Returns tags (list of dicts).
"""
tags = []
orig_time = ans.orig_time
if orig_time is None:
orig_time = 0
for onset, duration, desc in zip(ans.onset, ans.duration, ans.description):
# try to load annotations, as they would be exported by ReadManager
# if there is no our annotations reformat them to tags
......@@ -59,8 +64,8 @@ def tags_from_mne_annotations(ans):
except (json.decoder.JSONDecodeError, AssertionError):
# MNE created not in OBCI
tag = {'name': desc, 'desc': {}, 'channels': ''}
tag['start_timestamp'] = onset - orig_time
tag['end_timestamp'] = onset + duration - orig_time
tag['start_timestamp'] = onset
tag['end_timestamp'] = onset + duration
tags.append(tag)
return tags
......@@ -132,9 +137,10 @@ class ReadManagerMNEMixin:
def from_mne(cls, mne_raw):
"""Read Raw mne object to ReadManager."""
assert isinstance(mne_raw, mne.io.BaseRaw)
mne_raw.drop_channels(['OBCI_STIM'])
data = mne_raw.get_data() * 1e6 # mne keeps signals in Volts
return cls._rm_from_mne_data(data, mne_raw.info, mne_raw.annotations)
mne_raw_copy = deepcopy(mne_raw)
mne_raw_copy.drop_channels(['OBCI_STIM'])
data = mne_raw_copy.get_data() * 1e6 # mne keeps signals in Volts
return cls._rm_from_mne_data(data, mne_raw_copy.info, mne_raw_copy.annotations)
@classmethod
def _rm_from_mne_data(cls, data, info, annotations=None):
......@@ -178,7 +184,7 @@ class ReadManagerMNEMixin:
except ValueError:
meas_date_us = 0
meas_date = numpy.array([meas_date_s, meas_date_us], dtype=numpy.int32)
meas_date = (meas_date_s, meas_date_us)
info['meas_date'] = meas_date
def mne_annotations(self):
......@@ -225,5 +231,5 @@ class ReadManagerMNEMixin:
add_stim_chnl(raw)
ans = self.mne_annotations()
events, _ = self.mne_events()
raw.annotations = ans
raw.set_annotations(ans)
raw.add_events(events)
......@@ -13,8 +13,14 @@ test_requirements = [
'pytest-catchlog>=1.2.2',
'flaky>=3.3.0',
'nose>=1.3.7',
'mne==0.14.1'
'mne~=0.17.0',
'scipy',
'matplotlib',
]
install_requires = ['mne~=0.17.0',
]
needs_pytest = {'pytest', 'test', 'ptr'}.intersection(sys.argv)
pytest_runner_requirement = ['pytest-runner>=2.9']
......@@ -44,15 +50,10 @@ setup(
packages=find_packages(exclude=['scripts', ]),
include_package_data=True,
exclude_package_data={'': ['.gitignore', '.gitlab-ci.yml']},
install_requires=[],
install_requires=install_requires,
tests_require=test_requirements,
setup_requires=setup_requires,
# entry_points={
# 'console_scripts': [
# 'obci_readmanager = obci_readmanager.cmd.run_readmanager_preset:run',
# ],
# },
extras_require={
'test': pytest_runner_requirement + test_requirements,
},
)
'test': pytest_runner_requirement + test_requirements,
},
)
......@@ -3,10 +3,13 @@ import os
import math
from obci_readmanager.signal_processing.balance.wii_preprocessing import wii_filter_signal, wii_downsample_signal, wii_cut_fragments
from obci_readmanager.signal_processing.balance.wii_analysis import (wii_COP_path, wii_max_sway_AP_MP, wii_mean_COP_sway_AP_ML,
wii_RMS_AP_ML, wii_confidence_ellipse_area, wii_mean_velocity,
wii_get_percentages_values)
from obci_readmanager.signal_processing.balance.wii_preprocessing import wii_filter_signal, wii_downsample_signal, \
wii_cut_fragments
from obci_readmanager.signal_processing.balance.wii_analysis import (wii_COP_path, wii_max_sway_AP_MP,
wii_mean_COP_sway_AP_ML,
wii_RMS_AP_ML, wii_confidence_ellipse_area,
wii_mean_velocity,
wii_get_percentages_values)
from obci_readmanager.signal_processing.balance.wii_read_manager import WBBReadManager
import matplotlib.pyplot as py
......@@ -35,7 +38,6 @@ def test_wii_analysis(plot=False):
def test_wbb_tutorial():
# inicjalizacja klasy WBBReadManager - utwórz obiekt podając na wejściu ścieżki do
# odpowiednich plików :
pth = __file__
......@@ -225,5 +227,6 @@ def test_wbb_tutorial_calculations():
assert math.isclose(bottom_right, 23.487261146496824)
assert math.isclose(bottom_left, 18.232484076433121)
if __name__ == '__main__':
test_wii_analysis(plot=True)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2016-2018 Braintech Sp. z o.o. [Ltd.] <http://www.braintech.pl>
# All rights reserved.
"""
>>> import numpy
>>> import time
import time
>>> from obci_readmanager.signal_processing.buffers import auto_blink_buffer as R
import numpy
>>> per, ch, sm = (4, 3, 20)
from obci_readmanager.signal_processing.buffers import auto_blink_buffer as R
from obci_readmanager.signal_processing.signal.data_generic_write_proxy import SamplePacket
>>> ts_step = 1/float(sm)
try:
from .test_auto_ring_buffer import ArrayCallback
except (SystemError, ImportError):
from test_auto_ring_buffer import ArrayCallback
>>> f = print_bufs
>>> b = R.AutoBlinkBuffer(0, 5, ch, sm, f, False)
class BlinkArrayCallback(ArrayCallback):
def __call__(self, blink, arr):
super(BlinkArrayCallback, self).__call__(arr.copy())
>>> t = time.time()
>>> ts = numpy.linspace(t, t + 200 * ts_step, 200)
def test_auto_blink_buffer():
per, ch, sm = (4, 3, 20)
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[0 * per], ts_step))#1-4
ts_step = 1 / float(sm)
callback = BlinkArrayCallback()
>>> b.handle_blink(B(ts[1 * per]))
AutoBlinkBuffer - Got blink before buffer is full. Ignore!
b = R.AutoBlinkBuffer(0, 5, ch, sm, callback, False)
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[1 * per + 1] ,ts_step))#5-8
t = time.time()
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[2 * per + 1], ts_step))#9-12
ts = numpy.linspace(t, t + 200 * ts_step, 200)
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[3 * per + 1], ts_step))#13-16
b.handle_sample_packet(get_sample_packet(per, ch, ts[0 * per], ts_step)) # 1-4
>>> b.handle_blink(B(ts[4 * per + 1]))
b.handle_blink(B(ts[1 * per]))
# AutoBlinkBuffer - Got blink before buffer is full. Ignore!
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[4 * per + 2], ts_step)) #17-20
b.handle_sample_packet(get_sample_packet(per, ch, ts[1 * per + 1], ts_step)) # 5-8
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[5 * per + 2], ts_step))#21-24
FUNC
[[ 18. 19. 20. 21. 22.]
[ 180. 190. 200. 210. 220.]
[ 1800. 1900. 2000. 2100. 2200.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[2 * per + 1], ts_step)) # 9-12
b.handle_sample_packet(get_sample_packet(per, ch, ts[3 * per + 1], ts_step)) # 13-16
>>> b.handle_blink(B(ts[6 * per + 2]))
b.handle_blink(B(ts[4 * per + 1]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[6 * per + 3], ts_step)) #25-28
b.handle_sample_packet(get_sample_packet(per, ch, ts[4 * per + 2], ts_step)) # 17-20
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[7 * per + 3], ts_step))# 29-32
FUNC
[[ 26. 27. 28. 29. 30.]
[ 260. 270. 280. 290. 300.]
[ 2600. 2700. 2800. 2900. 3000.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[5 * per + 2], ts_step)) # 21-24
callback.assert_called([[18., 19., 20., 21., 22.],
[180., 190., 200., 210., 220.],
[1800., 1900., 2000., 2100., 2200.]])
>>> b.handle_blink(B(ts[8 * per + 3]))
b.handle_blink(B(ts[6 * per + 2]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[8 * per + 4], ts_step)) #33-36
b.handle_sample_packet(get_sample_packet(per, ch, ts[6 * per + 3], ts_step)) # 25-28
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[9 * per + 4], ts_step)) #37-40
FUNC
[[ 34. 35. 36. 37. 38.]
[ 340. 350. 360. 370. 380.]
[ 3400. 3500. 3600. 3700. 3800.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[7 * per + 3], ts_step)) # 29-32
callback.assert_called([[26., 27., 28., 29., 30.],
[260., 270., 280., 290., 300.],
[2600., 2700., 2800., 2900., 3000.]])
b.handle_blink(B(ts[8 * per + 3]))
>>> zero_count()
b.handle_sample_packet(get_sample_packet(per, ch, ts[8 * per + 4], ts_step)) # 33-36
>>> b = R.AutoBlinkBuffer(0, 10, ch, sm, f, False)
b.handle_sample_packet(get_sample_packet(per, ch, ts[9 * per + 4], ts_step)) # 37-40
callback.assert_called(
[[34., 35., 36., 37., 38.],
[340., 350., 360., 370., 380.],
[3400., 3500., 3600., 3700., 3800.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[10 * per + 4], ts_step))#1-4
zero_count()
>>> b.handle_blink(B(ts[11 * per + 4]))
AutoBlinkBuffer - Got blink before buffer is full. Ignore!
b = R.AutoBlinkBuffer(0, 10, ch, sm, callback, False)
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[11 * per + 5], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[10 * per + 4], ts_step)) # 1-4
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[12 * per + 5], ts_step))
b.handle_blink(B(ts[11 * per + 4]))
# AutoBlinkBuffer - Got blink before buffer is full. Ignore!
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[13 * per + 5], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[11 * per + 5], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[14 * per + 5], ts_step)) #17-20
b.handle_sample_packet(get_sample_packet(per, ch, ts[12 * per + 5], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[15 * per + 5], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[13 * per + 5], ts_step))
>>> b.handle_blink(B(ts[16 * per + 5]))
b.handle_sample_packet(get_sample_packet(per, ch, ts[14 * per + 5], ts_step)) # 17-20
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[16 * per + 6], ts_step)) #25-28
b.handle_sample_packet(get_sample_packet(per, ch, ts[15 * per + 5], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[17 * per + 6], ts_step))
b.handle_blink(B(ts[16 * per + 5]))
>>> b.handle_blink(B(ts[18 * per + 6]))
b.handle_sample_packet(get_sample_packet(per, ch, ts[16 * per + 6], ts_step)) # 25-28
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[18 * per + 7], ts_step)) #33-36
FUNC
[[ 26. 27. 28. 29. 30. 31. 32. 33. 34. 35.]
[ 260. 270. 280. 290. 300. 310. 320. 330. 340. 350.]
[ 2600. 2700. 2800. 2900. 3000. 3100. 3200. 3300. 3400. 3500.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[17 * per + 6], ts_step))
b.handle_blink(B(ts[18 * per + 6]))
>>> b.handle_blink(B(ts[19 * per + 7]))
b.handle_sample_packet(get_sample_packet(per, ch, ts[18 * per + 7], ts_step)) # 33-36
callback.assert_called(
[[26., 27., 28., 29., 30., 31., 32., 33., 34., 35.],
[260., 270., 280., 290., 300., 310., 320., 330., 340., 350.],
[2600., 2700., 2800., 2900., 3000., 3100., 3200., 3300., 3400., 3500.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[19 * per + 8], ts_step))
b.handle_blink(B(ts[19 * per + 7]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[20 * per + 8], ts_step))
FUNC
[[ 34. 35. 36. 37. 38. 39. 40. 41. 42. 43.]
[ 340. 350. 360. 370. 380. 390. 400. 410. 420. 430.]
[ 3400. 3500. 3600. 3700. 3800. 3900. 4000. 4100. 4200. 4300.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[19 * per + 8], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[21 * per + 8], ts_step))
FUNC
[[ 38. 39. 40. 41. 42. 43. 44. 45. 46. 47.]
[ 380. 390. 400. 410. 420. 430. 440. 450. 460. 470.]
[ 3800. 3900. 4000. 4100. 4200. 4300. 4400. 4500. 4600. 4700.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[20 * per + 8], ts_step))
callback.assert_called(
[[34., 35., 36., 37., 38., 39., 40., 41., 42., 43.],
[340., 350., 360., 370., 380., 390., 400., 410., 420., 430.],
[3400., 3500., 3600., 3700., 3800., 3900., 4000., 4100., 4200., 4300.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[22 * per + 8], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[21 * per + 8], ts_step))
callback.assert_called(
[[38., 39., 40., 41., 42., 43., 44., 45., 46., 47.],
[380., 390., 400., 410., 420., 430., 440., 450., 460., 470.],
[3800., 3900., 4000., 4100., 4200., 4300., 4400., 4500., 4600., 4700.]])
b.handle_sample_packet(get_sample_packet(per, ch, ts[22 * per + 8], ts_step))
>>> zero_count()
zero_count()
>>> b = R.AutoBlinkBuffer(0, 10, ch, sm, f, False)
b = R.AutoBlinkBuffer(0, 10, ch, sm, callback, False)
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[23 * per + 8], ts_step))#1-4
b.handle_sample_packet(get_sample_packet(per, ch, ts[23 * per + 8], ts_step)) # 1-4
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[24 * per + 8], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[24 * per + 8], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[25 * per + 8], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[25 * per + 8], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[26 * per + 8], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[26 * per + 8], ts_step))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[27 * per + 8], ts_step)) #17-20
b.handle_sample_packet(get_sample_packet(per, ch, ts[27 * per + 8], ts_step)) # 17-20
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[28 * per + 8], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[28 * per + 8], ts_step))
>>> b.handle_blink(B(ts[29 * per + 2]))
b.handle_blink(B(ts[29 * per + 2]))
>>> b.handle_blink(B(ts[29 * per + 3]))
b.handle_blink(B(ts[29 * per + 3]))
>>> b.handle_blink(B(ts[29 * per + 4]))
b.handle_blink(B(ts[29 * per + 4]))
>>> b.handle_blink(B(ts[29 * per + 5]))
b.handle_blink(B(ts[29 * per + 5]))
>>> b.handle_blink(B(ts[29 * per + 6]))
b.handle_blink(B(ts[29 * per + 6]))
>>> b.handle_blink(B(ts[29 * per + 7]))
b.handle_blink(B(ts[29 * per + 7]))
>>> b.handle_blink(B(ts[29 * per + 8]))
b.handle_blink(B(ts[29 * per + 8]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[29 * per + 10], ts_step)) #25-28
FUNC
[[ 19. 20. 21. 22. 23. 24. 25. 26. 27. 28.]
[ 190. 200. 210. 220. 230. 240. 250. 260. 270. 280.]
[ 1900. 2000. 2100. 2200. 2300. 2400. 2500. 2600. 2700. 2800.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[29 * per + 10], ts_step)) # 25-28
callback.assert_called(
[[19., 20., 21., 22., 23., 24., 25., 26., 27., 28.],
[190., 200., 210., 220., 230., 240., 250., 260., 270., 280.],
[1900., 2000., 2100., 2200., 2300., 2400., 2500., 2600., 2700., 2800.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[30 * per + 10], ts_step))
FUNC
[[ 20. 21. 22. 23. 24. 25. 26. 27. 28. 29.]
[ 200. 210. 220. 230. 240. 250. 260. 270. 280. 290.]
[ 2000. 2100. 2200. 2300. 2400. 2500. 2600. 2700. 2800. 2900.]]
FUNC
[[ 21. 22. 23. 24. 25. 26. 27. 28. 29. 30.]
[ 210. 220. 230. 240. 250. 260. 270. 280. 290. 300.]
[ 2100. 2200. 2300. 2400. 2500. 2600. 2700. 2800. 2900. 3000.]]
FUNC
[[ 22. 23. 24. 25. 26. 27. 28. 29. 30. 31.]
[ 220. 230. 240. 250. 260. 270. 280. 290. 300. 310.]
[ 2200. 2300. 2400. 2500. 2600. 2700. 2800. 2900. 3000. 3100.]]
FUNC
[[ 23. 24. 25. 26. 27. 28. 29. 30. 31. 32.]
[ 230. 240. 250. 260. 270. 280. 290. 300. 310. 320.]
[ 2300. 2400. 2500. 2600. 2700. 2800. 2900. 3000. 3100. 3200.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[30 * per + 10], ts_step))
callback.assert_called(
[[20., 21., 22., 23., 24., 25., 26., 27., 28., 29.],
[200., 210., 220., 230., 240., 250., 260., 270., 280., 290.],
[2000., 2100., 2200., 2300., 2400., 2500., 2600., 2700., 2800., 2900.]])
callback.assert_called(
[[21., 22., 23., 24., 25., 26., 27., 28., 29., 30.],
[210., 220., 230., 240., 250., 260., 270., 280., 290., 300.],
[2100., 2200., 2300., 2400., 2500., 2600., 2700., 2800., 2900., 3000.]])
callback.assert_called(
[[22., 23., 24., 25., 26., 27., 28., 29., 30., 31.],
[220., 230., 240., 250., 260., 270., 280., 290., 300., 310.],
[2200., 2300., 2400., 2500., 2600., 2700., 2800., 2900., 3000., 3100.]])
callback.assert_called(
[[23., 24., 25., 26., 27., 28., 29., 30., 31., 32.],
[230., 240., 250., 260., 270., 280., 290., 300., 310., 320.],
[2300., 2400., 2500., 2600., 2700., 2800., 2900., 3000., 3100., 3200.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[31 * per + 10], ts_step)) #33-36
FUNC
[[ 25. 26. 27. 28. 29. 30. 31. 32. 33. 34.]
[ 250. 260. 270. 280. 290. 300. 310. 320. 330. 340.]
[ 2500. 2600. 2700. 2800. 2900. 3000. 3100. 3200. 3300. 3400.]]
FUNC
[[ 26. 27. 28. 29. 30. 31. 32. 33. 34. 35.]
[ 260. 270. 280. 290. 300. 310. 320. 330. 340. 350.]
[ 2600. 2700. 2800. 2900. 3000. 3100. 3200. 3300. 3400. 3500.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[31 * per + 10], ts_step)) # 33-36
callback.assert_called(
[[25., 26., 27., 28., 29., 30., 31., 32., 33., 34.],
[250., 260., 270., 280., 290., 300., 310., 320., 330., 340.],
[2500., 2600., 2700., 2800., 2900., 3000., 3100., 3200., 3300., 3400.]])
callback.assert_called(
[[26., 27., 28., 29., 30., 31., 32., 33., 34., 35.],
[260., 270., 280., 290., 300., 310., 320., 330., 340., 350.],
[2600., 2700., 2800., 2900., 3000., 3100., 3200., 3300., 3400., 3500.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[32 * per + 10], ts_step)) #33-36
b.handle_sample_packet(get_sample_packet(per, ch, ts[32 * per + 10], ts_step)) # 33-36
>>> b.handle_blink(B(ts[33 * per + 10]))
b.handle_blink(B(ts[33 * per + 10]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[33 * per + 11], ts_step))
b.handle_sample_packet(get_sample_packet(per, ch, ts[33 * per + 11], ts_step))
>>> b.handle_blink(B(ts[34 * per + 11]))
b.handle_blink(B(ts[34 * per + 11]))
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[34 * per + 12], ts_step))#41-44
b.handle_sample_packet(get_sample_packet(per, ch, ts[34 * per + 12], ts_step)) # 41-44
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[35 * per + 12], ts_step))
FUNC
[[ 42. 43. 44. 45. 46. 47. 48. 49. 50. 51.]
[ 420. 430. 440. 450. 460. 470. 480. 490. 500. 510.]
[ 4200. 4300. 4400. 4500. 4600. 4700. 4800. 4900. 5000. 5100.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[35 * per + 12], ts_step))
callback.assert_called(
[[42., 43., 44., 45., 46., 47., 48., 49., 50., 51.],
[420., 430., 440., 450., 460., 470., 480., 490., 500., 510.],
[4200., 4300., 4400., 4500., 4600., 4700., 4800., 4900., 5000., 5100.]])
>>> b.handle_sample_packet(get_sample_packet(per, ch, ts[36 * per + 12], ts_step))
FUNC
[[ 46. 47. 48. 49. 50. 51. 52. 53. 54. 55.]
[ 460. 470. 480. 490. 500. 510. 520. 530. 540. 550.]
[ 4600. 4700. 4800. 4900. 5000. 5100. 5200. 5300. 5400. 5500.]]
b.handle_sample_packet(get_sample_packet(per, ch, ts[36 * per + 12], ts_step))
callback.assert_called(
[[46., 47., 48., 49., 50., 51., 52., 53., 54., 55.],
[460., 470., 480., 490., 500., 510., 520., 530., 540., 550.],
[4600., 4700., 4800., 4900., 5000., 5100., 5200., 5300., 5400., 5500.]])
"""
import numpy
from obci_readmanager.signal_processing.signal.data_raw_write_proxy import SamplePacket
COUNT = 0
......@@ -246,23 +248,6 @@ def get_sample_packet(per, ch_num, ts_start, ts_step):
for i in range(per):
COUNT += 1
for j in range(ch_num):
samples[i][j] = (10**j) * COUNT
samples[i][j] = (10 ** j) * COUNT
timestamps[i] = ts_start + ts_step * i
return SamplePacket(samples, timestamps)
def print_bufs(blink, bufs):
print("FUNC")
print(bufs)
def run():
import sys
import doctest
res = doctest.testmod(sys.modules[__name__])
if res.failed == 0:
print("All tests succeeded!")
if __name__ == '__main__':
run()
......@@ -2,94 +2,107 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2016-2018 Braintech Sp. z o.o. [Ltd.] <http://www.braintech.pl>
# All rights reserved.
from unittest.mock import Mock
"""
>>> from obci_readmanager.signal_processing.buffers import auto_ring_buffer as R
>>> per, ch = (4, 10)
>>> f = print_bufs
>>> b = R.AutoRingBuffer(10, 5, 5, ch, f, False)
>>> v1 = get_sample_packet(per, ch, 1)
>>> v2 = get_sample_packet(per, ch, 10)
>>> v3 = get_sample_packet(per, ch, 100)
>>> v4 = get_sample_packet(per, ch, 1000)
>>> b.handle_sample_packet(v1)
>>> b.handle_sample_packet(v2)
>>> b.handle_sample_packet(v3)
FUNC
[[ 1. 2. 3. 4. 10.]
[ 2. 3. 4. 5. 20.]
[ 3. 4. 5. 6. 30.]
[ 4. 5. 6. 7. 40.]
[ 5. 6. 7. 8. 50.]
[ 6. 7. 8. 9. 60.]
[ 7. 8. 9. 10. 70.]
[ 8. 9. 10. 11. 80.]
[ 9. 10. 11. 12. 90.]
[ 10. 11. 12. 13. 100.]]
>>> b.handle_sample_packet(v4)
FUNC
[[ 11. 12. 13. 100. 101.]
[ 21. 22. 23. 200. 201.]
[ 31. 32. 33. 300. 301.]
[ 41. 42. 43. 400. 401.]
[ 51. 52. 53. 500. 501.]
[ 61. 62. 63. 600. 601.]
[ 71. 72. 73. 700. 701.]
[ 81. 82. 83. 800. 801.]
[ 91. 92. 93. 900. 901.]
[ 101. 102. 103. 1000. 1001.]]
>>> b = R.AutoRingBuffer(10, 3, 5, ch, f, False)
>>> b.handle_sample_packet(v1)
>>> b.handle_sample_packet(v2)
>>> b.handle_sample_packet(v3)
FUNC
[[ 1. 2. 3.]
[ 2. 3. 4.]
[ 3. 4. 5.]
[ 4. 5. 6.]
[ 5. 6. 7.]
[ 6. 7. 8.]
[ 7. 8. 9.]
[ 8. 9. 10.]
[ 9. 10. 11.]
[ 10. 11. 12.]]
>>> b.handle_sample_packet(v4)
FUNC
[[ 11. 12. 13.]
[ 21. 22. 23.]
[ 31. 32. 33.]
[ 41. 42. 43.]
[ 51. 52. 53.]
[ 61. 62. 63.]
[ 71. 72. 73.]
[ 81. 82. 83.]
[ 91. 92. 93.]
[ 101. 102. 103.]]
"""
import numpy
from obci_readmanager.signal_processing.signal.data_raw_write_proxy import SamplePacket
class ArrayCallback(Mock):
def __init__(self, *args, **kwargs):
super(ArrayCallback, self).__init__(*args, **kwargs)
self.current_check = 0
def assert_called(self, arr2):
assert self.current_check < self.call_count
arr1 = numpy.array(self.call_args_list[self.current_check][0][0])
arr2 = numpy.array(arr2)
assert numpy.array_equal(arr1, arr2)
self.current_check += 1
def __call__(self, arr):
super(ArrayCallback, self).__call__(arr.copy())
def test_auto_ring_buffer():
from obci_readmanager.signal_processing.buffers import auto_ring_buffer as R
per, ch = (4, 10)
callback = ArrayCallback()
b = R.AutoRingBuffer(10, 5, 5, ch, callback, False)
v1 = get_sample_packet(per, ch, 1)
v2 = get_sample_packet(per, ch, 10)
v3 = get_sample_packet(per, ch, 100)
v4 = get_sample_packet(per, ch, 1000)
b.handle_sample_packet(v1)