wschrab / SpringClean.py

import pandas as pd # --------------------------------------------------------- # STEP 1 - Get user inputs # --------------------------------------------------------- spring_name = input( "Enter spring number (example: 1040_02): " ).strip() threshold = float( input( "Enter peak detection threshold (mm): " ) ) # --------------------------------------------------------- # STEP 2 - Open cleaned file # --------------------------------------------------------- input_file = f"Spring_{spring_name}_Cleaned.csv" print(f"\nLoading {input_file}...") df = pd.read_csv(input_file) # --------------------------------------------------------- # STEP 3 - Define important column names # --------------------------------------------------------- extension_column = "Extension from Preload (mm)" # --------------------------------------------------------- # STEP 4 - Create Above Threshold flag # # 1 = Extension is above threshold # 0 = Extension is below threshold # --------------------------------------------------------- print("Creating Above Threshold...") df["Above Threshold"] = ( df[extension_column] >= threshold ).astype(int) # --------------------------------------------------------- # STEP 5 - Create Previous Above Threshold column # # Shift the Above Threshold column down by one row # --------------------------------------------------------- print("Creating Previous Above Threshold...") df["Previous Above Threshold"] = ( df["Above Threshold"] .shift(1) .fillna(0) .astype(int) ) # --------------------------------------------------------- # STEP 6 - Create Rising Edge column # # Rising Edge occurs when: # # Previous = 0 # Current = 1 # --------------------------------------------------------- print("Creating Rising Edge...") df["Rising Edge"] = ( (df["Above Threshold"] == 1) & (df["Previous Above Threshold"] == 0) ).astype(int) # --------------------------------------------------------- # STEP 7 - Create Falling Edge column # # Falling Edge occurs when: # # Previous = 1 # Current = 0 # --------------------------------------------------------- print("Creating Falling Edge...") df["Falling Edge"] = ( (df["Above Threshold"] == 0) & (df["Previous Above Threshold"] == 1) ).astype(int) # --------------------------------------------------------- # STEP 8 - Create Cycle Number # # Every Rising Edge starts a new cycle. # # Example: # # Rising Edge: # 0 0 1 0 0 1 0 # # Cycle Number: # 0 0 1 1 1 2 2 # --------------------------------------------------------- print("Creating Cycle Number...") df["Cycle Number"] = ( df["Rising Edge"] .cumsum() ) # --------------------------------------------------------- # STEP 9 - Save analyzed file # --------------------------------------------------------- print("Saving Analyzed File...") output_file = f"Spring_{spring_name}_Analyzed.csv" df.to_csv(output_file, index=False) # --------------------------------------------------------- # STEP 11 - Create Peak Summary # --------------------------------------------------------- peak_summary = [] print("Creating Cycle Numbers...") # Get all valid cycle numbers cycle_numbers = sorted( df[df["Cycle Number"] > 0]["Cycle Number"].unique() ) for cycle in cycle_numbers: # Get only rows for this cycle that are above threshold cycle_data = df[ (df["Cycle Number"] == cycle) & (df["Above Threshold"] == 1) ] # Skip empty cycles just in case if len(cycle_data) == 0: continue # Find row with maximum load peak_index = cycle_data["Load (N)"].idxmax() peak_row = df.loc[peak_index] peak_summary.append({ "Cycle Number": cycle, "Test": peak_row["Test"], "Peak Load (N)": peak_row["Load (N)"], "Peak Extension (mm)": peak_row["Extension from Preload (mm)"], "Peak Time (s)": peak_row["Continuous Time (s)"], "Peak Row": peak_index }) # --------------------------------------------------------- # STEP 12 - Create dataframe # --------------------------------------------------------- peak_df = pd.DataFrame(peak_summary) # --------------------------------------------------------- # STEP 13 - Save peak summary # --------------------------------------------------------- print("Saving peak summary...") peak_output_file = ( f"Spring_{spring_name}_Peaks.csv" ) peak_df.to_csv( peak_output_file, index=False ) # --------------------------------------------------------- # STEP 14 - Report results # --------------------------------------------------------- total_cycles = int( df["Cycle Number"].max() ) print("\nAnalysis Complete") print( f"Total Cycles Found: " f"{total_cycles}" ) print( f"Analyzed File: " f"{output_file}" ) print( f"Peak Summary File: " f"{peak_output_file}" )

import pandas as pd import glob import os # --------------------------------------------------------- # STEP 1 - Ask user which spring to process # --------------------------------------------------------- spring_name = input("Enter spring number (example: 1040_01): ").strip() # --------------------------------------------------------- # STEP 2 - Find all matching test files # --------------------------------------------------------- file_pattern = f"Spring_{spring_name}_Test_*.csv" file_list = glob.glob(file_pattern) # Make sure files are processed in order file_list.sort() if len(file_list) == 0: print(f"No files found matching: {file_pattern}") exit() # --------------------------------------------------------- # STEP 3 - Variables used while combining files # --------------------------------------------------------- combined_data = [] time_offset = 0 # --------------------------------------------------------- # STEP 4 - Process each test file # --------------------------------------------------------- for test_number, file in enumerate(file_list, start=1): print(f"Processing: {file}") # Read CSV df = pd.read_csv(file) # Remove rows containing empty data df = df.dropna() # Time column name time_column = "Time (s)" # Create Test column df["Test"] = test_number # Create continuous time column df["Continuous Time (s)"] = df[time_column] + time_offset # Determine ending time for next test last_time = df[time_column].iloc[-1] time_offset += last_time # Store cleaned data combined_data.append(df) # --------------------------------------------------------- # STEP 5 - Combine all tests into one dataframe # --------------------------------------------------------- final_df = pd.concat(combined_data, ignore_index=True) # --------------------------------------------------------- # STEP 6 - Create output filename # --------------------------------------------------------- output_file = f"Spring_{spring_name}_Cleaned.csv" # --------------------------------------------------------- # STEP 7 - Save cleaned data # --------------------------------------------------------- final_df.to_csv(output_file, index=False) print() print(f"Finished!") print(f"Output saved as: {output_file}")

import pandas as pd import glob import os # --------------------------------------------------------- # STEP 1 - Ask user which spring to process # --------------------------------------------------------- spring_name = input("Enter spring number (example: 1040_01): ").strip() threshold = float( input( "Enter peak detection threshold (mm): " ) ) # --------------------------------------------------------- # STEP 2 - Find all matching test files # --------------------------------------------------------- file_pattern = f"Spring_{spring_name}_Test_*.csv" file_list = glob.glob(file_pattern) # Make sure files are processed in order file_list.sort() if len(file_list) == 0: print(f"No files found matching: {file_pattern}") exit() # --------------------------------------------------------- # STEP 3 - Variables used while combining files # --------------------------------------------------------- combined_data = [] time_offset = 0 # --------------------------------------------------------- # STEP 4 - Process each test file # --------------------------------------------------------- for test_number, file in enumerate(file_list, start=1): print(f"Processing: {file}") # Read CSV df = pd.read_csv(file) # Remove rows containing empty data df = df.dropna() # Time column name time_column = "Time (s)" # Create Test column df["Test"] = test_number # Create continuous time column df["Continuous Time (s)"] = df[time_column] + time_offset # Determine ending time for next test last_time = df[time_column].iloc[-1] time_offset += last_time # Store cleaned data combined_data.append(df) # --------------------------------------------------------- # STEP 5 - Combine all tests into one dataframe # --------------------------------------------------------- final_df = pd.concat(combined_data, ignore_index=True) # --------------------------------------------------------- # STEP 6 - Create output filename # --------------------------------------------------------- output_file = f"Spring_{spring_name}_Cleaned.csv" # --------------------------------------------------------- # STEP 7 - Save cleaned data # --------------------------------------------------------- final_df.to_csv(output_file, index=False) print() print(f"Finished Cleaning!") print(f"Output saved as: {output_file}") # --------------------------------------------------------------------- # ANALYZING SECTION # --------------------------------------------------------------------- # --------------------------------------------------------- # STEP 2 - Open cleaned file # --------------------------------------------------------- input_file = f"Spring_{spring_name}_Cleaned.csv" print(f"\nLoading {input_file}...") df = pd.read_csv(input_file) # --------------------------------------------------------- # STEP 3 - Define important column names # --------------------------------------------------------- extension_column = "Extension (mm)" # --------------------------------------------------------- # STEP 4 - Create Above Threshold flag # # 1 = Extension is above threshold # 0 = Extension is below threshold # --------------------------------------------------------- print("Creating Above Threshold...") df["Above Threshold"] = ( df[extension_column] >= threshold ).astype(int) # --------------------------------------------------------- # STEP 5 - Create Previous Above Threshold column # # Shift the Above Threshold column down by one row # --------------------------------------------------------- print("Creating Previous Above Threshold...") df["Previous Above Threshold"] = ( df["Above Threshold"] .shift(1) .fillna(0) .astype(int) ) # --------------------------------------------------------- # STEP 6 - Create Rising Edge column # # Rising Edge occurs when: # # Previous = 0 # Current = 1 # --------------------------------------------------------- print("Creating Rising Edge...") df["Rising Edge"] = ( (df["Above Threshold"] == 1) & (df["Previous Above Threshold"] == 0) ).astype(int) # --------------------------------------------------------- # STEP 7 - Create Falling Edge column # # Falling Edge occurs when: # # Previous = 1 # Current = 0 # --------------------------------------------------------- print("Creating Falling Edge...") df["Falling Edge"] = ( (df["Above Threshold"] == 0) & (df["Previous Above Threshold"] == 1) ).astype(int) # --------------------------------------------------------- # STEP 8 - Create Cycle Number # # Every Rising Edge starts a new cycle. # # Example: # # Rising Edge: # 0 0 1 0 0 1 0 # # Cycle Number: # 0 0 1 1 1 2 2 # --------------------------------------------------------- print("Creating Cycle Number...") df["Cycle Number"] = ( df["Rising Edge"] .cumsum() ) # --------------------------------------------------------- # STEP 9 - Save analyzed file # --------------------------------------------------------- print("Saving Analyzed File...") output_file = f"Spring_{spring_name}_Analyzed.csv" df.to_csv(output_file, index=False) # --------------------------------------------------------- # STEP 11 - Create Peak Summary # --------------------------------------------------------- peak_summary = [] print("Creating Cycle Numbers...") # Get all valid cycle numbers cycle_numbers = sorted( df[df["Cycle Number"] > 0]["Cycle Number"].unique() ) for cycle in cycle_numbers: # Get only rows for this cycle that are above threshold cycle_data = df[ (df["Cycle Number"] == cycle) & (df["Above Threshold"] == 1) ] # Skip empty cycles just in case if len(cycle_data) == 0: continue # Find row with maximum load peak_index = cycle_data["Load (N)"].idxmax() peak_row = df.loc[peak_index] peak_summary.append({ "Cycle Number": cycle, "Test": peak_row["Test"], "Peak Load (N)": peak_row["Load (N)"], "Peak Extension (mm)": peak_row[extension_column], "Peak Time (s)": peak_row["Continuous Time (s)"], "Peak Row": peak_index }) # --------------------------------------------------------- # STEP 12 - Create dataframe # --------------------------------------------------------- peak_df = pd.DataFrame(peak_summary) # --------------------------------------------------------- # STEP 13 - Save peak summary # --------------------------------------------------------- print("Saving peak summary...") peak_output_file = ( f"Spring_{spring_name}_Peaks.csv" ) peak_df.to_csv( peak_output_file, index=False ) # Cycles per test count cycles_per_test = ( peak_df.groupby("Test").size() ) print("Cycles Per Test:") for test, count in cycles_per_test.items(): print(f"Test {int(test)}: {count}") # --------------------------------------------------------- # STEP 14 - Report results # --------------------------------------------------------- total_cycles = int( df["Cycle Number"].max() ) print("\nAnalysis Complete") print( f"Total Cycles Found: " f"{total_cycles}" ) print( f"Total Peaks Found: " f"{len(peak_summary)}" ) print( f"Analyzed File: " f"{output_file}" ) print( f"Peak Summary File: " f"{peak_output_file}" ) input("\nPress Enter to exit")