0

GNU ラジオで Analog Discovery 2 オシロスコープ用のカスタム ソース ブロックを実装しようとしています。Analog Discovery 2デバイスからWAVファイルにサンプルを記録するためのPythonスクリプトが既にあります（質問の最後のコード）。

このサンプル ソースを直接 GNUradio コンパニオンに接続できるようにしたいと考えています。公式のチュートリアルに従ってカスタム ブロックを作成し、ブロックのテンプレート コードを生成しました。

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関数がどのように呼び出されるか、またそのタイミングがわかりません。サンプルレートを設定するにはどうすればよいですか??

サンプルを WAV ファイルに録音する Python コード:

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

1 に答える 1