working TimeMap implementation !

jubeatools/formats/eve/tests/test_timemap.py

@@ -0,0 +1,55 @@
+from fractions import Fraction
+
+from hypothesis import given
+
+from jubeatools import song
+from jubeatools.formats.eve.timemap import TimeMap
+from jubeatools.testutils import strategies as jbst
+from jubeatools.utils import fraction_to_decimal, group_by
+
+
+@given(jbst.timing_info(bpm_changes=True), jbst.beat_time())
+def test_that_seconds_at_beat_works_like_the_naive_approach(
+    timing: song.Timing, beat: song.BeatsTime
+) -> None:
+    time_map = TimeMap.from_timing(timing)
+    expected = naive_approach(timing, beat)
+    actual = time_map._frac_seconds_at(beat)
+    assert actual == expected
+
+
+def naive_approach(beats: song.Timing, beat: song.BeatsTime) -> Fraction:
+    if beat < 0:
+        raise ValueError("Can't compute seconds at negative beat")
+
+    if not beats.events:
+        raise ValueError("No BPM defined")
+
+    grouped_by_time = group_by(beats.events, key=lambda e: e.time)
+    for time, events in grouped_by_time.items():
+        if len(events) > 1:
+            raise ValueError(f"Multiple BPMs defined on beat {time} : {events}")
+
+    sorted_events = sorted(beats.events, key=lambda e: e.time)
+    first_event = sorted_events[0]
+    if first_event.time != song.BeatsTime(0):
+        raise ValueError("First BPM event is not on beat zero")
+    
+    if beat > sorted_events[-1].time:
+        events_before = sorted_events
+    else:
+        last_index = next(
+            i for i, e in enumerate(sorted_events)
+            if e.time >= beat
+        )
+        events_before = sorted_events[:last_index]
+    total_seconds = Fraction(0)
+    current_beat = beat
+    for event in reversed(events_before):
+        beats_since_previous = current_beat - event.time
+        seconds_since_previous = (60 * beats_since_previous) / Fraction(event.BPM)
+        total_seconds += seconds_since_previous
+        current_beat = event.time
+
+    total_seconds = total_seconds + Fraction(beats.beat_zero_offset)
+    return total_seconds

jubeatools/formats/eve/timemap.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from fractions import Fraction
+from typing import List
+
+from more_itertools import windowed
+from sortedcontainers import SortedKeyList
+
+from jubeatools import song
+from jubeatools.utils import fraction_to_decimal, group_by
+
+
+@dataclass
+class BPMAtSecond:
+    seconds: Fraction
+    BPM: Fraction
+
+
+@dataclass
+class BPMChange:
+    beats: song.BeatsTime
+    seconds: Fraction
+    BPM: Fraction
+
+
+@dataclass
+class TimeMap:
+    """Wraps a song.Timing to allow converting symbolic time (in beats)
+    to clock time (in seconds) and back"""
+
+    beat_zero_offset: song.SecondsTime
+    events_by_beats: SortedKeyList[BPMChange, song.BeatsTime]
+    events_by_seconds: SortedKeyList[BPMChange, Fraction]
+
+    @classmethod
+    def from_timing(cls, beats: song.Timing) -> TimeMap:
+        """Create a time map from a song.Timing object"""
+        if not beats.events:
+            raise ValueError("No BPM defined")
+
+        grouped_by_time = group_by(beats.events, key=lambda e: e.time)
+        for time, events in grouped_by_time.items():
+            if len(events) > 1:
+                raise ValueError(f"Multiple BPMs defined on beat {time} : {events}")
+
+        sorted_events = sorted(beats.events, key=lambda e: e.time)
+        first_event = sorted_events[0]
+        if first_event.time != song.BeatsTime(0):
+            raise ValueError("First BPM event is not on beat zero")
+
+        # set first BPM change then compute from there
+        current_second = Fraction(beats.beat_zero_offset)
+        bpm_changes = [
+            BPMChange(first_event.time, current_second, Fraction(first_event.BPM))
+        ]
+        for previous, current in windowed(sorted_events, 2):
+            if previous is None or current is None:
+                continue
+
+            beats_since_last_event = current.time - previous.time
+            seconds_since_last_event = (60 * beats_since_last_event) / Fraction(
+                previous.BPM
+            )
+            current_second += seconds_since_last_event
+            bpm_change = BPMChange(current.time, current_second, Fraction(current.BPM))
+            bpm_changes.append(bpm_change)
+
+        return cls(
+            beat_zero_offset=beats.beat_zero_offset,
+            events_by_beats=SortedKeyList(bpm_changes, key=lambda b: b.beats),
+            events_by_seconds=SortedKeyList(bpm_changes, key=lambda b: b.seconds),
+        )
+
+    @classmethod
+    def from_seconds(cls, events: List[BPMAtSecond]) -> TimeMap:
+        """Create a time map from a list of BPM changes with time positions
+        given in seconds. The first BPM implicitely happens at beat zero"""
+        if not events:
+            raise ValueError("No BPM defined")
+
+        grouped_by_time = group_by(events, key=lambda e: e.seconds)
+        for time, events in grouped_by_time.items():
+            if len(events) > 1:
+                raise ValueError(f"Multiple BPMs defined at {time} seconds : {events}")
+
+        # take the first BPM change then compute from there
+        sorted_events = sorted(events, key=lambda e: e.seconds)
+        first_event = sorted_events[0]
+        current_beat = Fraction(0)
+        bpm_changes = [BPMChange(current_beat, first_event.seconds, first_event.BPM)]
+        for previous, current in windowed(sorted_events, 2):
+            if previous is None or current is None:
+                continue
+
+            seconds_since_last_event = current.seconds - previous.seconds
+            beats_since_last_event = (
+                previous.BPM * seconds_since_last_event
+            ) / Fraction(60)
+            current_beat += beats_since_last_event
+            bpm_change = BPMChange(current_beat, current.seconds, current.BPM)
+            bpm_changes.append(bpm_change)
+
+        return cls(
+            beat_zero_offset=fraction_to_decimal(first_event.seconds),
+            events_by_beats=SortedKeyList(bpm_changes, key=lambda b: b.beats),
+            events_by_seconds=SortedKeyList(bpm_changes, key=lambda b: b.seconds),
+        )
+
+    def seconds_at(self, beat: song.BeatsTime) -> song.SecondsTime:
+        frac_seconds = self._frac_seconds_at(beat)
+        return fraction_to_decimal(frac_seconds)
+    
+    def _frac_seconds_at(self, beat):
+        if beat < 0:
+            raise ValueError("Can't compute seconds at negative beat")
+
+        # find previous bpm change
+        index = self.events_by_beats.bisect_key_right(beat) - 1
+        bpm_change: BPMChange = self.events_by_beats[index]
+
+        # compute seconds since last bpm change
+        beats_since_last_event = beat - bpm_change.beats
+        seconds_since_last_event = (60 * beats_since_last_event) / bpm_change.BPM
+        return bpm_change.seconds + seconds_since_last_event
+
+    def beats_at(self, seconds: song.SecondsTime) -> song.BeatsTime:
+        if seconds < self.beat_zero_offset:
+            raise ValueError(
+                f"Can't compute beat time at {seconds} seconds, since it predates "
+                f"beat zero, which happens at {self.beat_zero_offset} seconds"
+            )
+
+        # find previous bpm change
+        frac_seconds = Fraction(seconds)
+        index = self.events_by_seconds.bisect_key_right(frac_seconds) - 1
+        bpm_change: BPMChange = self.events_by_seconds[index]
+
+        # compute beats since last bpm change
+        seconds_since_last_event = frac_seconds - bpm_change.seconds
+        beats_since_last_event = (bpm_change.BPM * seconds_since_last_event) / Fraction(
+            60
+        )
+        return bpm_change.beats + beats_since_last_event

jubeatools/testutils/strategies.py

@@ -50,9 +50,10 @@ def beat_time(
         max_value = None
 
     if max_numerator is not None:
-        max_value = (
-            max_numerator if max_value is None else min(max_numerator, max_value)
-        )
+        if max_value is not None:
+            max_value = min(max_numerator, max_value)
+        else:
+            max_value = max_numerator
 
     numerator = draw(st.integers(min_value=min_value, max_value=max_value))
     return BeatsTime(numerator, denominator)

jubeatools/utils.py

@@ -1,9 +1,12 @@
+"""General utility functions"""
+
 import unicodedata
 from decimal import Decimal
 from fractions import Fraction
 from functools import reduce
 from math import gcd
-from typing import Callable, Optional, TypeVar
+from typing import Callable, Optional, TypeVar, List, Dict
+from collections import defaultdict
 
 
 def single_lcm(a: int, b: int) -> int:
@@ -24,7 +27,7 @@ def charinfo(c: str) -> str:
 A = TypeVar("A")
 B = TypeVar("B")
 
-
+# Monadic stuff !
 def none_or(c: Callable[[A], B], e: Optional[A]) -> Optional[B]:
     if e is None:
         return None
@@ -35,3 +38,15 @@ def none_or(c: Callable[[A], B], e: Optional[A]) -> Optional[B]:
 def fraction_to_decimal(frac: Fraction) -> Decimal:
     "Thanks stackoverflow ! https://stackoverflow.com/a/40468867/10768117"
     return frac.numerator / Decimal(frac.denominator)
+
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+
+def group_by(elements: List[V], key: Callable[[V], K]) -> Dict[K, List[V]]:
+    res = defaultdict(list)
+    for e in elements:
+        res[key(e)].append(e)
+
+    return res