Baby Language Lab Scripts
A collection of data processing tools.
 All Classes Namespaces Files Functions Variables Pages
reliability_exporter.py
Go to the documentation of this file.
1 ## @package parsers.reliability_exporter
2 
3 import csv
4 import logging
5 import traceback
6 
7 from utils.ui_utils import UIUtils
8 from db.bll_database import BLLDatabase, DBConstants
9 
10 ## The class writes details about a particular check run (from the Reliability App) to a CSV file.
11 class ReliabilityExporter(object):
12  ## Constructor
13  # @param self
14  # @param check_run_id (int) the db_id of the run we're writing details for
15  # @param filename (string) full path to the destination CSV file
16  def __init__(self, check, filename):
17  #ensure file suffix is present
18  if not filename.lower().endswith('.csv'):
19  filename += '.csv'
20 
21  self.check = check
22  self.filename = filename
23  self.logger = logging.getLogger(__name__)
24 
25  ## Performs the write to the CSV file.
26  # @param self
27  # @returns (boolean) True if write was successful, False if an error occurred
28  def export(self):
29  success = True #whether write succeeded or not
30 
31  try:
32  #write some basic info about the check
33  csv_file = open(self.filename, 'wb')
34  csv_writer = csv.writer(csv_file)
35  csv_writer.writerow(['Check name: "%s"' % (self.check.name)])
36  csv_writer.writerow(['Last run on %s' % (UIUtils.get_db_timestamp_str(str(self.check.last_run)))])
37  csv_writer.writerow(['Created on %s' % (UIUtils.get_db_timestamp_str(str(self.check.created)))])
38  csv_writer.writerow(['TRS / CSV file: %s' % (self.check.input_filename)])
39  csv_writer.writerow(['WAV file: %s' % (self.check.wav_filename)])
40  csv_writer.writerow(['Number of Segs: %s' % (self.check.num_segs)])
41  csv_writer.writerow(['Default context padding (sec): %s' % (self.check.default_context_padding)])
42  csv_writer.writerow(['Randomly Pick Segments: %s' % (str(self.check.pick_randomly)) ])
43 
44  #write filter descriptions
45  if self.check.filters:
46  csv_writer.writerow(['Filters:'])
47  for cur_filter in self.check.filters:
48  csv_writer.writerow([cur_filter.get_filter_desc_str()])
49  else:
50  csv_writer.writerow(['Filters: None'])
51 
52  #write the actual data from the tests
53  csv_writer.writerow([])
54  headers = ['Test Number', 'LENA Start Time (w/o padding)', 'LENA End Time (w/o padding)', 'Context Padding', 'User-Adjusted Start Time', 'User-Adjusted End Time', 'Uncertain/Other', 'Syllables (with context)', 'Syllables (w/o context)', 'Actual Codes', 'Category Selection', 'Category Correct']
55  csv_writer.writerow(headers) #write out the column headings
56 
57  #run through all of the tests, writing their data to the file and keeping track of how many times the user's selection was correct
58  correct_count = 0
59  db = BLLDatabase()
60  for i in range(self.check.num_segs):
61  cur_test = self.check.tests[i]
62  row = []
63 
64  row.append(str(i + 1))
65  row.append('%0.3f' % (cur_test.seg.start))
66  row.append('%0.3f' % (cur_test.seg.end))
67  row.append(str(cur_test.context_padding))
68  row.append('%0.3f' % (cur_test.seg.user_adj_start))
69  row.append('%0.3f' % (cur_test.seg.user_adj_end))
70  row.append(str(bool(cur_test.is_uncertain)))
71  row.append(str(cur_test.syllables_w_context))
72  row.append(str(cur_test.syllables_wo_context) if cur_test.syllables_wo_context != None else '')
73 
74  actual_codes = self._get_actual_speaker_codes(cur_test, db)
75  codes_str = ''
76  for cur_codeinfo in actual_codes:
77  codes_str += cur_codeinfo.code + ' '
78  if codes_str.endswith(' '):
79  codes_str = codes_str[:-1]
80  row.append(codes_str)
81 
82  cat_sel = self._get_cat_sel(cur_test)
83  row.append(cat_sel.disp_desc)
84 
85  cat_correct = self._get_cat_correct(cur_test, actual_codes, cat_sel)
86  correct_count += int(cat_correct)
87  row.append(str(bool(cat_correct)))
88 
89  csv_writer.writerow(row)
90 
91  db.close()
92  csv_writer.writerow([])
93  csv_writer.writerow(['Ratio correct: %0.2f' % ( float(correct_count) / float(self.check.num_segs) )])
94  csv_file.close()
95  success = True
96 
97  except Exception as err:
98  self.logger.error('Error exporting check: %s' % (err))
99 
100  return success
101 
102  def _get_cat_sel(self, test):
103  return DBConstants.COMBOS[DBConstants.COMBO_GROUPS.RELIABILITY_CATEGORIES][test.category_input]
104 
105  def _get_actual_speaker_codes(self, test, db):
106  codes = []
107  actual_code_rows = db.select('segs_to_speaker_codes rel join speaker_codes c on rel.speaker_code_id = c.id',
108  ['c.code'],
109  'rel.seg_id=?',
110  [test.seg.db_id],
111  )
112 
113  #lookup the corresponding CodeInfo object and append it to the list
114  for cur_row in actual_code_rows:
115  codes.append(DBConstants.SPEAKER_CODES.get_option(cur_row[0]))
116 
117  return codes
118 
119  ## Determines whether or not the users's selected option for a given test is correct.
120  # @param self
121  # @param cur_test (dictionary) a sub-dictionary from to get_tests_validation_data() - containing data about the test determine correctness for
122  # @param test_fcn (function) a function that tests whether an actual speaker code matches the user's combo selection. Function accepts a CodeInfo object representing an actual speaker code, and returns a boolean (False if user was incorrect, True if they were correct).
123  # @returns (boolean) True if one of the elements in cur_test's 'actual codes' list correctly corresponds to its 'category_input'
124  def search_actual_codes(self, cur_test, test_fcn, actual_codes):
125  result = False
126 
127  #run the function for each of the actual speaker codes
128  i = 0
129  while not result and i < len(actual_codes):
130  result = test_fcn(actual_codes[i])
131  i += 1
132 
133  return result
134 
135  ## This method updates the 'category_valid' key of a sub-dictionary obtained from get_tests_validation_data().
136  # i.e. it determines whether or not the user-selected option matches a speaker.
137  def _get_cat_correct(self, test, actual_codes, cat_sel):
138  #map category options to functions that check if they are correct
139  #each function accepts a codeinfo object and returns a boolean
140  cats_enum = DBConstants.COMBO_OPTIONS[DBConstants.COMBO_GROUPS.RELIABILITY_CATEGORIES]
141  validate_fcns = {
142  cats_enum.SILENCE: lambda speaker_code: speaker_code.is_speaker_type(DBConstants.SPEAKER_TYPES.SILENCE),
143 
144  cats_enum.OVERLAPPING_SPEECH: lambda speaker_code: speaker_code.has_property(DBConstants.SPEAKER_PROPS.OVERLAPPING),
145 
146  cats_enum.MEDIA: lambda speaker_code: speaker_code.has_property(DBConstants.SPEAKER_PROPS.MEDIA),
147 
148  cats_enum.TARGET_CHILD: lambda speaker_code: speaker_code.is_speaker_type(DBConstants.SPEAKER_TYPES.TARGET_CHILD),
149 
150  cats_enum.OTHER_CHILD: lambda speaker_code: speaker_code.is_speaker_type(DBConstants.SPEAKER_TYPES.OTHER_CHILD),
151 
152  cats_enum.FEMALE_ADULT: lambda speaker_code: speaker_code.is_speaker_type(DBConstants.SPEAKER_TYPES.FEMALE_ADULT),
153 
154  cats_enum.MALE_ADULT: lambda speaker_code: speaker_code.is_speaker_type(DBConstants.SPEAKER_TYPES.MALE_ADULT),
155 
156  cats_enum.DISTANT: lambda speaker_code: speaker_code.is_distance(DBConstants.SPEAKER_DISTANCES.FAR),
157 
158  cats_enum.NON_VERBAL_NOISE: lambda speaker_code: speaker_code.has_property(DBConstants.SPEAKER_PROPS.NON_VERBAL_NOISE),
159 
160  cats_enum.FUZ: lambda speaker_code: speaker_code.code == DBConstants.SPEAKER_CODES.get_option('FUZ').code,
161  }
162 
163  #set the 'category valid' key based upon the result from search_actual_codes()
164  cat_correct = self.search_actual_codes(test, #this is the sub-dictionary
165  validate_fcns[cat_sel.db_id], #this is the function that validates the <em>user's selected option</em> (not the actual code). The actual code's codeinfo object is passed to this function to determine whether or not the user is correct
166  actual_codes)
167 
168  return cat_correct