Implement mixing board affordances

.github/actions/build-supercollider-linux/action.yaml

@@ -66,6 +66,7 @@ runs:
         mkdir /tmp/supercollider/build
         cd /tmp/supercollider/build
         cmake \
+          -DCMAKE_POLICY_VERSION_MINIMUM=3.5 \
           -DRULE_LAUNCH_COMPILE=ccache \
           -DSC_ED=OFF \
           -DSC_EL=OFF \

supriya/mixers/__init__.py

@@ -0,0 +1,3 @@
+from .sessions import Session
+
+__all__ = ["Session"]

supriya/mixers/components.py

@@ -0,0 +1,313 @@
+import asyncio
+from typing import (
+    TYPE_CHECKING,
+    Awaitable,
+    Generator,
+    Generic,
+    Iterator,
+    Literal,
+    Optional,
+    Type,
+    TypeAlias,
+    TypeVar,
+    cast,
+)
+
+from ..contexts import AsyncServer, Buffer, BusGroup, Context, Group, Node
+from ..contexts.responses import QueryTreeGroup
+from ..enums import BootStatus, CalculationRate
+from ..ugens import SynthDef
+from ..utils import iterate_nwise
+
+C = TypeVar("C", bound="Component")
+
+A = TypeVar("A", bound="AllocatableComponent")
+
+# TODO: Integrate this with channel logic
+ChannelCount: TypeAlias = Literal[1, 2, 4, 8]
+
+if TYPE_CHECKING:
+    from .mixers import Mixer
+    from .sessions import Session
+
+
+class ComponentNames:
+    ACTIVE = "active"
+    CHANNEL_STRIP = "channel-strip"
+    DEVICES = "devices"
+    FEEDBACK = "feedback"
+    GAIN = "gain"
+    GROUP = "group"
+    INPUT = "input"
+    INPUT_LEVELS = "input-levels"
+    MAIN = "main"
+    OUTPUT = "output"
+    OUTPUT_LEVELS = "output-levels"
+    SYNTH = "synth"
+    TRACKS = "tracks"
+
+
+class Component(Generic[C]):
+
+    def __init__(
+        self,
+        *,
+        parent: C | None = None,
+    ) -> None:
+        self._lock = asyncio.Lock()
+        self._parent: C | None = parent
+        self._dependents: set[Component] = set()
+        self._is_active = True
+        self._feedback_dependents: set[Component] = set()
+
+    def __repr__(self) -> str:
+        return f"<{type(self).__name__}>"
+
+    async def _allocate_deep(self, *, context: AsyncServer) -> None:
+        if self.session is None:
+            raise RuntimeError
+        fifo: list[tuple[Component, int]] = []
+        current_synthdefs = self.session._synthdefs[context]
+        desired_synthdefs: set[SynthDef] = set()
+        for component in self._walk():
+            fifo.append((component, 0))
+            desired_synthdefs.update(component._get_synthdefs())
+        if required_synthdefs := sorted(
+            desired_synthdefs - current_synthdefs, key=lambda x: x.effective_name
+        ):
+            for synthdef in required_synthdefs:
+                context.add_synthdefs(synthdef)
+            await context.sync()
+            current_synthdefs.update(required_synthdefs)
+        while fifo:
+            component, attempts = fifo.pop(0)
+            if attempts > 2:
+                raise RuntimeError(component, attempts)
+            if not component._allocate(context=context):
+                fifo.append((component, attempts + 1))
+
+    def _allocate(self, *, context: AsyncServer) -> bool:
+        return True
+
+    def _deallocate(self) -> None:
+        pass
+
+    def _deallocate_deep(self) -> None:
+        for component in self._walk():
+            component._deallocate()
+
+    def _delete(self) -> None:
+        self._deallocate_deep()
+        self._parent = None
+
+    def _get_synthdefs(self) -> list[SynthDef]:
+        return []
+
+    def _iterate_parentage(self) -> Iterator["Component"]:
+        component = self
+        while component.parent is not None:
+            yield component
+            component = component.parent
+        yield component
+
+    def _reconcile(self, context: AsyncServer | None = None) -> bool:
+        return True
+
+    def _register_dependency(self, dependent: "Component") -> None:
+        self._dependents.add(dependent)
+
+    def _register_feedback(
+        self, context: AsyncServer | None, dependent: "Component"
+    ) -> BusGroup | None:
+        self._dependents.add(dependent)
+        self._feedback_dependents.add(dependent)
+        return None
+
+    def _unregister_dependency(self, dependent: "Component") -> bool:
+        self._dependents.discard(dependent)
+        return self._unregister_feedback(dependent)
+
+    def _unregister_feedback(self, dependent: "Component") -> bool:
+        had_feedback = bool(self._feedback_dependents)
+        self._feedback_dependents.discard(dependent)
+        return had_feedback and not self._feedback_dependents
+
+    def _walk(
+        self, component_class: Type["Component"] | None = None
+    ) -> Generator["Component", None, None]:
+        component_class_ = component_class or Component
+        if isinstance(self, component_class_):
+            yield self
+        for child in self.children:
+            yield from child._walk(component_class_)
+
+    @property
+    def address(self) -> str:
+        raise NotImplementedError
+
+    @property
+    def children(self) -> list["Component"]:
+        return []
+
+    @property
+    def context(self) -> AsyncServer | None:
+        if (mixer := self.mixer) is not None:
+            return mixer.context
+        return None
+
+    @property
+    def graph_order(self) -> tuple[int, ...]:
+        # TODO: Cache this
+        graph_order = []
+        for parent, child in iterate_nwise(reversed(list(self._iterate_parentage()))):
+            graph_order.append(parent.children.index(child))
+        return tuple(graph_order)
+
+    @property
+    def mixer(self) -> Optional["Mixer"]:
+        # TODO: Cache this
+        from .mixers import Mixer
+
+        for component in self._iterate_parentage():
+            if isinstance(component, Mixer):
+                return component
+        return None
+
+    @property
+    def parent(self) -> C | None:
+        return self._parent
+
+    @property
+    def parentage(self) -> list["Component"]:
+        # TODO: Cache this
+        return list(self._iterate_parentage())
+
+    @property
+    def session(self) -> Optional["Session"]:
+        # TODO: Cache this
+        from .sessions import Session
+
+        for component in self._iterate_parentage():
+            if isinstance(component, Session):
+                return component
+        return None
+
+    @property
+    def short_address(self) -> str:
+        address = self.address
+        for from_, to_ in [
+            ("session.", ""),
+            ("tracks", "t"),
+            ("devices", "d"),
+            ("mixers", "m"),
+        ]:
+            address = address.replace(from_, to_)
+        return address
+
+
+class AllocatableComponent(Component[C]):
+
+    def __init__(
+        self,
+        *,
+        parent: C | None = None,
+    ) -> None:
+        super().__init__(parent=parent)
+        self._audio_buses: dict[str, BusGroup] = {}
+        self._buffers: dict[str, Buffer] = {}
+        self._context: Context | None = None
+        self._control_buses: dict[str, BusGroup] = {}
+        self._is_active: bool = True
+        self._nodes: dict[str, Node] = {}
+
+    def _can_allocate(self) -> AsyncServer | None:
+        if (
+            context := self.context
+        ) is not None and context.boot_status == BootStatus.ONLINE:
+            return context
+        return None
+
+    def _deallocate(self) -> None:
+        super()._deallocate()
+        for key in tuple(self._audio_buses):
+            self._audio_buses.pop(key).free()
+        for key in tuple(self._control_buses):
+            self._control_buses.pop(key).free()
+        if group := self._nodes.get(ComponentNames.GROUP):
+            if not self._is_active:
+                group.free()
+            else:
+                group.set(gate=0)
+        self._nodes.clear()
+        for key in tuple(self._buffers):
+            self._buffers.pop(key).free()
+
+    def _get_audio_bus(
+        self,
+        context: AsyncServer | None,
+        name: str,
+        can_allocate: bool = False,
+        channel_count: int = 2,
+    ) -> BusGroup:
+        return self._get_buses(
+            calculation_rate=CalculationRate.AUDIO,
+            can_allocate=can_allocate,
+            channel_count=channel_count,
+            context=context,
+            name=name,
+        )
+
+    def _get_buses(
+        self,
+        context: AsyncServer | None,
+        name: str,
+        *,
+        calculation_rate: CalculationRate,
+        can_allocate: bool = False,
+        channel_count: int = 1,
+    ) -> BusGroup:
+        if calculation_rate == CalculationRate.CONTROL:
+            buses = self._control_buses
+        elif calculation_rate == CalculationRate.AUDIO:
+            buses = self._audio_buses
+        else:
+            raise ValueError(calculation_rate)
+        if (name not in buses) and can_allocate and context:
+            buses[name] = context.add_bus_group(
+                calculation_rate=calculation_rate,
+                count=channel_count,
+            )
+        return buses[name]
+
+    def _get_control_bus(
+        self,
+        context: AsyncServer | None,
+        name: str,
+        can_allocate: bool = False,
+        channel_count: int = 1,
+    ) -> BusGroup:
+        return self._get_buses(
+            calculation_rate=CalculationRate.CONTROL,
+            can_allocate=can_allocate,
+            channel_count=channel_count,
+            context=context,
+            name=name,
+        )
+
+    async def dump_tree(self, annotated: bool = True) -> QueryTreeGroup:
+        if self.session and self.session.status != BootStatus.ONLINE:
+            raise RuntimeError
+        tree = await cast(
+            Awaitable[QueryTreeGroup],
+            cast(Group, self._nodes[ComponentNames.GROUP]).dump_tree(),
+        )
+        if annotated:
+            annotations: dict[int, str] = {}
+            for component in self._walk():
+                if not isinstance(component, AllocatableComponent):
+                    continue
+                address = component.address
+                for name, node in component._nodes.items():
+                    annotations[node.id_] = f"{address}:{name}"
+            return tree.annotate(annotations)
+        return tree

supriya/mixers/devices.py

@@ -0,0 +1,64 @@
+from ..contexts import AsyncServer
+from ..enums import AddAction
+from ..ugens import SynthDef
+from .components import AllocatableComponent, C, ComponentNames
+from .synthdefs import DEVICE_DC_TESTER_2
+
+
+class DeviceContainer(AllocatableComponent[C]):
+
+    def __init__(self) -> None:
+        self._devices: list[Device] = []
+
+    def _add_device(self) -> "Device":
+        self._devices.append(device := Device(parent=self))
+        return device
+
+    def _delete_device(self, device: "Device") -> None:
+        self._devices.remove(device)
+
+    async def add_device(self) -> "Device":
+        async with self._lock:
+            device = self._add_device()
+            if context := self._can_allocate():
+                await device._allocate_deep(context=context)
+            return device
+
+    @property
+    def devices(self) -> list["Device"]:
+        return self._devices[:]
+
+
+class Device(AllocatableComponent):
+
+    def _allocate(self, *, context: AsyncServer) -> bool:
+        if not super()._allocate(context=context):
+            return False
+        elif self.parent is None:
+            raise RuntimeError
+        main_audio_bus = self.parent._get_audio_bus(context, name=ComponentNames.MAIN)
+        target_node = self.parent._nodes[ComponentNames.DEVICES]
+        with context.at():
+            self._nodes[ComponentNames.GROUP] = group = target_node.add_group(
+                add_action=AddAction.ADD_TO_TAIL
+            )
+            self._nodes[ComponentNames.SYNTH] = group.add_synth(
+                add_action=AddAction.ADD_TO_TAIL,
+                out=main_audio_bus,
+                synthdef=DEVICE_DC_TESTER_2,
+            )
+        return True
+
+    def _get_synthdefs(self) -> list[SynthDef]:
+        return [DEVICE_DC_TESTER_2]
+
+    async def set_active(self, active: bool = True) -> None:
+        async with self._lock:
+            pass
+
+    @property
+    def address(self) -> str:
+        if self.parent is None:
+            return "devices[?]"
+        index = self.parent.devices.index(self)
+        return f"{self.parent.address}.devices[{index}]"