当前位置: 首页 > 面试题库 >

试图模拟恒定字节率。与睡眠时间混淆

蒋寒
2023-03-14
问题内容

语境

我在我的计算机(播放器)上使用Windows 7,而在我的大学计算机(流媒体)上使用linux(debian),使用ssh控制。
我试图通过读取波形文件来模拟麦克风的恒定字节速率,就像有人在说话一样。问题在于字节速率低于目标。

选择32KB / s的速率和0.020秒的捕获时间。
我使用time.sleep实现了模拟麦克风,以每0.020秒产生每个数据块。但是获得的速率约为27KB / s,而不是32KB / s

问题

我决定通过阅读此问题,使用一些想法来测试linux机器上time.sleep的精确度。

我做了2种测试。1)繁忙的睡眠2)正常的睡眠

平均而言,从我得到的样本中可以看出,Linux机器的睡眠分辨率为4ms。在Windows上时,其小于/等于1毫秒。

问题

  1. 有什么可能会限制linux机器上的睡眠分辨率?
  2. (在Linux上)为什么繁忙的睡眠与time.sleep具有相同的分辨率?
  3. 如何通过读取波形文件成功模拟麦克风?

import time

def busy_sleep(t):
    s=time.time()
    while time.time() - s < t:
        pass
    e=time.time()
    return e-s

def normal_sleep(t):
    s=time.time()
    time.sleep(t)
    e=time.time()
    return e-s

def test(fun):
    f = lambda x: sum(fun(x) for d in range(10))/10
    print("0.100:{}".format(f(0.100)))
    print("0.050:{}".format(f(0.050)))
    print("0.025:{}".format(f(0.025)))
    print("0.010:{}".format(f(0.010)))
    print("0.009:{}".format(f(0.010)))
    print("0.008:{}".format(f(0.008)))
    print("0.007:{}".format(f(0.007)))
    print("0.006:{}".format(f(0.006)))
    print("0.005:{}".format(f(0.005)))
    print("0.004:{}".format(f(0.004)))
    print("0.003:{}".format(f(0.003)))
    print("0.002:{}".format(f(0.002)))
    print("0.001:{}".format(f(0.001)))

if __name__=="__main__":
    print("Testing busy_sleep:")
    test(busy_sleep)
    print("Testing normal_sleep:")
    test(normal_sleep)

结果

"""
Debian
Testing busy_sleep:
0.100:0.10223722934722901
0.050:0.051996989250183104
0.025:0.027996940612792967
0.020:0.02207831859588623
0.010:0.011997451782226562
0.009:0.011997222900390625
0.008:0.009998440742492676
0.007:0.007997279167175292
0.006:0.0079974365234375
0.005:0.007997465133666993
0.004:0.005918483734130859
0.003:0.003997836112976074
0.002:0.0039977550506591795
0.001:0.003997611999511719
Testing normal_sleep:
0.100:0.1020797061920166
0.050:0.051999988555908205
0.025:0.028000001907348634
0.020:0.02192000865936279
0.010:0.011999979019165039
0.009:0.012000055313110351
0.008:0.010639991760253906
0.007:0.008000001907348633
0.006:0.00799997329711914
0.005:0.008000059127807617
0.004:0.006159958839416504
0.003:0.004000000953674317
0.002:0.00399998664855957
0.001:0.004000091552734375

$ uname -a
Linux 3.2.0-4-amd64 #1 SMP Debian 3.2.57-3+deb7u2 x86_64 GNU/Linux
"""

"""
Windows 7

Testing busy_sleep:
0.100:0.10000572204589844
0.050:0.05000288486480713
0.025:0.0250014066696167
0.010:0.010500597953796388
0.009:0.010500597953796388
0.008:0.008000493049621582
0.007:0.00740041732788086
0.006:0.006400299072265625
0.005:0.005400300025939942
0.004:0.004700303077697754
0.003:0.003200197219848633
0.002:0.002700185775756836
0.001:0.0016000032424926759
Testing normal_sleep:
0.100:0.10000579357147217
0.050:0.0500028133392334
0.025:0.02500150203704834
0.010:0.01000049114227295
0.009:0.0100006103515625
0.008:0.008000493049621582
0.007:0.007000398635864257
0.006:0.006000304222106934
0.005:0.00500030517578125
0.004:0.0040001869201660155
0.003:0.0030002117156982424
0.002:0.0020000934600830078
0.001:0.0010000944137573243
"""

实码

导入os导入wave导入sys导入io导入时间

FORMAT = 8 #get_format_from_width(2)
NCHANNELS = 1
FRAMERATE = 16000 # samples per second
SAMPWIDTH = 2 # bytes in a sample
BYTE_RATE = FRAMERATE*SAMPWIDTH
CHUNK_DURATION = 0.020
CHUNK_BYTES = int(CHUNK_DURATION*BYTE_RATE)

class StreamSimulator:
    def __init__(self):
        wf = wave.open("Kalimba.wav","rb")
        buf = io.BytesIO()
        buf.write(wf.readframes(wf.getnframes()))
        wf.close()
        buf.seek(0)
        self.buf = buf
        self.step = time.time()

    def delay(self):
        #delay
        delta = time.time() - self.step 
        self.step=time.time()
        delay = CHUNK_DURATION - delta      
        if delay > 0.001:
            time.sleep(delay)


    def read(self):
        buf = self.buf  
        data = buf.read(CHUNK_BYTES)
        if len(data) == 0:
            buf.seek(0)
            data = buf.read(CHUNK_BYTES)

        self.delay()
        return data

    def close(self):
        self.buf.close()

class DynamicPainter:
    def __init__(self):
        self.l=0

    def paint(self,obj):        
        str1=str(obj)
        l1=len(str1)
        bs="\b"*self.l
        clean=" "*self.l
        total = bs+clean+bs+str1
        sys.stdout.write(total)
        sys.stdout.flush()
        self.l=l1

if __name__=="__main__":
    painter = DynamicPainter()
    stream = StreamSimulator()
    produced = 0
    how_many = 0
    painted = time.time()
    while True:
        while time.time()-painted < 1:
            d = stream.read()
            produced += len(d)
            how_many += 1
        producing_speed = int(produced/(time.time()-painted))       
        painter.paint("Producing speed: {} how many: {}".format(producing_speed,how_many))
        produced=0
        how_many=0
        painted = time.time()

编辑

更改了“实码”,添加了包括睡眠时间在内的时间度量。
但是现在我有了DOUBLE Byte Rate:Producing speed: 63996 how many: 100
这让我非常困惑。我尝试了不同的方法,最终每次都是两倍。

结论

感谢@JFSebastian和他的代码,我了解到:

  • 最好使用截止日期作为时间参考,而不是在每个循环中创建新的参考
  • 使用截止日期可以使time.sleep的不精确度“摊销”,使比特率在所需的比特率附近振荡,但会得出正确的(并且要更稳定)的平均值。
  • 您只需要使用一次time.time(),这意味着更少的计算误差。

结果,我有时会以恒定的32000 B / s振荡到31999,很少会振荡到31745。
现在我可以听到音乐而没有任何滞后或抖动!

最终密码

def read(self):
    buf = self.buf  
    data = buf.read(CHUNK_BYTES)
    if len(data) == 0:
        buf.seek(0)
        data = buf.read(CHUNK_BYTES)

    self.deadline += CHUNK_DURATION 
    delay = self.deadline - time.time()
    if delay > 0:
        time.sleep(delay)
    return data

问题答案:

结论

感谢@JFSebastian和他的代码,我了解到:

  • 最好使用截止日期作为时间参考,而不是在每个循环中创建新的参考
  • 使用截止日期可以使time.sleep的不精确度“摊销”,使比特率在所需的比特率附近振荡,但会得出正确的(并且要更稳定)的平均值。
  • 您只需要使用一次time.time(),这意味着更少的计算误差。

结果,我有时会以恒定的32000 B / s振荡到31999,很少会振荡到31745。
现在我可以听到音乐而没有任何滞后或抖动!

我尝试仅使用%运算符来使用@JFSebastian强制执行来使其余部分休眠,但KB /
s奇怪地波动,因此我决定保留最后期限实施,因为通过增加浮点值会造成不精确性。但是,总体结果足以满足我的需求。
谢谢大家。

最终密码

def read(self):
    self.deadline += 0.020  
    delay = self.deadline - time.perf_counter()
    if delay > 0:
        time.sleep(delay)
    return self._read()


 类似资料:
  • 问题内容: 语境 我在计算机(播放器)上使用Windows 7,而在大学计算机(流媒体)上使用linux(debian),我使用ssh对其进行控制。 我试图通过读取波形文件来模拟麦克风的恒定字节速率,就像有人在说话一样。问题在于字节速率低于目标。 选择32KB / s的速率和0.020秒的捕获时间。 我使用time.sleep实现了模拟麦克风,以每0.020秒产生每个数据块。但是获得的速率约为27

  • 我正在调用while循环内的线程Hibernate1秒。当标志为true时,循环将运行(标志为true无限时间)。在循环内,线程应Hibernate1秒,唤醒并增加计数器,检查IF条件,如果为FALSE,则应再次Hibernate1秒并继续29次。在第30次迭代中,IF条件为true,IF语句中调用的方法将收集并存储数据。最后,在第32次迭代中,第二个IF语句将把存储的数据发送到服务器,并将计数设

  • 问题内容: 有人可以告诉我这为什么起作用: 但这失败了: 这是错误: 问题答案: 这行: 是一个简短的变量声明,右侧表达式为:。是包中的类型常量,其类型为;并且是一种 无类型数值常量 ,其在表达时,将可能的话采取适当的类型。由于具有作为其基础类型,因此可以在不损失精度的情况下将常量转换为常量,从而可以执行常量表达式。 另一行: 第一行是一个简短的变量声明,其中的类型将从右边的表达式中推断出来,这是

  • 问题内容: 我一直在想做一个睡眠功能,它会一直睡眠到某个日期为止。我的想法基于日期,例如:2019-01-20 12:00:00。 我还没有真正弄清楚如何开始解决这个问题。我的想法是类似的东西 所以问题是我怎么可能能够睡到2019-01-20 12:00:00所给定的某个时间? 问题答案: 您可以计算出达到唤醒时间并进入增量时间睡眠所需的时间。 Python可以按时间间隔进行计算。如果您从另一个时

  • 问题内容: 关于PHP中的函数,我有两个问题: 睡眠时间是否会影响我的PHP脚本的最大执行时间限制?有时,PHP显示消息“超过30秒的最大执行时间”。如果我使用此消息会出现吗? 使用该功能有没有风险?它会消耗很多CPU性能吗? 问题答案: 您应该尝试一下,只需让脚本hibernate超过您的最大执行时间即可。 Spoiler:在Linux下,睡眠时间被忽略,但是在Windows下,它计为执行时间。

  • 我使用 C 和 POSIX 线程创建了一个多线程应用程序。我现在应该阻塞一个线程(主线程),直到设置了布尔标志(变为真)。 我找到了两种方法来完成这件事。 > 在没有睡眠的情况下旋转。 在睡眠中旋转循环。 如果我应该遵循第一种方式,为什么有些人编写代码遵循第二种方式?如果应该使用第二种方法,为什么要让当前线程Hibernate呢?这种方式的缺点是什么?