创建 GNURadio 源块:如何调整采样率
Create GNURadio source block: how to tune the sample rate
我正在尝试在 GNU radio 中为 Analog Discovery 2 示波器实现自定义源代码块。我已经有一个工作的 python 脚本来将 Analog Discovery 2 设备的样本记录到 WAV 文件(问题末尾的代码)。
我希望能够直接在 GNUradio companion 中连接这个样本源。我已经按照 the official tutorial 创建自定义块来为我的块生成模板代码:
import numpy
from gnuradio import gr
class AnalogDiscovery2(gr.sync_block):
def __init__(self, sample_rate):
gr.sync_block.__init__(self,
name="Analog Discovery 2",
in_sig=None,
out_sig=[numpy.float32])
self.sample_rate = sample_rate
def work(self, input_items, output_items):
out = output_items[0]
# <+signal processing here+>
out[:] = whatever
return len(output_items[0])
我知道我必须修改 work
函数来获取样本并将它们复制到 out
变量,但是我想知道如何调整采样率?我不知道 work
函数是如何调用的,它的时机是什么。如何设置采样率??
Python 将样本录制到 WAV 文件中的代码:
from ctypes import *
from dwfconstants import *
import math
import time
import matplotlib.pyplot as plt
import sys
import wave
import struct
if sys.platform.startswith("win"):
dwf = cdll.dwf
elif sys.platform.startswith("darwin"):
dwf = cdll.LoadLibrary("/Library/Frameworks/dwf.framework/dwf")
else:
dwf = cdll.LoadLibrary("libdwf.so")
#declare ctype variables
hdwf = c_int()
sts = c_byte()
hzAcq = c_double(48000)
nSamples = 96000
rgdSamples = (c_double*nSamples)()
cAvailable = c_int()
cLost = c_int()
cCorrupted = c_int()
fLost = 0
fCorrupted = 0
#print DWF version
version = create_string_buffer(16)
dwf.FDwfGetVersion(version)
print "DWF Version: "+version.value
#open device
print "Opening first device"
dwf.FDwfDeviceOpen(c_int(-1), byref(hdwf))
if hdwf.value == hdwfNone.value:
szerr = create_string_buffer(512)
dwf.FDwfGetLastErrorMsg(szerr)
print szerr.value
print "failed to open device"
quit()
print "Preparing to read sample..."
#print "Generating sine wave..."
#dwf.FDwfAnalogOutNodeEnableSet(hdwf, c_int(0), AnalogOutNodeCarrier, c_bool(True))
#dwf.FDwfAnalogOutNodeFunctionSet(hdwf, c_int(0), AnalogOutNodeCarrier, funcSine)
#dwf.FDwfAnalogOutNodeFrequencySet(hdwf, c_int(0), AnalogOutNodeCarrier, c_double(1))
#dwf.FDwfAnalogOutNodeAmplitudeSet(hdwf, c_int(0), AnalogOutNodeCarrier, c_double(2))
#dwf.FDwfAnalogOutConfigure(hdwf, c_int(0), c_bool(True))
# enable positive supply
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(0), c_int(0), c_double(True))
# set voltage to 3 V
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(0), c_int(1), c_double(3.0))
# enable negative supply
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(1), c_int(0), c_double(True))
# set voltage to -1 V
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(1), c_int(1), c_double(-1.0))
# master enable
dwf.FDwfAnalogIOEnableSet(hdwf, c_int(True))
#set up acquisition
dwf.FDwfAnalogInChannelEnableSet(hdwf, c_int(0), c_bool(True))
dwf.FDwfAnalogInChannelRangeSet(hdwf, c_int(0), c_double(0.1))
dwf.FDwfAnalogInAcquisitionModeSet(hdwf, acqmodeRecord)
dwf.FDwfAnalogInFrequencySet(hdwf, hzAcq)
dwf.FDwfAnalogInRecordLengthSet(hdwf, c_double(nSamples/hzAcq.value))
#wait at least 2 seconds for the offset to stabilize
time.sleep(2)
#begin acquisition
dwf.FDwfAnalogInConfigure(hdwf, c_int(0), c_int(1))
print " waiting to finish"
cSamples = 0
while cSamples < nSamples:
dwf.FDwfAnalogInStatus(hdwf, c_int(1), byref(sts))
if cSamples == 0 and (sts == DwfStateConfig or sts == DwfStatePrefill or sts == DwfStateArmed) :
# Acquisition not yet started.
continue
dwf.FDwfAnalogInStatusRecord(hdwf, byref(cAvailable), byref(cLost), byref(cCorrupted))
cSamples += cLost.value
if cLost.value :
fLost = 1
if cCorrupted.value :
fCorrupted = 1
if cAvailable.value==0 :
continue
if cSamples+cAvailable.value > nSamples :
cAvailable = c_int(nSamples-cSamples)
# get samples
dwf.FDwfAnalogInStatusData(hdwf, c_int(0), byref(rgdSamples, 8*cSamples), cAvailable)
cSamples += cAvailable.value
print "Recording finished"
if fLost:
print "Samples were lost! Reduce frequency"
if cCorrupted:
print "Samples could be corrupted! Reduce frequency"
#f = open("record.bin", "w")
#for v in rgdSamples:
# f.write("%s\n" % v)
#f.close()
# Write samples to file
wav_output = wave.open('record.wav', 'w')
wav_output.setparams((1, 2, 48000, nSamples, 'NONE', 'not compressed'))
values = []
for v in rgdSamples:
packed_value = struct.pack('h', 32768*v)
values.append(packed_value)
value_str = ''.join(values)
wav_output.writeframes(value_str)
wav_output.close()
你误会了。
采样率对于 GNU Radio 来说不是一个有意义的概念。 GNU Radio 块被安排得尽可能快;这意味着重复调用源块,直到输出缓冲区已满。
"wall clock" 与处理样本的速度无关。
例如,信号源仅使用有关采样率的信息来计算有多少样本一个周期。正弦有。信号源配置为以 500 的采样率产生频率为 10 的正弦波 完全 与将其配置为频率为 50 和采样率为 2500 的样本相同。有绝对没有行为差异。
我只能强调 GNU Radio 没有任何采样率的概念。块之间传递的信号只是 数字序列 。原始音频以特定速率采样的事实可能是正确参数化处理样本的块所必需的,但这不是样本固有的。
顺便说一下,没有理由编写自己的 wav 文件源:gr-audio 已经包含一个。
我正在尝试在 GNU radio 中为 Analog Discovery 2 示波器实现自定义源代码块。我已经有一个工作的 python 脚本来将 Analog Discovery 2 设备的样本记录到 WAV 文件(问题末尾的代码)。
我希望能够直接在 GNUradio companion 中连接这个样本源。我已经按照 the official tutorial 创建自定义块来为我的块生成模板代码:
import numpy
from gnuradio import gr
class AnalogDiscovery2(gr.sync_block):
def __init__(self, sample_rate):
gr.sync_block.__init__(self,
name="Analog Discovery 2",
in_sig=None,
out_sig=[numpy.float32])
self.sample_rate = sample_rate
def work(self, input_items, output_items):
out = output_items[0]
# <+signal processing here+>
out[:] = whatever
return len(output_items[0])
我知道我必须修改 work
函数来获取样本并将它们复制到 out
变量,但是我想知道如何调整采样率?我不知道 work
函数是如何调用的,它的时机是什么。如何设置采样率??
Python 将样本录制到 WAV 文件中的代码:
from ctypes import *
from dwfconstants import *
import math
import time
import matplotlib.pyplot as plt
import sys
import wave
import struct
if sys.platform.startswith("win"):
dwf = cdll.dwf
elif sys.platform.startswith("darwin"):
dwf = cdll.LoadLibrary("/Library/Frameworks/dwf.framework/dwf")
else:
dwf = cdll.LoadLibrary("libdwf.so")
#declare ctype variables
hdwf = c_int()
sts = c_byte()
hzAcq = c_double(48000)
nSamples = 96000
rgdSamples = (c_double*nSamples)()
cAvailable = c_int()
cLost = c_int()
cCorrupted = c_int()
fLost = 0
fCorrupted = 0
#print DWF version
version = create_string_buffer(16)
dwf.FDwfGetVersion(version)
print "DWF Version: "+version.value
#open device
print "Opening first device"
dwf.FDwfDeviceOpen(c_int(-1), byref(hdwf))
if hdwf.value == hdwfNone.value:
szerr = create_string_buffer(512)
dwf.FDwfGetLastErrorMsg(szerr)
print szerr.value
print "failed to open device"
quit()
print "Preparing to read sample..."
#print "Generating sine wave..."
#dwf.FDwfAnalogOutNodeEnableSet(hdwf, c_int(0), AnalogOutNodeCarrier, c_bool(True))
#dwf.FDwfAnalogOutNodeFunctionSet(hdwf, c_int(0), AnalogOutNodeCarrier, funcSine)
#dwf.FDwfAnalogOutNodeFrequencySet(hdwf, c_int(0), AnalogOutNodeCarrier, c_double(1))
#dwf.FDwfAnalogOutNodeAmplitudeSet(hdwf, c_int(0), AnalogOutNodeCarrier, c_double(2))
#dwf.FDwfAnalogOutConfigure(hdwf, c_int(0), c_bool(True))
# enable positive supply
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(0), c_int(0), c_double(True))
# set voltage to 3 V
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(0), c_int(1), c_double(3.0))
# enable negative supply
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(1), c_int(0), c_double(True))
# set voltage to -1 V
dwf.FDwfAnalogIOChannelNodeSet(hdwf, c_int(1), c_int(1), c_double(-1.0))
# master enable
dwf.FDwfAnalogIOEnableSet(hdwf, c_int(True))
#set up acquisition
dwf.FDwfAnalogInChannelEnableSet(hdwf, c_int(0), c_bool(True))
dwf.FDwfAnalogInChannelRangeSet(hdwf, c_int(0), c_double(0.1))
dwf.FDwfAnalogInAcquisitionModeSet(hdwf, acqmodeRecord)
dwf.FDwfAnalogInFrequencySet(hdwf, hzAcq)
dwf.FDwfAnalogInRecordLengthSet(hdwf, c_double(nSamples/hzAcq.value))
#wait at least 2 seconds for the offset to stabilize
time.sleep(2)
#begin acquisition
dwf.FDwfAnalogInConfigure(hdwf, c_int(0), c_int(1))
print " waiting to finish"
cSamples = 0
while cSamples < nSamples:
dwf.FDwfAnalogInStatus(hdwf, c_int(1), byref(sts))
if cSamples == 0 and (sts == DwfStateConfig or sts == DwfStatePrefill or sts == DwfStateArmed) :
# Acquisition not yet started.
continue
dwf.FDwfAnalogInStatusRecord(hdwf, byref(cAvailable), byref(cLost), byref(cCorrupted))
cSamples += cLost.value
if cLost.value :
fLost = 1
if cCorrupted.value :
fCorrupted = 1
if cAvailable.value==0 :
continue
if cSamples+cAvailable.value > nSamples :
cAvailable = c_int(nSamples-cSamples)
# get samples
dwf.FDwfAnalogInStatusData(hdwf, c_int(0), byref(rgdSamples, 8*cSamples), cAvailable)
cSamples += cAvailable.value
print "Recording finished"
if fLost:
print "Samples were lost! Reduce frequency"
if cCorrupted:
print "Samples could be corrupted! Reduce frequency"
#f = open("record.bin", "w")
#for v in rgdSamples:
# f.write("%s\n" % v)
#f.close()
# Write samples to file
wav_output = wave.open('record.wav', 'w')
wav_output.setparams((1, 2, 48000, nSamples, 'NONE', 'not compressed'))
values = []
for v in rgdSamples:
packed_value = struct.pack('h', 32768*v)
values.append(packed_value)
value_str = ''.join(values)
wav_output.writeframes(value_str)
wav_output.close()
你误会了。
采样率对于 GNU Radio 来说不是一个有意义的概念。 GNU Radio 块被安排得尽可能快;这意味着重复调用源块,直到输出缓冲区已满。
"wall clock" 与处理样本的速度无关。
例如,信号源仅使用有关采样率的信息来计算有多少样本一个周期。正弦有。信号源配置为以 500 的采样率产生频率为 10 的正弦波 完全 与将其配置为频率为 50 和采样率为 2500 的样本相同。有绝对没有行为差异。
我只能强调 GNU Radio 没有任何采样率的概念。块之间传递的信号只是 数字序列 。原始音频以特定速率采样的事实可能是正确参数化处理样本的块所必需的,但这不是样本固有的。
顺便说一下,没有理由编写自己的 wav 文件源:gr-audio 已经包含一个。