为什么我的剧本呈现音频刺激有一个不可预测的行为？

Question

我试着用python做一个简单的实验。我想展示两种不同类型的音频刺激，一个较高的和一个较低的音调。较高的音高具有固定的200ms持续时间，而较低的音高是成对的，第一个音高的持续时间是固定的，第二个音高的持续时间是可变的，可以采用以下值[.4, .6, .8, 1, 1.2]。我需要知道什么时候(机器)刺激开始和结束，它们的持续时间(精度不是最重要的问题，我有一个~10 is的容忍度)，因此我记录这些信息。

我使用库audiomath来创建和呈现刺激，并创建了几个自定义函数来管理任务的其他方面。我有三个脚本:一个是我定义函数的脚本，一个是为每个主题(源)设置实验的具体参数的脚本，另一个是用main()编写的脚本。

我的问题是，main()的工作不稳定:它有时工作，有时它似乎进入一个无限的循环，一个特定的声音出现，永远不会停止播放。关键是，这种行为似乎是随机的，问题出现在不同的试验中，或者根本不存在，即使是完全相同的参数。

这是我的密码：

源文件

#%%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))]

函数

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)

主

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:])

这3个文件位于同一个目录中，我使用目录中的终端浏览并运行主文件，如下所示：python experiment.py -s source -o /whatever/output/directory。任何帮助都将不胜感激。

Answer 1

这是一个太大/复杂的程序，不希望在堆栈溢出的非特定“不稳定”行为方面寻求帮助。您需要将其简化为一个小的可重复的示例，该示例的行为出乎意料。如果它有时起作用，而不是其他人，那么系统地回到使它失败的条件中去。我确实尝试过运行整个程序，但是在修复了一些缺失的导入之后，仍然存在未指定的“源文件”内容的问题。

所以我不知道你具体的问题是什么。但是，从audiomath和一般的实时性能角度来看，我肯定可以确定一些您不应该做的事情：

1. 虽然Player实例被设计为在关键时刻被播放、停止或操作，但它们(默认情况下)并不是在关键时刻被创建和销毁的。如果您想要快速创建/销毁它们，请预先初始化一个持久的Stream()实例，并在创建Player时将其作为stream参数传递，就像在https://audiomath.readthedocs.io/en/release/auto/Examples.html#play-sounds末尾所描述的那样。
2. 如果使用Synth实例，则可以利用它们的.duration属性，而不是在while循环中显式地检查时钟。例如，您可以设置tone880Hz.duration = 0.5，然后与p.Play(wait=True)同步播放声音。观察时钟的while循环的最大问题是，它们目前处于“繁忙-等待”循环，这将破坏CPU，可能会导致声音的零星中断(Python的多线程处理还远远不够完美)。但是，在你解决这个问题之前你应该知道.
3. "Play()，等待，睡眠，Play()“的策略永远无法实现一个刺激相对于另一个刺激的精确时机。首先，当您发出命令在任何软件中播放声音时，不可避免地会出现非零(并且随机变化！)命令与声音的物理启动之间的延迟。其次，sleep()不太可能像你想象的那样精确。这既适用于您一直用于创建空白的sleep()，也适用于Play(wait=True)内部将使用的sleep()。睡眠实现暂停操作“至少”指定的时间，但它们不能保证上限。这是非常依赖于硬件和操作系统的；在某些Windows系统上，您甚至会发现粒度从未超过10 OS。

如果您真的想使用Synth方法，我想您可以按步骤将gap编程到tone440Hz()和tone880Hz()的函数定义中，将cue_duration、duration和spacing作为全局变量访问(实际上，当您正在使用它时，为什么不把频率也作为一个全局变量，只编写一个函数)。但是，无论是在性能上还是在代码可维护性方面，我都没有看到任何巨大的优势。

我要做的是预先初始化以下内容(一次，在您的程序开始时)：

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

然后使用所需的参数将每个“长眨眼”刺激组合为静态Sound。这将确保音调和间隙持续时间是准确的：

s = tone440Hz[:cue_duration] % spacing % tone440Hz[:duration]

为了获得最佳的实时性能，您可以用不同的参数预先计算一组这些刺激。或者，如果发现这些组合操作(切片和剪接)发生得足够快，那么您可能会决定在试用时，在long_blink()函数中完成这些操作。

不管是哪种情况，在试用期发挥刺激作用时：

p = am.Player(s, stream=m)  # to make Player() initialization fast, use a pre-initialized Stream() instance
p.Play(wait=True)

最后:在实现这一点时，从零开始--从简单开始，在复合之前测试几个简单案例的性能。