#!/usr/bin/env python3 """ Script to calculate timing analysis for meditation recordings including: - Time from first HR recording to first timestamp marker - Time differences between consecutive timestamps - Time from last timestamp to end of recording Only processes recordings that have exactly 4 timestamps. """ import pandas as pd import os def get_recording_timing_analysis(recording_id): """ Calculate comprehensive timing analysis for a given recording. Args: recording_id (str): Recording ID (e.g., '01', '02', etc.) Returns: dict: Complete analysis results or None if invalid """ recording_dir = f"SingleRecordings/{recording_id}" # Check if files exist hr_file = f"{recording_dir}/hr.csv" timestamps_file = f"{recording_dir}/timestamps.csv" if not os.path.exists(hr_file) or not os.path.exists(timestamps_file): return None try: # Read HR data hr_data = pd.read_csv(hr_file) if hr_data.empty: return None # Read timestamp data timestamps_data = pd.read_csv(timestamps_file) if timestamps_data.empty: return None # Check if there are exactly 4 timestamps if len(timestamps_data) != 4: return None # Get timestamps first_hr_timestamp = hr_data['timestamp'].iloc[0] last_hr_timestamp = hr_data['timestamp'].iloc[-1] timestamp_markers = timestamps_data['timestamp'].tolist() # Calculate time from first HR to first timestamp time_to_first_marker_ms = timestamp_markers[0] - first_hr_timestamp time_to_first_marker_s = time_to_first_marker_ms / 1000.0 # Calculate intervals between consecutive timestamps intervals_ms = [] for i in range(len(timestamp_markers) - 1): interval = timestamp_markers[i + 1] - timestamp_markers[i] intervals_ms.append(interval) intervals_s = [interval / 1000.0 for interval in intervals_ms] # Calculate time from last timestamp to end of recording time_from_last_marker_ms = last_hr_timestamp - timestamp_markers[-1] time_from_last_marker_s = time_from_last_marker_ms / 1000.0 return { 'recording_id': recording_id, 'time_to_first_marker_ms': time_to_first_marker_ms, 'time_to_first_marker_s': time_to_first_marker_s, 'interval_1_2_ms': intervals_ms[0], 'interval_1_2_s': intervals_s[0], 'interval_2_3_ms': intervals_ms[1], 'interval_2_3_s': intervals_s[1], 'interval_3_4_ms': intervals_ms[2], 'interval_3_4_s': intervals_s[2], 'time_from_last_marker_ms': time_from_last_marker_ms, 'time_from_last_marker_s': time_from_last_marker_s, 'first_hr_timestamp': first_hr_timestamp, 'last_hr_timestamp': last_hr_timestamp, 'timestamp_markers': timestamp_markers } except Exception as e: print(f"Error processing recording {recording_id}: {e}") return None def main(): """Main function to process all recordings and print results.""" print("Meditation Recording Timing Analysis") print("=" * 80) print("Processing recordings with exactly 4 timestamps") print("=" * 80) valid_recordings = [] # Process recordings 01-11 for i in range(1, 12): recording_id = f"{i:02d}" result = get_recording_timing_analysis(recording_id) if result: valid_recordings.append(result) print(f"\nRecording {recording_id}:") print(f" Time to first marker: {result['time_to_first_marker_s']:.3f}s ({result['time_to_first_marker_ms']:.0f}ms)") print(f" Interval 1→2: {result['interval_1_2_s']:.3f}s ({result['interval_1_2_ms']:.0f}ms)") print(f" Interval 2→3: {result['interval_2_3_s']:.3f}s ({result['interval_2_3_ms']:.0f}ms)") print(f" Interval 3→4: {result['interval_3_4_s']:.3f}s ({result['interval_3_4_ms']:.0f}ms)") print(f" Time from last marker to end: {result['time_from_last_marker_s']:.3f}s ({result['time_from_last_marker_ms']:.0f}ms)") else: print(f"\nRecording {recording_id}: SKIPPED (does not have exactly 4 timestamps)") # Create CSV output if valid_recordings: print(f"\n" + "=" * 80) print(f"Creating CSV file with timing analysis...") # Prepare data for CSV csv_data = [] for result in valid_recordings: csv_data.append({ 'recording_id': result['recording_id'], 'time_to_first_marker_s': result['time_to_first_marker_s'], 'time_to_first_marker_ms': result['time_to_first_marker_ms'], 'interval_1_2_s': result['interval_1_2_s'], 'interval_1_2_ms': result['interval_1_2_ms'], 'interval_2_3_s': result['interval_2_3_s'], 'interval_2_3_ms': result['interval_2_3_ms'], 'interval_3_4_s': result['interval_3_4_s'], 'interval_3_4_ms': result['interval_3_4_ms'], 'time_from_last_marker_s': result['time_from_last_marker_s'], 'time_from_last_marker_ms': result['time_from_last_marker_ms'] }) # Create DataFrame and save to CSV df = pd.DataFrame(csv_data) csv_filename = 'timestamp_timing_analysis.csv' df.to_csv(csv_filename, index=False) print(f"CSV file saved as: {csv_filename}") # Summary statistics print(f"\n" + "=" * 80) print("SUMMARY STATISTICS") print("=" * 80) print(f"Total recordings processed: {len(valid_recordings)}") # Time to first marker statistics times_to_first = [r['time_to_first_marker_s'] for r in valid_recordings] print(f"\nTime to first marker:") print(f" Average: {sum(times_to_first)/len(times_to_first):.3f}s") print(f" Min: {min(times_to_first):.3f}s") print(f" Max: {max(times_to_first):.3f}s") # Interval statistics interval_names = ['interval_1_2_s', 'interval_2_3_s', 'interval_3_4_s'] interval_labels = ['Interval 1→2', 'Interval 2→3', 'Interval 3→4'] for interval_key, label in zip(interval_names, interval_labels): interval_times = [r[interval_key] for r in valid_recordings] print(f"\n{label}:") print(f" Average: {sum(interval_times)/len(interval_times):.3f}s") print(f" Min: {min(interval_times):.3f}s") print(f" Max: {max(interval_times):.3f}s") # Time from last marker statistics times_from_last = [r['time_from_last_marker_s'] for r in valid_recordings] print(f"\nTime from last marker to end:") print(f" Average: {sum(times_from_last)/len(times_from_last):.3f}s") print(f" Min: {min(times_from_last):.3f}s") print(f" Max: {max(times_from_last):.3f}s") else: print("\nNo valid recordings found with exactly 4 timestamps.") if __name__ == "__main__": main()