Python Block for GNURadio streaming samples from fx2lafw USB oscilloscopes

fx2lawfw-source.py

# https://ubfx.dev/p/usb-oscilloscope-gnu-radio-sdr/

import numpy
import array
from gnuradio import gr

import usb.core
import usb.util

class HantekFx2Source(gr.sync_block):
    def __init__(self):
        gr.sync_block.__init__(self,
            name="HantekFx2Source",
            in_sig=None,
            out_sig=[numpy.uint8])

        # Needs fx2lafw firmware
        self.dev = usb.core.find(idVendor=0x1d50, idProduct=0x608e)

        if self.dev is None:
            raise RuntimeError("Failed to open USB device")

        self.write_control(0xe4, b'\x01') # CHANELS_REG = 1
        self.write_control(0xe0, chr(10)) # VDIV_REG (amplifier feedback) = x10
        self.write_control(0xe2, chr(16)) # sample rate = 16 MHz

        self.write_control(0xe3, b'\x01') # TRIGGER_REG = 1, start capture
    
    def work(self, input_items, output_items):
        out = output_items[0]

        buf = self.dev.read(0x86, len(out), None)

        if buf is None:
            raise RuntimeError("Failed to read data")

        out[:] = buf
        return len(output_items[0])

    def write_control(self, req, value):
        if self.dev.ctrl_transfer(2 << 5, req, 0x0000, 0x0000, value) != len(value):
            raise RuntimeError("Failed control transfer")