1import os 2import numpy as np 3from ont_fast5_api.fast5_file import Fast5File 4from ont_fast5_api.analysis_tools.event_detection import EventDetectionTools 5from test.helpers import TestFast5ApiHelper, test_data 6 7 8class TestEventDetectionTools(TestFast5ApiHelper): 9 10 def test_001_read_events(self): 11 # Check that it is recognized properly. 12 fname = os.path.join(test_data, 'read_file_v1.0_single.fast5') 13 with EventDetectionTools(fname, mode='r', ) as fh: 14 self.assertTrue(fh.has_event_data) 15 self.assertTrue(fh.has_event_data(read_number=59)) 16 self.assertEqual('EventDetection_000', fh.group_name) 17 data, attrs = fh.get_event_data() 18 self.assertDictEqual({'read_number': 59, 19 'strand_id': 60, 20 'start_mux': 1, 21 'end_mux': 1, 22 'start_time': 32463855, 23 'duration': 729468}, attrs) 24 self.assertEqual(7875, data.size) 25 self.assertEqual(118, data[0]['length']) 26 data, attrs = fh.get_event_data(time_in_seconds=True) 27 self.assertEqual(0.0236, data[0]['length']) 28 29 def test_002_write_events(self): 30 fname = os.path.join(self.save_path, 'test.fast5') 31 with Fast5File(fname, 'w') as fh: 32 fh.add_channel_info({'channel_number': 1, 'sampling_rate': 4000}) 33 fh.add_read(12, 'unique_snowflake', 12345, 111, 0, 120.75) 34 with EventDetectionTools(fh, group_name='EventDetection_000', 35 meta={'name': 'test', 'version': '0.1.0'}) as evdet: 36 data = np.zeros(100, dtype=[('start', int), ('length', int), ('mean', float), ('stdv', float)]) 37 read_attrs = {'read_number': 12} 38 evdet.set_event_data(data, read_attrs) 39 with Fast5File(fname, 'r') as fh: 40 self.assertEqual(1, len(fh.status.read_info)) 41 read_info = fh.status.read_info[0] 42 self.assertEqual(12, read_info.read_number) 43 group = fh.get_latest_analysis('EventDetection') 44 self.assertEqual('EventDetection_000', group) 45 with EventDetectionTools(fh) as evdet: 46 self.assertTrue(evdet.has_event_data()) 47 data, attrs = evdet.get_event_data() 48 self.assertDictEqual({u'read_number': 12, 49 u'read_id': 'unique_snowflake', 50 u'start_time': 12345, 51 u'duration': 111, 52 u'start_mux': 0, 53 u'median_before': 120.75}, attrs) 54 self.assertEqual(100, data.size) 55