I am trying to run a simple experiment using python. I want to present two different types of audio stimuli, an higher and a lower pitch. The higher pitch has a fixed duration of 200ms while the lower pitch come in pairs, the first with a fixed duration of 250ms and the second with a variable duration that can take the following values [.4, .6, .8, 1, 1.2]. I need to know at what time (machine) the stimuli start and end, and their duration (precision is not the most important issue, I have a tolerance of ~ 10ms), thus I log this information
I am using the library audiomath to create and present the stimuli and I have create several custom functions to manage the other aspects of the task. I have 3 scripts: one in which I define the functions, one in which I set the specific parameters for the experiment for each subject (source) and one with the main()
My problem is that the main() works erratically: it works sometimes, some other times it seems to enter into an infinite loop and a certain sound is presented and never stop playing. The point is that this behavior seems to be really random, with the problem that presents itself at different trials, or not at all, even with the exact same parameter.
This is my code:
source file
#%%imports
from exp_funcs import tone440Hz, tone880Hz
import numpy as np
#%%global var
n_long = 10
n_short = 10
short_duration = .2
long_durations = [.4, .6, .8, 1, 1.2]
#%%calculations
n_tot = n_long + n_long
trial_types = ['short_blink'] * n_short + ['long_blink'] * n_long
sounds = [tone880Hz] * n_short + [tone440Hz] * n_long
np.random.seed(10)
durations = [short_duration] * n_short + [el for el in np.random.choice(long_durations, n_long)]
durations = [.5 if el < .2 else el for el in durations]
cue_duration = [.25] * n_tot
spacing = [1.25] * n_tot
np.random.seed(10)
iti = [el for el in (3 + np.random.normal(0, .25, n_tot))]
functions
import numpy as np
import audiomath as am
import time
import pandas as pd
TWO_PI = 2.0 * np.pi
@am.Synth(fs=22050)
def tone880Hz(fs, sampleIndices, channelIndices):
timeInSeconds = sampleIndices / fs
return np.sin(TWO_PI * 880 * timeInSeconds)
@am.Synth(fs=22050)
def tone440Hz(fs, sampleIndices, channelIndices):
timeInSeconds = sampleIndices / fs
return np.sin(TWO_PI * 440 * timeInSeconds)
def short_blink(sound, duration):
p = am.Player(sound)
init = time.time()
while time.time() < init + duration:
p.Play()
end = time.time()
p.Stop()
print(f'start {init} end {end} duration {end - init}')
return(init, end, end - init)
def long_blink(sound, duration, cue_duration, spacing):
p = am.Player(sound)
i_ = time.time()
while time.time() < i_ + cue_duration:
p.Play()
p.Stop()
time.sleep(spacing)
init = time.time()
while time.time() < init + duration:
p.Play()
end = time.time()
p.Stop()
print(f'start {init} end {end} duration {end - init}')
return(init, end, end - init)
def run_trial(ttype, sound, duration, cue_duration, spacing):
if ttype == 'short_blink':
init, end, effective_duration = short_blink(sound, duration)
else:
init, end, effective_duration = long_blink(sound, duration,
cue_duration, spacing)
otp_df = pd.DataFrame([[ttype, init, end, effective_duration]],
columns = ['trial type', 'start', 'stop',
'effective duration'])
return(otp_df)
main
import pandas as pd
import sys
import getopt
import os
import time
import random
from exp_funcs import run_trial
from pathlib import PurePath
def main(argv):
try:
opts, args = getopt.getopt(argv,'hs:o:',['help', 'source_file=', 'output_directory='])
except getopt.GetoptError:
print ('experiment.py -s source file -o output directory')
sys.exit(2)
for opt, arg in opts:
if opt == '-h':
print ('experiment.py -s source file')
sys.exit()
elif opt in ("-s", "--source_file"):
source_file = arg
elif opt in ("-o", "--output_directory"):
output_dir = arg
os.chdir(os.getcwd())
if not os.path.isfile(f'{source_file}.py'):
raise FileNotFoundError('{source_file} does not exist')
else:
source = __import__('source')
complete_param = list(zip(source.trial_types,
source.sounds,
source.durations,
source.cue_duration,
source.spacing,
source.iti))
# shuffle_param = random.sample(complete_param, len(complete_param))
shuffle_param = complete_param
dfs = []
for ttype, sound, duration, cue_duration, spacing, iti in shuffle_param:
time.sleep(iti)
df = run_trial(ttype, sound, duration, cue_duration, spacing)
dfs.append(df)
dfs = pd.concat(dfs)
dfs.to_csv(PurePath(f'{output_dir}/{source_file}.csv'), index = False)
if __name__ == "__main__":
main(sys.argv[1:])
The 3 files are in the same directory, I browse with the terminal within the directory and run the main as follow python experiment.py -s source -o /whatever/output/directory.
Any help would be more than appreciated
This is too big/complex a program to hope for help on non-specific "erratic" behavior here on stackoverflow. You need to boil it down into a small reproducible example that behaves unexpectedly. If it works sometimes and not others, systematically home in on the conditions that make it fail. I did make one attempt to run the whole thing, but after fixing a few missing imports there was still the matter of the unspecified "source file" content.
So I don't know specifically what your problem is. However, from the audiomath and general real-time-performance perspectives, I can certainly identify a few things you shouldn't be doing:
Although Player instances are designed to be played, stopped or manipulated at time-critical moments, they are not (by default) designed to be created and destroyed at time-critical moments. If you want to create/destroy them fast, pre-initialize a persistent Stream() instance and pass it as the stream argument when creating the Player, as described towards the end of https://audiomath.readthedocs.io/en/release/auto/Examples.html#play-sounds
If you are using Synth instances, you could take advantage of their .duration attribute instead of checking the clock explicitly in a while loop. For example, you can set tone880Hz.duration = 0.5, and then play the sound synchronously with p.Play(wait=True). The big problem with your clock-watching while loops is that they are currently "busy-wait" loops that will thrash the CPU, likely leading to sporadic disruption to your sound (Python's multithreading is far from perfect). However, before you fix this problem you should know...
The strategy "Play(), wait, sleep, Play()" is never going to achieve precise timing of one stimulus relative to the other anyway. First, whenever you issue a command to play a sound in any software, there will unavoidably be a non-zero (and randomly varying!) latency between the command and the physical onset of the sound. Second, sleep() is unlikely to be as precise as you think it is. This applies both to the sleep() you’ve been using to create a gap, and also to the sleep() that would be used internally by Play(wait=True). Sleep implementations suspend operation for "at least" the specified amount of time but they don't guarantee an upper bound on that. This is very hardware- and OS-dependent; on some Windows systems you may even find that the granularity never gets any better than 10ms.
If you really want to use the Synth approach I suppose you could program the gap procedurally into the function definitions of tone440Hz() and tone880Hz(), accessing cue_duration, duration and spacing as global variables (in fact, while you're at it, why not make frequency a global variable too, and only write one function). But I don't see any great advantage in this, either in performance or in code maintainability.
What I would do instead is pre-initialize the following (once, at the start of your program):
max_duration = 1 # length, in seconds, of the longest continuous tone we'll need
tone440Hz = am.Sound(fs=22050).GenerateWaveform(freq_hz=440, duration_msec=max_duration*1000)
tone880Hz = am.Sound(fs=22050).GenerateWaveform(freq_hz=880, duration_msec=max_duration*1000)
m = am.Stream()
Then compose each "long blink" stimulus as a static Sound using the parameters you want.
This will ensure that the tone and gap durations are precise:
s = tone440Hz[:cue_duration] % spacing % tone440Hz[:duration]
For best real-time performance, you could pre-compute a whole set of these stimuli with different parameters. Or, if it turns out that those composition operations (slicing and splicing) happen fast enough, you might decide you can get away with doing that at trial time, in your long_blink() function.
Either way, when it comes to playing the stimulus at trial time:
p = am.Player(s, stream=m) # to make Player() initialization fast, use a pre-initialized Stream() instance
p.Play(wait=True)
Finally: in implementing this, start from scratch—start simple, and test the performance of a few simple cases before compounding things.