Sketching

supriya/mixers/mixers.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, List, Optional, Tuple
+from typing import TYPE_CHECKING, List, Optional, Set, Tuple
 
 from ..contexts import AsyncServer, BusGroup
 from ..enums import AddAction
@@ -64,6 +64,7 @@ def __init__(self, *, parent: Optional["Session"]) -> None:
         TrackContainer.__init__(self)
         self._tracks.append(Track(parent=self))
         self._output = MixerOutput(parent=self)
+        self._soloed_tracks: Set[Track] = set()
 
     def _allocate(self, context: AsyncServer) -> bool:
         if not super()._allocate(context=context):

supriya/mixers/routing.py

@@ -32,6 +32,7 @@ class Connection(AllocatableComponent[A], Generic[A, S, T]):
     @dataclasses.dataclass
     class State:
         feedsback: Optional[bool] = None
+        inverted: bool = False
         postfader: bool = True
         source_bus: Optional[BusGroup] = None
         source_component: Optional[AllocatableComponent] = None
@@ -80,7 +81,7 @@ def _allocate_synth(
             in_=new_state.source_bus,
             out=new_state.target_bus,
             synthdef=PATCH_CABLE_2_2,
-            multiplier=-1 if new_state.inverted else 1
+            multiplier=-1 if new_state.inverted else 1,
         )
 
     def _deallocate(self) -> None:

supriya/mixers/tracks.py

@@ -225,6 +225,8 @@ def __init__(
         TrackContainer.__init__(self)
         self._feedback = TrackFeedback(parent=self)
         self._input = TrackInput(parent=self)
+        self._is_muted: bool = False
+        self._is_soloed: bool = False
         self._output = TrackOutput(parent=self)
         # TODO: Are sends the purview of track containers in general?
         self._sends: List[TrackSend] = []
@@ -257,7 +259,7 @@ def _allocate(self, *, context: AsyncServer) -> bool:
         )
         target_node = self.parent._nodes[ComponentNames.TRACKS]
         with context.at():
-            active_control_bus.set(float(self._is_active))
+            self._update_activation(True)
             gain_control_bus.set(0.0)
             input_levels_control_bus.set(0.0)
             output_levels_control_bus.set(0.0)
@@ -312,6 +314,9 @@ def _register_feedback(
             )
         return self._get_audio_bus(context, name=ComponentNames.FEEDBACK)
 
+    def _set_soloed(self, soloed: bool) -> None:
+        self._is_soloed = soloed
+
     def _unregister_feedback(self, dependent: "Component") -> bool:
         if should_tear_down := super()._unregister_feedback(dependent):
             # check if feedback should be torn down
@@ -320,6 +325,12 @@ def _unregister_feedback(self, dependent: "Component") -> bool:
             self._feedback._set_source(None)
         return should_tear_down
 
+    def _update_activation(self, force=True) -> None:
+        was_active = self._is_active
+        self._is_active = not self._is_muted
+        if self._can_allocate() and (force or (was_active != self._is_active)):
+            self._control_buses[ComponentNames.ACTIVE].set(float(self._is_active))
+
     async def add_send(
         self, target: TrackContainer, postfader: bool = True, inverted: bool = False
     ) -> TrackSend:
@@ -378,15 +389,14 @@ async def move(self, parent: TrackContainer, index: int) -> None:
             for component in sorted(self._dependents, key=lambda x: x.graph_order):
                 component._reconcile(context)
 
-    async def set_active(self, is_active: bool = True) -> None:
-        async with self._lock:
-            self._is_active = is_active
-            if self._can_allocate():
-                self._control_buses[ComponentNames.ACTIVE].set(float(self._is_active))
-
     async def set_input(self, input_: Optional[Union[BusGroup, "Track"]]) -> None:
         await self._input.set_source(input_)
 
+    async def set_muted(self, muted: bool = True) -> None:
+        async with self._lock:
+            self._is_muted = muted
+            self._update_activation()
+
     async def set_output(
         self, output: Optional[Union[BusGroup, Default, TrackContainer]]
     ) -> None:

tests/mixers/test_DeviceContainer.py

@@ -25,7 +25,7 @@
                      1002 group (session.mixers[0]:devices)
             +            1066 group (session.mixers[0].devices[0]:group)
             +                1067 supriya:device-dc-tester:2 (session.mixers[0].devices[0]:synth)
-            +                    dc: 1.0, out: 0.0
+            +                    dc: 1.0, out: 16.0
                      1003 supriya:channel-strip:2 (session.mixers[0]:channel-strip)
                          active: 1.0, bus: 16.0, gain: c0, gate: 1.0
                      1005 supriya:meters:2 (session.mixers[0]:output-levels)
@@ -37,7 +37,13 @@
                     contents=(
                         OscMessage("/g_new", 1066, 1, 1002),
                         OscMessage(
-                            "/s_new", "supriya:device-dc-tester:2", 1067, 1, 1066
+                            "/s_new",
+                            "supriya:device-dc-tester:2",
+                            1067,
+                            1,
+                            1066,
+                            "out",
+                            16.0,
                         ),
                     ),
                 ),
@@ -54,7 +60,7 @@
                              1008 group (session.mixers[0].tracks[0]:devices)
             +                    1066 group (session.mixers[0].tracks[0].devices[0]:group)
             +                        1067 supriya:device-dc-tester:2 (session.mixers[0].tracks[0].devices[0]:synth)
-            +                            dc: 1.0, out: 0.0
+            +                            dc: 1.0, out: 18.0
                              1009 supriya:channel-strip:2 (session.mixers[0].tracks[0]:channel-strip)
                                  active: c5, bus: 18.0, gain: c6, gate: 1.0
                              1051 supriya:patch-cable:2x2 (session.mixers[0].tracks[0].sends[0]:synth)
@@ -66,7 +72,13 @@
                     contents=(
                         OscMessage("/g_new", 1066, 1, 1008),
                         OscMessage(
-                            "/s_new", "supriya:device-dc-tester:2", 1067, 1, 1066
+                            "/s_new",
+                            "supriya:device-dc-tester:2",
+                            1067,
+                            1,
+                            1066,
+                            "out",
+                            18.0,
                         ),
                     ),
                 ),

tests/mixers/test_Track.py

@@ -1078,80 +1078,6 @@ async def test_Track_move(
     assert commands == expected_commands
 
 
-@pytest.mark.parametrize("online", [False, True])
-@pytest.mark.parametrize(
-    "target, active, expected_commands, expected_levels",
-    [
-        (
-            "mixers[0].tracks[0]",
-            False,
-            [OscMessage("/c_set", 5, 0.0)],
-            [
-                ("m[0].t[0]", (0.0, 0.0)),
-                ("m[0].t[0].t[0]", (1.0, 1.0)),
-                ("m[0].t[0].t[0].t[0]", (1.0, 1.0)),
-                ("m[0].t[0].t[1]", (0.0, 0.0)),
-                ("m[0].t[1]", (0.0, 0.0)),
-                ("m[0].t[2]", (0.0, 0.0)),
-                ("m[1].t[0]", (0.0, 0.0)),
-            ],
-        ),
-        (
-            "mixers[0].tracks[0].tracks[0]",
-            False,
-            [OscMessage("/c_set", 11, 0.0)],
-            [
-                ("m[0].t[0]", (0.0, 0.0)),
-                ("m[0].t[0].t[0]", (0.0, 0.0)),
-                ("m[0].t[0].t[0].t[0]", (1.0, 1.0)),
-                ("m[0].t[0].t[1]", (0.0, 0.0)),
-                ("m[0].t[1]", (0.0, 0.0)),
-                ("m[0].t[2]", (0.0, 0.0)),
-                ("m[1].t[0]", (0.0, 0.0)),
-            ],
-        ),
-    ],
-)
-@pytest.mark.asyncio
-async def test_Track_set_active(
-    active: bool,
-    complex_session: Tuple[Session, str],
-    expected_commands: List[Union[OscBundle, OscMessage]],
-    expected_levels: List[Tuple[float, ...]],
-    online: bool,
-    target: str,
-) -> None:
-    # Pre-conditions
-    session, _ = complex_session
-    if online:
-        await session.boot()
-        await session.mixers[0].tracks[0].tracks[0].tracks[0].add_device()
-        await session.mixers[0].tracks[1].sends[0].delete()
-        await asyncio.sleep(0.25)
-        initial_tree = await debug_tree(session)
-    target_ = session[target]
-    assert isinstance(target_, Track)
-    # Operation
-    with capture(session["mixers[0]"].context) as commands:
-        await target_.set_active(False)
-    # Post-conditions
-    if not online:
-        return
-    await assert_diff(
-        session,
-        "",
-        expected_initial_tree=initial_tree,
-    )
-    assert commands == expected_commands
-    assert [
-        (
-            track.short_address,
-            tuple(round(x, 6) for x in cast(Track, track).output_levels),
-        )
-        for track in session._walk(Track)
-    ] == expected_levels
-
-
 @pytest.mark.parametrize("online", [False, True])
 @pytest.mark.parametrize(
     "source, target, maybe_raises, expected_commands, expected_diff",
@@ -1404,6 +1330,80 @@ async def test_Track_set_input(
     assert commands == expected_commands
 
 
+@pytest.mark.parametrize("online", [False, True])
+@pytest.mark.parametrize(
+    "target, muted, expected_commands, expected_levels",
+    [
+        (
+            "mixers[0].tracks[0]",
+            True,
+            [OscMessage("/c_set", 5, 0.0)],
+            [
+                ("m[0].t[0]", (0.0, 0.0)),
+                ("m[0].t[0].t[0]", (1.0, 1.0)),
+                ("m[0].t[0].t[0].t[0]", (1.0, 1.0)),
+                ("m[0].t[0].t[1]", (0.0, 0.0)),
+                ("m[0].t[1]", (0.0, 0.0)),
+                ("m[0].t[2]", (0.0, 0.0)),
+                ("m[1].t[0]", (0.0, 0.0)),
+            ],
+        ),
+        (
+            "mixers[0].tracks[0].tracks[0]",
+            True,
+            [OscMessage("/c_set", 11, 0.0)],
+            [
+                ("m[0].t[0]", (0.0, 0.0)),
+                ("m[0].t[0].t[0]", (0.0, 0.0)),
+                ("m[0].t[0].t[0].t[0]", (1.0, 1.0)),
+                ("m[0].t[0].t[1]", (0.0, 0.0)),
+                ("m[0].t[1]", (0.0, 0.0)),
+                ("m[0].t[2]", (0.0, 0.0)),
+                ("m[1].t[0]", (0.0, 0.0)),
+            ],
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_Track_set_muted(
+    muted: bool,
+    complex_session: Tuple[Session, str],
+    expected_commands: List[Union[OscBundle, OscMessage]],
+    expected_levels: List[Tuple[float, ...]],
+    online: bool,
+    target: str,
+) -> None:
+    # Pre-conditions
+    session, _ = complex_session
+    if online:
+        await session.boot()
+        await session.mixers[0].tracks[0].tracks[0].tracks[0].add_device()
+        await session.mixers[0].tracks[1].sends[0].delete()
+        await asyncio.sleep(0.25)
+        initial_tree = await debug_tree(session)
+    target_ = session[target]
+    assert isinstance(target_, Track)
+    # Operation
+    with capture(session["mixers[0]"].context) as commands:
+        await target_.set_muted(muted)
+    # Post-conditions
+    if not online:
+        return
+    await assert_diff(
+        session,
+        "",
+        expected_initial_tree=initial_tree,
+    )
+    assert commands == expected_commands
+    assert [
+        (
+            track.short_address,
+            tuple(round(x, 6) for x in cast(Track, track).output_levels),
+        )
+        for track in session._walk(Track)
+    ] == expected_levels
+
+
 @pytest.mark.parametrize("online", [False, True])
 @pytest.mark.parametrize(
     "source, target, maybe_raises, expected_commands, expected_diff",