USRP (x310) AGC Code for a GnuRadio Embedded Python Block

agc.py

"""
USRP AGC Control

This block is a PI controller to the gain of the USRP based on the 
received signal. This particular implementation uses the peak of 
the signal, but other metrics, such as variance, os standard deviation
can be employed. The gain is controled in 0.5 steps and an hysteresis
can be set.
The main goal of such block is to avoid saturation at the ADC or
analog frontent.
"""

import numpy as np
from gnuradio import gr
import pmt

class blk(gr.sync_block):  # other base classes are basic_block, decim_block, interp_block
    def __init__(self, alpha=1.0, proportional=0.7, integral=1e-3, deadband = 1.0, setpoint = 0.3, maxgain=31.5):  # only default arguments here
        gr.sync_block.__init__(
            self,
            name='USRP Frontend AGC',   # will show up in GRC
            in_sig=[np.complex64],
            out_sig=[]
        )
        # if an attribute with the same name as a parameter is found,
        # a callback is registered (properties work, too).
        self.maxgain = maxgain
        self.alpha = alpha
        self.meas = 0
        self.message_port_register_out(pmt.intern("cmd"))
        self.ctrl = maxgain
        self.p = proportional
        self.i = integral
        self.setpoint = setpoint
        self.deadband = deadband
        cmd = pmt.cons(pmt.intern("gain"), pmt.from_double(self.ctrl))
        self.message_port_pub(pmt.intern("cmd"), cmd)
        self.counter = 0
        self.lastctrl = 0

    def work(self, input_items, output_items):
        in0 = input_items[0]
        if (self.counter==0):
            #self.meas = (self.alpha)*in0.std() + (1-self.alpha)*self.meas
            #self.meas = (self.alpha)*abs(in0[-1]) + (1-self.alpha)*self.meas
            self.meas = (self.alpha)*abs(in0).max() + (1-self.alpha)*self.meas
            error = self.setpoint - self.meas
            error_db = 20*np.log10(self.meas/self.setpoint)
            if (abs(error_db) < self.deadband) :
                error = 0

            self.ctrl = (self.ctrl + error*self.i) + error*self.p 
            self.ctrl = np.clip(self.ctrl, 0, self.maxgain)

            if (abs(self.ctrl - self.lastctrl) > 0.5):
                command = pmt.cons(
                    pmt.intern("gain"), 
                    pmt.from_double(int(self.ctrl*2)/2.0))
                self.message_port_pub(pmt.intern("cmd"), command)
                self.lastctrl = self.ctrl
        self.counter = (self.counter + 1)%17
        return len(in0)