Skip to content

Commit af8d79f

Browse files
committed
fix: correct all_pulses property to return ALL pulses from finest layer
The all_pulses property was incorrectly returning only pulses from the first structure of the last layer (e.g., 2 pulses) instead of ALL pulses from all structures in that layer (e.g., 256 pulses). This matches the TypeScript implementation which correctly does: lastLayer.map(ps => ps.pulses).flat() This was the root cause of Issue #34 - not actually sparse pulse data, but incorrect pulse retrieval. The proportional timing fallback is now only needed for truly sparse cases (which shouldn't normally occur). Also adds SPARSE_PULSE_THRESHOLD constant and documentation for clarity.
1 parent caf12a7 commit af8d79f

2 files changed

Lines changed: 123 additions & 1 deletion

File tree

idtap/classes/meter.py

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,12 @@
1010
MIN_TEMPO_BPM = 20 # Very slow musical pieces (e.g., some alap sections)
1111
MAX_TEMPO_BPM = 300 # Very fast musical pieces
1212

13+
# Sparse pulse detection threshold
14+
SPARSE_PULSE_THRESHOLD = 0.5 # Fallback to proportional timing if <50% of expected pulses present
15+
# When real transcription data has manually annotated pulses, often only key beats
16+
# are marked rather than all subdivisions. If we have less than this threshold,
17+
# we can't reliably interpolate between pulses and must use mathematical proportional timing.
18+
1319

1420
def find_closest_idxs(trials: List[float], items: List[float]) -> List[int]:
1521
"""Return indexes of items closest to each trial (greedy)."""
@@ -273,6 +279,8 @@ def __init__(self, hierarchy: Optional[List[int | List[int]]] = None,
273279
self.repetitions = repetitions
274280
self.pulse_structures: List[List[PulseStructure]] = []
275281
self._generate_pulse_structures()
282+
# Cache for sparse pulse detection (performance optimization)
283+
self._is_sparse_cached: Optional[bool] = None
276284

277285
def _validate_parameters(self, opts: Dict) -> None:
278286
"""Validate constructor parameters and provide helpful error messages."""
@@ -364,7 +372,15 @@ def _generate_pulse_structures(self) -> None:
364372

365373
@property
366374
def all_pulses(self) -> List[Pulse]:
367-
return self.pulse_structures[-1][0].pulses
375+
"""Get all pulses from the finest layer (lowest level) of the hierarchy.
376+
377+
This concatenates pulses from all pulse structures in the last layer,
378+
matching the TypeScript implementation: lastLayer.map(ps => ps.pulses).flat()
379+
"""
380+
if not self.pulse_structures or not self.pulse_structures[-1]:
381+
return []
382+
# Flatten all pulses from all structures in the finest layer
383+
return [pulse for ps in self.pulse_structures[-1] for pulse in ps.pulses]
368384

369385
@property
370386
def real_times(self) -> List[float]:

test_transcription_pulses.py

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
#!/usr/bin/env python
2+
"""Test script to examine pulse data in real transcription."""
3+
4+
from idtap import SwaraClient
5+
from idtap.classes.meter import Meter
6+
import json
7+
8+
# Initialize client
9+
client = SwaraClient()
10+
11+
# Fetch the transcription
12+
transcription_id = "68a3a79fffd9b2d478ee11e8"
13+
print(f"Fetching transcription: {transcription_id}")
14+
15+
try:
16+
piece_data = client.get_piece(transcription_id)
17+
18+
# Convert from JSON to Piece object
19+
from idtap.classes.piece import Piece
20+
piece = Piece.from_json(piece_data)
21+
22+
print(f"✓ Successfully loaded: {piece.title}")
23+
print(f" Instrumentation: {piece.instrumentation}")
24+
25+
# Examine meters
26+
if piece.meters:
27+
print(f"\n Meters found: {len(piece.meters)}")
28+
29+
for i, meter in enumerate(piece.meters):
30+
print(f"\n Meter {i}:")
31+
print(f" Hierarchy: {meter.hierarchy}")
32+
print(f" Tempo: {meter.tempo} BPM")
33+
print(f" Start time: {meter.start_time}s")
34+
print(f" Repetitions: {meter.repetitions}")
35+
print(f" Cycle duration: {meter.cycle_dur}s")
36+
37+
# Debug: Check pulse_structures
38+
print(f"\n Pulse structures layers: {len(meter.pulse_structures)}")
39+
for layer_idx, layer in enumerate(meter.pulse_structures):
40+
print(f" Layer {layer_idx}: {len(layer)} structures")
41+
for struct_idx, struct in enumerate(layer):
42+
print(f" Structure {layer_idx}.{struct_idx}: {len(struct.pulses)} pulses")
43+
44+
# Calculate expected vs actual pulses
45+
expected_pulses_per_cycle = meter._pulses_per_cycle
46+
expected_total = expected_pulses_per_cycle * meter.repetitions
47+
actual_total = len(meter.all_pulses)
48+
49+
# Check what all_pulses SHOULD be
50+
print(f"\n What all_pulses returns: {len(meter.all_pulses)} pulses")
51+
print(f" Should be all pulses from layer -1: {sum(len(ps.pulses) for ps in meter.pulse_structures[-1])} pulses")
52+
53+
print(f" Pulses per cycle (expected): {expected_pulses_per_cycle}")
54+
print(f" Total pulses expected: {expected_total}")
55+
print(f" Total pulses actual: {actual_total}")
56+
print(f" Pulse density: {actual_total / expected_total * 100:.1f}%")
57+
58+
# Check if this would be considered "sparse"
59+
is_sparse = actual_total < expected_total * 0.5
60+
print(f" Would be considered sparse (<50%)?: {'YES ⚠️' if is_sparse else 'NO ✓'}")
61+
62+
# Show first few pulse times if sparse
63+
if is_sparse or actual_total < expected_total:
64+
print(f" First 10 pulse times: {[round(p.real_time, 3) for p in meter.all_pulses[:10]]}")
65+
66+
# Check pulse spacing
67+
if len(meter.all_pulses) > 1:
68+
spacings = []
69+
for j in range(1, min(10, len(meter.all_pulses))):
70+
spacing = meter.all_pulses[j].real_time - meter.all_pulses[j-1].real_time
71+
spacings.append(round(spacing, 3))
72+
print(f" Pulse spacings: {spacings}")
73+
74+
# Try to understand the pattern
75+
if actual_total > 0:
76+
print(f" Analyzing pulse pattern...")
77+
# Check if pulses align with beats only
78+
beat_duration = meter.cycle_dur / meter.hierarchy[0] if meter.hierarchy else 0
79+
if beat_duration > 0:
80+
for j, pulse in enumerate(meter.all_pulses[:10]):
81+
relative_time = pulse.real_time - meter.start_time
82+
beat_position = relative_time / beat_duration
83+
print(f" Pulse {j}: {pulse.real_time:.3f}s (beat position: {beat_position:.2f})")
84+
85+
# Test get_musical_time with a few sample points
86+
print(f"\n Testing get_musical_time():")
87+
test_times = [
88+
meter.start_time + 0.1,
89+
meter.start_time + meter.cycle_dur * 0.25,
90+
meter.start_time + meter.cycle_dur * 0.5,
91+
meter.start_time + meter.cycle_dur * 0.75
92+
]
93+
94+
for test_time in test_times:
95+
result = meter.get_musical_time(test_time)
96+
if result:
97+
print(f" Time {test_time:.3f}s → {result} (frac: {result.fractional_beat:.3f})")
98+
else:
99+
print(f" Time {test_time:.3f}s → False (out of bounds)")
100+
else:
101+
print(" No meters found in this transcription")
102+
103+
except Exception as e:
104+
print(f"✗ Error loading transcription: {e}")
105+
import traceback
106+
traceback.print_exc()

0 commit comments

Comments
 (0)