211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
|
|
|
|
import time
|
|
|
|
from datetime import datetime
|
|
import device_commands as cmd
|
|
|
|
|
|
#### ---- Constants
|
|
|
|
WAIT_AFTER_SEND = 0.15 # Wait after sending command, before requesting input (in seconds).
|
|
|
|
|
|
#### ---- High-level port commands
|
|
'''
|
|
def create_port_connection():
|
|
prt = None
|
|
for port, _, _ in sorted(cmd.list_ports.comports()):
|
|
try:
|
|
prt = cmd.setup_port_connection(port=port, baudrate=115200, timeout_sec=1)
|
|
cmd.open_port(prt)
|
|
reset_port_settings(prt)
|
|
except:
|
|
prt.close()
|
|
continue
|
|
break
|
|
return prt
|
|
'''
|
|
def create_port_connection():
|
|
prt = None
|
|
print()
|
|
ports = []
|
|
for port, _,_ in sorted(cmd.list_ports.comports()):
|
|
ports.append(port)
|
|
|
|
#ONLY FOR LINUX!!!
|
|
have_ttyUSB = False
|
|
USB_ports = []
|
|
for port in ports:
|
|
if "USB" in port:
|
|
USB_ports.append(port)
|
|
if len(USB_ports):
|
|
ports = USB_ports
|
|
# print("ports:", ports)
|
|
|
|
|
|
# for port, _, _ in sorted(cmd.list_ports.comports()):
|
|
for port in ports:
|
|
try:
|
|
print("PORT:", port)
|
|
prt = cmd.setup_port_connection(port=port, baudrate=115200, timeout_sec=1)
|
|
cmd.open_port(prt)
|
|
reset_port_settings(prt)
|
|
except:
|
|
prt.close()
|
|
continue
|
|
break
|
|
return prt
|
|
|
|
|
|
|
|
|
|
# def setup_connection():
|
|
# prt = cmd.setup_port_connection()
|
|
# cmd.open_port(prt)
|
|
# return prt
|
|
|
|
|
|
def reset_port_settings(prt):
|
|
# Reset port settings and check status
|
|
cmd.send_DEFAULT_ENABLE(prt)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
|
|
|
|
def request_state(prt):
|
|
# Request data
|
|
cmd.send_STATE(prt)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
|
|
|
|
def send_control_parameters(prt, params):
|
|
# Send control parameters
|
|
hexstring = cmd.encode_Input(params)
|
|
cmd.send_DECODE_ENABLE(prt,hexstring)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
else:
|
|
print("")
|
|
|
|
def send_task_command(prt, sending_param):
|
|
# Send task command (TASK_ENABLE state in firmware)
|
|
hexstring = cmd.create_TaskEnableCommand(sending_param)
|
|
cmd.send_TASK_ENABLE(prt,hexstring)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
else:
|
|
print("")
|
|
|
|
|
|
|
|
def start_ramp_max(prt, freq_hz=None, duty=None, saw_step=None, pat_period=None, pat_period_base=None, dac_clk_hz=None, triangle=True, sram_mode=False, sram_samples=None, sram_hold=None, sram_amplitude=None):
|
|
# Start AD9102 sawtooth with configurable frequency/duty or SRAM ramp mode
|
|
if sram_mode:
|
|
if sram_hold is None:
|
|
sram_hold = cmd.AD9102_SRAM_HOLD_DEFAULT
|
|
if sram_samples is None and freq_hz is not None:
|
|
if dac_clk_hz is None:
|
|
dac_clk_hz = cmd.AD9102_DAC_CLK_HZ
|
|
sram_samples = cmd.calc_sram_samples_for_freq(freq_hz, dac_clk_hz, sram_hold)
|
|
hexstring = cmd.create_AD9102_ramp_command(enable=True, triangle=triangle, sram_mode=True,
|
|
sram_samples=sram_samples, sram_hold=sram_hold,
|
|
sram_amplitude=sram_amplitude)
|
|
else:
|
|
if pat_period_base is None:
|
|
pat_period_base = cmd.AD9102_PAT_PERIOD_BASE_DEFAULT
|
|
if saw_step is None and freq_hz is not None:
|
|
if dac_clk_hz is None:
|
|
dac_clk_hz = cmd.AD9102_DAC_CLK_HZ
|
|
saw_step = cmd.calc_saw_step_for_freq(freq_hz, dac_clk_hz, triangle)
|
|
if saw_step is None:
|
|
saw_step = cmd.AD9102_SAW_STEP_DEFAULT
|
|
if pat_period is None and duty is not None:
|
|
pat_period = cmd.calc_pat_period_for_duty(saw_step, duty, pat_period_base, triangle)
|
|
if pat_period is None:
|
|
pat_period = cmd.AD9102_PAT_PERIOD_DEFAULT
|
|
hexstring = cmd.create_AD9102_ramp_command(saw_step, pat_period, pat_period_base,
|
|
enable=True, triangle=triangle)
|
|
cmd.send_AD9102(prt, hexstring)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
else:
|
|
print("")
|
|
|
|
|
|
def start_ad9833_ramp(prt, freq_hz=None, mclk_hz=None, triangle=True, enable=True):
|
|
if freq_hz is None:
|
|
freq_hz = 0.0
|
|
hexstring = cmd.create_AD9833_ramp_command(freq_hz=freq_hz, mclk_hz=mclk_hz,
|
|
enable=enable, triangle=triangle)
|
|
cmd.send_AD9833(prt, hexstring)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
else:
|
|
print("")
|
|
|
|
|
|
def send_ds1809_pulse(prt, uc=False, dc=False, count=1, pulse_ms=None):
|
|
hexstring = cmd.create_DS1809_pulse_command(uc=uc, dc=dc, count=count, pulse_ms=pulse_ms)
|
|
cmd.send_DS1809(prt, hexstring)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
status = cmd.get_STATE(prt).hex()
|
|
if status is not None:
|
|
print("Received: STATE. State status:", cmd.decode_STATE(status), "("+cmd.flipfour(status)+")")
|
|
print("")
|
|
else:
|
|
print("")
|
|
|
|
|
|
def request_data(prt):
|
|
# Request data
|
|
cmd.send_TRANS_ENABLE(prt)
|
|
time.sleep(WAIT_AFTER_SEND)
|
|
data = cmd.get_DATA(prt).hex()
|
|
data_dict = []
|
|
if data is not None:
|
|
data_dict = cmd.decode_DATA(data)
|
|
return data_dict
|
|
|
|
|
|
def print_data(data):
|
|
def shorten(i):
|
|
return str(round(i, 2))
|
|
|
|
print("Data from device (time: "+datetime.now().strftime("%H:%M:%S:%f")+"):")
|
|
print("Message Header:", data['Header'], " Message ID:", data['Message_ID'])
|
|
print("Photodiode Current 1 ("+str(len(data['I1']))+" values):", \
|
|
shorten(data['I1']), shorten(data['I1'][1]), "...", \
|
|
shorten(data['I1']), shorten(data['I1'][-1]), "mA")
|
|
print("Photodiode Current 2 ("+str(len(data['I2']))+" values):", \
|
|
shorten(data['I2']), shorten(data['I2'][1]), "...", \
|
|
shorten(data['I2']), shorten(data['I2'][-1]), "mA")
|
|
print("Laser Temperature 1:", shorten(data['Temp_1']), "C")
|
|
print("Laser Temperature 2:", shorten(data['Temp_2']), "C")
|
|
print("Temperature of external thermistor 1:", shorten(data['Temp_Ext_1']), "C")
|
|
print("Temperature of external thermistor 2:", shorten(data['Temp_Ext_2']), "C")
|
|
print("Voltages 3V3: "+shorten(data['MON_3V3'])+"V 5V1: "+shorten(data['MON_5V1'])+ \
|
|
"V 5V2: "+shorten(data['MON_5V2'])+"V 7V0: "+shorten(data['MON_7V0'])+"V.")
|
|
|
|
def close_connection(prt):
|
|
cmd.close_port(prt)
|