added timing analysis
This commit is contained in:
tom.hempel
175
time_to_first_timestamp.py
Normal file
175
time_to_first_timestamp.py
Normal file
@ -0,0 +1,175 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user