Skip to content

Commit

Permalink
set/mutate_parameter:Fix for >1 refs of same event
Browse files Browse the repository at this point in the history
Before this patch it was dangerous to call 'set_parameter' or
'mutate_parameter' on a 'mutwo.core_events.ComplexEvent' which containes
multiple references of the same event. It was dangerous, because
set_parameter would could many times on the same event, which would lead
to unexpected results. For instance:

>>> simple_event = core_events.SimpleEvent(2)
>>> s = core_events.SequentialEvent([simple_event, simple_event])
>>> s.set_parameter('duration', lambda old_duration: old_duration * 2)
>>> s[0].duration
8

So instead of the expected return value '4' (which would be the
old_duration 2 multiplied by 2) we get 8, because 'set_parameter'
has been called twice since there are two references of the same event
'simple_event'.

After this patch the above code would return the expected 4.

This patch changes the API of inheriting from the abstract base class
'mutwo.core_events.abc.Event'. Instead of overriding the abstract
methods "set_parameter" and "mutate_parameter", users need to override
the methods "_set_parameter" and "_mutate_parameter". Because no outside
library currently depends on this ABC API detail (and 99.999% of all
time users will inherit from SimpleEvent or SequentialEvent or
SimultaneousEvent) this change is not considered to break the major
version of mutwo, but only to be a bug fix actually.

---

This patch fixes issue #23.
Additional context can be found there.
  • Loading branch information
levinericzimmermann committed Nov 12, 2022
1 parent 5f028bc commit 8c17c27
Show file tree
Hide file tree
Showing 2 changed files with 157 additions and 61 deletions.
148 changes: 115 additions & 33 deletions mutwo/core_events/abc.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,32 @@ def _event_to_metrized_event(self):
"mutwo.core_converters"
).core_converters.EventToMetrizedEvent()

@abc.abstractmethod
def _set_parameter(
self,
parameter_name: str,
object_or_function: typing.Union[
typing.Callable[
[core_constants.ParameterType], core_constants.ParameterType
],
core_constants.ParameterType,
],
set_unassigned_parameter: bool,
id_set: set[int],
) -> Event:
...

@abc.abstractmethod
def _mutate_parameter(
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
id_set: set[int],
) -> Event:
...

# ###################################################################### #
# public properties #
# ###################################################################### #
Expand Down Expand Up @@ -193,7 +219,6 @@ def get_parameter(
None
"""

@abc.abstractmethod
def set_parameter(
self,
parameter_name: str,
Expand All @@ -204,6 +229,7 @@ def set_parameter(
core_constants.ParameterType,
],
set_unassigned_parameter: bool = True,
mutate: bool = True,
) -> typing.Optional[Event]:
"""Sets parameter to new value for all children events.
Expand Down Expand Up @@ -232,15 +258,35 @@ def set_parameter(
>>> sequential_event.set_parameter('duration', lambda duration: duration * 2)
>>> sequential_event.get_parameter('duration')
(4, 6)
**Warning:**
If there are multiple references of the same Event inside a
:class:`~mutwo.core_events.SequentialEvent` or a
~mutwo.core_events.SimultaneousEvent`, ``set_parameter`` will
only be called once for each Event. So multiple references
of the same event will be ignored. This behaviour ensures,
that on a big scale level each item inside the
:class:`mutwo.core_events.abc.ComplexEvent` is treated equally
(so for instance the duration of each item is doubled, and
nor for some doubled and for those with references which
appear twice quadrupled).
"""
return self._set_parameter(
parameter_name,
object_or_function,
set_unassigned_parameter=set_unassigned_parameter,
mutate=mutate,
id_set=set([]),
)

@abc.abstractmethod
def mutate_parameter(
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
mutate: bool = True,
) -> typing.Optional[Event]:
"""Mutate parameter with a function.
Expand Down Expand Up @@ -280,7 +326,26 @@ def mutate_parameter(
>>> # now all pitches should be one octave higher (from 4 to 5)
>>> sequential_event.get_parameter('pitch_list')
([WesternPitch(c5), WesternPitch(e5)],)
**Warning:**
If there are multiple references of the same Event inside a
:class:`~mutwo.core_events.SequentialEvent` or a
~mutwo.core_events.SimultaneousEvent`, ``mutate_parameter`` will
only be called once for each Event. So multiple references
of the same event will be ignored. This behaviour ensures,
that on a big scale level each item inside the
:class:`mutwo.core_events.abc.ComplexEvent` is treated equally
(so for instance the duration of each item is doubled, and
nor for some doubled and for those with references which
appear twice quadrupled).
"""
return self._mutate_parameter(
parameter_name,
function,
mutate=mutate,
id_set=set([]),
)

@core_utilities.add_copy_option
def reset_tempo_envelope(self) -> Event:
Expand Down Expand Up @@ -517,6 +582,54 @@ def _assert_start_in_range(self, start: core_parameters.abc.Duration):
if self.duration < start:
raise core_utilities.InvalidStartValueError(start, self.duration)

def _apply_once_per_event(
self, method_name: str, *args, id_set: set[int], **kwargs
):
print('')
print(method_name, args, kwargs)
for event in self:
print(event, id(event), id_set)
if (event_id := id(event)) not in id_set:
id_set.add(event_id)
getattr(event, method_name)(*args, id_set=id_set, **kwargs)

@core_utilities.add_copy_option
def _set_parameter( # type: ignore
self,
parameter_name: str,
object_or_function: typing.Union[
typing.Callable[
[core_constants.ParameterType], core_constants.ParameterType
],
core_constants.ParameterType,
],
set_unassigned_parameter: bool,
id_set: set[int],
) -> ComplexEvent[T]:
self._apply_once_per_event(
"_set_parameter",
parameter_name,
object_or_function,
id_set=id_set,
set_unassigned_parameter=set_unassigned_parameter,
)

@core_utilities.add_copy_option
def _mutate_parameter( # type: ignore
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
id_set: set[int],
) -> ComplexEvent[T]:
self._apply_once_per_event(
"_mutate_parameter",
parameter_name,
function,
id_set=id_set,
)

# ###################################################################### #
# public methods #
# ###################################################################### #
Expand Down Expand Up @@ -606,37 +719,6 @@ def get_parameter(
parameter_value_list.append(parameter_value_tuple)
return tuple(parameter_value_list)

@core_utilities.add_copy_option
def set_parameter( # type: ignore
self,
parameter_name: str,
object_or_function: typing.Union[
typing.Callable[
[core_constants.ParameterType], core_constants.ParameterType
],
core_constants.ParameterType,
],
set_unassigned_parameter: bool = True,
) -> ComplexEvent[T]:
[
event.set_parameter(
parameter_name,
object_or_function,
set_unassigned_parameter=set_unassigned_parameter,
)
for event in self
]

@core_utilities.add_copy_option
def mutate_parameter( # type: ignore
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
) -> ComplexEvent[T]:
[event.mutate_parameter(parameter_name, function) for event in self]

@core_utilities.add_copy_option
def remove_by( # type: ignore
self, condition: typing.Callable[[Event], bool]
Expand Down
70 changes: 42 additions & 28 deletions mutwo/core_events/basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,44 @@ def __repr__(self) -> str:
)
return "{}({})".format(type(self).__name__, ", ".join(attribute_iterator))

# ###################################################################### #
# private methods #
# ###################################################################### #

@core_utilities.add_copy_option
def _set_parameter(
self,
parameter_name: str,
object_or_function: typing.Union[
typing.Callable[
[core_constants.ParameterType], core_constants.ParameterType
],
core_constants.ParameterType,
],
set_unassigned_parameter: bool,
id_set: set[int],
) -> SimpleEvent:
old_parameter = self.get_parameter(parameter_name)
if set_unassigned_parameter or old_parameter is not None:
if hasattr(object_or_function, "__call__"):
new_parameter = object_or_function(old_parameter)
else:
new_parameter = object_or_function
setattr(self, parameter_name, new_parameter)

@core_utilities.add_copy_option
def _mutate_parameter(
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
id_set: set[int],
) -> SimpleEvent:
parameter = self.get_parameter(parameter_name)
if parameter is not None:
function(parameter)

# ###################################################################### #
# properties #
# ###################################################################### #
Expand Down Expand Up @@ -139,17 +177,11 @@ def get_parameter(
) -> core_constants.ParameterType:
return getattr(self, parameter_name, None)

@core_utilities.add_copy_option
# Update docstring
def set_parameter( # type: ignore
self,
parameter_name: str,
object_or_function: typing.Union[
typing.Callable[
[core_constants.ParameterType], core_constants.ParameterType
],
core_constants.ParameterType,
],
set_unassigned_parameter: bool = True,
*args,
**kwargs,
) -> SimpleEvent:
"""Sets event parameter to new value.
Expand Down Expand Up @@ -191,25 +223,7 @@ def set_parameter( # type: ignore
10
"""

old_parameter = self.get_parameter(parameter_name)
if set_unassigned_parameter or old_parameter is not None:
if hasattr(object_or_function, "__call__"):
new_parameter = object_or_function(old_parameter)
else:
new_parameter = object_or_function
setattr(self, parameter_name, new_parameter)

@core_utilities.add_copy_option
def mutate_parameter( # type: ignore
self,
parameter_name: str,
function: typing.Union[
typing.Callable[[core_constants.ParameterType], None], typing.Any
],
) -> SimpleEvent:
parameter = self.get_parameter(parameter_name)
if parameter is not None:
function(parameter)
return super().set_parameter(*args, **kwargs)

def metrize(self, mutate: bool = True) -> SimpleEvent:
metrized_event = self._event_to_metrized_event(self)
Expand Down

0 comments on commit 8c17c27

Please sign in to comment.