Set control buses

supriya/contexts/entities.py

@@ -568,6 +568,21 @@ def free(self):
         """
         self.context.free_bus_group(self)
 
+    def get(
+        self, sync: bool = True
+    ) -> Union[Awaitable[Optional[Sequence[float]]], Optional[Sequence[float]]]:
+        """
+        Get the control bus group's values.
+
+        Emit ``/c_getn`` requests.
+
+        :param sync: If true, communicate the request immediately. Otherwise bundle it
+            with the current request context.
+        """
+        return cast(Union["AsyncServer", "Server"], self.context).get_bus_range(
+            bus=self[0], count=len(self), sync=sync
+        )
+
     def map_symbol(self) -> str:
         """
         Get the bus group's map symbol.
@@ -578,6 +593,22 @@ def map_symbol(self) -> str:
             return f"c{self.id_}"
         raise InvalidCalculationRate
 
+    def set(self, values: Union[float, Sequence[float]]) -> None:
+        """
+        Set a range of control buses.
+
+        Emit ``/c_setn`` or ``/c_fill`` requests.
+
+        :param values: The values to write. If a float is passed, use that as a fill.
+        """
+        if isinstance(values, float):
+            if len(self) == 1:
+                self.context.set_bus(bus=self[0], value=values)
+            else:
+                self.context.fill_bus_range(bus=self[0], count=len(self), value=values)
+        else:
+            self.context.set_bus_range(bus=self[0], values=values)
+
 
 @dataclasses.dataclass(frozen=True)
 class Node(ContextObject):

supriya/mixers/components.py

@@ -261,7 +261,7 @@ def _get_audio_bus(
         name: str,
         can_allocate: bool = False,
         channel_count: int = 2,
-    ) -> Optional[BusGroup]:
+    ) -> BusGroup:
         return self._get_buses(
             calculation_rate=CalculationRate.AUDIO,
             can_allocate=can_allocate,
@@ -278,7 +278,7 @@ def _get_buses(
         calculation_rate: CalculationRate,
         can_allocate: bool = False,
         channel_count: int = 1,
-    ) -> Optional[BusGroup]:
+    ) -> BusGroup:
         if calculation_rate == CalculationRate.CONTROL:
             buses = self._control_buses
         elif calculation_rate == CalculationRate.AUDIO:
@@ -290,15 +290,15 @@ def _get_buses(
                 calculation_rate=calculation_rate,
                 count=channel_count,
             )
-        return buses.get(name)
+        return buses[name]
 
     def _get_control_bus(
         self,
         context: Optional[AsyncServer],
         name: str,
         can_allocate: bool = False,
         channel_count: int = 1,
-    ) -> Optional[BusGroup]:
+    ) -> BusGroup:
         return self._get_buses(
             calculation_rate=CalculationRate.CONTROL,
             can_allocate=can_allocate,

supriya/mixers/mixers.py

@@ -92,6 +92,9 @@ def _allocate(self, context: AsyncServer) -> bool:
         )
         target_node = context.default_group
         with context.at():
+            gain_control_bus.set(0.0)
+            input_levels_control_bus.set(0.0)
+            output_levels_control_bus.set(0.0)
             self._nodes[ComponentNames.GROUP] = group = target_node.add_group(
                 add_action=AddAction.ADD_TO_TAIL
             )
@@ -104,7 +107,7 @@ def _allocate(self, context: AsyncServer) -> bool:
             self._nodes[ComponentNames.CHANNEL_STRIP] = channel_strip = group.add_synth(
                 add_action=AddAction.ADD_TO_TAIL,
                 bus=main_audio_bus,
-                gain=gain_control_bus.map_symbol() if gain_control_bus else None,
+                gain=gain_control_bus.map_symbol(),
                 synthdef=CHANNEL_STRIP_2,
             )
             self._nodes[ComponentNames.INPUT_LEVELS] = tracks.add_synth(

supriya/mixers/tracks.py

@@ -256,6 +256,10 @@ def _allocate(self, *, context: AsyncServer) -> bool:
         )
         target_node = self.parent._nodes[ComponentNames.TRACKS]
         with context.at():
+            active_control_bus.set(1.0)
+            gain_control_bus.set(0.0)
+            input_levels_control_bus.set(0.0)
+            output_levels_control_bus.set(0.0)
             self._nodes[ComponentNames.GROUP] = group = target_node.add_group(
                 add_action=AddAction.ADD_TO_TAIL
             )
@@ -268,8 +272,8 @@ def _allocate(self, *, context: AsyncServer) -> bool:
             self._nodes[ComponentNames.CHANNEL_STRIP] = channel_strip = group.add_synth(
                 add_action=AddAction.ADD_TO_TAIL,
                 bus=main_audio_bus,
-                active=active_control_bus.map_symbol() if active_control_bus else None,
-                gain=gain_control_bus.map_symbol() if gain_control_bus else None,
+                active=active_control_bus.map_symbol(),
+                gain=gain_control_bus.map_symbol(),
                 synthdef=CHANNEL_STRIP_2,
             )
             self._nodes[ComponentNames.INPUT_LEVELS] = tracks.add_synth(
@@ -297,6 +301,8 @@ def _register_feedback(
     ) -> Optional[BusGroup]:
         super()._register_feedback(context, dependent)
         # check if feedback should be setup
+        if not context:
+            return None
         if self._feedback_dependents:
             self._feedback._set_source(
                 self._get_audio_bus(

tests/mixers/test_Mixer.py

@@ -17,6 +17,8 @@
             [
                 OscBundle(
                     contents=(
+                        OscMessage("/c_set", 11, 1.0, 12, 0.0),
+                        OscMessage("/c_fill", 13, 2, 0.0, 15, 2, 0.0),
                         OscMessage(
                             "/g_new", 1014, 1, 1001, 1015, 0, 1014, 1016, 1, 1014
                         ),

tests/mixers/test_Track.py

@@ -463,6 +463,8 @@ async def test_Track_add_track(
     assert commands == [
         OscBundle(
             contents=(
+                OscMessage("/c_set", 11, 1.0, 12, 0.0),
+                OscMessage("/c_fill", 13, 2, 0.0, 15, 2, 0.0),
                 OscMessage("/g_new", 1014, 1, 1007, 1015, 0, 1014, 1016, 1, 1014),
                 OscMessage(
                     "/s_new",