|
ME 405 Term Project
|
Functions | |
| eff_to_key (eff) | |
| key_to_eff (key) | |
| create_run (control_mode, control_val, driving_mode, run_num, size) | |
| clean_data (df, cols=None, mode='all') | |
| start_stream_handshake (ser, timeout=2.0, retries=3) | |
Variables | |
| str | key = '' |
| bool | running = False |
| bool | streaming = False |
| bool | print_stream_frames = True |
| bool | stream_expected = True |
| bool | first_frame = True |
| dict | runs = {} |
| int | run_count = 0 |
| str | frame_buffer = "" |
| int | control_mode = 0 |
| int | driving_mode = 0 |
| int | effort = 0 |
| int | setpoint = 0 |
| float | kp = 0.0 |
| float | ki = 0.0 |
| float | k_line = 0.0 |
| bool | first = True |
| bool | done = False |
| queue = local_queue.Queue() | |
| bool | queuing = False |
| bool | creating_run = False |
| str | test_origin = "manual" |
| dict | control_mode_dict = {0: "effort", 1: "velocity", 2: "line following"} |
| dict | driving_mode_dict = {0: "straight", 1: "pivot", 2: "arc"} |
| str | user_prompt |
| int | GEAR_RATIO = 3952/33 |
| int | CPR_MOTOR = 12 |
| int | CPR_WHEEL = GEAR_RATIO*CPR_MOTOR |
| int | RAD_PER_COUNT = 2 * math.pi / CPR_WHEEL |
| int | WHEEL_RADIUS_MM = 35 |
| exist_ok | |
| ser = Serial('COM3', baudrate=115200, timeout=1) | |
| selected = input("Enter choice (e, v, l, or q to quit): ") | |
| key_in = input("Enter effort key (0-9 or 'a' for 100%) or percent (0-100): ").strip() | |
| eff_val = key_to_eff(key_in) | |
| str | line = 'e' + eff_to_key(effort) |
| gains = input("Enter setpoint (mm/s), Kp, and Ki separated by a comma (e.g., 40,1.5,0.1): ") | |
| sp_str | |
| kp_str | |
| ki_str | |
| setpoint_mm_s = float(sp_str) | |
| int | setpoint_rad_s = setpoint_mm_s / WHEEL_RADIUS_MM |
| setpoint_scaled = int(setpoint_rad_s * 100) | |
| kp_scaled = int(kp * 100) | |
| ki_scaled = int(ki * 100) | |
| k_line_str | |
| target_str | |
| target_scaled = int(setpoint_rad_s * 100) | |
| k_line_scaled = int(k_line * 100) | |
| eff = input("Enter effort test in the following format: start, end, step (e.g., 0,100,10): ") | |
| start_str | |
| end_str | |
| step_str | |
| start = int(start_str) | |
| end = int(end_str) | |
| step = int(step_str) | |
| csv_dir = os.path.join('runs', 'csvs') | |
| plots_dir = os.path.join('runs', 'plots') | |
| df1 = meta.get('motor_data') | |
| df2 = meta.get('obsv_data', None) | |
| df = pd.concat([df1, df2], axis=1) | |
| str | run_code = 'E' |
| str | driving_code = "STR" |
| str | base_name = f"{run_name}_{run_code}_{driving_code}" |
| csv_name = os.path.join(csv_dir, base_name + ".csv") | |
| list | cols = ["_time", "_left_pos", "_right_pos", "_left_vel", "_right_vel", "_obsv_time", "_obsv_left_vel", "_obsv_right_vel", "_obsv_s", "_obsv_yaw"] |
| index | |
| next_test = queue.dequeue() | |
| mode = next_test[0] | |
| str | mode_name = "effort" |
| params = None | |
| int | sample_size = 200 |
| str | run_name = f'run{run_count}' |
| chunk = ser.read(ser.in_waiting or 1).decode() | |
| str | frame = frame_buffer[start+3 : end] |
| idx_str | |
| time_s | |
| left_pos | |
| right_pos | |
| left_vel | |
| right_vel | |
| idx = int(idx_str) | |
@file main.py @brief Script to run on PC to interface with Romi robot over Bluetooth serial. Sends commands to Romi to run tests in different control modes (effort, velocity, line following), toggles data streaming, and saves received data to CSV files with optional plotting.
| test.clean_data | ( | df, | |
| cols = None, | |||
| mode = 'all' ) |
Clean DataFrame by removing all zero rows except initial zeros up to the first non-zero data.
Parameters:
df: pandas.DataFrame
cols: list of columns to check. If None, defaults to all motor_data columns.
mode: 'all' -> remove rows where ALL specified cols are zero;
'any' -> remove rows where ANY specified cols are zero.
Returns: (cleaned_df, removed_count)
| test.create_run | ( | control_mode, | |
| control_val, | |||
| driving_mode, | |||
| run_num, | |||
| size ) |
| test.eff_to_key | ( | eff | ) |
| test.key_to_eff | ( | key | ) |
| test.start_stream_handshake | ( | ser, | |
| timeout = 2.0, | |||
| retries = 3 ) |
Send START and wait for ACK from MCU. Returns True if ACK received.
| str test.base_name = f"{run_name}_{run_code}_{driving_code}" |
| test.chunk = ser.read(ser.in_waiting or 1).decode() |
| list test.cols = ["_time", "_left_pos", "_right_pos", "_left_vel", "_right_vel", "_obsv_time", "_obsv_left_vel", "_obsv_right_vel", "_obsv_s", "_obsv_yaw"] |
| int test.control_mode = 0 |
| dict test.control_mode_dict = {0: "effort", 1: "velocity", 2: "line following"} |
| int test.CPR_MOTOR = 12 |
| int test.CPR_WHEEL = GEAR_RATIO*CPR_MOTOR |
| bool test.creating_run = False |
| test.csv_dir = os.path.join('runs', 'csvs') |
| test.df1 = meta.get('motor_data') |
| test.df2 = meta.get('obsv_data', None) |
| bool test.done = False |
| str test.driving_code = "STR" |
| tuple test.driving_mode = 0 |
| dict test.driving_mode_dict = {0: "straight", 1: "pivot", 2: "arc"} |
| test.eff_val = key_to_eff(key_in) |
| test.effort = 0 |
| str test.end = int(end_str) |
| test.end_str |
| test.exist_ok |
| bool test.first = True |
| bool test.first_frame = True |
| str test.frame = frame_buffer[start+3 : end] |
| str test.frame_buffer = "" |
| test.gains = input("Enter setpoint (mm/s), Kp, and Ki separated by a comma (e.g., 40,1.5,0.1): ") |
| int test.GEAR_RATIO = 3952/33 |
| test.idx = int(idx_str) |
| test.idx_str |
| test.index |
| test.k_line = 0.0 |
| test.k_line_scaled = int(k_line * 100) |
| test.k_line_str |
| test.key = '' |
| test.ki = 0.0 |
| test.ki_scaled = int(ki * 100) |
| test.ki_str |
| test.kp = 0.0 |
| test.kp_scaled = int(kp * 100) |
| test.kp_str |
| test.left_pos |
| test.left_vel |
| str test.line = 'e' + eff_to_key(effort) |
| test.mode = next_test[0] |
| str test.mode_name = "effort" |
| test.next_test = queue.dequeue() |
| dict test.params = None |
| test.plots_dir = os.path.join('runs', 'plots') |
| bool test.print_stream_frames = True |
| test.queue = local_queue.Queue() |
| bool test.queuing = False |
| int test.RAD_PER_COUNT = 2 * math.pi / CPR_WHEEL |
| test.right_pos |
| test.right_vel |
| str test.run_code = 'E' |
| int test.run_count = 0 |
| str test.run_name = f'run{run_count}' |
| bool test.running = False |
| dict test.runs = {} |
| int test.sample_size = 200 |
| test.selected = input("Enter choice (e, v, l, or q to quit): ") |
| test.ser = Serial('COM3', baudrate=115200, timeout=1) |
| test.setpoint = 0 |
| test.setpoint_mm_s = float(sp_str) |
| int test.setpoint_rad_s = setpoint_mm_s / WHEEL_RADIUS_MM |
| test.setpoint_scaled = int(setpoint_rad_s * 100) |
| test.sp_str |
| str test.start = int(start_str) |
| test.start_str |
| test.step = int(step_str) |
| test.step_str |
| bool test.stream_expected = True |
| bool test.streaming = False |
| test.target_scaled = int(setpoint_rad_s * 100) |
| test.target_str |
| str test.test_origin = "manual" |
| test.time_s |
| str test.user_prompt |
| int test.WHEEL_RADIUS_MM = 35 |