main Module¶

class AudioPlayer:

    def __init__(self):
        """Creates an audio player that can play sine beeps at various frequencies, volumes and with various durations.
        Automatically detects current samplerate of selected sound device.
        """
        self.fs = self.get_device_samplerate()
        self.beep_duration = 10
        self.volume = 0
        self.frequency = 440
        self.stream = None
        self.is_playing = False

    def generate_tone(self)->np.array:
        """Generates a sine tone with current audio player settings.

        Returns:
            array: sine wave as numpy array
        """
        t = np.linspace(start=0, 
                        stop=self.beep_duration, 
                        num=int(self.fs * self.beep_duration), 
                        endpoint=False)
        tone = np.sin(2 * np.pi * self.frequency * t) * self.volume

        # Create fade-out envelope
        fade_duration = 0.003  # 3 ms fade-out
        fade_samples = int(self.fs * fade_duration)
        fade_out = np.linspace(1, 0, fade_samples)
        envelope = np.ones_like(tone)
        envelope[-fade_samples:] = fade_out

        # Apply the envelope to the tone
        tone = tone * envelope

        return tone

    def play_beep(self, frequency:int, volume:float, duration:int, channel:str='lr'):
        """Sets the frequency, volume and beep duration of the audio player and then plays a beep with those parameters.

        Args:
            frequency (int): frequency in Hz
            volume (float): volume multiplier (between 0 and 1)
            duration (int): duration of the beep in seconds
            channel (string): 'l', 'r' or 'lr' for only left, only right or both channels respectively
        """
        self.frequency = frequency
        self.volume = volume
        self.beep_duration = duration
        tone = self.generate_tone()
        if channel == 'l':
            sd.play(np.array([tone, np.zeros(len(tone))]).T, self.fs)
        elif channel == 'r':
            sd.play(np.array([np.zeros(len(tone)), tone]).T, self.fs)
        else:
            sd.play(tone, self.fs)

    def stop(self):
        """Stops the current playback.
        """
        sd.stop()

    def int_or_str(self, text: str)->int:
        """Helper function for argument parsing.
        """
        try:
            return int(text)
        except ValueError:
            return text

    def get_device_samplerate(self):
        """Gets current samplerate from the selected audio output device.

        Returns:
            float: samplerate of current sound device
        """
        parser = argparse.ArgumentParser(add_help=False)
        parser.add_argument(
            '-l', '--list-devices', action='store_true',
            help='show list of audio devices and exit')
        args, remaining = parser.parse_known_args()
        if args.list_devices:
            print(sd.query_devices())
            parser.exit(0)
        parser = argparse.ArgumentParser(
            description=__doc__,
            formatter_class=argparse.RawDescriptionHelpFormatter,
            parents=[parser])
        parser.add_argument(
            'frequency', nargs='?', metavar='FREQUENCY', type=float, default=500,
            help='frequency in Hz (default: %(default)s)')
        parser.add_argument(
            '-d', '--device', type=self.int_or_str,
            help='output device (numeric ID or substring)')
        parser.add_argument(
            '-a', '--amplitude', type=float, default=0.2,
            help='amplitude (default: %(default)s)')
        args = parser.parse_args(remaining)
        return sd.query_devices(args.device, 'output')['default_samplerate']

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 Controller():

    def __init__(self):
        """Controller class (MVC architecture) that combines model and view of the Audiometer.
        """
        self.selected_program = ""
        program_functions = {"Klassisches Audiogramm" : self.start_standard_procedure,
                             "Kurzes Screening" : self.start_screen_procedure,
                             "Kalibrierung" : self.start_calibration}

        self.calibration_funcs = [self.start_calibration, self.calibration_next_freq, self.calibration_repeat_freq, self.stop_sound, self.calibration_set_level]
        self.view = setup_ui(self.start_familiarization, 
                             program_functions, self.calibration_funcs, self.get_progress)

        # helper variable for calibration
        self.button_changed = False

    def run_app(self):
        """Starts the app by running the tkinter mainloop of the view.
        """
        self.view.mainloop()

    def start_familiarization(self, id:str="", headphone:str="Sennheiser_HDA200", calibrate:bool=True, **additional_data)->bool:
        """Creates a Familiarization object and uses it to start the familiarization process.

        Args:
            id (str, optional): ID of test subject. Defaults to "".
            headphone (str, optional): Name of headphone model being used. Defaults to "Sennheiser_HDA200".
            calibrate (bool, optional): Whether to use calibration file. Defaults to True.
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done

        Returns:
            bool: Whether familiarization was successful
        """
        self.selected_program = "familiarization"
        self.familiarization = Familiarization(id=id, headphone_name=headphone, calibrate=calibrate, **additional_data)
        return self.familiarization.familiarize()

    def start_standard_procedure(self, binaural:bool=False, headphone:str="Sennheiser_HDA200", calibrate:bool=True, **additional_data):
        """Creates a StandardProcedure object and uses it to start the standard procedure.

        Args:
            binaural (bool, optional): Whether to test both ears at the same time. Defaults to False.
            headphone (str, optional): Name of headphone model being used. Defaults to "Sennheiser_HDA200".
            calibrate (bool, optional): Whether to use calibration file. Defaults to True.
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done
        """
        self.selected_program = "standard"
        self.standard_procedure = StandardProcedure(self.familiarization.get_temp_csv_filename(), headphone_name=headphone, calibrate=calibrate, **additional_data)
        self.standard_procedure.standard_test(binaural)

    def start_screen_procedure(self, binaural:bool=False, headphone:str="Sennheiser_HDA200", calibrate:bool=True, **additional_data):
        """Creates a ScreeningProcedure object and uses it to start the screening procedure.

        Args:
            binaural (bool, optional): Whether to test both ears at the same time. Defaults to False.
            headphone (str, optional): Name of headphone model being used. Defaults to "Sennheiser_HDA200".
            calibrate (bool, optional): Whether to use calibration file. Defaults to True.
            **additional_data: additional key/value pairs to be stored in CSV file after procedure is done
        """
        self.selected_program = "screening"
        self.screen_procedure = ScreeningProcedure(self.familiarization.get_temp_csv_filename(), headphone_name=headphone, calibrate=calibrate, **additional_data)
        self.screen_procedure.screen_test(binaural)

    def start_calibration(self, level:int, headphone:str="Sennheiser_HDA200")->tuple:
        """Creates a Calibration object and uses it to start calibration.

        Args:
            level (int): Level of calibration in dB HL.
            headphone (str, optional): Name of headphone model being used. Defaults to "Sennheiser_HDA200".
        """
        self.selected_program = "calibration"
        self.calibration = Calibration(startlevel=level, headphone_name=headphone)
        _, current_freq, current_spl = self.calibration_next_freq()
        return current_freq, current_spl

    def calibration_next_freq(self)->tuple:
        """Goes to next frequency in calibration process and play it.

        Returns:
            bool: Whether there are more frequencies left after this one.
            int: current frequency
            float: expected SPL value in dB
        """
        more_freqs, current_freq, current_spl = self.calibration.play_one_freq()
        if more_freqs:
            return True, current_freq, current_spl
        elif self.button_changed == False:
            self.button_changed = True
            return False, current_freq, current_spl
        else:
            self.calibration.finish_calibration()
            return False, current_freq, current_spl

    def calibration_repeat_freq(self):
        """Repeats the current frequency during calibration process.
        """
        self.calibration.repeat_freq()

    def calibration_set_level(self, spl:float):
        """Sets the measured level in dB during calbration process.

        Args:
            spl (float): Sound pressure level that was measured in dB
        """
        self.calibration.set_calibration_value(spl)

    def stop_sound(self):
        """Stops the sound during calibration process.
        """
        self.calibration.stop_playing()

    def get_progress(self)->float:
        """Gets current progress in curent procedure for progress bar.

        Returns:
            float: progress value between 0.0 and 1.0
        """
        if self.selected_program == "familiarization":
            return self.familiarization.get_progress()
        elif self.selected_program == "standard":
            return self.standard_procedure.get_progress()
        elif self.selected_program == "screening":
            return self.screen_procedure.get_progress()
        elif self.selected_program == "calibration":
            return 0.0
        else:
            return 0.0

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