Pattern Improvements (#350)

* Typing patterns

* Add Pattern.play(target_bus=...) arg

* Implement PinPattern

* Integrating PinPattern

* Simplify events

* Maintain WeakSet of PatternPlayers

* More typing, remove unimplemented sequence patterns

supriya/patterns/__init__.py

@@ -22,7 +22,6 @@
     UnaryOpPattern,
 )
 from .players import PatternPlayer
-from .sequences import GatePattern, RepeatPattern, RestartPattern, StutterPattern
 from .structure import BusPattern, FxPattern, GroupPattern, ParallelPattern
 
 __all__ = [
@@ -36,7 +35,6 @@
     "Event",
     "EventPattern",
     "FxPattern",
-    "GatePattern",
     "GroupAllocateEvent",
     "GroupPattern",
     "MonoEventPattern",
@@ -48,14 +46,11 @@
     "PatternPlayer",
     "Priority",
     "RandomPattern",
-    "RepeatPattern",
-    "RestartPattern",
     "SeedPattern",
     "SequencePattern",
     "ShufflePattern",
     "StartEvent",
     "StopEvent",
-    "StutterPattern",
     "SynthAllocateEvent",
     "UnaryOpPattern",
     "UpdatePattern",

supriya/patterns/eventpatterns.py

@@ -12,9 +12,9 @@ class EventPattern(Pattern):
     Akin to SuperCollider's Pbind.
     """
 
-    def __init__(self, event_type: Type[Event] = NoteEvent, **patterns):
-        self._patterns = patterns
+    def __init__(self, event_type: Type[Event] = NoteEvent, **patterns) -> None:
         self._event_type = event_type
+        self._patterns = patterns
 
     def _iterate(self, state=None):
         patterns = self._prepare_patterns()
@@ -42,7 +42,7 @@ def event_type(self) -> Type[Event]:
         return self._event_type
 
     @property
-    def is_infinite(self):
+    def is_infinite(self) -> bool:
         for value in self._patterns.values():
             if isinstance(value, Pattern) and not value.is_infinite:
                 return False
@@ -69,7 +69,7 @@ def _iterate(self, state=None):
                 return
 
     @property
-    def is_infinite(self):
+    def is_infinite(self) -> bool:
         for value in self._patterns.values():
             if isinstance(value, Pattern) and not value.is_infinite:
                 return False
@@ -81,7 +81,7 @@ class UpdatePattern(Pattern):
     Akin to SuperCollider's Pbindf.
     """
 
-    def __init__(self, pattern, **patterns):
+    def __init__(self, pattern: Pattern, **patterns) -> None:
         self._pattern = pattern
         self._patterns = patterns
 
@@ -112,7 +112,7 @@ def _prepare_patterns(self):
         return patterns
 
     @property
-    def is_infinite(self):
+    def is_infinite(self) -> bool:
         for value in self._patterns.values():
             if isinstance(value, Pattern) and not value.is_infinite:
                 return False
@@ -124,7 +124,7 @@ class ChainPattern(Pattern):
     Akin to SuperCollider's Pchain.
     """
 
-    def __init__(self, *patterns):
+    def __init__(self, *patterns: Pattern) -> None:
         self._patterns = tuple(patterns)
 
     def _iterate(self, state=None):
@@ -143,7 +143,7 @@ def _iterate(self, state=None):
                 return
 
     @property
-    def is_infinite(self):
+    def is_infinite(self) -> bool:
         for value in self._patterns:
             if isinstance(value, Pattern) and not value.is_infinite:
                 return False

supriya/patterns/events.py

@@ -44,46 +44,22 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
+        **kwargs,
     ) -> None:
-        raise NotImplementedError
+        pass
 
 
 class StartEvent(Event):
     """
     The first event injected by a pattern player.
     """
 
-    def perform(
-        self,
-        context: Context,
-        proxy_mapping: Dict[Union[UUID, Tuple[UUID, int]], ContextObject],
-        *,
-        current_offset: float,
-        notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
-        priority: Priority,
-        target_node: Optional[Node] = None,
-    ) -> None:
-        pass
-
 
 class StopEvent(Event):
     """
     The last event injected by a pattern player.
     """
 
-    def perform(
-        self,
-        context: Context,
-        proxy_mapping: Dict[Union[UUID, Tuple[UUID, int]], ContextObject],
-        *,
-        current_offset: float,
-        notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
-        priority: Priority,
-        target_node: Optional[Node] = None,
-    ) -> None:
-        pass
-
 
 class BusAllocateEvent(Event):
     def __init__(
@@ -108,7 +84,7 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
+        **kwargs,
     ) -> None:
         proxy_mapping[self.id_] = context.add_bus_group(
             calculation_rate=self.calculation_rate, count=self.channel_count
@@ -130,7 +106,7 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
+        **kwargs,
     ) -> None:
         if not isinstance(bus_group := proxy_mapping.pop(self.id_), BusGroup):
             raise RuntimeError(bus_group)
@@ -169,7 +145,6 @@ def __init__(
 
     def _resolve_target_node(
         self,
-        target_node: Optional[Node],
         proxy_mapping: Dict[Union[UUID, Tuple[UUID, int]], ContextObject],
     ) -> Optional[Node]:
         if isinstance(self.target_node, UUID):
@@ -178,7 +153,7 @@ def _resolve_target_node(
             ):
                 raise RuntimeError(target_node_)
             return target_node_
-        return self.target_node or target_node
+        return self.target_node
 
 
 class GroupAllocateEvent(NodeEvent):
@@ -190,13 +165,11 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
         **kwargs,
     ) -> None:
         proxy_mapping[self.id_] = context.add_group(
             add_action=self.add_action,
             target_node=self._resolve_target_node(
-                target_node,
                 proxy_mapping,
             ),
         )
@@ -217,7 +190,7 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
+        **kwargs,
     ) -> None:
         if not isinstance(node := proxy_mapping.pop(self.id_), Node):
             raise RuntimeError(node)
@@ -279,7 +252,6 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
         **kwargs,
     ) -> None:
         if priority == Priority.START:
@@ -291,12 +263,12 @@ def perform(
             for key, value in settings.items():
                 if isinstance(value, UUID):
                     settings[key] = proxy_mapping[value]
+            # add the synth
             if self.id_ not in proxy_mapping:
                 proxy_mapping[self.id_] = context.add_synth(
                     add_action=self.add_action,
                     synthdef=self.synthdef or default,
                     target_node=self._resolve_target_node(
-                        target_node,
                         proxy_mapping,
                     ),
                     **settings,
@@ -357,17 +329,17 @@ def perform(
         current_offset: float,
         notes_mapping: Dict[Union[UUID, Tuple[UUID, int]], float],
         priority: Priority,
-        target_node: Optional[Node] = None,
+        **kwargs,
     ) -> None:
         settings = self.kwargs.copy()
         for key, value in settings.items():
             if isinstance(value, UUID):
                 settings[key] = proxy_mapping[value]
+        # add the synth
         proxy_mapping[self.id_] = context.add_synth(
             add_action=self.add_action,
             synthdef=self.synthdef,
             target_node=self._resolve_target_node(
-                target_node,
                 proxy_mapping,
             ),
             **settings,

supriya/patterns/noise.py

@@ -1,10 +1,18 @@
+from typing import Optional, Sequence, Union
+
 from uqbar.enums import IntEnumeration
 
 from .patterns import Pattern, SequencePattern
 
 
 class ChoicePattern(SequencePattern):
-    def __init__(self, sequence, iterations=1, forbid_repetitions=False, weights=None):
+    def __init__(
+        self,
+        sequence: Sequence,
+        iterations: Optional[int] = 1,
+        forbid_repetitions: bool = False,
+        weights: Optional[Sequence[float]] = None,
+    ) -> None:
         super().__init__(sequence, iterations=iterations)
         self._forbid_repetitions = bool(forbid_repetitions)
         if weights:
@@ -46,11 +54,11 @@ def _find_index_weighted(self, rng):
         return index
 
     @property
-    def forbid_repetitions(self):
+    def forbid_repetitions(self) -> bool:
         return self._forbid_repetitions
 
     @property
-    def weights(self):
+    def weights(self) -> Optional[Sequence[float]]:
         return self._weights
 
 
@@ -59,8 +67,12 @@ class Distribution(IntEnumeration):
         WHITE_NOISE = 0
 
     def __init__(
-        self, minimum=0.0, maximum=1.0, iterations=None, distribution="WHITE_NOISE"
-    ):
+        self,
+        minimum: float = 0.0,
+        maximum: float = 1.0,
+        iterations: Optional[int] = None,
+        distribution: Union["RandomPattern.Distribution", str] = "WHITE_NOISE",
+    ) -> None:
         if iterations is not None:
             iterations = int(iterations)
             if iterations < 1:
@@ -83,28 +95,33 @@ def procedure(one, two):
                 return
 
     @property
-    def distribution(self):
+    def distribution(self) -> Union["RandomPattern.Distribution", str]:
         return self._distribution
 
     @property
-    def is_infinite(self):
+    def is_infinite(self) -> bool:
         return self._iterations is None
 
     @property
-    def iterations(self):
+    def iterations(self) -> Optional[int]:
         return self._iterations
 
     @property
-    def minimum(self):
+    def minimum(self) -> float:
         return self._minimum
 
     @property
-    def maximum(self):
+    def maximum(self) -> float:
         return self._maximum
 
 
 class ShufflePattern(SequencePattern):
-    def __init__(self, sequence, iterations=1, forbid_repetitions=False):
+    def __init__(
+        self,
+        sequence: Sequence,
+        iterations: Optional[int] = 1,
+        forbid_repetitions: bool = False,
+    ) -> None:
         super().__init__(sequence, iterations=iterations)
         self._forbid_repetitions = bool(forbid_repetitions)
 
@@ -141,5 +158,5 @@ def _shuffle(self, length, rng, previous_index=None):
         return shuffled_indices
 
     @property
-    def forbid_repetitions(self):
+    def forbid_repetitions(self) -> bool:
         return self._forbid_repetitions