|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import threading |
| 4 | +import time |
| 5 | + |
| 6 | +import numpy as np |
| 7 | + |
| 8 | +from openlifu.bf.pulse import Pulse |
| 9 | +from openlifu.bf.sequence import Sequence |
| 10 | +from openlifu.geo import Point |
| 11 | +from openlifu.io.LIFUInterface import LIFUInterface |
| 12 | +from openlifu.plan.solution import Solution |
| 13 | + |
| 14 | +# set PYTHONPATH=%cd%\src;%PYTHONPATH% |
| 15 | +# python notebooks/test_watertank.py |
| 16 | + |
| 17 | +""" |
| 18 | +Test script to automate: |
| 19 | +1. Connect to the device. |
| 20 | +2. Test HVController: Turn HV on/off and check voltage. |
| 21 | +3. Test Device functionality. |
| 22 | +""" |
| 23 | + |
| 24 | +log_interval = 1 # seconds; you can adjust this variable as needed |
| 25 | +stop_logging = False # flag to signal the logging thread to stop |
| 26 | + |
| 27 | +def log_temperature(): |
| 28 | + # Create a file with the current timestamp in the name |
| 29 | + timestamp = time.strftime("%Y%m%d_%H%M%S") |
| 30 | + filename = f"{timestamp}_temp.log" |
| 31 | + with open(filename, "w") as logfile: |
| 32 | + while not stop_logging: |
| 33 | + temperature = interface.txdevice.get_temperature() |
| 34 | + a_temp = interface.txdevice.get_ambient_temperature() |
| 35 | + current_time = time.strftime("%Y-%m-%d %H:%M:%S") |
| 36 | + log_line = f"{current_time}: Temperature: {temperature}, Ambient Temperature: {a_temp}\n" |
| 37 | + logfile.write(log_line) |
| 38 | + logfile.flush() # Ensure the data is written immediately |
| 39 | + time.sleep(log_interval) |
| 40 | + |
| 41 | + |
| 42 | +print("Starting LIFU Test Script...") |
| 43 | +interface = LIFUInterface(test_mode=False) |
| 44 | +tx_connected, hv_connected = interface.is_device_connected() |
| 45 | +if tx_connected and hv_connected: |
| 46 | + print("LIFU Device Fully connected.") |
| 47 | +else: |
| 48 | + print(f'LIFU Device NOT Fully Connected. TX: {tx_connected}, HV: {hv_connected}') |
| 49 | + |
| 50 | +# Ask the user if they want to log temperature |
| 51 | +log_choice = input("Do you want to log temperature before starting trigger? (y/n): ").strip().lower() |
| 52 | +log_temp = (log_choice == "y") |
| 53 | + |
| 54 | +print("Ping the device") |
| 55 | +interface.txdevice.ping() |
| 56 | + |
| 57 | +print("Set Trigger") |
| 58 | +json_trigger_data = { |
| 59 | + "TriggerFrequencyHz": 10, |
| 60 | + "TriggerPulseCount": 0, |
| 61 | + "TriggerPulseWidthUsec": 20000, |
| 62 | + "TriggerPulseTrainInterval": 0, |
| 63 | + "TriggerPulseTrainCount": 0, |
| 64 | + "TriggerMode": 1, |
| 65 | + "ProfileIndex": 0, |
| 66 | + "ProfileIncrement": 0 |
| 67 | +} |
| 68 | + |
| 69 | +trigger_setting = interface.txdevice.set_trigger_json(data=json_trigger_data) |
| 70 | +if trigger_setting: |
| 71 | + print(f"Trigger Setting: {trigger_setting}") |
| 72 | +else: |
| 73 | + print("Failed to set trigger setting.") |
| 74 | + |
| 75 | +print("Enumerate TX7332 chips") |
| 76 | +num_tx_devices = interface.txdevice.enum_tx7332_devices() |
| 77 | +if num_tx_devices > 0: |
| 78 | + print(f"Number of TX7332 devices found: {num_tx_devices}") |
| 79 | +else: |
| 80 | + raise Exception("No TX7332 devices found.") |
| 81 | + |
| 82 | +# set focus |
| 83 | +xInput = 0 |
| 84 | +yInput = 0 |
| 85 | +zInput = 50 |
| 86 | + |
| 87 | +frequency = 400e3 |
| 88 | +voltage = 12.0 |
| 89 | +duration = 2e-5 |
| 90 | + |
| 91 | +pulse = Pulse(frequency=frequency, amplitude=voltage, duration=duration) |
| 92 | +pt = Point(position=(xInput,yInput,zInput), units="mm") |
| 93 | +sequence = Sequence( |
| 94 | + pulse_interval=0.1, |
| 95 | + pulse_count=10, |
| 96 | + pulse_train_interval=1, |
| 97 | + pulse_train_count=1 |
| 98 | +) |
| 99 | + |
| 100 | +solution = Solution( |
| 101 | + id="solution", |
| 102 | + name="Solution", |
| 103 | + protocol_id="example_protocol", |
| 104 | + transducer_id="example_transducer", |
| 105 | + delays = np.zeros((1,64)), |
| 106 | + apodizations = np.ones((1,64)), |
| 107 | + pulse = pulse, |
| 108 | + sequence = sequence, |
| 109 | + target=pt, |
| 110 | + foci=[pt], |
| 111 | + approved=True |
| 112 | +) |
| 113 | + |
| 114 | +sol_dict = solution.to_dict() |
| 115 | +profile_index = 1 |
| 116 | +profile_increment = True |
| 117 | +interface.txdevice.set_solution( |
| 118 | + pulse = sol_dict['pulse'], |
| 119 | + delays = sol_dict['delays'], |
| 120 | + apodizations= sol_dict['apodizations'], |
| 121 | + sequence= sol_dict['sequence'], |
| 122 | + profile_index=profile_index, |
| 123 | + profile_increment=profile_increment |
| 124 | +) |
| 125 | + |
| 126 | +print("Get Trigger") |
| 127 | +trigger_setting = interface.txdevice.get_trigger_json() |
| 128 | +if trigger_setting: |
| 129 | + print(f"Trigger Setting: {trigger_setting}") |
| 130 | +else: |
| 131 | + print("Failed to get trigger setting.") |
| 132 | + |
| 133 | +# If logging is enabled, start the logging thread |
| 134 | +if log_temp: |
| 135 | + t = threading.Thread(target=log_temperature) |
| 136 | + t.start() |
| 137 | +else: |
| 138 | + print("Get Temperature") |
| 139 | + temperature = interface.txdevice.get_temperature() |
| 140 | + print(f"Temperature: {temperature} °C") |
| 141 | + |
| 142 | + print("Get Ambient") |
| 143 | + a_temp = interface.txdevice.get_ambient_temperature() |
| 144 | + print(f"Ambient Temperature: {a_temp} °C") |
| 145 | + |
| 146 | +print("Press enter to START trigger:") |
| 147 | +input() # Wait for the user to press Enter |
| 148 | +print("Starting Trigger...") |
| 149 | +if interface.txdevice.start_trigger(): |
| 150 | + print("Trigger Running Press enter to STOP:") |
| 151 | + input() # Wait for the user to press Enter |
| 152 | + if interface.txdevice.stop_trigger(): |
| 153 | + print("Trigger stopped successfully.") |
| 154 | + else: |
| 155 | + print("Failed to stop trigger.") |
| 156 | +else: |
| 157 | + print("Failed to get trigger setting.") |
| 158 | + |
| 159 | +# Stop the temperature logging before starting the trigger |
| 160 | +if log_temp: |
| 161 | + stop_logging = True |
| 162 | + t.join() |
0 commit comments