baumax-132600620-xmas-tree-lights-signal/serial_source_block.py

import gnuradio
import numpy
import serial


class SerialSourceBlock(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="Serial Source Block", in_sig=None, out_sig=[numpy.uint8]
        )
        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(
            min(len(output_items[0]), self._buffer_max_length)
        )
        output_items[0][: len(buffer)] = numpy.frombuffer(buffer, dtype="u1")
        assert output_items[0][: len(buffer)].max() <= 1, "invalid value"
        return len(buffer)