model Module¶

class Calibration(Procedure):

    def __init__(self, startlevel:int=60, signal_length:int=10, headphone_name:str="Sennheiser_HDA200", **additional_data):
        """Process for calibrating system.

        Args:
            startlevel (int, optional): starting level of procedure in dB HL. Defaults to 60.
            signal_length (int, optional): length of played signals in seconds. Defaults to 10.
            headphone_name (str, optional): Name of headphone model being used. Defaults to Sennheiser_HDA200.
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done
        """
        super().__init__(startlevel, signal_length, headphone_name=headphone_name, calibrate=False)      
        self.tempfile = self.create_temp_csv(id="", **additional_data) # create a temporary file to store level at frequencies
        self.generator = self.get_next_freq()
        self.dbspl = self.level + self.retspl[self.frequency]

    def get_next_freq(self):
        """Generator that goes through all frequencies twice.
        Changes self.side to 'r' after going through all frequencies the first time.

        Yields:
            int: frequency
        """
        self.side = 'l'
        frequency = 125
        while frequency <= 8000:
            yield frequency
            frequency *= 2

        frequency = 125
        self.side = 'r'
        while frequency <= 8000:
            yield frequency
            frequency *= 2

    def play_one_freq(self)->tuple:
        """Get the next frequency and play it.

        Returns:
            bool: False if no more frequencies left
            int: current frequency
            float: expected SPL value in dB
        """
        self.ap.stop()

        try:
            self.frequency = next(self.generator)
        except:
            return False, self.frequency, self.dbspl

        self.dbspl = self.level + self.retspl[self.frequency]
        print(f"Side {self.side} at {self.frequency} Hz: The SPL value should be {self.dbspl} dB.")
        self.ap.play_beep(self.frequency, self.dbhl_to_volume(self.level), self.signal_length, self.side)
        if self.frequency >= 8000 and self.side == 'r':
            return False, self.frequency, self.dbspl
        else:
            return True, self.frequency, self.dbspl

    def repeat_freq(self):
        """Repeats the last played frequency.
        """
        self.ap.stop()
        print(f" Repeating side {self.side} at {self.frequency} Hz: The SPL value should be {self.dbspl} dB.")
        self.ap.play_beep(self.frequency, self.dbhl_to_volume(self.level), self.signal_length, self.side)

    def set_calibration_value(self, measured_value:float):
        """Rights the given calibration value into temporary CSV file

        Args:
            measured_value (float): measured SPL value in dB
        """
        value = measured_value - self.dbspl
        self.add_to_temp_csv(str(value), str(self.frequency), self.side, self.tempfile)

    def finish_calibration(self):
        """Makes a permanent CSV file from the temporary file that overwrites calibration.csv.

        Args:
            temp_filename (str): name of temporary CSV file
        """
        self.ap.stop()
        # read temp file
        with open(self.tempfile, mode='r', newline='') as temp_file:
            dict_reader = csv.DictReader(temp_file)
            rows = list(dict_reader)

        filename = "calibration.csv"

        with open(filename, mode='w', newline='') as final_file:
            dict_writer = csv.DictWriter(final_file, fieldnames=self.freq_bands)
            dict_writer.writeheader()
            dict_writer.writerows(rows)

        print("Datei gespeicher als " + filename)

    def stop_playing(self):
        """Stops the audio player.
        """
        self.ap.stop()

class Familiarization(Procedure):

    def __init__(self, startlevel:int=40, signal_length:int=1, headphone_name:str="Sennheiser_HDA200",calibrate:bool=True, id:str="", **additional_data):
        """Creates the Familiarization process.

        Args:
            startlevel (int, optional): starting level of procedure in dB HL. Defaults to 40.
            signal_length (int, optional): length of played signals in seconds. Defaults to 1.
            headphone_name (str, optional): Name of headphone model being used. Defaults to Sennheiser_HDA200.
            calibrate (bool, optional): Use calibration file. Defaults to True.
            id (str, optional): id to be stored, that will later be used for naming exported CSV file
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done
        """
        super().__init__(startlevel, signal_length, headphone_name=headphone_name, calibrate=calibrate)      
        self.fails = 0 # number of times familiarization failed
        self.tempfile = self.create_temp_csv(id=id, **additional_data) # create a temporary file to store level at frequencies

    def get_temp_csv_filename(self)->str:
        """Gets name of temp CSV file.

        Returns:
            str: name of CSV file
        """
        return self.tempfile

    def familiarize(self)->bool:
        """Main function.

        Returns:
            bool: familiarization successful
        """
        self.progress = 0.01
        while True:
            self.tone_heard = True

            # first loop (always -20dBHL)
            while self.tone_heard:
                self.play_tone()

                if self.jump_to_end == True:
                    for f in self.freq_bands:
                        self.add_to_temp_csv(20, f, 'lr', self.get_temp_csv_filename())
                    return True

                if self.tone_heard:
                    self.level -= 20

                    if self.progress < 1/5:
                        self.progress = 1/5

                else:
                    self.level += 10

            if self.progress < 1/3:
                self.progress = 1/3        

            # second loop (always +10dBHL)
            while not self.tone_heard:
                self.play_tone()
                if not self.tone_heard:
                    self.level += 10

            self.progress = 2/3

            # replay tone with same level
            self.play_tone()

            if not self.tone_heard:
                self.fails += 1
                if self.fails >= 2:
                    self.progress = 1
                    print("Familiarization unsuccessful. Please read rules and start again.")
                    return False
                else:
                    self.level = self.startlevel

            else:
                print("Familiarization successful!")
                self.progress = 1
                self.add_to_temp_csv(self.level, '1000', 'l', self.tempfile)
                return True

class Procedure:

    def __init__(self, startlevel:float, signal_length:float, headphone_name:str="Sennheiser_HDA200", calibrate:bool=True):
        """Creates the parent class for the familiarization, the main procedure, and the screening.

        Args:
            startlevel (float): starting level of procedure in dB HL
            signal_length (float): length of played signals in seconds
            headphone_name (str, optional): Name of headphone model being used. Defaults to Sennheiser_HDA200.
            calibrate (bool, optional): Use calibration file. Defaults to True.
        """
        self.ap = AudioPlayer()
        self.startlevel = startlevel
        self.level = startlevel
        self.signal_length = signal_length
        self.frequency = 1000
        self.zero_dbhl = 0.000005 # zero_dbhl in absolute numbers. This is a rough guess for uncalibrated systems and will be adjusted through the calibration file
        self.tone_heard = False
        self.freq_bands = ['125', '250', '500', '1000', '2000', '4000', '8000']
        self.freq_levels = {125: 20, 250: 20, 500: 20, 1000: 20, 2000: 20, 4000: 20, 8000: 20} # screening levels
        self.side = 'l'
        self.test_mode = False # set True to be able to skip procedures with right arrow key
        self.jump_to_end = False
        self.use_calibration = calibrate
        self.progress = 0 # value for progressbar
        self.retspl = self.get_retspl_values(headphone_name)
        self.calibration = self.get_calibration_values()
        self.save_path = self.get_save_path()  # Initialize save_path

    def get_retspl_values(self, headphone_name:str):
        """Reads the correct RETSPL values from the retspl.csv file.

        Args:
            headphone_name (str): exact name of headphone as it appears in CSV file

        Returns:
            dict of int:float : RETSPL values for each frequency band from 125 Hz to 8000 Hz
        """
        file_name = 'retspl.csv'

        # Check if the CSV file exists
        if not os.path.isfile(file_name):
            print(f"File '{file_name}' not found.")
            return

        retspl_values = {}

        try:
            with open(file_name, mode='r') as file:
                reader = csv.DictReader(file)
                for row in reader:
                    if row['headphone_model'] == headphone_name:
                        retspl_values[int(row['frequency'])] = float(row['retspl'])
        except Exception as e:
            print(f"Error reading the file: {e}")
            return

        # Check if the headphone model was found
        if not retspl_values:
            print(f"Headphone model '{headphone_name}' not found.")
            return

        print(retspl_values)
        return retspl_values

    def get_calibration_values(self)->dict:
        """Read the correct calibration values from the calibration.csv file.

        Returns:
            dict of int:float : calibration values for each frequency band from 125 Hz to 8000 Hz
        """
        file_name = 'calibration.csv'

        # Check if the CSV file exists
        if not os.path.isfile(file_name):
            print(f"File '{file_name}' not found.")
            return

        try:
            with open(file_name, mode='r') as file:
                reader = csv.DictReader(file)
                calibration_str_values_l = next(reader)
                calibration_str_values_r = next(reader)

                # convert dictionary to int:float and put into extra dictionary for left and right side
                calibration_values = {}
                calibration_values['l'] = {int(k): float(v) for k, v in calibration_str_values_l.items()}
                calibration_values['r'] = {int(k): float(v) for k, v in calibration_str_values_r.items()}

                # if both sides are used, calculate average between both sides
                calibration_values['lr'] = {}
                for k, v in calibration_values['l'].items():
                    calibration_values['lr'][k] = (10 * np.log10((10 ** (v / 10) + 10 ** (calibration_values['r'][k] / 10)) / 2))

        except Exception as e:
            print(f"Error reading the file: {e}")
            return

        print(calibration_values)
        return calibration_values

    def dbhl_to_volume(self, dbhl:float)->float:
        """Calculate dB HL into absolute numbers.

        Args:
            dbhl (float): value in dB HL

        Returns:
            float: value in absolute numbers
        """
        if self.use_calibration:
            # add RETSPL and values from calibration file at that frequency
            dbspl = dbhl + self.retspl[self.frequency] - self.calibration[self.side][self.frequency] 
        else:
            # only add RETSPL
            dbspl = dbhl + self.retspl[self.frequency] 

        return self.zero_dbhl * 10 ** (dbspl / 20) # calculate from dB to absolute numbers using the reference point self.zero_dbhl

    def key_press(self, key:keyboard.Key):
        """Function for pynputto be called on key press

        Args:
            key (keyboard.Key): key that was pressed
        """
        if key == keyboard.Key.space:
            self.tone_heard = True
            print("Tone heard!")
        elif self.test_mode and key == keyboard.Key.right:
            self.jump_to_end = True

    def play_tone(self):
        """Sets tone_heard to False, play beep, then waits 4 s (max) for keypress.
        Sets tone_heard to True if key is pressed.
        Then waits for around 1 s to 2.5 s (randomized).
        """
        self.tone_heard = False
        print(self.frequency, "Hz - playing tone at", self.level, "dBHL.")
        self.ap.play_beep(self.frequency, self.dbhl_to_volume(self.level), self.signal_length, self.side)
        listener = keyboard.Listener(on_press=self.key_press, on_release=None)
        listener.start()
        current_wait_time = 0
        max_wait_time = 4000 # in ms 
        step_size = 50 # in ms

        while current_wait_time < max_wait_time and not self.tone_heard: # wait for keypress
            time.sleep(step_size / 1000)
            current_wait_time += step_size
        listener.stop()
        self.ap.stop()

        if not self.tone_heard:
            print("Tone not heard :(")
        else:
            sleep_time = random.uniform(1, 2.5) # random wait time between 1 and 2.5
            time.sleep(sleep_time) # wait before next tone is played. #TODO test times

    def create_temp_csv(self, id:str="", **additional_data)->str:
        """Creates a temporary CSV file with the relevant frequency bands as a header
        and NaN in the second and third line as starting value for each band.
        (second line: left ear, third line: right ear)
        ID and additional data will be stored in subsequent lines in the format: key, value.

        Args:
            id (str, optional): id to be stored, that will later be used for naming exported csv file
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done

        Returns:
            str: name of temporary file
        """
        with tfile.NamedTemporaryFile(mode='w+', delete=False, newline='', suffix='.csv') as temp_file:
            # Define the CSV writer
            csv_writer = csv.writer(temp_file)

            # Write header
            csv_writer.writerow(self.freq_bands)

            # Write value NaN for each frequency in second and third row
            csv_writer.writerow(['NaN' for _ in range(len(self.freq_bands))])
            csv_writer.writerow(['NaN' for _ in range(len(self.freq_bands))])

            # Write id and additional data
            if id:
                csv_writer.writerow(["id", id])
            if additional_data:
                for key, value in additional_data.items():
                    csv_writer.writerow([key, value])

            return temp_file.name

    def add_to_temp_csv(self, value:str, frequency:str, side:str, temp_filename:str):
        """Add a value in for a specific frequency to the temporary CSV file

        Args:
            value (str): level in dB HL at specific frequency
            frequency (str): frequency where value should be added
            side (str): specify which ear ('l' or 'r')
            temp_filename (str): name of temporary CSV file
        """
        # Read all rows from the CSV file
        with open(temp_filename, mode='r', newline='') as temp_file:
            dict_reader = csv.DictReader(temp_file)
            rows = list(dict_reader)

        # Update the relevant row based on the side parameter
        if side == 'l':
            rows[0][frequency] = value
        elif side == 'r':
            rows[1][frequency] = value
        else:
            rows[0][frequency] = value
            rows[1][frequency] = value

        # Write all rows back to the CSV file
        with open(temp_filename, mode='w', newline='') as temp_file:
            dict_writer = csv.DictWriter(temp_file, fieldnames=self.freq_bands)
            dict_writer.writeheader()
            dict_writer.writerows(rows)

        print(rows[0], rows[1])

        for row in rows[2:]:
            print(row['125'], row['250'])

    def get_value_from_csv(self, frequency:str, temp_filename:str, side:str='l')->str:
        """Get the value at a specific frequency from the temporary CSV file.

        Args:
            frequency (str): frequency where value is stored
            temp_filename (str): name of temporary CSV file
            side (str, optional): specify which ear ('l' or 'r'). Defaults to 'l'.

        Returns:
            str: dB HL value at specified frequency
        """
        with open(temp_filename, mode='r', newline='') as temp_file:
            dict_reader = csv.DictReader(temp_file)
            freq_dict = next(dict_reader) # left ear
            if side == 'r': # go to next line if right side
                freq_dict = next(dict_reader)    
            return freq_dict[frequency]

    def create_final_csv_and_audiogram(self, temp_filename:str, binaural:bool=False):
        """Creates a permanent CSV file and audiogram from the temporary file.

        Args:
            temp_filename (str): Name of the temporary CSV file.
            binaural (bool): If the test is binaural.
        """
        # Read the temporary file
        with open(temp_filename, mode='r', newline='') as temp_file:
            dict_reader = csv.DictReader(temp_file)
            rows = list(dict_reader)

        # Get date and time
        now = datetime.now()
        date_str = now.strftime("%Y%m%d_%H%M%S")
        try:
            id = rows[2]['250']
        except:
            id = "missingID"

        # Create folder for the subject
        folder_name = os.path.join(self.save_path, f"{id}")
        if not os.path.exists(folder_name):
            os.makedirs(folder_name)

        final_csv_filename = os.path.join(folder_name, f"{id}_audiogramm_{date_str}.csv")

        # Write the permanent CSV file
        with open(final_csv_filename, mode='x', newline='') as final_file:
            dict_writer = csv.DictWriter(final_file, fieldnames=self.freq_bands)
            dict_writer.writeheader()
            dict_writer.writerows(rows)

        freqs = [int(x) for x in self.freq_bands]

        left_levels = [self.parse_dbhl_value(rows[0][freq]) for freq in self.freq_bands]
        right_levels = [self.parse_dbhl_value(rows[1][freq]) for freq in self.freq_bands]

        # Generate the audiogram filename
        audiogram_filename = os.path.join(folder_name, f"{id}_audiogram_{date_str}.png")
        print(left_levels, right_levels)
        create_audiogram(freqs, left_levels, right_levels, binaural=binaural, name=audiogram_filename, freq_levels=self.freq_levels)

    def parse_dbhl_value(self, value:str)->int:
        """Parses the dBHL value from the CSV file.

        Args:
            value (str): the value from the CSV file

        Returns:
            int or None: the parsed value or None if 'NH'
        """
        if value == 'NH':
            return 'NH'
        try:
            return int(value)
        except ValueError:
            return None

    def get_progress(self)->float:
        """Gets the current progress.

        Returns:
            float: progress value between 0.0 and 1.0
        """
        return self.progress

    def get_save_path(self)->str:
        """Gets selected path from settings.csv file for saving files.

        Returns:
            str: save path
        """
        file_name = 'settings.csv'

        # Check if the CSV file exists
        if not os.path.isfile(file_name):
            print(f"File '{file_name}' not found.")
            return

        save_path = ""

        try:
            with open(file_name, mode='r') as file:
                reader = csv.DictReader(file)
                settings = next(reader)
                if settings['file path']:
                    save_path = settings['file path']
                else:
                    save_path = os.getcwd()

        except Exception as e:
            print(f"Error reading the file: {e}")
            return

        return save_path