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