Skip to content

Fix fractional_beat clustering with sparse pulse data (Issue #28) #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jon-myers merged 1 commit into from
Sep 9, 2025
Merged

Fix fractional_beat clustering with sparse pulse data (Issue #28) #30

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 66 additions & 28 deletions idtap/classes/meter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,6 +465,12 @@ def _hierarchical_position_to_pulse_index(self, positions: List[int], cycle_numb

def _calculate_level_start_time(self, positions: List[int], cycle_number: int, reference_level: int) -> float:
"""Calculate start time of hierarchical unit at reference level."""
# Check if we have sufficient pulse data for accurate calculation
expected_pulses = self._pulses_per_cycle * self.repetitions
if len(self.all_pulses) < expected_pulses * 0.5: # Less than 50% of expected pulses
# Fall back to proportional timing calculation for sparse pulse data
return self._calculate_proportional_level_start_time(positions, cycle_number, reference_level)

# Create positions for start of reference-level unit
# Ensure we have positions up to reference_level
start_positions = list(positions[:reference_level + 1])
Expand All @@ -476,20 +482,48 @@ def _calculate_level_start_time(self, positions: List[int], cycle_number: int, r

# Add bounds checking to prevent IndexError
if start_pulse_index < 0 or start_pulse_index >= len(self.all_pulses):
# This can happen when calculating duration of the last unit in a level
# In such cases, we should use the meter's end time
if start_pulse_index >= len(self.all_pulses):
# Beyond the last pulse - use meter end time
total_duration = self.repetitions * self.cycle_dur
return self.start_time + total_duration
else:
# Negative index (shouldn't happen but defensive)
return self.start_time
# Fall back to proportional calculation
return self._calculate_proportional_level_start_time(positions, cycle_number, reference_level)

return self.all_pulses[start_pulse_index].real_time

def _calculate_proportional_level_start_time(self, positions: List[int], cycle_number: int, reference_level: int) -> float:
"""Calculate level start time using proportional cycle division for sparse pulse data."""
cycle_start_time = self.start_time + cycle_number * self.cycle_dur

# Calculate proportional position within the cycle for this reference level
level_start_positions = list(positions[:reference_level + 1])
while len(level_start_positions) < reference_level + 1:
level_start_positions.append(0)

# Calculate cumulative position as fraction of cycle
cumulative_position = 0.0
current_subdivisions = 1

for level in range(reference_level + 1):
level_size = self.hierarchy[level]
if isinstance(level_size, list):
level_size = sum(level_size)

if level < len(level_start_positions):
position_at_level = level_start_positions[level]
else:
position_at_level = 0

# Add this level's contribution to the cumulative position
cumulative_position += position_at_level / current_subdivisions
current_subdivisions *= level_size

return cycle_start_time + cumulative_position * self.cycle_dur

def _calculate_level_duration(self, positions: List[int], cycle_number: int, reference_level: int) -> float:
"""Calculate actual duration of hierarchical unit based on pulse timing."""
# Check if we have sufficient pulse data for accurate calculation
expected_pulses = self._pulses_per_cycle * self.repetitions
if len(self.all_pulses) < expected_pulses * 0.5: # Less than 50% of expected pulses
# Fall back to proportional duration calculation
return self._calculate_proportional_level_duration(positions, cycle_number, reference_level)

# Get start time of current unit
start_time = self._calculate_level_start_time(positions, cycle_number, reference_level)

Expand All @@ -503,23 +537,30 @@ def _calculate_level_duration(self, positions: List[int], cycle_number: int, ref
hierarchy_size = sum(hierarchy_size)

if next_positions[reference_level] >= hierarchy_size:
if reference_level == 0:
# Next beat is in next cycle
next_cycle_number = cycle_number + 1
if next_cycle_number >= self.repetitions:
# Use meter end time
return self.start_time + self.repetitions * self.cycle_dur - start_time
next_positions[0] = 0
return self._calculate_level_start_time(next_positions, next_cycle_number, reference_level) - start_time
else:
# Carry over to higher level
next_positions[reference_level] = 0
next_positions[reference_level - 1] += 1
return self._calculate_level_duration(next_positions, cycle_number, reference_level - 1)
# Fall back to proportional calculation for overflow cases
return self._calculate_proportional_level_duration(positions, cycle_number, reference_level)

end_time = self._calculate_level_start_time(next_positions, cycle_number, reference_level)
return end_time - start_time

def _calculate_proportional_level_duration(self, positions: List[int], cycle_number: int, reference_level: int) -> float:
"""Calculate level duration using proportional cycle division."""
# Duration of a unit at this reference level is 1/size of that level within its parent
level_size = self.hierarchy[reference_level]
if isinstance(level_size, list):
level_size = sum(level_size)

# Calculate the duration of the parent unit
if reference_level == 0:
# Beat level - parent is the cycle
parent_duration = self.cycle_dur
else:
# Subdivision level - calculate parent unit duration recursively
parent_positions = positions[:reference_level]
parent_duration = self._calculate_proportional_level_duration(parent_positions, cycle_number, reference_level - 1)

return parent_duration / level_size

def get_musical_time(self, real_time: float, reference_level: Optional[int] = None) -> Union['MusicalTime', Literal[False]]:
"""
Convert real time to musical time within this meter.
Expand Down Expand Up @@ -599,14 +640,11 @@ def get_musical_time(self, real_time: float, reference_level: Optional[int] = No
fractional_beat = max(0.0, min(1.0, fractional_beat))

else:
# Reference level behavior - preserve full positions for fractional_beat calculation
# but truncate for final result
# Reference level behavior
truncated_positions = positions[:ref_level + 1]

# Use full positions for accurate fractional_beat calculation
# This prevents clustering when reference_level=0 (Issue #28)
current_level_start_time = self._calculate_level_start_time(positions, cycle_number, ref_level)
level_duration = self._calculate_level_duration(positions, cycle_number, ref_level)
current_level_start_time = self._calculate_level_start_time(truncated_positions, cycle_number, ref_level)
level_duration = self._calculate_level_duration(truncated_positions, cycle_number, ref_level)

if level_duration <= 0:
fractional_beat = 0.0
Expand Down
Loading