# Arbitray waveform generator (AWG) communication to remote control UI # # # updated by wolf2018: version_date = "10-Mar-2022" # # v25-Feb: initial version # v10-Mar-2022: comment out blink(3) and blink after sending status to speed up communication with Remote UI import gc gc.collect() # precaution to free up unused RAM from machine import Pin, soft_reset, freq # set CPU frequency freq(250000000) # set GP23 to high to switch Pico power supply from PFM to PWM to reduce noise GP23 = Pin(23, Pin.OUT) GP23.value(1) #import AWG functions from wave_gen import * from sys import stdin, stdout import utime import json import io # initialize on-board LED led = Pin(25, Pin.OUT) # define wave with default values # wave values will be updated based on input from remote UI wave = {"func" : sine, "frequency" : 2000, "amplitude" : 0.48, "offset" : 0.5, "phase" : 0, "replicate" : 1, "pars" : [0.2, 0.4, 0.2], "maxsamp" : 512, } # maxsamp must be a multiple of 4. # AWG_status flag # status: Meaning: # ------- -------- # stopped generator output stopped, new set up trigger is allowed, initialization status # calc wave generetor set up is trigger, wave form is calculated, no further trigger is allowed # running calculation finished, generator output active, new set up trigger is allowed # -- init -- intitialization, generator not yet started using start button AWGstat = {"AWG_status" : "- init -", "nsamp" : 0, "F_out" : 0, } # Connection status to Remote UI Conn_stat = {"version" : version_date, "connection" : "not connected", "CPU" : str(freq()), } # Map function name to function Function = {"sine" : sine, "pulse" : pulse, "gauss" : gaussian, "sinc" : sinc, "expo" : exponential, "noise" : noise} # make buffers for the waveform. Here we reserve the max number of bytes # The Remote UI can transmit smaller number of buffer bytes to be used for calculation # large buffers give better results but are slower to fill wavbuf={} wavbuf[0]=bytearray(4096) wavbuf[1]=bytearray(4096) # define simple communication functions in_char="" in_text="" out_text="" Handshake_char = "=" def connect(): in_char="0" in_text="" while in_char != Handshake_char: in_char = stdin.read(1) #print(in_char) in_text += in_char #print(in_text) return in_text.strip() def send(out_file): #out_file expected as dict, packaged in Json format and dump it to stream stdout json.dump(out_file, stdout, separators=(",",":")) #Close sending with a newline character print() def receive(): #response = {"func": "pulse", "frequency": 10000, "amplitude": 0.89, "offset": 0, "phase": 0, "replicate": 1, "pars": [0.05, 0.45, 0.05], "frequency_value": 10000, "freq_range": 1, "command": "setup", "maxsamp": 512} #return response #print("receiving") bytes_in = stdin.readline() # repeat read if received data is not not json while str(bytes_in[0]) != "{": bytes_in = stdin.readline() #print("bytes_in= ", str(bytes_in)) #print("Json load now") response = json.load(io.BytesIO(bytes_in)) #print("response= ", response) return response def blink(number): blink = number * 2 for _ in range (blink): led.toggle() utime.sleep_ms(150) # program execution: # when program starts AWG listens for first communication # first command from RemoteUI should be (?????????) # loop to wait until connected to remote UI try: # signal starting state blink(1) # start connection hand shake by RemoteUI sending a "=" character connected = 0 while not connected: # signal waiting for connection text=connect() if text == "=": Conn_stat["connection"] = "READY" Conn_stat["version"] = version_date send(Conn_stat) connected = 1 # now we are connected and start main loop blink(2) # start main loop while True: #receive command response = receive() #blink(3) if response["command"] == "setup": # load wave from received values wave["func"] = Function[response["func"]] wave["frequency"] = response["frequency"] wave["amplitude"] = response["amplitude"] wave["offset"] = response["offset"] wave["phase"] = response["phase"] wave["replicate"] = response["replicate"] wave["pars"] = response["pars"] wave["maxsamp"] = response["maxsamp"] AWGstat["AWG_status"]="calc wave" setup_status = setupwave(wavbuf[0],wave) # setupwave returned to main program # setupwave returns AWG status, nsamp and Frequency out AWGstat["AWG_status"] = setup_status[0] AWGstat["nsamp"] = setup_status[1] AWGstat["F_out"] = setup_status[2] # send status to RemoteUI send(AWGstat) # signal running AWG blink(4) elif response["command"] == "stop": AWGstat["AWG_status"]="stopped" AWGstat["nsamp"] = 0 AWGstat["F_out"] = 0 # stop the generator output and send status to RemoteUI stopDMA() send(AWGstat) elif response["command"] == "file": #.....implement receipt of wave file later AWGstat["AWG_status"]="command: file not implemented" AWGstat["nsamp"] = 0 AWGstat["F_out"] = 0 send(AWGstat) elif response["command"] == "disconnect": AWGstat["AWG_status"]="connection reset" AWGstat["nsamp"] = 0 AWGstat["F_out"] = 0 # stop the generator output, send status to RemoteUI and restart the AWG stopDMA() send(AWGstat) soft_reset() else: AWGstat["AWG_status"]="error" AWGstat["nsamp"] = 0 AWGstat["F_out"] = 0 send(AWGstat) except KeyboardInterrupt: # set connection status and send to RemoteUI Conn_stat["version"] = "0.0.0" Conn_stat["connection"] = "closed" send(Conn_stat) connected = 0 except Exception as e: #print("0: mainloop crashed: ", e) led.on() Conn_stat["version"] = "mainloop crashed" Conn_stat["connection"] = e send(Conn_stat) connected = 0 finally: print("0: finally: cleaning up") stopDMA() #soft_reset()