Re: [Discuss-gnuradio] Rearranging a running flow graph

[Josh Blum]

So, I went the python way and created a valve hier block. It is a 
message sink and a message source with a thread that only forwards 
messages when the valve is closed. (code attached)

I noticed that calling fg.stop() blocks because my thread is stuck in a 
loop. I am using the gr.gr_threading framework. Is there a proper way to 
setup a gr.gr_threading.Thread so that fg.stop() can kill/stop the 
running threads?

Thanks,
-Josh

Josh Blum wrote:
> Eric Blossom wrote:
> > On Mon, Apr 16, 2007 at 11:51:09AM +0200, Trond Danielsen wrote:
> > > 2007/3/17, Josh Blum <address@hidden>:
> > > > I think gnuradio needs a mux and a demux block. A mux has N inputs 
> > > > and a
> > > > set_n method. The mux will only feed the output with the nth input
> > > > stream (throw out/ignore the other inputs). A demux has N outputs and a
> > > > set_n method. The demux will only feed the nth output with the input 
> > > > stream.
> > > Has anybody written something like this allready?
> >
> > No. 
> > I think this should be implemented using the work-in-progress
> > that Johnathan's doing for hier_block2.  It may be about a week too
> > early to use ;)
>
> How would you go about doing this? I had thought that one could use a 
> bunch of message sources and sinks, and a thread could facilitate data 
> transfer for only one source-sink pair. But how could several blocks 
> with single inputs appear as one block with multiple inputs? Seems like 
> trouble...
>
> -------
>
> On another note: The original problem was trying to toggle a section of 
> a running flow graph on and off. The gr.skiphead seems to be able to do 
> this by starving the data stream for the first n samples. Could there be 
> a gr.valve block that can starve a data stream based on a boolean flag? 
> The flag could set by an open/close method (or something to that effect).
>
> Or a block like this exists and I missed it?
>
> -Josh
>
> > Eric
> >
> >
> > _______________________________________________
> > Discuss-gnuradio mailing list
> > address@hidden
> > http://lists.gnu.org/mailman/listinfo/discuss-gnuradio

#######################################################################################
##      Valve Def, Hier Block, and Helper Thread
#######################################################################################

class ValveThread(threading.Thread):
        """Thread to forward data between the input message queue and the 
output message queue."""
        def __init__(self, valve_helper):
                """!
                ValveThread contructor.
                @param valve_helper the valve helper block
                """
                self.valve_helper = valve_helper
                threading.Thread.__init__(self)
                self.start()
                print 'Created valve thread.'
                
        def run(self):
                """In an endless while loop: read the msgq_out and write to the 
msgq_in when closed."""
                while True:
                        msg = self.valve_helper.msgq_out.delete_head()  
#blocking read of message queue
                        if self.valve_helper.open: del msg      #delete the 
message
                        else: self.valve_helper.msgq_in.insert_tail(msg) 
#forward message                       

class ValveHelper(gr.hier_block):
        """A hier block used to intercept data from a message sink, 
        and relay it to a message source based on the state of the valve: 
open/closed."""
        def __init__(self, fg, item_size, open):
                """!
                ValveHelper constructor.
                @param fg the gr flow graph
                @param item_size the size of the gr data stream in bytes
                @param open the valve is open if bool(open) evaluates true
                """
                self.set_open(open)
                #create blocks          
                self.msgq_out = gr.msg_queue(DEFAULT_QUEUE_LIMIT)               
                msg_sink = gr.message_sink(item_size, self.msgq_out, False)
                msg_source = gr.message_source(item_size, DEFAULT_QUEUE_LIMIT)
                self.msgq_in = msg_source.msgq()
                #connect blocks
                gr.hier_block.__init__(self, fg, msg_sink, msg_source)          
                #start the thread               
                ValveThread(self)
                
        def set_open(self, open):
                """!
                Set the open state of the valve.
                @param open the new open state
                """
                self.open = open

def Valve(sb):
        type = Enum(all_choices, 1)
        vlen = Int(1, min=1)
        sb.add_input_socket('in', Variable(type), vlen=vlen)
        sb.add_output_socket('out', Variable(type), vlen=vlen)
        sb.add_param('Type', type, False, type=True)
        sb.add_param('Sampling Rate', Float(default_samp_rate)) 
        sb.add_param('Open', Int(0))    
        sb.add_param('Vector Length', vlen)
        sb.set_docs('''\
When open is 1, the valve will not forward data.
The valve has a throttle automatically attatched to it at runtime to save the 
CPU.
''')    
        def make(fg, type, samp_rate, open, vlen):
                item_size = type.parse().get_num_bytes()*vlen.parse()
                block = ValveHelper(fg, item_size, open.parse())
                fg.add_callback(block.set_open, open)
                return ThrottleHelper(fg, item_size, samp_rate.parse(), block, 
True)
        return sb, make