# # Uses flask to create a web server that shows FFT in real-time from # multiple bands of a receiver. Optionally save the data to a se # of text files - data in CSV format. # # Copyright Trimble Inc. 2023 - 2024 # import requests import xmltodict import numpy as np from pylab import where import threading import time import sys import signal import datetime import argparse from flask import Flask, jsonify, request, Response from flask_compress import Compress import zlib #import zstandard as zstd import datetime # Create the flask app object app = Flask(__name__) Compress(app) # Enable gzip compression RFBands = ['L1', 'L2', 'L5', 'E6', 'B1'] # Global variable that holds the latest FFT data FFTData = [''] * len(RFBands) # Center Freq for each band centFreq = [0]*len(RFBands) allFreq = [] # When True gets the "raw" FFT, this is very noisy so the thresholds # need adjusting. The default has some very mild filtering raw = False # = False matches the web GUI high rate FFT # Save data to a set of text files? saveAll = False # From local network testing the throughput (FFT rate) is roughly # the same (~5Hz) on the Linux end limited by the rate the receiver # generates the FFTs. However, while gzip will reduce the amount # of data passed over the network, it will approximately double the # HTTPD load on an Alloy (5% vs 10%). Therefore, for local testing, # disabling gzip is the best option. A remote receiver may be # different as the Internet bandwidth may become a driving factor. headers={'Accept-Encoding':'gzip'} #headers={'Accept-Encoding':'None'} proxies={'http':None,'https':None} # Used to control the Ctrl-C program termination ThreadingActive = False # Called when CTRL=C is pressed. Kills the threads and exits def signal_handler(signal,frame): global ThreadingActive print('Ctrl+C Detected - shutting down ...') ThreadingActive = False for num in range(len(threadHandle)): threadHandle[num].join() sys.exit(0) # Called to get the FFT data from the receiver def getFFT(URL,session,showTimeTag=True): r = session.get(URL, auth=(user,password), headers=headers, proxies={'http':None}, timeout=5) r.raise_for_status() tokens = r.text.split(',') timeTag = tokens[0] # in milliseconds in the first field if(showTimeTag): with mutex: print(timeTag) data = np.array(tokens[1:2049]).astype(int) return(timeTag, data) # Called by each thread to get the FFT data from a band of a single receiver def fftWorker(bands): global FFTData global centFreq bandToken = '' bandFreqLUT = {} index = 0 for thisBandStr in bands: # Derived constants url = 'http://' + IPAddr + '/xml/dynamic/rfSpectrumAnalyzer.xml?rfBand=' + thisBandStr + '&filterMode=NoFilter' r = requests.get(url, auth=(user,password), proxies='', timeout=5) r.raise_for_status() d = xmltodict.parse(r.text) numZeroes = 0 for thisData in d['saData']['points']['y']: if(thisData == '0'): numZeroes +=1 if(numZeroes == 2048): # All zeroes == band not supported print(thisBandStr,': No FFT data') continue if(len(bandToken) > 0): bandToken += ',' bandToken += thisBandStr centFreq[index] = float(d['saData']['center_freq_mhz']) bandFreqLUT[centFreq[index]] = thisBandStr allFreq.append({thisBandStr:centFreq[index]}) index +=1 #url = 'http://' + IPAddr + '/xml/dynamic/push_rfSpectrumAnalyzer.csv?rfBand=L1,L2,L5,E6,B1&minRate=200&dB=1' url = 'http://' + IPAddr + '/xml/dynamic/push_rfSpectrumAnalyzer.csv?rfBand=' + bandToken + '&minRate=200&dB=1' with requests.get(url, auth=(user,password), stream=True, headers=headers, proxies=proxies, timeout=60) as resp: for line in resp.iter_lines(): if(ThreadingActive == False): return if line: thisLine = line.decode('utf-8') if(thisLine.startswith('data: ')): # Dump "data: " which is part of the streamed standard #with mutex: # print(datetime.datetime.now().isoformat(),thisLine[:30]) thisLine = thisLine[6:] tokens = thisLine.split(',') # Data is CSV #print(len(tokens)) if(len(tokens) == 2053): # col 0 = GPS time of week in milliseconds # col 1 = Center frequency (middle of FFT) in MHz # col 2 = pre or post mitigation FFT (M7 will always be pre == 0) # col 3 = Position or Vector antenna # col 4 = RSSI (place holder for M8) # 2048 columns of FFT data # Total = 2053 columns # From the center frequency, determine which band we are # processing gotBand = False for thisFreq, thisBandStr in bandFreqLUT.items(): if(thisFreq == float(tokens[1])): gotBand = True break if(gotBand == False): continue with mutex: # Copy the FFT data into the global variable FFTData[RFBands.index(thisBandStr)] = thisLine if(saveAll): # Save if enabled # Create hourly files now = datetime.datetime.now(datetime.timezone.utc) dateTimeStr = str(now.year) + '-' + str(now.month).zfill(2) + '-' + str(now.day).zfill(2) dateTimeStr += '-' + str(now.hour).zfill(2) dateTimeStr += '-' + thisBandStr + '.txt' with open(dateTimeStr,'a') as fid: fid.write(thisLine) fid.write('\n') print('Processing Complete') def event_stream(compressionType): if(compressionType == 'gzip'): stream = zlib.compressobj(9) # Set to maximum zlib compression # elif(compressionType == 'zstd'): # ToDo - zstd compression is not as good as zlib - why? # compression_level = 7 # stream = zstd.ZstdCompressor(level=compression_level) lastTimeTag = [''] * len(RFBands) while ThreadingActive: dataStr = '' for band in range(len(RFBands)): with mutex: thisLine = FFTData[band] tokens = thisLine.split(',') # Data is CSV if(len(tokens) == 2053): if(lastTimeTag[band] == tokens[0]): continue else: lastTimeTag[band] = tokens[0] #diff = [(5.0 + 10.0* np.log10(int(j))-10.0*np.log10(int(i))) for i, j in zip( tokens[2:-1], tokens[3:])] #dataStr = tokens[0] + ',' + tokens[1] + ',' + ','.join(f'{(num):.1f}' for num in diff) # For Linear FFT #dataStr = tokens[0] + ',' + tokens[1] + ',' + ','.join(f'{(10*np.log10(int(num))):.1f}' for num in tokens[2:]) # For Log FFT, just copy over the data dataStr = thisLine # Just copy over the dB data if(compressionType == 'gzip'): data = dataStr.encode('utf-8') + b'\n' cdata = stream.compress(b"data: %s\n\n" % data) + stream.flush(zlib.Z_SYNC_FLUSH) with mutex: print('gzip',tokens[0],tokens[1],len(thisLine),len(dataStr),len(data),len(cdata)) yield cdata # elif(compressionType == 'zstd'): # data = dataStr.encode('utf-8') + b'\n' # cdata = stream.compress(b"data: %s\n\n" % dataStr.encode('utf-8')) #+ stream.flush() # print('zstd',len(thisLine),len(dataStr),len(data),len(cdata)) # yield cdata else: yield dataStr.encode('utf-8') time.sleep(0.01) # Ensure we don't get into a tight loop # Provide the FFT data (all bands) to the web server @app.route("/getFFT") def getTestTrack(): print('In getFFT') compressionType = 'None' if('gzip' in request.headers.get('Accept-Encoding')): compressionType = 'gzip' # elif('zstd' in request.headers.get('Accept-Encoding')): # compressionType = 'zstd' print(compressionType) resp = Response(event_stream(compressionType),mimetype="text/event-stream") resp.headers["Transfer-Encoding"] = "chunked" if(compressionType == 'gzip'): resp.headers["Content-Encoding"] = "deflate" # elif(compressionType == 'zstd'): # resp.headers["Content-Encoding"] = "zstd" return resp # Return the center frequency for each band. This is used to label the # x-axis of the plot @app.route("/freq.json") def freq(): # Round to 2 significant digits #jsonData = jsonify(centFreq) jsonData = jsonify(allFreq) return jsonData # Return the dimensions of the plot to the web server so that # the canvas, which contains the plot, can be sized correctly @app.route("/dimensions.json") def dimensions(): # Round to 2 significant digits jsonData = jsonify(dimensions) return jsonData @app.route('/favicon.ico') def favicon(): return app.send_static_file('favicon.ico') # Map / to index.html @app.route('/') def main(): return app.send_static_file('index.html') # index.html is in the static directory, so we need to map that # to the root of the web server @app.route('/index.html') def index(): return app.send_static_file('index.html') # Main entry point if __name__ == '__main__': # Parse arguments parser = argparse.ArgumentParser(description='coerBuild Spectrogram viewer') parser.add_argument("IPAddr", help="IP Address of the receiver") parser.add_argument('-p','--port', help='Optional HTTP port, by default 9876 e.g. --port 80') parser.add_argument('-u','--user', help='Option username, by default uses "admin". To change e.g. --user stinger') # Note the width/height control the canvas (where we draw the Spectrogram). This gives the fundamental resolution of # the plot. We apply a "100%" style to the canvas, so it will scale to the screen. If you set the width/height to # a small value, the plot will be very pixelated. Conversely, if you set it to a very high value, you are wasting memory/CPU # resources in the browser as the monitor may not be able to display the high resolution data. The default values will be # good for most applications parser.add_argument('-x','--width', help='Optional width of an individual Spectogram - plot will be larger, by default 900 e.g. --width 1024') parser.add_argument('-y','--height', help='Optional height of an individual Spectogram - plot will be larger, by default 400 e.g. --height 400') parser.add_argument('-c','--password', help='Password, by default uses "password". To change e.g. --password mypassword') parser.add_argument('-a','--saveall', help='Default is false, when this flag is present, all FFT data will be saved to files',action='store_true') args = parser.parse_args() if(args.port): webPort = int(args.port) else: #webPort = 9876 #webPort = 8080 webPort = 8081 if(args.width): width = int(args.width) else: width = 900 if(args.height): height = int(args.height) else: height = 400 dimensions = {"width":width, "height":height} print(dimensions) if(args.user): user = args.user else: user = 'admin' if(args.saveall): saveAll = args.saveall if(args.password): password = args.password else: password = 'password' IPAddr = args.IPAddr # Setup a Ctrl-C handler signal.signal(signal.SIGINT, signal_handler) # Setup a mutex - we'll use this to protect stdout printing. mutex = threading.Lock() # Kick off a thread to get the FFT data ThreadingActive = True # We are about to enable the threading threadHandle = [] thisThread = threading.Thread(target=fftWorker, args =([RFBands])) threadHandle.append(thisThread) thisThread.start() # Now kick off the flask app app.run(host='0.0.0.0',port=webPort,threaded=True)