core_events: Add method ComplexEvent.extend_until

With this method it is very easy to ensure that a ComplexEvent has at
least a specific duration. This can be helpful when converting events to
another outside-world format.

mutwo/core_events/abc.py

@@ -946,3 +946,47 @@ def split_child_at(
         >>> sequential_event
         SequentialEvent([SimpleEvent(duration = 1), SimpleEvent(duration = 2)])
         """
+
+    @abc.abstractmethod
+    def extend_until(
+        self,
+        duration: core_parameters.abc.Duration,
+        duration_to_white_space: typing.Optional[
+            typing.Callable[[core_parameters.abc.Duration], Event]
+        ] = None,
+        prolong_simple_event: bool = True,
+    ) -> ComplexEvent:
+        """Prolong event until at least `duration` by appending an empty event.
+
+        :param duration: Until which duration the event shall be extended.
+            If event is already longer than or equal to given `duration`,
+            nothing will be changed. For :class:`~mutwo.core_events.SimultaneousEvent`
+            the default value is `None` which is equal to the duration of
+            the `SimultaneousEvent`.
+        :type duration: core_parameters.abc.Duration
+        :param duration_to_white_space: A function which creates the 'rest' or
+            'white space' event from :class:`~mutwo.core_parameters.abc.Duration`.
+            If this is ``None`` `mutwo` will fall back to use the default function
+            which is `mutwo.core_events.configurations.DEFAULT_DURATION_TO_WHITE_SPACE`.
+            Default to `None`.
+        :type duration_to_white_space: typing.Optional[typing.Callable[[core_parameters.abc.Duration], Event]]
+        :param prolong_simple_event: If set to ``True`` `mutwo` will prolong a single
+            :class:`~mutwo.core_events.SimpleEvent` inside a :class:`~mutwo.core_events.SimultaneousEvent`.
+            If set to ``False`` `mutwo` will raise an :class:`~mutwo.core_utilities.ImpossibleToExtendUntilError`
+            in case it finds a single `SimpleEvent` inside a `SimultaneousEvent`.
+            This doesn't effect `SimpleEvent` inside a `SequentialEvent`, here we can
+            simply append a new white space event.
+        :type prolong_simple_event: bool
+        :param mutate: If ``False`` the function will return a copy of the given object.
+            If set to ``True`` the object itself will be changed and the function will
+            return the changed object. Default to ``True``.
+        :type mutate: bool
+
+        **Example:**
+
+        >>> from mutwo import core_events
+        >>> s = core_events.SequentialEvent([core_events.SimpleEvent(1)])
+        >>> s.extend_until(10)
+        >>> print(s)
+        SequentialEvent([SimpleEvent(duration = DirectDuration(duration = 1)), SimpleEvent(duration = DirectDuration(duration = 9))])
+        """

mutwo/core_events/basic.py

@@ -702,6 +702,23 @@ def split_at(
         else:
             return self[:index].copy(), self[index:].copy()
 
+    @core_utilities.add_copy_option
+    def extend_until(
+        self,
+        duration: core_parameters.abc.Duration,
+        duration_to_white_space: typing.Optional[
+            typing.Callable[[core_parameters.abc.Duration], core_events.abc.Event]
+        ] = None,
+        prolong_simple_event: bool = True,
+    ) -> SequentialEvent:
+        duration = core_events.configurations.UNKNOWN_OBJECT_TO_DURATION(duration)
+        duration_to_white_space = (
+            duration_to_white_space
+            or core_events.configurations.DEFAULT_DURATION_TO_WHITE_SPACE
+        )
+        if (difference := duration - self.duration) > 0:
+            self.append(duration_to_white_space(difference))
+
 
 class SimultaneousEvent(core_events.abc.ComplexEvent, typing.Generic[T]):
     """Event-Object which contains other Event-Objects which happen at the same time."""
@@ -791,6 +808,37 @@ def split_child_at(
                 split_event = event.split_at(absolute_time)
                 self[event_index] = SequentialEvent(split_event)
 
+    @core_utilities.add_copy_option
+    def extend_until(
+        self,
+        duration: typing.Optional[core_parameters.abc.Duration] = None,
+        duration_to_white_space: typing.Optional[
+            typing.Callable[[core_parameters.abc.Duration], core_events.abc.Event]
+        ] = None,
+        prolong_simple_event: bool = True,
+    ) -> SequentialEvent:
+        duration = (
+            self.duration
+            if duration is None
+            else core_events.configurations.UNKNOWN_OBJECT_TO_DURATION(duration)
+        )
+        duration_to_white_space = (
+            duration_to_white_space
+            or core_events.configurations.DEFAULT_DURATION_TO_WHITE_SPACE
+        )
+        for event in self:
+            try:
+                event.extend_until(
+                    duration, duration_to_white_space, prolong_simple_event
+                )
+            # SimpleEvent
+            except AttributeError:
+                if prolong_simple_event:
+                    if (difference := duration - event.duration) > 0:
+                        event.duration += difference
+                else:
+                    raise core_utilities.ImpossibleToExtendUntilError(event)
+
 
 @core_utilities.add_tag_to_class
 class TaggedSimpleEvent(SimpleEvent):

mutwo/core_events/configurations.py

@@ -1,6 +1,7 @@
 """Configurations which are shared for all event classes in :mod:`mutwo.core_events`."""
 
 import fractions
+import functools
 import typing
 
 import quicktions
@@ -68,4 +69,17 @@ def __unknown_object_to_duration(unknown_object):
 """Default property parameter name for events in
 :class:`mutwo.core_events.TempoEnvelope`."""
 
-del typing
+
+# Avoid circular import problem
+@functools.cache
+def __simpleEvent():
+    return __import__("mutwo.core_events").core_events.SimpleEvent
+
+
+DEFAULT_DURATION_TO_WHITE_SPACE = lambda duration: __simpleEvent()(duration)
+"""Default conversion for parameter `duration_to_white_space` in
+:func:`mutwo.core_events.abc.ComplexEvent.extend_until`. This simply
+returns a :class:`mutwo.core_events.SimpleEvent` with the given
+duration."""
+
+del functools, typing

mutwo/core_events/envelopes.py

@@ -548,6 +548,27 @@ def cut_off(
 
         return self
 
+    @core_utilities.add_copy_option
+    def extend_until(
+        self,
+        duration: core_parameters.abc.Duration,
+        duration_to_white_space: typing.Optional[
+            typing.Callable[[core_parameters.abc.Duration], core_events.abc.Event]
+        ] = None,
+        prolong_simple_event: bool = True,
+    ) -> Envelope[T]:
+        self_duration = self.duration
+        super().extend_until(
+            duration,
+            duration_to_white_space=duration_to_white_space
+            or (
+                lambda duration: self._make_event(
+                    duration, self.parameter_at(self_duration), 0
+                )
+            ),
+            prolong_simple_event=prolong_simple_event,
+        )
+
 
 class RelativeEnvelope(Envelope, typing.Generic[T]):
     __parent_doc_string = Envelope.__doc__.split("\n")[2:]  # type: ignore

mutwo/core_utilities/exceptions.py

@@ -9,8 +9,10 @@
     "InvalidAverageValueStartAndEndWarning",
     "InvalidStartValueError",
     "InvalidPointError",
+    "ImpossibleToPutInError",
     "ImpossibleToSquashInError",
     "ImpossibleToSlideInError",
+    "ImpossibleToExtendUntilError",
     "InvalidStartAndEndValueError",
     "InvalidCutOutStartAndEndValuesError",
     "SplitUnavailableChildError",
@@ -81,6 +83,16 @@ def __init__(self, event_to_be_slided_into, event_to_slide_in):
         super().__init__(event_to_be_slided_into, event_to_slide_in, "slide")
 
 
+class ImpossibleToExtendUntilError(TypeError):
+    def __init__(self, event_to_extend_until):
+        super().__init__(
+            f"Can't extend '{event_to_extend_until}' of type"
+            f"'{type(event_to_extend_until)}' which resides inside a "
+            "SimultaneousEvent. Set 'prolong_simple_event' to 'True' in"
+            "case you want simple events to be prolonged."
+        )
+
+
 class InvalidStartAndEndValueError(Exception):
     def __init__(self, start, end):
         super().__init__(

tests/events/basic_tests.py

@@ -681,6 +681,29 @@ def test_start_and_end_time_per_event(self):
             ),
         )
 
+    def test_extend_until(self):
+        s, se = core_events.SimpleEvent, core_events.SequentialEvent
+
+        self.assertEqual(
+            self.sequence.extend_until(100, mutate=False), se([s(1), s(2), s(3), s(94)])
+        )
+
+        # Do nothing if already long enough
+        self.assertEqual(self.sequence.extend_until(6), se([s(1), s(2), s(3)]))
+
+        # Change in place
+        self.assertEqual(self.sequence.extend_until(7), se([s(1), s(2), s(3), s(1)]))
+        self.assertEqual(self.sequence, se([s(1), s(2), s(3), s(1)]))
+
+        # Do nothing if already longer
+        self.assertEqual(self.sequence.extend_until(4), se([s(1), s(2), s(3), s(1)]))
+
+        # Use custom event generator
+        self.assertEqual(
+            self.sequence.extend_until(8, duration_to_white_space=lambda d: se([s(d)])),
+            se([s(1), s(2), s(3), s(1), se([s(1)])]),
+        )
+
 
 class SimultaneousEventTest(unittest.TestCase, EventTest):
     class DummyParameter(object):
@@ -1004,6 +1027,40 @@ def test_remove_by(self):
             core_events.SimultaneousEvent([core_events.SimpleEvent(3)]),
         )
 
+    def test_extend_until(self):
+        s, se, si = (
+            core_events.SimpleEvent,
+            core_events.SequentialEvent,
+            core_events.SimultaneousEvent,
+        )
+
+        # Extend simple events inside simultaneous event..
+        self.assertEqual(
+            self.sequence.extend_until(10, mutate=False), si([s(10), s(10), s(10)])
+        )
+
+        # ..should raise if flag is set to False
+        self.assertRaises(
+            core_utilities.ImpossibleToExtendUntilError,
+            self.sequence.extend_until,
+            10,
+            prolong_simple_event=False,
+        )
+
+        # Extend sequential events inside simultaneous event..
+        ese = se([s(1), s(2), s(3), s(4)])  # extended sequential event
+        self.assertEqual(
+            self.nested_sequence.extend_until(10, mutate=False), si([ese, ese])
+        )
+
+        # Nothing happens if already long enough
+        self.assertEqual(
+            self.nested_sequence.extend_until(4, mutate=False), self.nested_sequence
+        )
+
+        # Check default value for SimultaneousEvent
+        self.assertEqual(si([s(1), s(3)]).extend_until(), si([s(3), s(3)]))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/events/envelopes_test.py

@@ -232,6 +232,15 @@ def test_cut_off(self):
             self.envelope.value_at(1.6), cut_off_envelope.value_at(0.6)
         )
 
+    def test_extend_until(self):
+        self.assertEqual(len(self.envelope), 5)
+        self.envelope.extend_until(10)
+        self.assertEqual(len(self.envelope), 6)
+        self.assertEqual(self.envelope[-1].duration, 4)
+        self.assertEqual(
+            self.envelope.parameter_tuple[-1] == self.envelope.parameter_tuple[-2]
+        )
+
 
 class RelativeEnvelopeTest(unittest.TestCase):
     def setUp(cls):