I would like to use GNU radio to begin recording a predetermined number of complex samples to a file sink automatically as soon as a signal magnitude threshold is reached. For example, as the gnu radio flow graph is running, it should be monitoring a signal and waiting for the amplitude to reach a threshold. As soon as the threshold is reached, the next 10,000 complex samples should be written to a file sink. No more samples should ever be written to the file.

What is the best way of doing this?

Is this possible to accomplish using a custom embedded python block that has one input and one output that only allows a predetermined number of samples through to the output after the threshold has been reached and never allows any more samples through?


2 Answers 2


For a signal that you expect to stay above the threshold once reached, you can use the "Power Squelch" block, with the "Gate" parameter set to "True", followed by a "Head" Block, set to 10,000 samples.

  • $\begingroup$ hahaha, oh yeah, that would indeed be easier. $\endgroup$ May 19, 2020 at 14:13
  • 1
    $\begingroup$ @MarcusMüller yeah, I've had to use it before to cut down on CPU resource exhaustion with a multichannel audio receiver flowgraph. In that application most of the channels were silent most of the time, so the gate functionality worked nicely in cutting down the sample processing load for the CPU. Of course you need to keep squelch broken to keep the samples flowing for OP's application. $\endgroup$
    – Andy Walls
    May 19, 2020 at 15:06

Sounds like you have a very workable approach:

Write a GNU Radio, Embedded Python, out-of-tree Python or C++, doesn't matter, which:

  • is a general block (not a sync_block)
  • has a member property triggered or similar, which is initialized to False in the constructor
  • has a member property threshold or similar, which is initialized to the value passed as constructor argument in the constructor
  • If triggered, then
    • copy input to output (specifically, min(len(input_items[0], noutput_items)), because in a general block, there's no guarantee that in- and output buffers have the same free space)
    • consume_each the number of items copied (tell GNU Radio you've processed this much input), and
    • return the number of items copied (you produced this much output).
  • else,
    • look for a super-threshold value in the input (numpy.any(numpy.abs(input_items[0]) > threshold)
    • if no values are found:
      • consume_each(len(input_items[0]))
      • return 0 (you didn't produce any output)
    • if a value is found:
      • triggered = True
      • find the position of the first value larger than the threshold
      • consume_each as much items as are before that position
      • return 0 (you didn't produce any output, but your work method will be called again, with triggered set)

... and connect its output to a "Head" block, set to pass 10000 items; connect that to your file sink.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.