arduino-oscilloscope/epy_block_0.py

import gnuradio
import numpy
import serial


class ArduinoAnalogReadings(gnuradio.gr.sync_block):
    def __init__(
        self, port="/dev/ttyUSB0", baud_rate=115200, buffer_max_length=10, timeout=2
    ):
        gnuradio.gr.sync_block.__init__(
            self, name="Arduino Analog Readings", in_sig=None, out_sig=[numpy.int16]
        )
        self._serial_port = serial.Serial(
            port=port, baudrate=baud_rate, bytesize=serial.EIGHTBITS, timeout=timeout
        )
        self._buffer_max_length = buffer_max_length

    def work(self, input_items, output_items):
        # pylint: disable=unused-argument
        buffer = self._serial_port.read(
            2 * min(len(output_items[0]), self._buffer_max_length)
        )
        assert len(buffer) % 2 == 0
        buffer_samples_count = len(buffer) // 2
        output_items[0][:buffer_samples_count] = numpy.frombuffer(buffer, dtype=">u2")
        assert output_items[0][:buffer_samples_count].max() < 1024, "lost sync?"
        return buffer_samples_count