KATCP wrapper for beamformer

scripts/corr_katcp_interface.py

@@ -3,10 +3,10 @@
 Author: Jason Manley
 Date: 2010-11-11"""
 
-import logging,corr,sys,Queue,katcp
+import logging,corr,sys,Queue,katcp,time
 from optparse import OptionParser
 from katcp.kattypes import request, return_reply, Float, Int, Str, Bool
-import struct
+import struct, re, string
 
 logging.basicConfig(level=logging.WARN,
                     stream=sys.stderr,
@@ -87,6 +87,7 @@ def request_initialise(self, sock, n_retries):
         # Next, if beamformer object is instantiated, initiate the beamformer
         if self.b is not None:
             try:
+              time.sleep(1) # allow little time for corr init to close
               self.b.initialise()
             except:
               return("fail","Beamformer could not initialise. Check the log.")
@@ -140,6 +141,37 @@ def request_label_input(self, sock, input_n, ant_str):
             #return("fail","it broke.")
             return("fail","Sorry, your input number is invalid. Valid range: 0 to %i."%(self.c.config['n_inputs']-1))
 
+
+    @return_reply(Str(),Str())
+    def request_bf_destination(self, sock, orgmsg):
+        """Set destination for a given stream to a new IP address. The first argument should be the stream name, the second meta/data, the third the IP address in dotted-quad notation. An optional fourth parameters is the port."""
+        if self.c is None:
+            return ("fail","... you haven't connected yet!")
+        if self.b is None:
+            return ("fail","... no beamformer available!")
+        if len(orgmsg.arguments) < 3: return ("fail", "... usage: <stream> <meta/data> <ip> [port]")
+        stream     = orgmsg.arguments[0]
+        identifier = orgmsg.arguments[1]
+        ip         = orgmsg.arguments[2]
+
+        if not stream in self.b.get_beams():
+            return ("fail", "... name %s not a known beam!"%(stream))
+        if len(ip.split('.')) != 4: return ("fail", "Not an expected ip address format")
+        if len(orgmsg.arguments)>3:
+            try: port=int(orgmsg.arguments[3])
+            except Exception as e: return ("fail", "... Exception %s" % e)
+        else: port=None
+
+        if string.lower(identifier) == "meta":
+            self.b.config_meta_output(beams=stream,dest_ip_str=ip,dest_port=port, issue_spead=False)
+        if string.lower(identifier) == "data":
+            self.b.config_udp_output(beams=stream,dest_ip_str=ip,dest_port=port, issue_spead=False)
+        time.sleep(3)
+        return ("ok",
+        "data %s:%i"%(self.b.config['bf_rx_udp_ip_str_beam%d'%self.b.beam2index(stream)[0]], self.b.config['bf_rx_udp_port_beam%d'%self.b.beam2index(stream)[0]]),
+        "meta %s:%i"%(self.b.config['bf_rx_meta_ip_str_beam%d'%self.b.beam2index(stream)[0]], self.b.config['bf_rx_udp_port_beam%d'%self.b.beam2index(stream)[0]])
+        )
+
     @return_reply(Str(),Str())
     def request_tx_start(self, sock, orgmsg):
         """Start transmission to the given IP address and port, or use the defaults from the config file if not specified. The first argument should be the IP address in dotted-quad notation. The second is the port."""
@@ -149,6 +181,20 @@ def request_tx_start(self, sock, orgmsg):
         beam=None
         dest_ip_str=None
         dest_port=None
+
+        if len(orgmsg.arguments)>2:
+            # beamformer port
+            try: dest_port=int(orgmsg.arguments[2])
+            except Exception as e: return ("fail", "... %s" % e)
+
+        if len(orgmsg.arguments)>1:
+            # second argument can be either port of ip
+            if (len(orgmsg.arguments[1].split('.')) == 4) and (self.b is not None): # ip address
+                dest_ip_str=orgmsg.arguments[1]
+            else:
+                try: dest_port=int(orgmsg.arguments[1])
+                except Exception as e: return ("fail", "... %s" % e)
+
         if len(orgmsg.arguments)>0:
             # first argument can be either stream name or ip
             if len(orgmsg.arguments[0].split('.')) == 4: # ip address
@@ -157,28 +203,14 @@ def request_tx_start(self, sock, orgmsg):
             elif (self.b is not None): # stream name = beam name
                 beam=orgmsg.arguments[0]
                 if not beam in self.b.get_beams():
+                    self.reply_inform(sock, katcp.Message.inform(orgmsg.name, "Name %s not a known beam, assuming correlator output"%(beam)),orgmsg)
                     beam=None
-                    self.reply_inform(sock, katcp.Message.inform(orgmsg.name, "Name %s not a know beam, assuming correlator output"%(beam)),orgmsg)
-                else:
-                    # default destination for selected beam
-                    dest_port=None
-                    dest_ip_str=None
-        if len(orgmsg.arguments)>1:
-            # second argument can be either port of ip
-            if (len(orgmsg.arguments[1].split('.')) == 4) and (self.b is not None): # ip address
-                dest_ip_str=orgmsg.arguments[1]
-            else:
-                try: dest_port=int(orgmsg.arguments[1])
-                except Exception as e: return ("fail", "... %s" % e)
-        if len(orgmsg.arguments)>2:
-            # beamformer port
-            try: dest_port=int(orgmsg.arguments[2])
-            except Exception as e: return ("fail", "... %s" % e)
-
         try:
             if (self.b is not None) and (beam is not None):
-                self.b.config_udp_output(beams=beam,dest_ip_str=dest_ip_str,dest_port=dest_port)
-                self.b.spead_issue_all(beams=beam)
+                self.b.config_meta_output(beams=beam,dest_ip_str=dest_ip_str,dest_port=dest_port, issue_spead=False)
+                self.b.config_udp_output(beams=beam,dest_ip_str=dest_ip_str,dest_port=dest_port, issue_spead=False)
+                self.b.spead_issue_all(beams=beam, from_fpga=False)
+                time.sleep(1) # allow little time for meta data issue to finish
                 self.b.tx_start(beams=beam)
                 return ("ok",
                 "data %s:%i"%(self.b.config['bf_rx_udp_ip_str_beam%d'%self.b.beam2index(beam)[0]], self.b.config['bf_rx_udp_port_beam%d'%self.b.beam2index(beam)[0]]),

src/bf_functions.py

@@ -338,7 +338,7 @@ def frequency2fft_bin(self, frequencies=all):
         elif frequencies == all:
             fft_bins = range(n_chans)
         else:
-            bandwidth = (self.get_param('adc_clk')/2)
+            bandwidth = self.get_param('bandwidth')
             start_freq = 0
             channel_width = bandwidth/n_chans
             for frequency in frequencies:
@@ -350,7 +350,7 @@ def frequency2fft_bin(self, frequencies=all):
     def get_bf_bandwidth(self):
 	"""Returns the bandwidth for one bf engine"""
 
-	bandwidth = (self.get_param('adc_clk')/2)
+	bandwidth = self.get_param('bandwidth')
         bf_be_per_fpga = len(self.get_bfs()) 
         n_fpgas = len(self.get_fpgas())
 
@@ -370,7 +370,7 @@ def get_bf_fft_bins(self):
     def get_fft_bin_bandwidth(self):
 	"""get bandwidth of single fft bin"""
 	n_chans = self.get_param('n_chans')
-        bandwidth = self.get_param('adc_clk')/2
+        bandwidth = self.get_param('bandwidth')
 
 	fft_bin_bandwidth = bandwidth/n_chans
 	return fft_bin_bandwidth
@@ -391,7 +391,7 @@ def fft_bin2frequency(self, fft_bins=all):
                                'function %s, line no %s\n' %(__name__, inspect.currentframe().f_lineno), \
                                self.syslogger)
 
-        bandwidth = self.get_param('adc_clk')/2
+        bandwidth = self.get_param('bandwidth')
 
 	for fft_bin in fft_bins:
 	    frequencies.append((float(fft_bin)/n_chans)*bandwidth)
@@ -833,27 +833,44 @@ def bf_write_int(self, destination, data, offset=0, beams=all, antennas=None, fr
                 self.write_int('control', [control], 0, fft_bins=fft_bins, blindwrite=blindwrite)      
 
     def cf_bw2fft_bins(self, centre_frequency, bandwidth):
-        """returns fft bins associated with provided centre_frequency and bandwidth"""
+        """returns fft bins associated with provided centre_frequency and bandwidth
+        centre_frequency is assumed to be the centre of the 2^(N-1) fft bin"""
         bins = []
 
-        adc_clk = self.get_param('adc_clk')
         n_chans = self.get_param('n_chans')
+        max_bandwidth = self.get_param('bandwidth')
+        fft_bin_width = self.get_fft_bin_bandwidth()
 
         #TODO spectral line mode systems??
-        if (centre_frequency-bandwidth/2) < 0 or (centre_frequency+bandwidth/2) > adc_clk/2:
-            raise fbfException(1, 'Band specified out of range of our system', \
+        if (centre_frequency-bandwidth/2) < 0 or \
+           (centre_frequency+bandwidth/2) > max_bandwidth or \
+           (centre_frequency-bandwidth/2) >= (centre_frequency+bandwidth/2):
+            raise fbfException(1, 'Band specified not valid for our system', \
                                'function %s, line no %s\n' %(__name__, inspect.currentframe().f_lineno), \
                                self.syslogger)
 
-        #full band required
-        if bandwidth == adc_clk/2:
-            bins = range(n_chans) 
-        else:    
-            #get fft bin for edge frequencies
-            edge_bins = self.frequency2fft_bin(frequencies=[centre_frequency-bandwidth/2, centre_frequency+bandwidth/2])
-            bins = range(edge_bins[0], edge_bins[1]+1)
+        #get fft bin for edge frequencies
+        edge_bins = self.frequency2fft_bin(frequencies=[centre_frequency-bandwidth/2, centre_frequency+bandwidth/2-fft_bin_width])
+        bins = range(edge_bins[0], edge_bins[1]+1)
 
         return bins
+
+    def cf_bw2cf_bw(self, centre_frequency, bandwidth):
+        """converts specfied centre_frequency, bandwidth to values possible for our system"""
+
+        bins = self.cf_bw2fft_bins(centre_frequency, bandwidth)
+
+        bfs = self.frequency2bf_index(fft_bins=bins, unique=True)
+	bf_bandwidth = self.get_bf_bandwidth()
+	fft_bin_bandwidth = self.get_fft_bin_bandwidth()
+
+	#calculate start frequency accounting for frequency specified in centre of bin
+	start_frequency = min(bfs)*bf_bandwidth
+	beam_centre_frequency = start_frequency+bf_bandwidth*(float(len(bfs))/2)
+
+	beam_bandwidth = len(bfs) * bf_bandwidth
+
+        return beam_centre_frequency, beam_bandwidth
 
     def get_enabled_fft_bins(self, beam):
         """Returns fft bins representing band that is enabled for beam"""
@@ -898,7 +915,12 @@ def initialise(self, set_cal = True, config_output = True, send_spead = True):
         if set_cal: self.cal_set_all(all, spead_issue=False)
         else: self.syslogger.info('Skipped calibration config of beamformer.')
 
-        if send_spead: self.spead_issue_all(all)
+        #if we set up calibration weights, then config has these already so don't 
+        #read from fpga
+        if set_cal: from_fpga = False
+        else: from_fpga = True
+
+        if send_spead: self.spead_issue_all(beams=all, from_fpga=from_fpga)
         else: self.syslogger.info('Skipped issue of spead meta data.')
 
         self.syslogger.info("Beamformer initialisation complete.")
@@ -1066,7 +1088,7 @@ def set_passband(self, beams=all, centre_frequency=None, bandwidth=None, spead_i
 
         beams = self.beams2beams(beams)
 
-	max_bandwidth = self.get_param('adc_clk')/2
+	max_bandwidth = self.get_param('bandwidth')
 
         for beam in beams:
 
@@ -1081,18 +1103,15 @@ def set_passband(self, beams=all, centre_frequency=None, bandwidth=None, spead_i
 	    else:
 		b = bandwidth
 
-	    if ((cf-b/2) < 0) or ((cf+b/2) > max_bandwidth):
-                raise fbfException(1, 'Passband settings specified for beam %s out of range 0->%iMHz'%(beam, max_bandwidth/1000000), \
-                                   'function %s, line no %s\n' %(__name__, inspect.currentframe().f_lineno), \
-                                   self.syslogger)
+            cf_actual, b_actual = self.cf_bw2cf_bw(cf,b)
 
             if centre_frequency != None:
-                self.set_beam_param(beam, 'centre_frequency', centre_frequency)
-                self.syslogger.info('Centre frequency for beam %s set to %i Hz'%(beam, centre_frequency))
+                self.set_beam_param(beam, 'centre_frequency', cf_actual)
+                self.syslogger.info('Centre frequency for beam %s set to %i Hz'%(beam, cf_actual))
 
             if bandwidth != None:
-                self.set_beam_param(beam, 'bandwidth', bandwidth)
-                self.syslogger.info('Bandwidth for beam %s set to %i Hz'%(beam, bandwidth))
+                self.set_beam_param(beam, 'bandwidth', b_actual)
+                self.syslogger.info('Bandwidth for beam %s set to %i Hz'%(beam, b_actual))
 
             if centre_frequency != None or bandwidth != None:
                 #restart if currently transmitting
@@ -1110,18 +1129,14 @@ def set_passband(self, beams=all, centre_frequency=None, bandwidth=None, spead_i
     def get_passband(self, beam):
         """gets the centre frequency and bandwidth for the specified beam"""
 
-        fft_bins = self.get_enabled_fft_bins(beam)
-        bfs = self.frequency2bf_index(fft_bins=fft_bins, unique=True)
-	bf_bandwidth = self.get_bf_bandwidth()
-	fft_bin_bandwidth = self.get_fft_bin_bandwidth()
+        #parameter checking
+        cf = self.get_beam_param(beam, 'centre_frequency')
 
-	#calculate start frequency accounting for frequency specified in centre of bin
-	start_frequency = min(bfs)*bf_bandwidth-fft_bin_bandwidth/2
-	centre_frequency = start_frequency+bf_bandwidth*(float(len(bfs))/2)
+        b = self.get_beam_param(beam, 'bandwidth')
 
-	beam_bandwidth = len(bfs) * bf_bandwidth
-	
-        return centre_frequency, beam_bandwidth
+        cf_actual, b_actual = self.cf_bw2cf_bw(cf,b)
+
+        return cf_actual, b_actual
 
     def get_n_chans(self, beam):
         """gets the number of active channels for the specified beam"""       
@@ -1577,14 +1592,14 @@ def spead_cal_meta_issue(self, beams=all, from_fpga=True):
             self.send_spead_heap(beam, ig)
             self.syslogger.info("Issued SPEAD EQ metadata for beam %s" %beam)
 
-    def spead_issue_all(self, beams=all):
+    def spead_issue_all(self, beams=all, from_fpga=True):
         """Issues all SPEAD metadata."""
 
         self.spead_static_meta_issue(beams)
         self.spead_passband_meta_issue(beams)
         self.spead_destination_meta_issue(beams)
         self.spead_time_meta_issue(beams)
         self.spead_eq_meta_issue(beams)
-        self.spead_cal_meta_issue(beams)
+        self.spead_cal_meta_issue(beams=beams, from_fpga=from_fpga)
         self.spead_labelling_issue(beams)
 

src/corr_functions.py

@@ -237,8 +237,8 @@ def __init__(self, connect = True, config_file = '/etc/corr/default', log_handle
             logger.setLevel(log_level)
 
         self.syslogger.info('Configuration file %s parsed ok.' % config_file)
-        self.spead_tx=spead.Transmitter(spead.TransportUDPtx(self.config['rx_meta_ip_str'], self.config['rx_udp_port']))
-        self.spead_ig=spead.ItemGroup()
+        self.spead_tx = spead.Transmitter(spead.TransportUDPtx(self.config['rx_meta_ip_str'], self.config['rx_udp_port']))
+        self.spead_ig = spead.ItemGroup()
 
         if connect == True:
             self.connect()
@@ -1738,7 +1738,7 @@ def config_roach_10gbe_ports(self):
 #                # Assign an IP address to each XAUI port's associated 10GbE core.
 #                fpga.write_int('gbe_ip%i'%x, ip)
 
-    def config_udp_output(self,dest_ip_str=None,dest_port=None):
+    def config_udp_output(self, dest_ip_str=None, dest_port=None):
         """Configures the destination IP and port for X engine output. dest_port and dest_ip are optional parameters to override the config file defaults. dest_ip is string in dotted-quad notation."""
         if dest_ip_str==None:
             dest_ip_str=self.config['rx_udp_ip_str']
@@ -1756,6 +1756,10 @@ def config_udp_output(self,dest_ip_str=None,dest_port=None):
         self.xwrite_int_all('gbe_out_ip',struct.unpack('>L',socket.inet_aton(dest_ip_str))[0])
         self.xwrite_int_all('gbe_out_port',dest_port)
         self.syslogger.info("Correlator output configured to %s:%i." % (dest_ip_str, dest_port))
+
+        # need a new spead transmitter if the port and ip have changed
+        self.spead_tx = spead.Transmitter(spead.TransportUDPtx(self.config['rx_meta_ip_str'], self.config['rx_udp_port']))
+
         #self.xwrite_int_all('gbe_out_pkt_len',self.config['rx_pkt_payload_len']) now a compile-time option
 
         #Temporary for correlators with separate gbe core for output data: