You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

100 lines
3.7 KiB

from models import PeakModel, DropModel, TroughModel, JumpModel, GeneralModel
from models import PeakModelState, DropModelState, TroughModelState, JumpModelState, GeneralModelState
import utils.meta
import aiounittest
from analytic_unit_manager import AnalyticUnitManager
from collections import namedtuple
TestData = namedtuple('TestData', ['uid', 'type', 'values', 'segments'])
def get_random_id() -> str:
return str(id(list()))
class TestDataset(aiounittest.AsyncTestCase):
timestep = 50 #ms
def _fill_task(self, uid, data, task_type, analytic_unit_type, segments=None, cache=None):
task = {
'analyticUnitId': uid,
'type': task_type,
'payload': {
'data': data,
'from': data[0][0],
'to': data[-1][0],
'analyticUnitType': analytic_unit_type,
'detector': 'pattern',
'cache': cache
},
'_id': get_random_id()
}
if segments: task['payload']['segments'] = segments
return task
def _convert_values(self, values) -> list:
from_t = 0
to_t = len(values) * self.timestep
return list(zip(range(from_t, to_t, self.timestep), values))
def _index_to_test_time(self, idx) -> int:
return idx * self.timestep
def _get_learn_task(self, test_data):
uid, analytic_unit_type, values, segments = test_data
data = self._convert_values(values)
segments = [{
'analyticUnitId': uid,
'from': self._index_to_test_time(s[0]),
'to': self._index_to_test_time(s[1]),
'labeled': True,
'deleted': False
} for s in segments]
return self._fill_task(uid, data, 'LEARN', analytic_unit_type, segments=segments)
def _get_detect_task(self, test_data, cache):
uid, analytic_unit_type, values, _ = test_data
data = self._convert_values(values)
return self._fill_task(uid, data, 'DETECT', analytic_unit_type, cache=cache)
def _get_test_dataset(self, pattern) -> tuple:
"""
pattern name: ([dataset values], [list of segments])
segment - (begin, end) - indexes in dataset values
returns dataset in format (data: List[int], segments: List[List[int]])
"""
datasets = {
'PEAK': ([0, 0, 1, 2, 3, 4, 3, 2, 1, 0, 0], [[2, 8]]),
'JUMP': ([0, 0, 1, 2, 3, 4, 4, 4], [[1, 6]]),
'DROP': ([4, 4, 4, 3, 2, 1, 0, 0], [[1, 6]]),
'TROUGH': ([4, 4, 3, 2, 1, 0, 1, 2, 3, 4, 4], [[1, 9]]),
'GENERAL': ([0, 0, 1, 2, 3, 4, 3, 2, 1, 0, 0], [[2, 8]])
}
return datasets[pattern]
async def _learn(self, task, manager=None) -> dict:
if not manager: manager = AnalyticUnitManager()
result = await manager.handle_analytic_task(task)
return result['payload']['cache']
async def _detect(self, task, manager=None) -> dict:
if not manager: manager = AnalyticUnitManager()
result = await manager.handle_analytic_task(task)
return result
async def _test_detect(self, test_data, manager=None):
learn_task = self._get_learn_task(test_data)
cache = await self._learn(learn_task, manager)
detect_task = self._get_detect_task(test_data, cache)
result = await self._detect(detect_task, manager)
return result
async def test_unit_manager(self):
test_data = TestData(get_random_id(), 'PEAK', [0,1,2,5,10,5,2,1,1,1,0,0,0,0], [[1,7]])
manager = AnalyticUnitManager()
with_manager = await self._test_detect(test_data, manager)
without_manager = await self._test_detect(test_data)
self.assertEqual(with_manager, without_manager)